From 800113743aead4df540d24a9d73177d21e6a295e Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 24 Jul 2026 15:55:37 -0600 Subject: [PATCH 1/3] feat: route translate and to_csv through codegen dispatch by default `translate` (StringTranslate) and `to_csv` (StructsToCsv) have native implementations that are incompatible with Spark by default, so today the whole enclosing operator falls back to Spark unless allowIncompatible is set. Route them through the JVM codegen dispatcher by default instead, using the same NativeOptInAvailable pattern as `to_json` and `split`: the dispatcher runs Spark's own generated code (bit-exact) while keeping the operator on the native pipeline, and the faster native path stays available via allowIncompatible. Update the translate SQL test (previously asserting fallback) to verify native dispatch, and add a to_csv test. --- docs/source/user-guide/latest/expressions.md | 4 +- .../org/apache/comet/serde/strings.scala | 29 ++++++++- .../org/apache/comet/serde/structs.scala | 60 +++++++++---------- .../sql-tests/expressions/csv/to_csv.sql | 41 +++++++++++++ .../expressions/string/string_translate.sql | 18 +++--- 5 files changed, 108 insertions(+), 44 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 189033f3f72..792ee87d783 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/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 | 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 ebf45089882..81d6606740b 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -109,15 +109,38 @@ object CometOctetLength extends CometScalarFunction[OctetLength]("octet_length") } } -object CometStringTranslate extends CometScalarFunction[StringTranslate]("translate") { +object CometStringTranslate + extends CometExpressionSerde[StringTranslate] + with NativeOptInAvailable { 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" override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason) - override def getSupportLevel(expr: StringTranslate): SupportLevel = Incompatible( - Some(incompatReason)) + override def getSupportLevel(expr: StringTranslate): SupportLevel = + if (!CometConf.isExprAllowIncompat(getExprConfigName(expr))) { + Compatible(nativeOptIn = + Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr))))) + } else { + Compatible() + } + + override def convert( + expr: StringTranslate, + inputs: Seq[Attribute], + binding: Boolean): Option[Expr] = { + if (CometConf.isExprAllowIncompat(getExprConfigName(expr))) { + val childExprs = expr.children.map(exprToProtoInternal(_, inputs, binding)) + val optExpr = scalarFunctionExprToProto("translate", childExprs: _*) + optExprWithFallbackReason(optExpr, expr, expr.children: _*) + } else { + // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the + // Comet pipeline (bit-exact). The native path differs on graphemes vs code points, so it is + // opt-in via allowIncompatible. + CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding) + } + } } object CometLevenshtein extends CometExpressionSerde[Levenshtein] { diff --git a/spark/src/main/scala/org/apache/comet/serde/structs.scala b/spark/src/main/scala/org/apache/comet/serde/structs.scala index 2f9619d491f..9346612b218 100644 --- a/spark/src/main/scala/org/apache/comet/serde/structs.scala +++ b/spark/src/main/scala/org/apache/comet/serde/structs.scala @@ -258,7 +258,7 @@ object CometJsonToStructs extends CometCodegenDispatch[JsonToStructs] with Nativ } } -object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] { +object CometStructsToCsv extends CometCodegenDispatch[StructsToCsv] with NativeOptInAvailable { private val incompatibleDataTypes = Seq(DateType, TimestampType, TimestampNTZType, BinaryType) @@ -266,42 +266,42 @@ object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] { "Date, Timestamp, TimestampNTZ, and Binary data types may produce different results" + " (https://github.com/apache/datafusion-comet/issues/3232)") - override def getUnsupportedReasons(): Seq[String] = Seq( - "Complex types (arrays, maps, structs) in the schema are not supported") - - override def getSupportLevel(expr: StructsToCsv): SupportLevel = { + // The native ToCsv path only supports non-complex, compatible field types. Everything else + // (and the default, unless opted in) runs through the codegen dispatcher, which is bit-exact. + private def nativeSupported(expr: StructsToCsv): Boolean = { val dataTypes = expr.inputSchema.fields.map(_.dataType) - val containsComplexType = dataTypes.exists(DataTypeSupport.isComplexType) - if (containsComplexType) { - return Unsupported( - Some( - s"The schema ${expr.inputSchema} is not supported because it includes a complex type")) - } - val containsIncompatibleDataTypes = dataTypes.exists(incompatibleDataTypes.contains) - if (containsIncompatibleDataTypes) { - return Incompatible( - Some( - s"The schema ${expr.inputSchema} is not supported because " + - s"it includes a incompatible data types: $incompatibleDataTypes")) - } - // https://github.com/apache/datafusion-comet/issues/3232 - Incompatible() + !dataTypes.exists(DataTypeSupport.isComplexType) && + !dataTypes.exists(incompatibleDataTypes.contains) } + override def getSupportLevel(expr: StructsToCsv): SupportLevel = + if (!CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeSupported(expr)) { + Compatible(nativeOptIn = + Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr))))) + } else { + Compatible() + } + override def convert( expr: StructsToCsv, inputs: Seq[Attribute], binding: Boolean): Option[ExprOuterClass.Expr] = { - for { - childProto <- exprToProtoInternal(expr.child, inputs, binding) - } yield { - val optionsProto = options2Proto(expr.options, expr.timeZoneId) - val toCsv = ExprOuterClass.ToCsv - .newBuilder() - .setChild(childProto) - .setOptions(optionsProto) - .build() - ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build() + if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeSupported(expr)) { + for { + childProto <- exprToProtoInternal(expr.child, inputs, binding) + } yield { + val optionsProto = options2Proto(expr.options, expr.timeZoneId) + val toCsv = ExprOuterClass.ToCsv + .newBuilder() + .setChild(childProto) + .setOptions(optionsProto) + .build() + ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build() + } + } else { + // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the + // Comet pipeline (bit-exact). The native path is opt-in via allowIncompatible. + super.convert(expr, inputs, binding) } } diff --git a/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql b/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql new file mode 100644 index 00000000000..95c6dd55395 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql @@ -0,0 +1,41 @@ +-- 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))) diff --git a/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql b/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql index d9dabde9f51..f21bb974c22 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql @@ -15,11 +15,11 @@ -- 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 @@ -27,17 +27,17 @@ CREATE TABLE test_translate(s string, from_str string, to_str string) USING parq statement INSERT INTO test_translate VALUES ('hello', 'el', 'ip'), ('hello', 'aeiou', '12345'), ('', 'a', 'b'), (NULL, 'a', 'b'), ('hello', '', ''), ('abc', 'abc', 'x') -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') From e00ad5694ad0feb0fd9a603ac00133b7b813904f Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 31 Jul 2026 13:46:38 -0600 Subject: [PATCH 2/3] review: express default dispatcher routing declaratively Replace the hand-rolled getSupportLevel/convert branching with the `CodegenDispatchFallback` mixin, which is what the framework already provides for exactly this shape. `exprToProtoInternal` routes an `Unsupported` result to the dispatcher plus a logDebug, a non-opted-in `Incompatible` to the dispatcher plus the [COMET-INFO] opt-in hint, and an opted-in `Incompatible` to the native proto with a logWarning. For `CometStringTranslate` that makes the whole diff one mixin on the original declaration: the pre-existing `Incompatible(Some(incompatReason))` was already the right classification. This also restores the `CometScalarFunction` base, so the function name "translate" stops being duplicated and the opted-in logWarning comes back. For `CometStructsToCsv` the original three-way `getSupportLevel` was likewise already correct -- `Unsupported` for complex field types, `Incompatible` for the #3232 types, `Incompatible()` otherwise -- so the +30/-30 rewrite collapses to the mixin. That restores `getUnsupportedReasons`, drops the duplicated `nativeSupported` predicate that had to stay in sync between two call sites, and picks up the logDebug on the opted-in-but-unsupported path for free. Also drops `CometCodegenDispatch` as the base, which is for expressions with no native path at all. Tests: - string_translate.sql gains the two rows that actually distinguish the paths: "e" + U+0301 (one grapheme, two code points) and U+0000 as the `to` argument. Built with decode(X'65CC81') / decode(X'00') to keep the fixture ASCII. Verified these discriminate: adding them to string_translate_enabled.sql, which opts into the native path, fails. - to_csv.sql gains options coverage (`sep`, and `timestampFormat`/`dateFormat` over date and timestamp fields), which the dispatcher path had nowhere. - CometCsvExpressionSuite "to_csv - default options" drops the allowIncompatible=true wrapper and covers c6 Double, c10 Timestamp and c11 TimestampNTZ, which it previously had to skip. It now tests what its name says. Two field types are deliberately left unasserted, both Spark behavior rather than Comet's, confirmed against Spark 3.5 with Comet disabled: - Binary: Spark's CSV converter renders it with Java's default toString, e.g. "[B@731af74", an identity hash that differs between any two evaluations, so it is not comparable by any engine including Spark against itself. - Complex types: a non-null array/map/struct renders as "org.apache.spark.sql.vectorized.ColumnarArray@1ada50f0" and any null complex value throws NullPointerException inside Spark's UnsafeWriter.write. Before this change these schemas fell back to Spark; now they reach the dispatcher, which runs the same Spark code, so the observable result is unchanged. --- .../org/apache/comet/serde/strings.scala | 32 ++------- .../org/apache/comet/serde/structs.scala | 65 ++++++++++--------- .../sql-tests/expressions/csv/to_csv.sql | 36 ++++++++++ .../expressions/string/string_translate.sql | 14 +++- .../comet/CometCsvExpressionSuite.scala | 33 +++++----- 5 files changed, 106 insertions(+), 74 deletions(-) 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 81d6606740b..887e0a784ee 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -109,38 +109,20 @@ object CometOctetLength extends CometScalarFunction[OctetLength]("octet_length") } } +// 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 CometExpressionSerde[StringTranslate] - with NativeOptInAvailable { + 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" override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason) - override def getSupportLevel(expr: StringTranslate): SupportLevel = - if (!CometConf.isExprAllowIncompat(getExprConfigName(expr))) { - Compatible(nativeOptIn = - Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr))))) - } else { - Compatible() - } - - override def convert( - expr: StringTranslate, - inputs: Seq[Attribute], - binding: Boolean): Option[Expr] = { - if (CometConf.isExprAllowIncompat(getExprConfigName(expr))) { - val childExprs = expr.children.map(exprToProtoInternal(_, inputs, binding)) - val optExpr = scalarFunctionExprToProto("translate", childExprs: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) - } else { - // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the - // Comet pipeline (bit-exact). The native path differs on graphemes vs code points, so it is - // opt-in via allowIncompatible. - CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding) - } - } + override def getSupportLevel(expr: StringTranslate): SupportLevel = Incompatible( + Some(incompatReason)) } object CometLevenshtein extends CometExpressionSerde[Levenshtein] { diff --git a/spark/src/main/scala/org/apache/comet/serde/structs.scala b/spark/src/main/scala/org/apache/comet/serde/structs.scala index 9346612b218..a1cfcadd57e 100644 --- a/spark/src/main/scala/org/apache/comet/serde/structs.scala +++ b/spark/src/main/scala/org/apache/comet/serde/structs.scala @@ -258,7 +258,12 @@ object CometJsonToStructs extends CometCodegenDispatch[JsonToStructs] with Nativ } } -object CometStructsToCsv extends CometCodegenDispatch[StructsToCsv] with NativeOptInAvailable { +// 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) @@ -266,42 +271,42 @@ object CometStructsToCsv extends CometCodegenDispatch[StructsToCsv] with NativeO "Date, Timestamp, TimestampNTZ, and Binary data types may produce different results" + " (https://github.com/apache/datafusion-comet/issues/3232)") - // The native ToCsv path only supports non-complex, compatible field types. Everything else - // (and the default, unless opted in) runs through the codegen dispatcher, which is bit-exact. - private def nativeSupported(expr: StructsToCsv): Boolean = { - val dataTypes = expr.inputSchema.fields.map(_.dataType) - !dataTypes.exists(DataTypeSupport.isComplexType) && - !dataTypes.exists(incompatibleDataTypes.contains) - } + override def getUnsupportedReasons(): Seq[String] = Seq( + "Complex types (arrays, maps, structs) in the schema are not supported") - override def getSupportLevel(expr: StructsToCsv): SupportLevel = - if (!CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeSupported(expr)) { - Compatible(nativeOptIn = - Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr))))) - } else { - Compatible() + override def getSupportLevel(expr: StructsToCsv): SupportLevel = { + val dataTypes = expr.inputSchema.fields.map(_.dataType) + val containsComplexType = dataTypes.exists(DataTypeSupport.isComplexType) + if (containsComplexType) { + return Unsupported( + Some( + s"The schema ${expr.inputSchema} is not supported because it includes a complex type")) + } + val containsIncompatibleDataTypes = dataTypes.exists(incompatibleDataTypes.contains) + if (containsIncompatibleDataTypes) { + return Incompatible( + Some( + s"The schema ${expr.inputSchema} is not supported because " + + s"it includes a incompatible data types: $incompatibleDataTypes")) } + // https://github.com/apache/datafusion-comet/issues/3232 + Incompatible() + } override def convert( expr: StructsToCsv, inputs: Seq[Attribute], binding: Boolean): Option[ExprOuterClass.Expr] = { - if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeSupported(expr)) { - for { - childProto <- exprToProtoInternal(expr.child, inputs, binding) - } yield { - val optionsProto = options2Proto(expr.options, expr.timeZoneId) - val toCsv = ExprOuterClass.ToCsv - .newBuilder() - .setChild(childProto) - .setOptions(optionsProto) - .build() - ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build() - } - } else { - // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the - // Comet pipeline (bit-exact). The native path is opt-in via allowIncompatible. - super.convert(expr, inputs, binding) + for { + childProto <- exprToProtoInternal(expr.child, inputs, binding) + } yield { + val optionsProto = options2Proto(expr.options, expr.timeZoneId) + val toCsv = ExprOuterClass.ToCsv + .newBuilder() + .setChild(childProto) + .setOptions(optionsProto) + .build() + ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build() } } diff --git a/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql b/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql index 95c6dd55395..ccb83f3ae87 100644 --- a/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql +++ b/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql @@ -39,3 +39,39 @@ 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. diff --git a/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql b/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql index f21bb974c22..2f08775d360 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql @@ -24,8 +24,20 @@ 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 SELECT translate(s, from_str, to_str) FROM test_translate diff --git a/spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala index f7e6b03d33d..6e50e7cb9ae 100644 --- a/spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala +++ 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) } } From 7a7942c52c8c2da3f9209fc3fec594bcac4bef47 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 4 Aug 2026 18:50:26 -0600 Subject: [PATCH 3/3] review: pin to_csv dispatcher behavior for ANSI casts and complex field types Addresses review feedback on #5032. - to_csv_ansi.sql: a raising Cast inside the named_struct, under ANSI, in both the single-ordinal and the two-ordinal (#5219) shape, plus a try_cast counterpart and the sentinel query the ExpectError convention requires. - to_csv.sql: complex field types (array / map / nested struct) reach the runtime dispatch because isSupportedDataType recurses, so assert the dispatcher hands the nested InternalRow / ArrayData / MapData through without raising. On Spark 3.4 / 3.5 that is the CometSpecializedGettersDispatch generic get; verified the queries fail when its complex-type branches are removed. - to_csv_nested.sql (Spark 4.0+): Spark 4.0 gave UnivocityGenerator real array / map / struct converters, so from 4.0 on the rendered value is deterministic and can be asserted in full rather than only for nullness. - string_translate.sql: the two divergence rows used decode(X'..') inside an inline table, which Spark 4.1 rejects with INVALID_INLINE_TABLE. Use \u escapes instead; verified the rows still fail under allowIncompatible=true. --- .../sql-tests/expressions/csv/to_csv.sql | 56 +++++++++++++++---- .../sql-tests/expressions/csv/to_csv_ansi.sql | 56 +++++++++++++++++++ .../expressions/csv/to_csv_nested.sql | 56 +++++++++++++++++++ .../expressions/string/string_translate.sql | 19 ++++--- 4 files changed, 167 insertions(+), 20 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/csv/to_csv_ansi.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/csv/to_csv_nested.sql diff --git a/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql b/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql index ccb83f3ae87..6bc210b83fb 100644 --- a/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql +++ b/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql @@ -48,9 +48,11 @@ SELECT to_csv(named_struct('a', a, 'b', b, 'c', c), map('sep', ';')) FROM test_t -- 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. +-- 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 @@ -66,12 +68,42 @@ 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. +-- 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, m: map, n: struct>) 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), 'm', CAST(NULL AS map), 'n', CAST(NULL AS struct))), + (CAST(NULL AS struct, m: map, n: struct>)) + +-- 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 diff --git a/spark/src/test/resources/sql-tests/expressions/csv/to_csv_ansi.sql b/spark/src/test/resources/sql-tests/expressions/csv/to_csv_ansi.sql new file mode 100644 index 00000000000..1ba07ad49d0 --- /dev/null +++ b/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 diff --git a/spark/src/test/resources/sql-tests/expressions/csv/to_csv_nested.sql b/spark/src/test/resources/sql-tests/expressions/csv/to_csv_nested.sql new file mode 100644 index 00000000000..fe768ab16bc --- /dev/null +++ b/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, m: map, n: struct>) 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), 'm', CAST(NULL AS map), 'n', CAST(NULL AS struct))), + (CAST(NULL AS struct, m: map, n: struct>)) + +-- 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 diff --git a/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql b/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql index 2f08775d360..ca755cd70ac 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/string_translate.sql @@ -26,18 +26,21 @@ CREATE TABLE test_translate(s string, from_str string, to_str string) USING parq -- 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. +-- 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'), - (concat('caf', decode(X'65CC81', 'UTF-8')), 'e', 'E'), - ('hello', 'l', decode(X'00', 'UTF-8')) + ('cafe\u0301', 'e', 'E'), + ('hello', 'l', '\u0000') query SELECT translate(s, from_str, to_str) FROM test_translate