diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index b3c8e1334c6..72322253e3b 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 | Routes through the JVM codegen dispatcher by default; the native path is opt-in via allowIncompatible | --- 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..da4387fdbf0 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,14 @@ object CometJsonToStructs extends CometCodegenDispatch[JsonToStructs] with Nativ } } -object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] { +/** + * `to_csv` has a native implementation that is not Spark-compatible by default + * (https://github.com/apache/datafusion-comet/issues/3232). `CodegenDispatchFallback` keeps it in + * the Comet pipeline by running Spark's own `StructsToCsv.doGenCode` in the JVM codegen + * dispatcher. Set `spark.comet.expression.StructsToCsv.allowIncompatible=true` to opt into the + * native path. Nested array/map/struct fields have no native path and always use the dispatcher. + */ +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/csv.sql b/spark/src/test/resources/sql-tests/expressions/csv/csv.sql index c7738948a8d..1f8ead816c7 100644 --- a/spark/src/test/resources/sql-tests/expressions/csv/csv.sql +++ b/spark/src/test/resources/sql-tests/expressions/csv/csv.sql @@ -15,9 +15,10 @@ -- specific language governing permissions and limitations -- under the License. --- CSV structured-text functions (from_csv, schema_of_csv). These have no native (rust) --- implementation; they extend Spark's CodegenFallback and stay native via the codegen --- dispatcher. +-- CSV structured-text functions. `from_csv` and `schema_of_csv` have no native (rust) +-- implementation and stay native via the codegen dispatcher. `to_csv` has a native path that is +-- opt-in via allowIncompatible; by default Spark's own generated code runs through the JVM +-- codegen dispatcher so the projection stays in Comet. statement CREATE TABLE test_csv(s STRING) USING parquet @@ -32,3 +33,15 @@ SELECT from_csv(s, 'a INT, b STRING') FROM test_csv -- literal argument query SELECT schema_of_csv('1,abc') + +statement +CREATE TABLE test_to_csv(id INT, name STRING) USING parquet + +statement +INSERT INTO test_to_csv VALUES (1, 'abc'), (2, NULL), (NULL, 'def'), (NULL, NULL) + +query +SELECT to_csv(named_struct('id', id, 'name', name)) FROM test_to_csv + +query +SELECT to_csv(named_struct('id', id, 'name', name), map('sep', ';')) FROM test_to_csv diff --git a/spark/src/test/resources/sql-tests/expressions/csv/csv_fallback.sql b/spark/src/test/resources/sql-tests/expressions/csv/csv_fallback.sql index 427c54c8769..79c0f5078d6 100644 --- a/spark/src/test/resources/sql-tests/expressions/csv/csv_fallback.sql +++ b/spark/src/test/resources/sql-tests/expressions/csv/csv_fallback.sql @@ -15,7 +15,8 @@ -- specific language governing permissions and limitations -- under the License. --- With the codegen dispatcher disabled, from_csv has no native path and falls back to Spark. +-- With the codegen dispatcher disabled, from_csv and to_csv have no in-pipeline path and fall +-- back to Spark. -- Config: spark.comet.exec.scalaUDF.codegen.enabled=false @@ -27,3 +28,12 @@ INSERT INTO test_csv_fallback VALUES ('1,abc'), ('2,def'), (''), (NULL) query expect_fallback(spark.comet.exec.scalaUDF.codegen.enabled) SELECT from_csv(s, 'a INT, b STRING') FROM test_csv_fallback + +statement +CREATE TABLE test_to_csv_fallback(id INT, name STRING) USING parquet + +statement +INSERT INTO test_to_csv_fallback VALUES (1, 'abc'), (2, NULL) + +query expect_fallback(spark.comet.exec.scalaUDF.codegen.enabled) +SELECT to_csv(named_struct('id', id, 'name', name)) FROM test_to_csv_fallback 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..0be81d1efa1 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/csv/to_csv_nested.sql @@ -0,0 +1,31 @@ +-- 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. + +-- Nested array/map/struct fields have no native to_csv path and stay on the JVM codegen +-- dispatcher. Spark 3.4/3.5 stringify nested arrays via Object.toString(); Spark 4.0 +-- (SPARK-47497) prints pretty strings that match the dispatcher. + +-- MinSparkVersion: 4.0 + +statement +CREATE TABLE test_to_csv_nested(id INT, items ARRAY) USING parquet + +statement +INSERT INTO test_to_csv_nested VALUES (1, array(1, 2)), (2, array()), (3, NULL) + +query +SELECT to_csv(named_struct('id', id, 'items', items)) FROM test_to_csv_nested diff --git a/spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala index f7e6b03d33d..07bc35a3690 100644 --- a/spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala @@ -29,9 +29,14 @@ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions._ import org.apache.spark.sql.types.StringType +import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions} +import org.apache.comet.udf.codegen.CometScalaUDFCodegen -class CometCsvExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { +class CometCsvExpressionSuite + extends CometTestBase + with AdaptiveSparkPlanHelper + with CometCodegenAssertions { test("to_csv - default options") { withTempDir { dir => @@ -193,4 +198,71 @@ class CometCsvExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper } } } + + test("to_csv routes through the codegen dispatcher by default (issue #5578)") { + // Without allowIncompatible, getSupportLevel is Incompatible and CodegenDispatchFallback + // runs Spark's own StructsToCsv.doGenCode inside the Comet pipeline. assertCodegenRan pins + // that the dispatcher is what kept the projection native. + val data: Seq[(Option[Int], Option[String])] = + Seq((Some(1), Some("alice")), (Some(2), None), (None, Some("bob")), (None, None)) + withParquetTable(data, "tbl") { + assertCodegenRan { + checkSparkAnswerAndOperator("SELECT to_csv(named_struct('id', _1, 'name', _2)) FROM tbl") + checkSparkAnswerAndOperator( + "SELECT to_csv(named_struct('id', _1, 'name', _2), map('sep', ';')) FROM tbl") + } + } + } + + test("to_csv with nested array field routes through the codegen dispatcher (issue #5578)") { + // Spark 3.4/3.5 stringify nested arrays via Object.toString(), so Spark's ColumnarArray + // and the dispatcher's InputArray cannot match. SPARK-47497 (Spark 4.0) prints pretty + // strings. Nested array/map/struct fields are Unsupported on the native path; the + // dispatcher still compiles them: arrays are in CometBatchKernelCodegen.isSupportedDataType. + assume( + isSpark40Plus, + "to_csv pretty-prints nested arrays starting in Spark 4.0 (SPARK-47497)") + withTable("t") { + sql("CREATE TABLE t(id INT, items ARRAY) USING parquet") + sql("INSERT INTO t VALUES (1, array(1, 2, 3)), (2, array()), (3, NULL)") + assertCodegenRan { + checkSparkAnswerAndOperator( + "SELECT to_csv(named_struct('id', id, 'items', items)) FROM t") + } + } + } + + test("to_csv NULL struct routes through the codegen dispatcher (issue #5578)") { + withTable("t") { + sql("CREATE TABLE t(s STRUCT) USING parquet") + sql( + "INSERT INTO t VALUES (named_struct('id', 1, 'name', 'alice')), " + + "(named_struct('id', CAST(NULL AS INT), 'name', 'bob')), (NULL)") + assertCodegenRan { + checkSparkAnswerAndOperator("SELECT to_csv(s) FROM t") + } + } + } + + test("to_csv falls back to Spark when the codegen dispatcher is disabled (issue #5578)") { + withParquetTable(Seq((1, "alice"), (2, "bob")), "tbl") { + withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false") { + checkSparkAnswerAndFallbackReason( + "SELECT to_csv(named_struct('id', _1, 'name', _2)) FROM tbl", + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key) + } + } + } + + test("to_csv keeps the native path when allowIncompatible is enabled (issue #5578)") { + withParquetTable(Seq((1, "alice"), (2, "bob")), "tbl") { + withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[StructsToCsv]) -> "true") { + CometScalaUDFCodegen.resetStats() + checkSparkAnswerAndOperator("SELECT to_csv(named_struct('id', _1, 'name', _2)) FROM tbl") + assert( + CometScalaUDFCodegen.stats().totalLookups == 0, + "expected native to_csv, not codegen dispatch") + } + } + } }