From 37f12bed6ec897b18d344d7d8c409663f642d0b0 Mon Sep 17 00:00:00 2001 From: adibmbrk Date: Tue, 1 Sep 2026 07:21:22 +0530 Subject: [PATCH 1/2] fix: route length/bit_length/octet_length binary input through codegen dispatcher length, bit_length, and octet_length rejected BinaryType input and fell the whole projection back to Spark. Mix in CodegenDispatchFallback so the binary case routes through the JVM codegen dispatcher (Spark's own doGenCode) inside the Comet pipeline instead. Docs updated to Hybrid and the SQL fixtures now assert native parity on binary input. Closes #5584 Signed-off-by: adibmbrk --- docs/source/user-guide/latest/expressions.md | 12 ++++++------ .../scala/org/apache/comet/serde/strings.scala | 17 ++++++++++++++--- .../sql-tests/expressions/string/bit_length.sql | 11 +++++++---- .../sql-tests/expressions/string/length.sql | 17 +++++++++++++++++ .../expressions/string/octet_length.sql | 11 +++++++---- 5 files changed, 51 insertions(+), 17 deletions(-) diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 2ba71e93c4d..d7579bea45b 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -553,11 +553,11 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | --- | --- | --- | --- | | `ascii` | ✅ | Native | | | `base64` | ✅ | Native | | -| `bit_length` | ✅ | Native | | +| `bit_length` | ✅ | Hybrid | | | `btrim` | ✅ | — | | | `char` | ✅ | Native | | -| `char_length` | ✅ | Native | | -| `character_length` | ✅ | Native | | +| `char_length` | ✅ | Hybrid | | +| `character_length` | ✅ | Hybrid | | | `chr` | ✅ | Native | | | `collate` | 🔜 | — | Spark collation (umbrella [#2190](https://github.com/apache/datafusion-comet/issues/2190)) | | `collation` | ✅ | — | Constant-folded to a literal (Spark 4.0+) | @@ -574,8 +574,8 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `instr` | ✅ | Native | | | `lcase` | ✅ | Hybrid | | | `left` | ✅ | Native | | -| `len` | ✅ | Native | | -| `length` | ✅ | Native | | +| `len` | ✅ | Hybrid | | +| `length` | ✅ | Hybrid | | | `levenshtein` | ✅ | Native | | | `locate` | ✅ | Codegen dispatch | | | `lower` | ✅ | Hybrid | | @@ -583,7 +583,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `ltrim` | ✅ | Native | | | `luhn_check` | ✅ | — | Native via `StaticInvoke` (tests: luhn_check.sql) | | `mask` | ✅ | — | Routed through the JVM codegen dispatcher | -| `octet_length` | ✅ | Native | | +| `octet_length` | ✅ | Hybrid | | | `overlay` | ✅ | Codegen dispatch | | | `position` | ✅ | Codegen dispatch | | | `printf` | ✅ | Codegen dispatch | | diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index 665c0d3b54e..0051ac80f40 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -82,7 +82,10 @@ object CometUpper extends CometCaseConversionBase[Upper]("upper") object CometLower extends CometCaseConversionBase[Lower]("lower") -object CometLength extends CometScalarFunction[Length]("length") { +object CometLength extends CometScalarFunction[Length]("length") with CodegenDispatchFallback { + // The native `length` UDF has no path for BinaryType. Rather than fall the projection back to + // Spark, route the binary case through the JVM codegen dispatcher (Spark's own `doGenCode`, i.e. + // `numBytes()`) inside the Comet pipeline so the result stays native and matches Spark exactly. override def getUnsupportedReasons(): Seq[String] = Seq("`BinaryType` input is not supported") override def getSupportLevel(expr: Length): SupportLevel = expr.child.dataType match { @@ -91,7 +94,11 @@ object CometLength extends CometScalarFunction[Length]("length") { } } -object CometBitLength extends CometScalarFunction[BitLength]("bit_length") { +object CometBitLength + extends CometScalarFunction[BitLength]("bit_length") + with CodegenDispatchFallback { + // See CometLength: BinaryType has no native path, so route it through the codegen dispatcher + // (Spark's own `doGenCode`, i.e. `numBytes() * 8`) instead of falling back to Spark. override def getUnsupportedReasons(): Seq[String] = Seq("`BinaryType` input is not supported") override def getSupportLevel(expr: BitLength): SupportLevel = expr.child.dataType match { @@ -100,7 +107,11 @@ object CometBitLength extends CometScalarFunction[BitLength]("bit_length") { } } -object CometOctetLength extends CometScalarFunction[OctetLength]("octet_length") { +object CometOctetLength + extends CometScalarFunction[OctetLength]("octet_length") + with CodegenDispatchFallback { + // See CometLength: BinaryType has no native path, so route it through the codegen dispatcher + // (Spark's own `doGenCode`, i.e. `numBytes()`) instead of falling back to Spark. override def getUnsupportedReasons(): Seq[String] = Seq("`BinaryType` input is not supported") override def getSupportLevel(expr: OctetLength): SupportLevel = expr.child.dataType match { diff --git a/spark/src/test/resources/sql-tests/expressions/string/bit_length.sql b/spark/src/test/resources/sql-tests/expressions/string/bit_length.sql index a2327120163..b9fd2abcd23 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/bit_length.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/bit_length.sql @@ -15,6 +15,10 @@ -- specific language governing permissions and limitations -- under the License. +-- BinaryType has no native path, so it routes through the codegen dispatcher (Spark's own +-- `doGenCode`, i.e. `numBytes() * 8`) instead of falling back to Spark. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true + statement CREATE TABLE test_bit_length(s string) USING parquet @@ -28,16 +32,15 @@ SELECT bit_length(s) FROM test_bit_length query SELECT bit_length('hello'), bit_length(''), bit_length(NULL) --- BinaryType input falls back to Spark; the native DataFusion impl rejects Binary at runtime, --- so the serde gates Binary as Unsupported (matching the existing CometLength shape). +-- BinaryType input routes through the codegen dispatcher and stays inside Comet statement CREATE TABLE test_bit_length_binary(b binary) USING parquet statement INSERT INTO test_bit_length_binary VALUES (X'48656c6c6f'), (X''), (NULL), (X'FF') -query expect_fallback(bit_length on BinaryType is not supported) +query SELECT bit_length(b) FROM test_bit_length_binary -query expect_fallback(bit_length on BinaryType is not supported) +query SELECT bit_length(X'48656c6c6f'), bit_length(CAST(NULL AS BINARY)) diff --git a/spark/src/test/resources/sql-tests/expressions/string/length.sql b/spark/src/test/resources/sql-tests/expressions/string/length.sql index 1e1e8ccc7a5..fc7807dc74f 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/length.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/length.sql @@ -15,6 +15,10 @@ -- specific language governing permissions and limitations -- under the License. +-- BinaryType has no native path, so it routes through the codegen dispatcher (Spark's own +-- `doGenCode`, i.e. `numBytes()`) instead of falling back to Spark. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true + statement CREATE TABLE test_length(s string) USING parquet @@ -27,3 +31,16 @@ SELECT length(s), char_length(s) FROM test_length -- literal arguments query SELECT length('hello'), length(''), length(NULL) + +-- BinaryType input routes through the codegen dispatcher and stays inside Comet +statement +CREATE TABLE test_length_binary(b binary) USING parquet + +statement +INSERT INTO test_length_binary VALUES (X'48656c6c6f'), (X''), (NULL), (X'FF') + +query +SELECT length(b) FROM test_length_binary + +query +SELECT length(X'48656c6c6f'), length(CAST(NULL AS BINARY)) diff --git a/spark/src/test/resources/sql-tests/expressions/string/octet_length.sql b/spark/src/test/resources/sql-tests/expressions/string/octet_length.sql index e650950347c..5add3b10020 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/octet_length.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/octet_length.sql @@ -15,6 +15,10 @@ -- specific language governing permissions and limitations -- under the License. +-- BinaryType has no native path, so it routes through the codegen dispatcher (Spark's own +-- `doGenCode`, i.e. `numBytes()`) instead of falling back to Spark. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true + statement CREATE TABLE test_octet_length(s string) USING parquet @@ -28,16 +32,15 @@ SELECT octet_length(s) FROM test_octet_length query SELECT octet_length('hello'), octet_length(''), octet_length(NULL) --- BinaryType input falls back to Spark; the native DataFusion impl rejects Binary at runtime, --- so the serde gates Binary as Unsupported (matching the existing CometLength shape). +-- BinaryType input routes through the codegen dispatcher and stays inside Comet statement CREATE TABLE test_octet_length_binary(b binary) USING parquet statement INSERT INTO test_octet_length_binary VALUES (X'48656c6c6f'), (X''), (NULL), (X'FF') -query expect_fallback(octet_length on BinaryType is not supported) +query SELECT octet_length(b) FROM test_octet_length_binary -query expect_fallback(octet_length on BinaryType is not supported) +query SELECT octet_length(X'48656c6c6f'), octet_length(CAST(NULL AS BINARY)) From b517646ff95d1e9a2b086a1802fddbdd3b805d87 Mon Sep 17 00:00:00 2001 From: adibmbrk Date: Fri, 4 Sep 2026 00:32:53 +0530 Subject: [PATCH 2/2] test: add binary-input microbenchmark for length/bit_length/octet_length Measures the three roots on BinaryType across the two paths the serdes can take, per review request. CometStringExpressionBenchmark covers them only on StringType, which takes the native DataFusion kernel and never exercises the binary route. Three arms as cases of one Benchmark, so warmup, iteration count, data and SQL settings are matched by construction: Spark; Comet with the codegen dispatcher disabled (the Spark-fallback path this PR replaces); and Comet with it enabled. Shapes span payload width (8 B / 64 B / 1 KB) and null fraction (0 / 50 / 90%). Each case prints its physical plan and a result digest before the timings, and the harness warns if the arms disagree. The measured result is that the dispatcher path is slower than the Spark fallback it replaces at every shape, by an amount that tracks payload width (+42 ms at 64 B, +329 ms at 1 KB over 2M rows); at 1 KB it is also about twice as slow as Spark. The kernel's generated getBinary allocates a byte[] and copies the whole payload per row, because that is what Spark's numBytes() reads, so a length that should be an offset subtraction pays for the full value. The null short-circuit elides that copy on skipped rows and narrows the gap as nulls rise, but does not close it. Signed-off-by: adibmbrk --- .../CometBinaryLengthBenchmark.scala | 248 ++++++++++++++++++ 1 file changed, 248 insertions(+) create mode 100644 spark/src/test/scala/org/apache/spark/sql/benchmark/CometBinaryLengthBenchmark.scala diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometBinaryLengthBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometBinaryLengthBenchmark.scala new file mode 100644 index 00000000000..0438b62df78 --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometBinaryLengthBenchmark.scala @@ -0,0 +1,248 @@ +/* + * 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.spark.sql.benchmark + +import org.apache.spark.benchmark.Benchmark +import org.apache.spark.sql.catalyst.optimizer.ConstantFolding +import org.apache.spark.sql.internal.SQLConf + +import org.apache.comet.CometConf + +/** + * One binary-input shape under benchmark. Each shape is materialized as its own column in the + * source table so a single write covers the whole grid. + * + * @param name + * short label for the shape + * @param column + * name of the source column holding the shape + * @param width + * payload width in bytes for the non-null rows + * @param nullPercent + * percentage of rows that are NULL + */ +case class BinaryLengthShape(name: String, column: String, width: Int, nullPercent: Int) { + + /** SQL that materializes this shape as a `binary` column. */ + def sql: String = { + // RPAD to an exact width over the row id keeps values distinct (so Parquet does not collapse + // the column to a single dictionary entry) while holding the payload width fixed. The digits + // and the pad character are ASCII, so the UTF-8 encoding is exactly `width` bytes. + val payload = s"CAST(RPAD(CAST(value AS STRING), $width, 'x') AS BINARY)" + val expr = if (nullPercent == 0) { + payload + } else { + s"CASE WHEN PMOD(value, 100) < $nullPercent THEN CAST(NULL AS BINARY) ELSE $payload END" + } + s"$expr AS $column" + } +} + +// spotless:off +/** + * Benchmark to measure performance of Comet `length` / `bit_length` / `octet_length` on + * `BinaryType` input, across the two paths the serdes can take. Added in response to the review + * request on the PR that mixed `CodegenDispatchFallback` into the three serdes. + * `CometStringExpressionBenchmark` covers these three roots only on `StringType`, which takes the + * native DataFusion kernel and so never exercises the binary route. + * + * Three arms, all cases of the same `Benchmark` so they share warmup, iteration count, data and + * SQL settings by construction: + * + * - `Spark` -- Comet off entirely, the reference. + * - `Comet (Spark fallback)` -- Comet on with the codegen dispatcher disabled. Binary input is + * `Unsupported` and, with the dispatcher off, `CodegenDispatchFallback` produces no marker, + * so the enclosing projection falls back to Spark. This is the path this PR replaces, and it + * is what the pre-PR build does unconditionally. + * - `Comet (codegen dispatch)` -- Comet on with the dispatcher enabled, i.e. the path this PR + * adds: Spark's own `doGenCode` (`numBytes()`) compiled into a per-batch kernel that reads + * the Arrow vector directly, with no transition out of the Comet pipeline. + * + * The shapes span payload width (8 / 64 / 1024 bytes) at a fixed null fraction, and null fraction + * (0% / 50% / 90%) at a fixed width. Both axes matter, and the measured result is that the + * dispatcher path is the slower of the two Comet arms: + * + * - Payload width drives the size of the gap. The kernel's `getBinary` allocates a `byte[]` and + * copies the whole payload out of the Arrow buffer per row (see `emitBinaryBodyUnsafe` in + * `CometBatchKernelCodegenInput`), because that is what Spark's generated `numBytes()` reads. + * So a `length` that should be an offset subtraction pays for the full value, and the cost the + * dispatch arm adds over the Spark-fallback arm tracks the payload: tens of ms at 8-64 B, + * several hundred at 1 KB, where the dispatch arm also runs about twice as slow as Spark. + * - Null fraction pulls the other way. The kernel short-circuits null rows ahead of the + * generated code, so a skipped row costs no copy and the gap shrinks as nulls rise -- at the + * same 64 B width it is several times smaller at 90% nulls than at 0%. It narrows sharply but + * does not close: the dispatch arm stays the slower of the two Comet arms at every shape here. + * + * Keeping the projection inside the Comet pipeline is therefore not automatically a win for these + * roots; on wide binary it is a regression against the Spark fallback it replaces. + * + * Each case prints its physical plan and a digest of its result set before the timings, so the + * report shows which operators each arm actually ran and that all three arms agree on the output. + * + * To run this benchmark: + * {{{ + * SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometBinaryLengthBenchmark + * }}} + * Results will be written to "spark/benchmarks/CometBinaryLengthBenchmark-**results.txt". + */ +// spotless:on +object CometBinaryLengthBenchmark extends CometBenchmarkBase { + + private val shapes = List( + BinaryLengthShape("width_8B", "b_w8", width = 8, nullPercent = 0), + BinaryLengthShape("width_64B", "b_w64", width = 64, nullPercent = 0), + BinaryLengthShape("width_1KB", "b_w1024", width = 1024, nullPercent = 0), + BinaryLengthShape("width_64B_50pct_null", "b_w64_n50", width = 64, nullPercent = 50), + BinaryLengthShape("width_64B_90pct_null", "b_w64_n90", width = 64, nullPercent = 90)) + + private val roots = List("length", "bit_length", "octet_length") + + /** + * Excluding `ConstantFolding` matches the rest of the expression benchmarks. Nothing in these + * queries is foldable (every root reads a column), but the exclusion is applied to all three + * arms so the optimizer configuration is identical across them. + */ + private val noConstantFolding = + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> ConstantFolding.ruleName + + private val sparkConfigs = Seq(noConstantFolding, CometConf.COMET_ENABLED.key -> "false") + + private def cometConfigs(dispatch: Boolean): Seq[(String, String)] = Seq( + noConstantFolding, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> dispatch.toString) + + override def runCometBenchmark(mainArgs: Array[String]): Unit = { + // Twice the 1M default the other expression benchmarks use. `length` on binary is O(1) per + // row, so what separates the two Comet arms is only the per-row cost of leaving the pipeline + // (`CometColumnarToRow` then a Spark `Project`) versus staying in it (`CometProject`). The row + // count has to be large enough that fixed query-submission cost does not drown that out: at + // 8192 rows every arm reports hundreds of ns/row, which is overhead rather than expression + // work. + val values = 2 * 1024 * 1024 // 2M rows + + runBenchmarkWithTable("Binary length expressions", values) { v => + withTempPath { dir => + withTempTable("parquetV1Table") { + prepareTable(dir, spark.sql(s"SELECT ${shapes.map(_.sql).mkString(", ")} FROM $tbl")) + + for (root <- roots; shape <- shapes) { + val name = s"$root ${shape.name}" + val query = s"select $root(${shape.column}) as v from parquetV1Table" + runBenchmark(name) { + runBinaryLengthModes(name, v, query) + } + } + } + } + } + } + + /** Runs the three arms for a single root/shape pair. */ + private def runBinaryLengthModes(name: String, cardinality: Long, query: String): Unit = { + val benchmark = new Benchmark(name, cardinality, output = output) + + val arms = Seq( + ("Spark", sparkConfigs), + ("Comet (Spark fallback)", cometConfigs(dispatch = false)), + ("Comet (codegen dispatch)", cometConfigs(dispatch = true))) + + // Report the plan and the result digest for every arm before timing anything. This both + // documents what each arm ran and guards the comparison: an arm that silently planned + // differently, or that disagreed on its output, is visible in the results file rather than + // showing up as an unexplained speedup. + val digests = arms.map { case (label, configs) => + label -> describe(benchmark, label, query, configs) + } + val (referenceLabel, reference) = digests.head + digests.tail.foreach { case (label, result) => + if (result != reference) { + report( + benchmark, + s"""WARNING: "$label" does not agree with "$referenceLabel". The arms below are not + |computing the same thing, so the timings are not comparable. + | $referenceLabel: $reference + | $label: $result""".stripMargin) + } + } + + arms.foreach { case (label, configs) => + benchmark.addCase(label) { _ => + withSQLConf(configs: _*) { + spark.sql(query).noop() + } + } + } + + benchmark.run() + } + + /** + * Runs `query` under `configs`, writes its physical plan and a digest of its output to the + * results file, and returns the digest so the caller can compare arms. + */ + private def describe( + benchmark: Benchmark, + label: String, + query: String, + configs: Seq[(String, String)]): String = withSQLConf(configs: _*) { + val df = spark.sql(query) + // Execute the benchmarked query itself rather than the digest query below, so the plan + // reported is the one the timings measure. `noop()` runs it without collecting. Execute + // before reading the plan: AQE only settles the final plan once the query has run. + df.noop() + val plan = stripAQEPlan(df.queryExecution.executedPlan) + val summary = digest(query) + report(benchmark, s"$label plan:\n${plan.treeString}$summary") + summary + } + + /** + * A one-line summary of `query`'s output, printed under each arm's plan and compared across + * arms. Aggregated in Spark rather than collected to the driver: at this row count, pulling + * every value back for a row-by-row comparison would cost more than the benchmark it guards. + * Count, null count, sum and range together are enough to catch an arm that computes a + * different length, or that returns a value where another returns NULL. + */ + private def digest(query: String): String = { + val row = spark + .sql(s"""SELECT COUNT(1), COUNT(v), SUM(CAST(v AS BIGINT)), MIN(v), MAX(v) + |FROM ($query)""".stripMargin) + .collect() + .head + s"result: rows=${row.get(0)} nulls=${row.getLong(0) - row.getLong(1)} " + + s"sum=${row.get(2)} min=${row.get(3)} max=${row.get(4)}" + } + + /** + * Writes a line to the benchmark results file as well as the console. `Benchmark.out` tees the + * two, so writing through it keeps this ordered against the results table `Benchmark.run` + * writes to the same stream. + * + * This mirrors `CometBenchmarkBase.warn`, which is private to the base. Kept local rather than + * widening the shared base's API, since no other benchmark reports plans: only this one was + * asked to show which path each arm took. + */ + private def report(benchmark: Benchmark, message: String): Unit = { + val border = "-" * 80 + benchmark.out.println(s"\n$border\n$message\n$border") + } +}