fix: preserve concat binary string bytes

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

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))) {

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

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.