feat: route translate and to_csv through codegen dispatch by default

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

@@ -239,7 +239,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | --- | --- | --- | --- |
 | `from_csv` | ✅ | Codegen dispatch |  |
 | `schema_of_csv` | ✅ | Codegen dispatch |  |
-| `to_csv` | ✅ | Native |  |
+| `to_csv` | ✅ | Hybrid | Codegen dispatch by default; the native path is opt-in via allowIncompatible |
 
 ---
 
@@ -607,7 +607,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | `to_char` | ✅ | Codegen dispatch |  |
 | `to_number` | ✅ | Codegen dispatch |  |
 | `to_varchar` | ✅ | Codegen dispatch |  |
-| `translate` | ✅ | Native | 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 is opt-in via allowIncompatible |
+| `translate` | ✅ | Hybrid | Codegen dispatch 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 is opt-in via allowIncompatible |
 | `trim` | ✅ | Native |  |
 | `try_to_binary` | ✅ | — | Runs natively (rewrites to `try_eval(to_binary(...))`) |
 | `try_to_number` | ✅ | Codegen dispatch | 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

@@ -258,7 +258,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,109 @@
+-- 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: on Spark 3.4 / 3.5 the CSV
+-- converter has no binary branch and 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. (Spark 4.0 added a real binary formatter, but
+-- this fixture runs on every supported version.)
+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). `StructsToCsv.checkInputDataTypes` accepts
+-- them and `CometBatchKernelCodegen.isSupportedDataType` recurses into them, so they pass the
+-- plan-time gate and reach the runtime dispatch. That is the accepted-at-plan-time /
+-- rejected-at-runtime shape #5219 found for TIME types: `canHandle` greenlights the expression
+-- before the plan commits, so a runtime gap is an execute-time failure with no fallback left.
+--
+-- Only non-nullness is asserted here, because the rendered value is not comparable on every
+-- supported Spark version. Spark 3.4 / 3.5's `UnivocityGenerator.makeConverter` has no branch for
+-- complex types and lands on `getter.get(ordinal, dataType).toString`, an identity string such as
+-- "org.apache.spark.sql.vectorized.ColumnarArray@1ada50f0" whose hash differs between any two
+-- evaluations and whose class differs between Spark's converter input (`ColumnarArray` /
+-- `UnsafeArrayData`) and the kernel's (`InputArray_*`). That generic `get` is exactly what
+-- `CometSpecializedGettersDispatch` implements for `CometInternalRow` / `CometArrayData`, so on
+-- those versions these queries are the coverage for it. Spark 4.0 added real array/map/struct
+-- converters, which render deterministically; `to_csv_nested.sql` asserts those values in full.
+statement
+CREATE TABLE test_to_csv_nested(s struct<i: int, arr: array<int>, m: map<string, int>, n: struct<x: int, y: string>>) USING parquet
+
+statement
+INSERT INTO test_to_csv_nested VALUES
+  (named_struct('i', 1, 'arr', array(1, 2, 3), 'm', map('k', 10), 'n', named_struct('x', 5, 'y', 'z'))),
+  (named_struct('i', NULL, 'arr', CAST(NULL AS array<int>), 'm', CAST(NULL AS map<string, int>), 'n', CAST(NULL AS struct<x: int, y: string>))),
+  (CAST(NULL AS struct<i: int, arr: array<int>, m: map<string, int>, n: struct<x: int, y: string>>))
+
+-- Struct column straight from the scan: the kernel reads it with getStruct, so the converter's
+-- per-field reads of the array / map / struct fields all land on `CometInternalRow`. The all-null
+-- and null-struct rows keep the assertion from collapsing to a constant true.
+query
+SELECT to_csv(s) IS NOT NULL FROM test_to_csv_nested
+
+-- named_struct over the complex fields: the kernel's array / map / struct readers produce the
+-- values that CreateNamedStruct stores into the row the converter then reads.
+query
+SELECT to_csv(named_struct('arr', s.arr, 'm', s.m, 'n', s.n)) IS NOT NULL FROM test_to_csv_nested
+
+-- Complex types nested inside complex types, so the converter recurses through the kernel's
+-- element getters rather than stopping at the first level.
+query
+SELECT to_csv(named_struct('aa', array(array(1, 2), array(3)), 'ms', map('k', named_struct('x', 1)))) IS NOT NULL FROM test_to_csv_nested

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

@@ -0,0 +1,56 @@
+-- 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.
+
+-- ANSI mode with a raising subtree inside the struct: `to_csv` routes through the codegen
+-- dispatcher by default, and the whole tree (StructsToCsv over CreateNamedStruct over Cast) is
+-- bound into one kernel, so the Cast's ANSI throw has to survive the dispatcher rather than being
+-- swallowed into a NULL. That is the failure mode #5219 fixed for `CometBatchKernelCodegen`'s null
+-- short-circuit, which skipped a subtree Spark would have evaluated when every input ordinal was
+-- tested up front.
+--
+-- `to_csv` does not reach the short-circuit today (CreateNamedStruct is not NullIntolerant, so
+-- `allNullIntolerant` already fails), but the shape is one wire-up change away from mattering and
+-- was untested for this expression, so these queries pin it. The second query is the exact #5219
+-- shape: two input ordinals where the non-cast one is NULL on the row whose cast raises.
+-- Config: spark.sql.ansi.enabled=true
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_to_csv_ansi(s string, i int) USING parquet
+
+statement
+INSERT INTO test_to_csv_ansi VALUES
+  ('1', 10),
+  ('notanint', NULL)
+
+query expect_error(CAST_INVALID_INPUT)
+SELECT to_csv(named_struct('a', CAST(s AS INT))) FROM test_to_csv_ansi
+
+query expect_error(CAST_INVALID_INPUT)
+SELECT to_csv(named_struct('a', CAST(s AS INT), 'b', i)) FROM test_to_csv_ansi
+
+-- The struct field's cast is the only raising node, so restricting to the well-formed row makes
+-- the expression succeed. Doubles as the sentinel required by `ExpectError` fixtures: it uses
+-- checkSparkAnswerAndOperator, so a silent dispatcher rejection (which would make the queries
+-- above pass vacuously via a Spark fallback that raises the same error) fails here instead.
+query
+SELECT to_csv(named_struct('a', CAST(s AS INT), 'b', i)) FROM test_to_csv_ansi WHERE i IS NOT NULL
+
+-- try_cast does not raise under ANSI, so the surrounding to_csv must render the NULL field with
+-- the CSV nullValue rather than propagating an error.
+query
+SELECT to_csv(named_struct('a', TRY_CAST(s AS INT), 'b', i)) FROM test_to_csv_ansi

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

@@ -0,0 +1,56 @@
+-- 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.
+
+-- Full-value coverage of `to_csv` over complex field types, which `to_csv.sql` can only assert for
+-- non-nullness because Spark 3.4 / 3.5 render a complex field as a Java identity string. Spark 4.0
+-- gave `UnivocityGenerator.makeConverter` real array / map / struct branches ("[1, 2, 3]",
+-- "{k -> 10}", "{5, z}"), built from the typed `getArray` / `getMap` / `getStruct` getters, so from
+-- 4.0 on the value is deterministic and Comet's kernel-side `CometInternalRow` / `CometArrayData` /
+-- `CometMapData` getters have to agree with Spark's row-side ones element for element.
+-- MinSparkVersion: 4.0
+
+statement
+CREATE TABLE test_to_csv_nested4(s struct<i: int, arr: array<int>, m: map<string, int>, n: struct<x: int, y: string>>) USING parquet
+
+statement
+INSERT INTO test_to_csv_nested4 VALUES
+  (named_struct('i', 1, 'arr', array(1, 2, 3), 'm', map('k', 10), 'n', named_struct('x', 5, 'y', 'z'))),
+  (named_struct('i', 2, 'arr', array(), 'm', map(), 'n', named_struct('x', NULL, 'y', NULL))),
+  (named_struct('i', 3, 'arr', array(1, CAST(NULL AS int)), 'm', map('k', CAST(NULL AS int)), 'n', named_struct('x', 7, 'y', 'a,b'))),
+  (named_struct('i', NULL, 'arr', CAST(NULL AS array<int>), 'm', CAST(NULL AS map<string, int>), 'n', CAST(NULL AS struct<x: int, y: string>))),
+  (CAST(NULL AS struct<i: int, arr: array<int>, m: map<string, int>, n: struct<x: int, y: string>>))
+
+-- Struct column straight from the scan: the converter reads the array / map / struct fields off
+-- the kernel's CometInternalRow. Empty collections, null elements and a value needing CSV quoting
+-- are all in the data above.
+query
+SELECT to_csv(s) FROM test_to_csv_nested4
+
+-- nullValue changes how a null *element* inside a complex field renders (`appendNull` only emits
+-- when the option was set explicitly), which is a separate code path from a null top-level field.
+query
+SELECT to_csv(s, map('nullValue', 'NIL')) FROM test_to_csv_nested4
+
+-- named_struct over the complex fields, so the values pass through CreateNamedStruct's row before
+-- the converter reads them back.
+query
+SELECT to_csv(named_struct('arr', s.arr, 'm', s.m, 'n', s.n)) FROM test_to_csv_nested4
+
+-- Complex inside complex: the converter recurses, so an array-of-array read goes through the
+-- kernel's element getters rather than stopping at the first level.
+query
+SELECT to_csv(named_struct('aa', array(array(1, 2), array(3)), 'ms', map('k', named_struct('x', 1)))) FROM test_to_csv_nested4

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

@@ -15,29 +15,44 @@
 -- 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. Written with \u escapes rather than literal characters to
+-- keep the fixture ASCII. They have to be plain string literals: an inline table only accepts
+-- expressions the analyzer can evaluate, and `decode(X'..')` is not one of them on Spark 4.1
+-- (INVALID_INLINE_TABLE.CANNOT_EVALUATE_EXPRESSION_IN_INLINE_TABLE).
+--   'cafe\u0301' ends in "e" + U+0301 COMBINING ACUTE ACCENT: one grapheme, two code
+--     points. DataFusion iterates graphemes, Spark iterates code points, so translating "e"
+--     differs.
+--   '\u0000' as the `to` argument is U+0000, which Spark treats as a deletion sentinel and
+--     the native path substitutes it 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'),
+  ('cafe\u0301', 'e', 'E'),
+  ('hello', 'l', '\u0000')
 
-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)
     }
   }