diff --git a/docs/source/contributor-guide/spark_configs_support.md b/docs/source/contributor-guide/spark_configs_support.md index 4ec8a99edf..56e963de1a 100644 --- a/docs/source/contributor-guide/spark_configs_support.md +++ b/docs/source/contributor-guide/spark_configs_support.md @@ -34,6 +34,8 @@ The status column uses these values: Spark for others, or runs natively but with documented incompatibilities. - **Falls back** -- Comet does not run the affected expressions natively under this config and always defers to Spark. +- **JVM dispatch** -- Comet runs Spark's generated expression code inside the Comet + pipeline, without falling the surrounding operator back to Spark. - **Unaudited** -- the config's interaction with Comet has not yet been verified. ## Audited Configurations @@ -44,6 +46,12 @@ The status column uses these values: - Affected expression: `exists` - Spark versions checked: 3.4.3, 3.5.8, 4.0.2, 4.1.2 - Date: 2026-07-22 +- `spark.sql.function.concatBinaryAsString` + - Default: `false` + - Status: JVM dispatch + - Affected expressions: `concat` with all-binary inputs + - Spark versions checked: 3.4.3, 3.5.8, 4.0.2, 4.1.2 + - Date: 2026-07-23 - `spark.sql.legacy.timeParserPolicy` - Default: `EXCEPTION` - Status: Partial (see notes) @@ -76,6 +84,24 @@ wraps the divergent column-backed case in `assertCodegenRan` and `checkSparkAnswerAndOperator`, directly verifying dispatcher activity and preventing the test from passing through silent Spark fallback. +### `spark.sql.function.concatBinaryAsString` + +Spark applies this configuration during type coercion. With the default `false` +value, `concat` retains `BinaryType` when every input is binary. Comet does not have a +native binary `concat`, so it runs Spark's generated `Concat` code through the +Arrow-direct JVM codegen dispatcher while keeping the surrounding operator in the +Comet pipeline. With `true`, Spark inserts `BinaryType`-to-`StringType` casts before +`concat`. Spark preserves arbitrary bytes in those casts, but Arrow strings require +valid UTF-8, so Comet also dispatches this expression tree to avoid normalizing +malformed bytes in a native cast. + +The SQL config-matrix coverage checks byte-level results, nulls, empty values, and +malformed UTF-8 inputs under both values. `CometCodegenSuite` checks the resolved +result type and additionally asserts that both config values invoke the JVM codegen +dispatcher, preventing a result-only test from passing through an unnoticed Spark +fallback. If the JVM codegen dispatcher is disabled, all-binary `concat` falls back +to Spark under either config value. + ### `spark.sql.legacy.timeParserPolicy` **Source.** `SQLConf.LEGACY_TIME_PARSER_POLICY` selects the formatter used by 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 ebf4508988..5ae23680ae 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -258,6 +258,8 @@ object CometConcat with CometTypeShim with CodegenDispatchFallback { private val unsupportedReason = "CONCAT supports only string input parameters" + private val binaryToStringCastReason = + "CONCAT over BinaryType-to-StringType casts requires Spark's byte-preserving semantics" // Spark 4.0 widens Concat to accept collated strings and preserves the collation in the merged // result type. The native concat UDF always produces UTF8 (UTF8_BINARY semantics), so a @@ -266,16 +268,30 @@ object CometConcat "concat does not support non-UTF8_BINARY collations " + "(https://github.com/apache/datafusion-comet/issues/2190)" - override def getUnsupportedReasons(): Seq[String] = Seq(unsupportedReason) + override def getUnsupportedReasons(): Seq[String] = + Seq(unsupportedReason, binaryToStringCastReason) override def getIncompatibleReasons(): Seq[String] = Seq(collationReason) override def getSupportLevel(expr: Concat): SupportLevel = { + // spark.sql.function.concatBinaryAsString=true inserts BinaryType-to-StringType casts before + // Concat. Spark's cast preserves arbitrary bytes, whereas Arrow strings require valid UTF-8 + // and Comet's native cast replaces malformed sequences. Dispatch the whole Spark expression + // so downstream byte-sensitive expressions (for example, hex) observe the original bytes. + val containsBinaryToStringCast = expr.children.exists { + case cast: Cast => + cast.child.dataType.isInstanceOf[BinaryType] && + cast.dataType.isInstanceOf[StringType] + case _ => false + } + // Use isInstanceOf rather than `== DataTypes.StringType` so that collated strings (a // StringType with a non-default collationId, which is not == the default StringType) are still // recognised as string input and routed to the collation check below rather than reported as // an unsupported input type. - if (!expr.children.forall(_.dataType.isInstanceOf[StringType])) { + if (containsBinaryToStringCast) { + Unsupported(Some(binaryToStringCastReason)) + } else if (!expr.children.forall(_.dataType.isInstanceOf[StringType])) { Unsupported(Some(unsupportedReason)) } else if (hasNonDefaultStringCollation(expr.dataType) || expr.children.exists(c => hasNonDefaultStringCollation(c.dataType))) { diff --git a/spark/src/test/resources/sql-tests/expressions/string/concat_binary_as_string.sql b/spark/src/test/resources/sql-tests/expressions/string/concat_binary_as_string.sql new file mode 100644 index 0000000000..408cc28b8a --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/string/concat_binary_as_string.sql @@ -0,0 +1,46 @@ +-- 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. + +-- ConfigMatrix: spark.sql.function.concatBinaryAsString=false,true + +statement +CREATE TABLE test_concat_binary_as_string(id int, c1 binary, c2 binary, c3 binary) USING parquet + +statement +INSERT INTO test_concat_binary_as_string VALUES + (1, X'6162', X'6364', X'65'), + (2, X'FF', X'FE41', X'80'), + (3, X'', X'00', NULL), + (4, NULL, X'01', X'02') + +-- false keeps all-binary concat as BinaryType. true inserts BinaryType-to-StringType casts. +-- Both expression trees run through Spark's generated code inside the Comet pipeline so the +-- true path preserves malformed bytes instead of normalizing them to UTF-8 replacement bytes. +query +SELECT id, hex(concat(c1, c2)) +FROM test_concat_binary_as_string +ORDER BY id + +query +SELECT id, hex(concat(c1, c2, c3)) +FROM test_concat_binary_as_string +ORDER BY id + +query +SELECT + hex(concat(X'FF', X'FE41')), + concat(CAST(NULL AS BINARY), X'01') IS NULL diff --git a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala index 5806cb3501..6ede84eefd 100644 --- a/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala @@ -126,6 +126,32 @@ class CometCodegenSuite } } + test("spark.sql.function.concatBinaryAsString preserves types through dispatch") { + withTable("t") { + sql("CREATE TABLE t (id INT, c1 BINARY, c2 BINARY) USING parquet") + sql("""INSERT INTO t VALUES + | (1, X'6162', X'6364'), + | (2, X'FF', X'FE41'), + | (3, X'', X'00'), + | (4, NULL, X'01')""".stripMargin) + + Seq(false -> BinaryType, true -> StringType).foreach { case (enabled, expectedType) => + withSQLConf("spark.sql.function.concatBinaryAsString" -> enabled.toString) { + val concatenated = sql("SELECT concat(c1, c2) AS value FROM t ORDER BY id") + assert(concatenated.schema("value").dataType == expectedType) + checkSparkSchema(concatenated) + + val result = sql("SELECT hex(concat(c1, c2)) AS value FROM t ORDER BY id") + // The operator check rejects a Spark plan fallback; the counter check proves that the + // Spark expression itself ran through the in-pipeline JVM codegen dispatcher. + assertCodegenRan { + checkSparkAnswerAndOperator(result) + } + } + } + } + } + test("disabled mode bypasses the dispatcher") { // When the per-feature config is off, `CometScalaUDF.convert` returns None and the enclosing // operator falls back to Spark. The dispatcher's counters must not move.