diff --git a/dev/diffs/4.0.2.diff b/dev/diffs/4.0.2.diff index a2f6eac74a..0485a2d5f7 100644 --- a/dev/diffs/4.0.2.diff +++ b/dev/diffs/4.0.2.diff @@ -1380,6 +1380,28 @@ index 2e33f6505ab..be967565303 100644 } withTable("t1", "t2") { +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala +index a40e34d94d0..abc1f035d15 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala +@@ -21,6 +21,7 @@ import org.apache.spark.sql.QueryTest.sameRows + import org.apache.spark.sql.catalyst.InternalRow + import org.apache.spark.sql.catalyst.expressions.{Cast, Literal} + import org.apache.spark.sql.catalyst.expressions.variant.{ToVariantObject, VariantExpressionEvalUtils} ++import org.apache.spark.sql.comet.CometNativeColumnarToRowExec + import org.apache.spark.sql.execution.WholeStageCodegenExec + import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector + import org.apache.spark.sql.functions._ +@@ -358,7 +359,8 @@ class VariantEndToEndSuite extends QueryTest with SharedSparkSession { + s"cast(to_variant_object(s) as ${schema(2).dataType.sql})") + checkAnswer(df, input) + val plan = df.queryExecution.executedPlan +- assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY")) ++ assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY") ++ || plan.isInstanceOf[CometNativeColumnarToRowExec]) + } + } + } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala index 11e9547dfc5..637411056ae 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala diff --git a/dev/diffs/4.1.2.diff b/dev/diffs/4.1.2.diff index 30f6c2f23f..8bd4bfe732 100644 --- a/dev/diffs/4.1.2.diff +++ b/dev/diffs/4.1.2.diff @@ -1479,6 +1479,28 @@ index 3ba48da0e32..0313aaa9ec6 100644 } withTable("t1", "t2") { +diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala +index 8a0e2c29653..d276a51cbc6 100644 +--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala ++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala +@@ -21,6 +21,7 @@ import org.apache.spark.sql.QueryTest.sameRows + import org.apache.spark.sql.catalyst.InternalRow + import org.apache.spark.sql.catalyst.expressions.{Cast, Literal} + import org.apache.spark.sql.catalyst.expressions.variant.{ToVariantObject, VariantExpressionEvalUtils} ++import org.apache.spark.sql.comet.CometNativeColumnarToRowExec + import org.apache.spark.sql.execution.WholeStageCodegenExec + import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector + import org.apache.spark.sql.functions._ +@@ -358,7 +359,8 @@ class VariantEndToEndSuite extends QueryTest with SharedSparkSession { + s"cast(to_variant_object(s) as ${schema(2).dataType.sql})") + checkAnswer(df, input) + val plan = df.queryExecution.executedPlan +- assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY")) ++ assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY") ++ || plan.isInstanceOf[CometNativeColumnarToRowExec]) + } + } + } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala index fee375db10a..8c2c24e2c5f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala index 619b69912f..eba1e4b152 100644 --- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala +++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala @@ -25,12 +25,17 @@ import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType, import org.apache.comet.CometConf import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, withFallbackReason} -import org.apache.comet.serde.{CometExpressionSerde, Compatible, ExprOuterClass, Incompatible, SupportLevel, Unsupported} +import org.apache.comet.DataTypeSupport.isComplexType +import org.apache.comet.serde.{CodegenDispatchFallback, CometExpressionSerde, Compatible, ExprOuterClass, Incompatible, SupportLevel, Unsupported} import org.apache.comet.serde.ExprOuterClass.Expr import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProtoInternal, serializeDataType} -import org.apache.comet.shims.CometExprShim +import org.apache.comet.shims.{CometExprShim, CometTypeShim} -object CometCast extends CometExpressionSerde[Cast] with CometExprShim { +object CometCast + extends CometExpressionSerde[Cast] + with CometExprShim + with CometTypeShim + with CodegenDispatchFallback { // Shared with CometCastSuite so the asserted reason cannot drift from production. private[comet] val negativeScaleDecimalToStringReason: String = @@ -38,9 +43,10 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { // When `spark.sql.legacy.castComplexTypesToString.enabled` is true, Spark wraps maps and // structs with `[]` (instead of `{}`) when casting to string, and omits NULL elements of - // structs/maps/arrays (instead of rendering them as the literal "null"). Comet only - // implements the default formatting, so fall back to Spark for any array/map/struct to-string - // cast when the flag is enabled. The flag is internal in Spark 4.0 and defaults to false. + // structs/maps/arrays (instead of rendering them as the literal "null"). Comet's native + // cast only implements the default formatting, so with the flag enabled we route + // array/map/struct to-string casts through the JVM codegen dispatcher via + // `CodegenDispatchFallback`. The flag is internal in Spark 4.0 and defaults to false. private[comet] val legacyCastComplexTypesToStringReason: String = "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported" @@ -75,7 +81,12 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { if (cast.child.isInstanceOf[Literal]) { // A cast whose child is a literal is folded by Spark at planning time via `cast.eval()` // (see `convert`), so the cast never executes natively and the result matches Spark by - // definition. `CometLiteral` then validates the resulting literal's data type. + // definition. `CometLiteral` then validates the resulting literal's data type, except + // for `VariantType` which must be rejected here: the fold produces a `Literal[VariantType]` + // that no downstream Comet serde can serialize. + if (isVariantType(cast.child.dataType) || isVariantType(cast.dataType)) { + return unsupported(cast.child.dataType, cast.dataType) + } Compatible() } else { isSupported(cast.child.dataType, cast.dataType, cast.timeZoneId, evalMode(cast)) @@ -159,13 +170,20 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { timeZoneId: Option[String], evalMode: CometEvalMode.Value): SupportLevel = { + // Spark 4's `VariantType` (SPARK-45827) has no native counterpart in Comet, and the codegen + // dispatcher also cannot serialize `VariantType` in the data args or return type. The + // version-shimmed `isVariantType` returns false on Spark 3.x. Reporting `Unsupported` lets the + // `CodegenDispatchFallback` mixin try the dispatcher and then fall back to Spark cleanly. + if (isVariantType(fromType) || isVariantType(toType)) { + return unsupported(fromType, toType) + } + if (fromType == toType) { return Compatible() } - if (toType == DataTypes.StringType && legacyCastComplexTypesToString && (fromType - .isInstanceOf[ArrayType] || fromType.isInstanceOf[StructType] || - fromType.isInstanceOf[MapType])) { + if (toType == DataTypes.StringType && isComplexType(fromType) && + legacyCastComplexTypesToString) { return Unsupported(Some(legacyCastComplexTypesToStringReason)) } diff --git a/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala b/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala index 97320be9e7..b71476c3dd 100644 --- a/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala +++ b/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala @@ -36,6 +36,9 @@ trait CometTypeShim { @nowarn // Spark 4 feature; Variant shredding doesn't exist in Spark 3.x. def isVariantStruct(s: StructType): Boolean = false + @nowarn // Spark 4 feature; VariantType doesn't exist in Spark 3.x. + def isVariantType(dt: DataType): Boolean = false + @nowarn // Spark 4.1 feature; TimeType doesn't exist in Spark 3.x. def isTimeType(dt: DataType): Boolean = false } diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala index 1d4a9f601e..f48955a7da 100644 --- a/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala +++ b/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala @@ -20,7 +20,7 @@ package org.apache.comet.shims import org.apache.spark.sql.execution.datasources.VariantMetadata -import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType, StructType} +import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType, StructType, VariantType} trait CometTypeShim { // A `StringType` carries collation metadata in Spark 4.0. Only non-default (non-UTF8_BINARY) @@ -54,6 +54,12 @@ trait CometTypeShim { // and force scan fallback. def isVariantStruct(s: StructType): Boolean = VariantMetadata.isVariantStruct(s) + // Comet has no native execution path for Spark 4's `VariantType` (introduced in + // SPARK-45827). Serdes call this to route casts/expressions touching the type back to Spark + // rather than serializing an unsupported datatype into the native plan. Stubbed to `false` in + // Spark 3.x where `VariantType` does not exist. + def isVariantType(dt: DataType): Boolean = dt.isInstanceOf[VariantType] + def isTimeType(dt: DataType): Boolean = dt.getClass.getSimpleName.startsWith("TimeType") diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql index 8b1d989ae7..c9c1dc8ff6 100644 --- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql +++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql @@ -149,8 +149,7 @@ SELECT cast(named_struct('a', named_struct('b', named_struct('c', 1, 'd', 'leaf' query SELECT cast(named_struct('s1', '', 's2', ' ', 's3', cast(null as string)) as string) --- Map-valued field: not supported, falls back to Spark. -query expect_fallback(to StringType is not supported) +query SELECT cast(named_struct('m', map('k', 1)) as string) -- ---------------------------------------------------------------------------- @@ -270,69 +269,70 @@ SELECT cast(array(cast(1.5 as double), cast('NaN' as double), cast('-Infinity' a query SELECT cast(array(array(array(1, 2), array(3)), array(array(cast(null as int)))) as string) --- Array of map: not supported, falls back to Spark. -query expect_fallback(to StringType is not supported) +-- Array of map: map-to-string is routed through the codegen dispatcher via the outer array. +query SELECT cast(array(map('k', 1)) as string) -- ---------------------------------------------------------------------------- -- Map → string -- ---------------------------------------------------------------------------- --- Comet does not implement map-to-string casts, so every map → string falls back to Spark. +-- Comet has no native map-to-string cast; `CometCast` mixes in `CodegenDispatchFallback`, so +-- these stay native via the codegen dispatcher and match Spark exactly. -- Note: maps materialized through parquet have nondeterministic entry order, so map column -- tests use literal maps directly rather than reading from a parquet table. -- Map with string keys, int values. -query expect_fallback(Cast from MapType) +query SELECT cast(map('a', 1, 'b', 2, 'c', 3) as string) -- Map with NULL values rendered as "null". -query expect_fallback(Cast from MapType) +query SELECT cast(map('a', 1, 'b', cast(null as int), 'c', 3) as string) -- Map with int keys, string values. -query expect_fallback(Cast from MapType) +query SELECT cast(map(1, 'one', 2, 'two', 3, 'three') as string) -- Map with boolean values. -query expect_fallback(Cast from MapType) +query SELECT cast(map('t', true, 'f', false, 'n', cast(null as boolean)) as string) -- Map with bigint values at min/max. -query expect_fallback(Cast from MapType) +query SELECT cast(map('max', 9223372036854775807, 'min', -9223372036854775808, 'zero', cast(0 as bigint)) as string) -- Map with decimal values. -query expect_fallback(Cast from MapType) +query SELECT cast(map('pos', cast('1.234567890123456789' as decimal(38, 18)), 'neg', cast('-1.234567890123456789' as decimal(38, 18)), 'null', cast(null as decimal(38, 18))) as string) -- Map with date and timestamp values. -query expect_fallback(Cast from MapType) +query SELECT cast(map('a', date '2024-01-15', 'b', date '1970-01-01', 'c', cast(null as date)) as string) -query expect_fallback(Cast from MapType) +query SELECT cast(map('a', timestamp '2024-01-15 10:30:45', 'b', cast(null as timestamp)) as string) -- Map with binary values. -query expect_fallback(Cast from MapType) +query SELECT cast(map('a', X'616263', 'b', X'', 'c', cast(null as binary)) as string) -- Map with float / double values: NaN / ±0 / ±Infinity / NULL. -query expect_fallback(Cast from MapType) +query SELECT cast(map('nan', cast('NaN' as float), 'neg0', cast(-0.0 as float), 'null', cast(null as float)) as string) -query expect_fallback(Cast from MapType) +query SELECT cast(map('nan', cast('NaN' as double), 'inf', cast('Infinity' as double), 'ninf', cast('-Infinity' as double), 'null', cast(null as double)) as string) -- Map with struct values: each value rendered as `{f1, f2, ...}`. -query expect_fallback(Cast from MapType) +query SELECT cast(map('a', named_struct('x', 1, 'y', 'first'), 'b', cast(null as struct)) as string) -- Map with array values. -query expect_fallback(Cast from MapType) +query SELECT cast(map('a', array(1, 2, 3), 'b', array(cast(null as int)), 'c', cast(null as array)) as string) -- Empty map. -query expect_fallback(Cast from MapType) +query SELECT cast(map() as string) -- NULL map: Spark constant-folds this to a literal NULL, so the cast never reaches Comet @@ -341,5 +341,5 @@ query SELECT cast(cast(null as map) as string) -- Map of map. -query expect_fallback(Cast from MapType) +query SELECT cast(map('outer', map('inner', 1)) as string) diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql index 2c0bc19b3b..fcdb94b352 100644 --- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql +++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string_legacy.sql @@ -15,26 +15,25 @@ -- specific language governing permissions and limitations -- under the License. --- When `spark.sql.legacy.castComplexTypesToString.enabled` is true Spark wraps maps and --- structs with `[...]` (instead of `{...}`) and omits NULL elements of structs/maps/arrays --- (instead of rendering them as the literal "null"). Comet only implements the default --- formatting, so any array/map/struct → string cast must fall back to Spark. --- The flag is internal in Spark 4.0 and defaults to false. +-- When `spark.sql.legacy.castComplexTypesToString.enabled` is true, Spark wraps maps and +-- structs with `[...]` instead of `{...}` and omits NULL elements. `CometCast` mixes in +-- `CodegenDispatchFallback`, so these casts stay native via the codegen dispatcher and +-- match Spark exactly. The flag is internal in Spark 4.0 and defaults to false. -- Config: spark.sql.legacy.castComplexTypesToString.enabled=true --- Struct → string falls back. -query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported) +-- Struct → string. +query SELECT CAST(struct(1, 2, null) AS STRING) --- Array → string falls back (NULL elements rendered differently between modes). -query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported) +-- Array → string. +query SELECT CAST(array(1, 2, null) AS STRING) --- Map → string falls back (`[]` vs `{}` wrapping differs between modes). -query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported) +-- Map → string. +query SELECT CAST(map('a', 1, 'b', null) AS STRING) --- Nested complex types still fall back through the outer type. -query expect_fallback(spark.sql.legacy.castComplexTypesToString.enabled=true is not supported) +-- Nested complex types via the outer struct. +query SELECT CAST(struct(array(1, null), map('k', null)) AS STRING) diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala index abc04bfbb5..fa5178cdfd 100644 --- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala @@ -1797,7 +1797,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { Unsupported(Some(expectedMessage))) } - test("cast ArrayType(DateType) to unsupported ArrayType falls back") { + test("cast ArrayType(DateType) to unsupported ArrayType routes through codegen dispatch") { val fromType = ArrayType(DateType) val unsupportedElementTypes = Seq(BooleanType, ByteType, ShortType, LongType, FloatType, DoubleType, DecimalType(10, 2)) @@ -1814,9 +1814,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { assert( CometCast.isSupported(fromType, toType, None, CometEvalMode.LEGACY) == Unsupported(Some(expectedMessage))) - checkSparkAnswerAndFallbackReason( - data.select(col("a").cast(toType).as("converted")), - expectedMessage) + checkSparkAnswerAndOperator(data.select(col("a").cast(toType).as("converted"))) } } } @@ -1870,6 +1868,26 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + test("cast to and from VariantType matches Spark") { + // VariantType has no native path in Comet and the codegen dispatcher cannot serialize it + // either, so the operator falls back to Spark. Guarded on Spark 4.0+ (parse_json / VARIANT + // are unavailable in 3.x). + assume(CometSparkSessionExtensions.isSpark40Plus, "VariantType requires Spark 4.0+") + withTable("variant_cast") { + sql("CREATE TABLE variant_cast(id INT, s STRING) USING parquet") + sql(""" + |INSERT INTO variant_cast VALUES + | (1, '{"a": 1}'), + | (2, '{"b": [1, 2, 3]}'), + | (3, cast(null as string)) + """.stripMargin) + checkSparkAnswer( + "SELECT id, CAST(parse_json(s) AS STRING) AS v FROM variant_cast ORDER BY id") + checkSparkAnswer( + "SELECT id, CAST(CAST(s AS VARIANT) AS STRING) AS v FROM variant_cast ORDER BY id") + } + } + private def testArrayCastMatrix( elementTypes: Seq[DataType], wrapType: DataType => DataType, diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index b8a70760bf..5af03be9d1 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -2034,9 +2034,6 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { sql(s"insert into $table values(0, 1, 100.000001)") Seq( - ( - s"SELECT cast(make_interval(c0, c1, c0, c1, c0, c0, c2) as string) as C from $table", - Set("Cast from CalendarIntervalType to StringType is not supported")), ( "SELECT " + "date_part('YEAR', make_interval(c0, c1, c0, c1, c0, c0, c2))" @@ -2054,9 +2051,7 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { + s"(SELECT c1, sum(c0) as sum_c0, sum(c2) as sum_c2 from $table group by c1) as A, " + s"(SELECT c1, cast(make_interval(c0, c1, c0, c1, c0, c0, c2) as string) as casted from $table) as B " + "where A.c1 = B.c1 ", - Set( - "Cast from CalendarIntervalType to StringType is not supported", - "Comet shuffle is not enabled: spark.comet.shuffle.enabled is not enabled")), + Set("Comet shuffle is not enabled: spark.comet.shuffle.enabled is not enabled")), (s"select * from $table LIMIT 10 OFFSET 3", Set("Comet shuffle is not enabled"))) .foreach(test => { val qry = test._1