feat: route translate and to_csv through codegen dispatch by default

docs/source/user-guide/latest/expressions.md

@@ -226,7 +226,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | --- | --- | --- |
 | `from_csv` | ✅ |  |
 | `schema_of_csv` | ✅ |  |
-| `to_csv` | ✅ |  |
+| `to_csv` | ✅ | Runs through the JVM codegen dispatcher by default; the native path is opt-in via allowIncompatible |
 
 ---
 
@@ -594,7 +594,7 @@ expression-level). The `outer` variants are wired but marked `Incompatible`; the
 | `to_char` | ✅ |  |
 | `to_number` | ✅ |  |
 | `to_varchar` | ✅ |  |
-| `translate` | ✅ | Falls back by default; opt-in via allowIncompatible ([#4463](https://github.com/apache/datafusion-comet/issues/4463)) |
+| `translate` | ✅ | Runs through the JVM codegen dispatcher by default; the native path (grapheme vs code-point differences) is opt-in via allowIncompatible ([#4463](https://github.com/apache/datafusion-comet/issues/4463)) |
 | `trim` | ✅ |  |
 | `try_to_binary` | ✅ | Runs natively (rewrites to `try_eval(to_binary(...))`) |
 | `try_to_number` | ✅ | Routed through the JVM codegen dispatcher |

spark/src/main/scala/org/apache/comet/serde/strings.scala

@@ -109,7 +109,12 @@ object CometOctetLength extends CometScalarFunction[OctetLength]("octet_length")
   }
 }
 
-object CometStringTranslate extends CometScalarFunction[StringTranslate]("translate") {
+// Routes through the JVM codegen dispatcher by default via `CodegenDispatchFallback`: the
+// `Incompatible` result below reaches the dispatcher (Spark's own `doGenCode`, bit-exact) instead
+// of falling the projection back to Spark. The native path stays available via `allowIncompatible`.
+object CometStringTranslate
+    extends CometScalarFunction[StringTranslate]("translate")
+    with CodegenDispatchFallback {
   private val incompatReason =
     "DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and" +
       " substitutes U+0000 instead of treating it as a deletion sentinel"

spark/src/main/scala/org/apache/comet/serde/structs.scala

@@ -259,7 +259,12 @@ object CometJsonToStructs extends CometCodegenDispatch[JsonToStructs] with Nativ
   }
 }
 
-object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] {
+// Routes through the JVM codegen dispatcher by default: the `Unsupported` (complex field types)
+// and non-opted-in `Incompatible` (#3232 field types) results below are both handled by
+// `CodegenDispatchFallback`, which runs Spark's own `doGenCode` inside the Comet pipeline so the
+// projection stays native and bit-exact. The native ToCsv path remains available via
+// `allowIncompatible`, for the field types it can actually handle.
+object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] with CodegenDispatchFallback {
 
   private val incompatibleDataTypes = Seq(DateType, TimestampType, TimestampNTZType, BinaryType)
 

spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql

@@ -0,0 +1,77 @@
+-- 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.
+
+-- to_csv runs through the codegen dispatcher by default so results match Spark exactly, including
+-- quoting and escaping. The native path is opt-in via
+-- spark.comet.expression.StructsToCsv.allowIncompatible.
+
+statement
+CREATE TABLE test_to_csv(a int, b string, c double) USING parquet
+
+statement
+INSERT INTO test_to_csv VALUES
+  (1, 'x', 2.5),
+  (-3, 'hello,world', 0.0),
+  (0, 'has "quote"', -1.5),
+  (NULL, NULL, NULL),
+  (7, '', 3.0)
+
+-- column struct: values with delimiters and quotes exercise Spark's CSV quoting rules
+query
+SELECT to_csv(named_struct('a', a, 'b', b, 'c', c)) FROM test_to_csv
+
+-- literal struct (constant folding is disabled by the test suite)
+query
+SELECT
+  to_csv(named_struct('a', 1, 'b', 'x', 'c', 2.5)),
+  to_csv(named_struct('s', 'a,b', 'n', CAST(NULL AS INT)))
+
+-- Options are the main axis of behavior difference for this expression, and the dispatcher path
+-- had no options coverage anywhere (CometCsvExpressionSuite only varies them under
+-- allowIncompatible=true). `sep` also changes which values need quoting.
+query
+SELECT to_csv(named_struct('a', a, 'b', b, 'c', c), map('sep', ';')) FROM test_to_csv
+
+-- timestampFormat over date and timestamp fields: these are the #3232 types that were
+-- Incompatible before this change and now route through the dispatcher.
+-- BinaryType is the other #3232 type but is deliberately absent: Spark's CSV converter renders it
+-- with Java's default Object.toString(), e.g. "[B@10bc15e4", an identity hash that differs between
+-- any two evaluations, so the value is not assertable by any engine including Spark itself.
+statement
+CREATE TABLE test_to_csv_temporal(d date, t timestamp) USING parquet
+
+statement
+INSERT INTO test_to_csv_temporal VALUES
+  (DATE '2024-01-31', TIMESTAMP '2024-01-31 12:34:56.789'),
+  (DATE '1970-01-01', TIMESTAMP '1970-01-01 00:00:00'),
+  (NULL, NULL)
+
+query
+SELECT to_csv(named_struct('d', d, 't', t)) FROM test_to_csv_temporal
+
+query
+SELECT to_csv(named_struct('d', d, 't', t), map('timestampFormat', 'yyyy/MM/dd HH:mm', 'dateFormat', 'dd-MM-yyyy')) FROM test_to_csv_temporal
+
+-- Complex field types (arrays, maps, nested structs) are deliberately not asserted here.
+-- StructsToCsv.checkInputDataTypes accepts them, so they reach the converter, but Spark's own
+-- output for them is not a value that can be compared: a non-null array/map/struct renders as a
+-- Java identity string such as "org.apache.spark.sql.vectorized.ColumnarArray@1ada50f0", and any
+-- null complex value throws NullPointerException inside Spark's UnsafeWriter.write. Both were
+-- confirmed against Spark 3.5 with Comet disabled entirely, so they are Spark behavior, not
+-- Comet's. Before this change these schemas were Unsupported and fell the projection back to
+-- Spark; now they reach the dispatcher, which runs the same Spark code, so the observable result
+-- is unchanged either way.

spark/src/test/resources/sql-tests/expressions/string/string_translate.sql

@@ -15,29 +15,41 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
--- translate is gated as Incompatible by default. DataFusion's translate iterates over Unicode
--- graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion
--- sentinel, so the native path silently diverges from Spark for combining-mark inputs and for
--- to=NUL. These default-config tests assert that the expression falls back cleanly to Spark.
--- See string_translate_enabled.sql for the opt-in native path.
+-- translate runs through the codegen dispatcher by default so results match Spark exactly. The
+-- native path diverges from Spark (DataFusion iterates over Unicode graphemes where Spark uses code
+-- points, and substitutes U+0000 instead of treating it as a deletion sentinel), so it is opt-in
+-- via spark.comet.expression.StringTranslate.allowIncompatible. See string_translate_enabled.sql
+-- for the opt-in native path.
 
 statement
 CREATE TABLE test_translate(s string, from_str string, to_str string) USING parquet
 
+-- The last two rows are the regression test for this file's routing change: they are the inputs
+-- where the native path is known to disagree with Spark, so they are only assertable now that the
+-- default is the bit-exact dispatcher. Built with decode() rather than literal characters to keep
+-- the fixture ASCII.
+--   decode(X'65CC81') is "e" + U+0301 COMBINING ACUTE ACCENT: one grapheme, two code points.
+--     DataFusion iterates graphemes, Spark iterates code points, so translating "e" differs.
+--   decode(X'00') as the `to` argument is U+0000, which Spark treats as a deletion sentinel and
+--     the native path substitutes literally.
 statement
-INSERT INTO test_translate VALUES ('hello', 'el', 'ip'), ('hello', 'aeiou', '12345'), ('', 'a', 'b'), (NULL, 'a', 'b'), ('hello', '', ''), ('abc', 'abc', 'x')
+INSERT INTO test_translate VALUES
+  ('hello', 'el', 'ip'), ('hello', 'aeiou', '12345'), ('', 'a', 'b'), (NULL, 'a', 'b'),
+  ('hello', '', ''), ('abc', 'abc', 'x'),
+  (concat('caf', decode(X'65CC81', 'UTF-8')), 'e', 'E'),
+  ('hello', 'l', decode(X'00', 'UTF-8'))
 
-query expect_fallback(is not fully compatible with Spark)
+query
 SELECT translate(s, from_str, to_str) FROM test_translate
 
 -- column + literal + literal
-query expect_fallback(is not fully compatible with Spark)
+query
 SELECT translate(s, 'el', 'ip') FROM test_translate
 
 -- literal + column + column
-query expect_fallback(is not fully compatible with Spark)
+query
 SELECT translate('hello', from_str, to_str) FROM test_translate
 
 -- literal + literal + literal
-query expect_fallback(is not fully compatible with Spark)
+query
 SELECT translate('hello', 'el', 'ip'), translate('hello', 'aeiou', '12345'), translate('', 'a', 'b'), translate(NULL, 'a', 'b')

spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala

@@ -47,24 +47,21 @@ class CometCsvExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper
           SchemaGenOptions(generateArray = false, generateStruct = false, generateMap = false),
           DataGenOptions(allowNull = true, generateNegativeZero = true))
       }
-      withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[StructsToCsv]) -> "true") {
-        val df = spark.read
-          .parquet(filename)
-          .select(
-            to_csv(
-              struct(
-                col("c0"),
-                col("c1"),
-                col("c2"),
-                col("c3"),
-                col("c4"),
-                col("c5"),
-                col("c7"),
-                col("c8"),
-                col("c9"),
-                col("c12"))))
-        checkSparkAnswerAndOperator(df)
-      }
+      // Every column in the fuzz schema except c13 Binary, with no allowIncompatible opt-in:
+      // to_csv now routes through the codegen dispatcher by default, so Spark's own converter runs
+      // and all field types match by construction. This previously ran under
+      // allowIncompatible=true and had to skip c6 Double, c10 Timestamp and c11 TimestampNTZ,
+      // which are types the native path cannot match Spark on (the latter two are #3232).
+      //
+      // c13 Binary stays out, and not because of a divergence: Spark's CSV converter renders
+      // BinaryType with Java's default Object.toString(), so a row comes out as
+      // "...,[B@731af74". That is an identity hash of the byte array instance, which differs
+      // between any two evaluations, so the value can never be asserted against a second run --
+      // by Comet or by Spark against itself.
+      val df = spark.read
+        .parquet(filename)
+        .select(to_csv(struct((0 to 12).map(i => col(s"c$i")): _*)))
+      checkSparkAnswerAndOperator(df)
     }
   }