diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala
index efa46144180..889cb1eca7e 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala
@@ -89,6 +89,7 @@ import org.apache.texera.amber.operator.source.sql.postgresql.PostgreSQLSourceOp
 import org.apache.texera.amber.operator.split.SplitOpDesc
 import org.apache.texera.amber.operator.substringSearch.SubstringSearchOpDesc
 import org.apache.texera.amber.operator.symmetricDifference.SymmetricDifferenceOpDesc
+import org.apache.texera.amber.operator.binning.BinningOpDesc
 import org.apache.texera.amber.operator.typecasting.TypeCastingOpDesc
 import org.apache.texera.amber.operator.udf.java.JavaUDFOpDesc
 import org.apache.texera.amber.operator.udf.python._
@@ -220,6 +221,7 @@ trait StateTransferFunc
     new Type(value = classOf[PostgreSQLSourceOpDesc], name = "PostgreSQLSource"),
     new Type(value = classOf[AsterixDBSourceOpDesc], name = "AsterixDBSource"),
     new Type(value = classOf[TypeCastingOpDesc], name = "TypeCasting"),
+    new Type(value = classOf[BinningOpDesc], name = "Binning"),
     new Type(value = classOf[LimitOpDesc], name = "Limit"),
     new Type(value = classOf[SleepOpDesc], name = "Sleep"),
     new Type(value = classOf[LoopStartOpDesc], name = "LoopStart"),
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningMethod.java b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningMethod.java
new file mode 100644
index 00000000000..e0d9cbef61e
--- /dev/null
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningMethod.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.binning;
+
+import com.fasterxml.jackson.annotation.JsonValue;
+
+/** How the range of a numeric column is cut into bins. */
+public enum BinningMethod {
+
+    EQUAL_WIDTH("equal width"),
+
+    EQUAL_FREQUENCY("equal frequency");
+
+    private final String name;
+
+    BinningMethod(String name) {
+        this.name = name;
+    }
+
+    // use the name string instead of enum string in JSON
+    @JsonValue
+    public String getName() {
+        return this.name;
+    }
+
+}
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala
new file mode 100644
index 00000000000..58ff59ebfb0
--- /dev/null
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.binning
+
+import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
+import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle}
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema}
+import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity}
+import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName
+import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator}
+import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.{
+  PythonTemplateBuilderStringContext,
+  pyStringLiteral
+}
+
+@JsonSchemaInject(json = """
+{
+  "attributeTypeRules": {
+    "attribute": {
+      "enum": ["integer", "long", "double"]
+    }
+  }
+}
+""")
+class BinningOpDesc extends PythonOperatorDescriptor with StandaloneCodeGenerator {
+
+  @JsonProperty(required = true)
+  @JsonSchemaTitle("Attribute")
+  @JsonPropertyDescription("numeric column to cut into bins")
+  @AutofillAttributeName
+  var attribute: EncodableString = ""
+
+  @JsonProperty(required = true, defaultValue = "equal width")
+  @JsonSchemaTitle("Method")
+  @JsonPropertyDescription("how the bins are cut")
+  var method: BinningMethod = BinningMethod.EQUAL_WIDTH
+
+  @JsonProperty(required = true, defaultValue = "4")
+  @JsonSchemaTitle("Number of bins")
+  @JsonPropertyDescription("how many bins to cut the column into")
+  @JsonSchemaInject(json = """{"minimum": 2, "maximum": 100}""")
+  var bins: Int = 4
+
+  override def operatorInfo: OperatorInfo =
+    OperatorInfo(
+      userFriendlyName = "Binning",
+      operatorDescription =
+        "Cut a numeric column into bins, so rows can be grouped by range rather than by value",
+      operatorGroupName = OperatorGroupConstants.CLEANING_GROUP,
+      inputPorts = List(InputPort()),
+      // Blocking: an equal-frequency cut is made at the column's quantiles, which
+      // are not known until the last row has arrived.
+      outputPorts = List(OutputPort(blocking = true))
+    )
+
+  /** The bin a row fell in, named after the column it was cut from. */
+  private def resultColumn: String = s"${attribute}_bin"
+
+  override def getOutputSchemas(
+      inputSchemas: Map[PortIdentity, Schema]
+  ): Map[PortIdentity, Schema] =
+    Map(
+      operatorInfo.outputPorts.head.id ->
+        // The bin reads as its own range, "(2.5, 5.0]", rather than as a number:
+        // the point of binning is a label to group by, and a number would invite
+        // arithmetic on what is really a name.
+        inputSchemas.values.head.add(resultColumn, AttributeType.STRING)
+    )
+
+  /** `duplicates="drop"` because a quantile cut can put two edges in one place
+    * where the values repeat, and fewer bins beats raising.
+    */
+  private def cutArgs: String =
+    method match {
+      case BinningMethod.EQUAL_WIDTH     => s"bins=$bins"
+      case BinningMethod.EQUAL_FREQUENCY => s"""q=$bins, duplicates="drop""""
+    }
+
+  private def cutName: String =
+    method match {
+      case BinningMethod.EQUAL_WIDTH     => "pd.cut"
+      case BinningMethod.EQUAL_FREQUENCY => "pd.qcut"
+    }
+
+  /** An empty cell has no bin, and `astype(str)` would render its absence as the
+    * text "nan", so the hole is kept as one.
+    */
+  private val labelSuffix: String =
+    """.astype("string").astype("object").where(lambda s: s.notna(), None)"""
+
+  override def generatePythonCode(): String = {
+    val cut = cutName
+    val args = cutArgs
+    val suffix = labelSuffix
+    // The result column is named in PYTHON rather than here: joining it to
+    // `attribute` in Scala would hand `pyb` a plain string, and the value it was
+    // built to protect would be spliced into the template unguarded.
+    //
+    // The guard is there because edges cannot be found in a column that holds
+    // nothing: an equal-width cut raises both when every cell is empty and when
+    // no row arrived at all. Leaving those values uncut keeps the holes, and the
+    // suffix turns them into the empty bins they already are.
+    pyb"""from pytexera import *
+       |import pandas as pd
+       |
+       |class ProcessTableOperator(UDFTableOperator):
+       |
+       |    @overrides
+       |    def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]:
+       |        out = table.copy()
+       |        _column = $attribute
+       |        _binned = out[_column]
+       |        if _binned.notna().any():
+       |            _binned = ${cut}(_binned, ${args})
+       |        out[_column + "_bin"] = _binned${suffix}
+       |        yield out""".encode
+  }
+
+  override def generateStandaloneCode(): String = {
+    val column = pyStringLiteral(attribute)
+    val result = pyStringLiteral(resultColumn)
+    s"""out1df = in1df.copy()
+       |_binned = out1df[$column]
+       |if _binned.notna().any():
+       |    _binned = $cutName(_binned, $cutArgs)
+       |out1df[$result] = _binned$labelSuffix""".stripMargin
+  }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala
new file mode 100644
index 00000000000..5cc0027c2f4
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.binning
+
+import com.typesafe.config.ConfigFactory
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.util.Try
+
+class BinningOpDescSpec extends AnyFlatSpec with Matchers {
+
+  private val inputSchema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("age", AttributeType.DOUBLE)
+  )
+
+  private def desc(m: BinningMethod = BinningMethod.EQUAL_WIDTH, n: Int = 4): BinningOpDesc = {
+    val d = new BinningOpDesc
+    d.attribute = "age"
+    d.method = m
+    d.bins = n
+    d
+  }
+
+  "BinningOpDesc.operatorInfo" should "advertise the name and the Cleaning group" in {
+    val info = (new BinningOpDesc).operatorInfo
+    info.userFriendlyName shouldBe "Binning"
+    info.operatorGroupName shouldBe OperatorGroupConstants.CLEANING_GROUP
+    info.inputPorts should have length 1
+    info.outputPorts should have length 1
+  }
+
+  // A quantile cut is made at the column's own quantiles, which the last row can
+  // still move, so nothing may be emitted until the input ends.
+  it should "declare its output port blocking" in {
+    (new BinningOpDesc).operatorInfo.outputPorts.head.blocking shouldBe true
+  }
+
+  "The output schema" should "append one STRING column named after the source" in {
+    val schema = desc().getOutputSchemas(Map(PortIdentity() -> inputSchema))(PortIdentity())
+    schema.getAttributeNames shouldBe List("id", "age", "age_bin")
+    schema.getAttribute("age_bin").getType shouldBe AttributeType.STRING
+  }
+
+  it should "refuse a derived name the input already carries" in {
+    val clashing = inputSchema.add("age_bin", AttributeType.STRING)
+    a[RuntimeException] should be thrownBy
+      desc().getOutputSchemas(Map(PortIdentity() -> clashing))
+  }
+
+  "The generated code" should "cut on the count of bins asked for" in {
+    desc(BinningMethod.EQUAL_WIDTH, 7).generateStandaloneCode() should include("bins=7")
+  }
+
+  // A quantile cut can put two edges in one place where the values repeat, and
+  // dropping the duplicate yields fewer bins rather than raising.
+  it should "ask the quantile cut to drop a duplicate edge" in {
+    val code = desc(BinningMethod.EQUAL_FREQUENCY).generateStandaloneCode()
+    code should include("pd.qcut(")
+    code should include("""duplicates="drop"""")
+    code should include("q=4")
+  }
+
+  it should "keep an empty cell empty rather than rendering it as text" in {
+    desc().generateStandaloneCode() should include("where(lambda s: s.notna(), None)")
+  }
+
+  it should "hold the source and the derived name as escaped literals" in {
+    val d = desc()
+    d.attribute = "a\"b"
+    val code = d.generateStandaloneCode()
+    code should include("""in1df.copy()""")
+    code should include("""out1df["a\"b"]""")
+    code should include("""out1df["a\"b_bin"]""")
+  }
+
+  // Edges cannot be found in a column that holds nothing, and an equal-width cut
+  // raises rather than handing back empty bins. Both renderings skip the cut, which
+  // leaves the holes for the suffix to turn into the empty labels they already are.
+  it should "leave a column with nothing in it uncut" in {
+    val guard = "if _binned.notna().any():"
+    desc().generateStandaloneCode() should include(guard)
+    desc().generatePythonCode() should include(guard)
+  }
+
+  // Both paths make the same call, on a series both name the same, so a difference
+  // between them is a difference in this operator rather than in pandas.
+  "The two renderings" should "make the same pandas call" in {
+    val d = desc(BinningMethod.EQUAL_FREQUENCY, 5)
+    val call = """pd.qcut(_binned, q=5, duplicates="drop")"""
+    d.generateStandaloneCode() should include(call)
+    d.generatePythonCode() should include(call)
+  }
+
+  "The platform code" should "be a table operator, since a quantile needs every row" in {
+    val code = desc().generatePythonCode()
+    code should include("class ProcessTableOperator(UDFTableOperator)")
+    code should include("def process_table(")
+  }
+
+  // Python executable resolution, following FilledAreaPlotOpDescSpec:
+  // udf.conf python.path (UDF_PYTHON_PATH), then python3 / python / py.
+  private def resolvePythonExecutable(): Option[String] = {
+    def fromConfig: Option[String] = {
+      val configOpt =
+        Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption
+          .orElse(Try(ConfigFactory.load()).toOption)
+      configOpt
+        .flatMap(c => Try(c.getConfig("python").getString("path")).toOption)
+        .map(_.trim)
+        .filter(_.nonEmpty)
+    }
+
+    def isRunnable(exe: String): Boolean = {
+      val pTry = Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start())
+      pTry.toOption.exists { p =>
+        val finished = p.waitFor(5, TimeUnit.SECONDS)
+        if (!finished) { p.destroyForcibly(); false }
+        else p.exitValue() == 0
+      }
+    }
+
+    (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(isRunnable)
+  }
+
+  private def canImportPandas(python: String): Boolean = {
+    val pTry = Try(
+      new ProcessBuilder(python, "-c", "import pandas").redirectErrorStream(true).start()
+    )
+    pTry.toOption.exists { p =>
+      val finished = p.waitFor(60, TimeUnit.SECONDS)
+      if (!finished) { p.destroyForcibly(); false }
+      else p.exitValue() == 0
+    }
+  }
+
+  // Runs the exported code as written over three frames, and reports the length of
+  // the bin column and how much of it was filled. The exported code is plain pandas,
+  // so nothing about the operator has to be stubbed for it to run.
+  private val runtimeDriverScript: String =
+    """import sys
+      |import pandas as pd
+      |
+      |code = open(sys.argv[1]).read()
+      |cases = {
+      |    "allnull": pd.DataFrame({"age": pd.Series([None, None], dtype="float64")}),
+      |    "empty": pd.DataFrame({"age": pd.Series([], dtype="float64")}),
+      |    "mixed": pd.DataFrame({"age": pd.Series([1.0, 2.0, None])}),
+      |}
+      |for cid, frame in cases.items():
+      |    namespace = {"pd": pd, "in1df": frame}
+      |    try:
+      |        exec(code, namespace)
+      |        labels = namespace["out1df"]["age_bin"]
+      |        print("CASE %s OK:%d:%d" % (cid, len(labels), labels.notna().sum()))
+      |    except Exception as error:
+      |        print("CASE %s %s" % (cid, type(error).__name__))
+      |""".stripMargin
+
+  // An all-empty column used to raise on an equal-width cut, and so did an empty
+  // table, both before the suffix that keeps a hole a hole could run.
+  for (method <- BinningMethod.values) {
+    it should s"bin a column with nothing in it rather than raising, cutting by $method" in {
+      val python = resolvePythonExecutable().getOrElse(
+        cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+      )
+      if (!canImportPandas(python)) {
+        cancel(s"'$python' cannot import pandas; skipping runtime verification")
+      }
+
+      val moduleFile = Files.createTempFile("binning_op_", ".py")
+      val driverFile = Files.createTempFile("binning_driver_", ".py")
+      try {
+        val code = desc(method).generateStandaloneCode()
+        Files.write(moduleFile, code.getBytes(StandardCharsets.UTF_8))
+        Files.write(driverFile, runtimeDriverScript.getBytes(StandardCharsets.UTF_8))
+
+        val process = new ProcessBuilder(python, driverFile.toString, moduleFile.toString)
+          .redirectErrorStream(true)
+          .start()
+        val finished = process.waitFor(120, TimeUnit.SECONDS)
+        if (!finished) {
+          process.destroyForcibly()
+          fail("Runtime verification driver timed out after 120s")
+        }
+        val output = new String(process.getInputStream.readAllBytes(), StandardCharsets.UTF_8)
+        withClue(s"Exported code:\n$code\nDriver output:\n$output\n") {
+          process.exitValue() shouldBe 0
+          val verdicts = "CASE (\\S+) (\\S+)".r
+            .findAllMatchIn(output)
+            .map(m => m.group(1) -> m.group(2))
+            .toMap
+          verdicts shouldBe Map(
+            "allnull" -> "OK:2:0", // two rows, neither of them in a bin
+            "empty" -> "OK:0:0", // no rows to put in one
+            "mixed" -> "OK:3:2" // the hole is the only row left unlabelled
+          )
+        }
+      } finally {
+        Try(Files.deleteIfExists(moduleFile))
+        Try(Files.deleteIfExists(driverFile))
+        ()
+      }
+    }
+  }
+}
diff --git a/frontend/src/assets/operator_images/Binning.png b/frontend/src/assets/operator_images/Binning.png
new file mode 100644
index 00000000000..3f616cf9750
Binary files /dev/null and b/frontend/src/assets/operator_images/Binning.png differ