From 3abb614c037be09f3ec7f4e2e11b1f7c3fce8022 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 24 Jul 2026 15:55:37 -0600 Subject: [PATCH 1/2] 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 efbd33ba1c..8b0205be1b 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/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 | 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 1482d3e79e..529d2d4dc1 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 409ef38b4f..8d1b0f845a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/structs.scala +++ b/spark/src/main/scala/org/apache/comet/serde/structs.scala @@ -259,7 +259,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) @@ -267,42 +267,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 0000000000..95c6dd5539 --- /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 d9dabde9f5..f21bb974c2 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 26994c9197a4c03a4c33a9f045bcac6030ef941e Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 31 Jul 2026 13:46:38 -0600 Subject: [PATCH 2/2] 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 529d2d4dc1..62babdd6f8 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 8d1b0f845a..0ffbd6de35 100644 --- a/spark/src/main/scala/org/apache/comet/serde/structs.scala +++ b/spark/src/main/scala/org/apache/comet/serde/structs.scala @@ -259,7 +259,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) @@ -267,42 +272,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 95c6dd5539..ccb83f3ae8 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 f21bb974c2..2f08775d36 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 f7e6b03d33..6e50e7cb9a 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) } }