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..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,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" 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..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 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) 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..ccb83f3ae8 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/csv/to_csv.sql @@ -0,0 +1,77 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- to_csv runs through the codegen dispatcher by default so results match Spark exactly, including +-- quoting and escaping. The native path is opt-in via +-- spark.comet.expression.StructsToCsv.allowIncompatible. + +statement +CREATE TABLE test_to_csv(a int, b string, c double) USING parquet + +statement +INSERT INTO test_to_csv VALUES + (1, 'x', 2.5), + (-3, 'hello,world', 0.0), + (0, 'has "quote"', -1.5), + (NULL, NULL, NULL), + (7, '', 3.0) + +-- column struct: values with delimiters and quotes exercise Spark's CSV quoting rules +query +SELECT to_csv(named_struct('a', a, 'b', b, 'c', c)) FROM test_to_csv + +-- literal struct (constant folding is disabled by the test suite) +query +SELECT + to_csv(named_struct('a', 1, 'b', 'x', 'c', 2.5)), + to_csv(named_struct('s', 'a,b', 'n', CAST(NULL AS INT))) + +-- Options are the main axis of behavior difference for this expression, and the dispatcher path +-- had no options coverage anywhere (CometCsvExpressionSuite only varies them under +-- allowIncompatible=true). `sep` also changes which values need quoting. +query +SELECT to_csv(named_struct('a', a, 'b', b, 'c', c), map('sep', ';')) FROM test_to_csv + +-- timestampFormat over date and timestamp fields: these are the #3232 types that were +-- Incompatible before this change and now route through the dispatcher. +-- BinaryType is the other #3232 type but is deliberately absent: Spark's CSV converter renders it +-- with Java's default Object.toString(), e.g. "[B@10bc15e4", an identity hash that differs between +-- any two evaluations, so the value is not assertable by any engine including Spark itself. +statement +CREATE TABLE test_to_csv_temporal(d date, t timestamp) USING parquet + +statement +INSERT INTO test_to_csv_temporal VALUES + (DATE '2024-01-31', TIMESTAMP '2024-01-31 12:34:56.789'), + (DATE '1970-01-01', TIMESTAMP '1970-01-01 00:00:00'), + (NULL, NULL) + +query +SELECT to_csv(named_struct('d', d, 't', t)) FROM test_to_csv_temporal + +query +SELECT to_csv(named_struct('d', d, 't', t), map('timestampFormat', 'yyyy/MM/dd HH:mm', 'dateFormat', 'dd-MM-yyyy')) FROM test_to_csv_temporal + +-- Complex field types (arrays, maps, nested structs) are deliberately not asserted here. +-- StructsToCsv.checkInputDataTypes accepts them, so they reach the converter, but Spark's own +-- output for them is not a value that can be compared: a non-null array/map/struct renders as a +-- Java identity string such as "org.apache.spark.sql.vectorized.ColumnarArray@1ada50f0", and any +-- null complex value throws NullPointerException inside Spark's UnsafeWriter.write. Both were +-- confirmed against Spark 3.5 with Comet disabled entirely, so they are Spark behavior, not +-- Comet's. Before this change these schemas were Unsupported and fell the projection back to +-- Spark; now they reach the dispatcher, which runs the same Spark code, so the observable result +-- is unchanged either way. 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..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 @@ -15,29 +15,41 @@ -- specific language governing permissions and limitations -- under the License. --- translate is gated as Incompatible by default. DataFusion's translate iterates over Unicode --- graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion --- sentinel, so the native path silently diverges from Spark for combining-mark inputs and for --- to=NUL. These default-config tests assert that the expression falls back cleanly to Spark. --- See string_translate_enabled.sql for the opt-in native path. +-- translate runs through the codegen dispatcher by default so results match Spark exactly. The +-- native path diverges from Spark (DataFusion iterates over Unicode graphemes where Spark uses code +-- points, and substitutes U+0000 instead of treating it as a deletion sentinel), so it is opt-in +-- via spark.comet.expression.StringTranslate.allowIncompatible. See string_translate_enabled.sql +-- for the opt-in native path. statement CREATE TABLE test_translate(s string, from_str string, to_str string) USING parquet +-- The last two rows are the regression test for this file's routing change: they are the inputs +-- where the native path is known to disagree with Spark, so they are only assertable now that the +-- default is the bit-exact dispatcher. Built with decode() rather than literal characters to keep +-- the fixture ASCII. +-- decode(X'65CC81') is "e" + U+0301 COMBINING ACUTE ACCENT: one grapheme, two code points. +-- DataFusion iterates graphemes, Spark iterates code points, so translating "e" differs. +-- decode(X'00') as the `to` argument is U+0000, which Spark treats as a deletion sentinel and +-- the native path substitutes literally. statement -INSERT INTO test_translate VALUES ('hello', 'el', 'ip'), ('hello', 'aeiou', '12345'), ('', 'a', 'b'), (NULL, 'a', 'b'), ('hello', '', ''), ('abc', 'abc', 'x') +INSERT INTO test_translate VALUES + ('hello', 'el', 'ip'), ('hello', 'aeiou', '12345'), ('', 'a', 'b'), (NULL, 'a', 'b'), + ('hello', '', ''), ('abc', 'abc', 'x'), + (concat('caf', decode(X'65CC81', 'UTF-8')), 'e', 'E'), + ('hello', 'l', decode(X'00', 'UTF-8')) -query expect_fallback(is not fully compatible with Spark) +query SELECT translate(s, from_str, to_str) FROM test_translate -- column + literal + literal -query expect_fallback(is not fully compatible with Spark) +query SELECT translate(s, 'el', 'ip') FROM test_translate -- literal + column + column -query expect_fallback(is not fully compatible with Spark) +query SELECT translate('hello', from_str, to_str) FROM test_translate -- literal + literal + literal -query expect_fallback(is not fully compatible with Spark) +query SELECT translate('hello', 'el', 'ip'), translate('hello', 'aeiou', '12345'), translate('', 'a', 'b'), translate(NULL, 'a', 'b') 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) } }