From 0f579eafb594a6582e447c7e44f7184f84dde007 Mon Sep 17 00:00:00 2001 From: adibmbrk Date: Tue, 1 Sep 2026 07:21:22 +0530 Subject: [PATCH 1/3] 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 2cf0894ceb8..2b21a1ce600 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` | ✅ | Hybrid | Non-UTF8_BINARY collated input routes through the JVM codegen dispatcher; other input runs natively | | `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 32cb6b66414..c0e23045b1f 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 60aae0abb9f6b129223d7b3d07fff989f49eefc9 Mon Sep 17 00:00:00 2001 From: adibmbrk Date: Fri, 4 Sep 2026 00:32:53 +0530 Subject: [PATCH 2/3] 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") + } +} From 4feac8447fa325102bafffb5e40fec5d9255038b Mon Sep 17 00:00:00 2001 From: adibmbrk Date: Mon, 7 Sep 2026 13:05:20 +0530 Subject: [PATCH 3/3] fix: keep Spark fallback for length/bit_length/octet_length on binary input The codegen dispatcher measured slower than the Spark fallback it replaced on BinaryType at every shape (+25.6% at 64 B, +27.6% at 1 KB), so CometLength, CometBitLength, and CometOctetLength no longer mix in CodegenDispatchFallback: BinaryType input falls back to Spark again, as before the previous change. Also fixes the benchmark's Scala 3.4/3.5 compile break (withSQLConf returns Unit there, not the block's value), captures the plan of the actual timed write command instead of the pre-noop SELECT, and adds a downstream-consumer case and a combined-projection case across all three roots. Signed-off-by: adibmbrk --- docs/source/user-guide/latest/expressions.md | 12 +-- .../org/apache/comet/serde/strings.scala | 21 ++--- .../expressions/string/bit_length.sql | 10 +-- .../sql-tests/expressions/string/length.sql | 10 +-- .../expressions/string/octet_length.sql | 10 +-- .../CometBinaryLengthBenchmark.scala | 77 ++++++++++++++++--- 6 files changed, 87 insertions(+), 53 deletions(-) diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 2b21a1ce600..2cf0894ceb8 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` | ✅ | Hybrid | | +| `bit_length` | ✅ | Native | | | `btrim` | ✅ | — | | | `char` | ✅ | Native | | -| `char_length` | ✅ | Hybrid | | -| `character_length` | ✅ | Hybrid | | +| `char_length` | ✅ | Native | | +| `character_length` | ✅ | Native | | | `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` | ✅ | Hybrid | | -| `length` | ✅ | Hybrid | | +| `len` | ✅ | Native | | +| `length` | ✅ | Native | | | `levenshtein` | ✅ | Hybrid | Non-UTF8_BINARY collated input routes through the JVM codegen dispatcher; other input runs natively | | `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` | ✅ | Hybrid | | +| `octet_length` | ✅ | Native | | | `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 c0e23045b1f..3899e551a02 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -82,10 +82,9 @@ object CometUpper extends CometCaseConversionBase[Upper]("upper") object CometLower extends CometCaseConversionBase[Lower]("lower") -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. +object CometLength extends CometScalarFunction[Length]("length") { + // Benchmarked: the codegen dispatcher is slower than Spark fallback on BinaryType (see + // CometBinaryLengthBenchmark), so BinaryType stays on plain Spark fallback for now. override def getUnsupportedReasons(): Seq[String] = Seq("`BinaryType` input is not supported") override def getSupportLevel(expr: Length): SupportLevel = expr.child.dataType match { @@ -94,11 +93,8 @@ object CometLength extends CometScalarFunction[Length]("length") with CodegenDis } } -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. +object CometBitLength extends CometScalarFunction[BitLength]("bit_length") { + // See CometLength: benchmarked slower via the dispatcher, so BinaryType stays on Spark fallback. override def getUnsupportedReasons(): Seq[String] = Seq("`BinaryType` input is not supported") override def getSupportLevel(expr: BitLength): SupportLevel = expr.child.dataType match { @@ -107,11 +103,8 @@ object CometBitLength } } -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. +object CometOctetLength extends CometScalarFunction[OctetLength]("octet_length") { + // See CometLength: benchmarked slower via the dispatcher, so BinaryType stays on Spark fallback. 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 b9fd2abcd23..3d58c69a136 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,10 +15,6 @@ -- 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 @@ -32,15 +28,15 @@ SELECT bit_length(s) FROM test_bit_length query SELECT bit_length('hello'), bit_length(''), bit_length(NULL) --- BinaryType input routes through the codegen dispatcher and stays inside Comet +-- BinaryType input falls back to Spark; benchmarked slower via the codegen dispatcher. 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 +query expect_fallback(bit_length on BinaryType is not supported) SELECT bit_length(b) FROM test_bit_length_binary -query +query expect_fallback(bit_length on BinaryType is not supported) 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 fc7807dc74f..e8151371ae9 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/length.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/length.sql @@ -15,10 +15,6 @@ -- 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 @@ -32,15 +28,15 @@ SELECT length(s), char_length(s) FROM test_length query SELECT length('hello'), length(''), length(NULL) --- BinaryType input routes through the codegen dispatcher and stays inside Comet +-- BinaryType input falls back to Spark; benchmarked slower via the codegen dispatcher. statement CREATE TABLE test_length_binary(b binary) USING parquet statement INSERT INTO test_length_binary VALUES (X'48656c6c6f'), (X''), (NULL), (X'FF') -query +query expect_fallback(Length on BinaryType is not supported) SELECT length(b) FROM test_length_binary -query +query expect_fallback(Length on BinaryType is not supported) 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 5add3b10020..09196e37853 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,10 +15,6 @@ -- 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 @@ -32,15 +28,15 @@ SELECT octet_length(s) FROM test_octet_length query SELECT octet_length('hello'), octet_length(''), octet_length(NULL) --- BinaryType input routes through the codegen dispatcher and stays inside Comet +-- BinaryType input falls back to Spark; benchmarked slower via the codegen dispatcher. 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 +query expect_fallback(octet_length on BinaryType is not supported) SELECT octet_length(b) FROM test_octet_length_binary -query +query expect_fallback(octet_length on BinaryType is not supported) SELECT octet_length(X'48656c6c6f'), octet_length(CAST(NULL AS BINARY)) 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 index 0438b62df78..3d07bc86930 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometBinaryLengthBenchmark.scala +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometBinaryLengthBenchmark.scala @@ -20,8 +20,11 @@ package org.apache.spark.sql.benchmark import org.apache.spark.benchmark.Benchmark +import org.apache.spark.sql.DataFrame import org.apache.spark.sql.catalyst.optimizer.ConstantFolding +import org.apache.spark.sql.execution.{QueryExecution, SparkPlan} import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.util.QueryExecutionListener import org.apache.comet.CometConf @@ -91,7 +94,9 @@ case class BinaryLengthShape(name: String, column: String, width: Int, nullPerce * 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. + * roots; on wide binary it is a regression against the Spark fallback it replaces. Given these + * results, the three serdes no longer mix in `CodegenDispatchFallback` for `BinaryType`; both + * Comet arms below now take the Spark-fallback path until a narrower implementation lands. * * 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. @@ -151,6 +156,31 @@ object CometBinaryLengthBenchmark extends CometBenchmarkBase { runBinaryLengthModes(name, v, query) } } + + // Width shapes only: null fraction is orthogonal to what these two extra scenarios test. + val widthShapes = shapes.take(3) // 8B / 64B / 1KB + + // A native downstream consumer (aggregation) so an arm that leaves the Comet pipeline + // for the projection also pays a transition back in for the sum. + for (shape <- widthShapes) { + val name = s"length downstream-sum ${shape.name}" + val query = + s"select sum(cast(length(${shape.column}) as bigint)) as v from parquetV1Table" + runBenchmark(name) { + runBinaryLengthModes(name, v, query) + } + } + + // All three roots in one projection, to amortize the dispatcher's per-expression cost. + for (shape <- widthShapes) { + val name = s"combined roots ${shape.name}" + val c = shape.column + val query = + s"select (length($c) + bit_length($c) + octet_length($c)) as v from parquetV1Table" + runBenchmark(name) { + runBinaryLengthModes(name, v, query) + } + } } } } @@ -196,25 +226,48 @@ object CometBinaryLengthBenchmark extends CometBenchmarkBase { } /** - * 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. + * Runs `query` under `configs`, writes the actually-timed plan and a digest of its output to + * the results file, and returns the digest so the caller can compare arms. `withSQLConf` + * returns `Unit` on Spark 3.4/3.5, so the result is captured inside the block and returned + * after it. */ 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") + configs: Seq[(String, String)]): String = { + var summary: String = null + withSQLConf(configs: _*) { + val plan = stripAQEPlan(timedPlan(spark.sql(query))) + summary = digest(query) + report(benchmark, s"$label plan:\n${plan.treeString}$summary") + } summary } + /** + * Runs `df.noop()` and returns the executed plan of the write command it actually times. + * `noop()` runs a separate command `QueryExecution` from `df`'s own, so `df.queryExecution` is + * the pre-write SELECT plan, not what the timings measure; a listener captures the command's. + */ + private def timedPlan(df: DataFrame): SparkPlan = { + @volatile var captured: SparkPlan = null + val listener = new QueryExecutionListener { + override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = + captured = qe.executedPlan + override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit = + () + } + spark.listenerManager.register(listener) + try { + df.noop() + } finally { + spark.listenerManager.unregister(listener) + } + require(captured != null, "no QueryExecution captured for noop() write command") + captured + } + /** * 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