Skip to content
Merged
22 changes: 22 additions & 0 deletions dev/diffs/4.0.2.diff
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
22 changes: 22 additions & 0 deletions dev/diffs/4.1.2.diff
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
38 changes: 28 additions & 10 deletions spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
Original file line number Diff line number Diff line change
Expand Up @@ -25,22 +25,28 @@ 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 {
Comment thread
andygrove marked this conversation as resolved.

// Shared with CometCastSuite so the asserted reason cannot drift from production.
private[comet] val negativeScaleDecimalToStringReason: String =
"Negative-scale decimal requires spark.sql.legacy.allowNegativeScaleOfDecimal=true"

// 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"

Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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))
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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")

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)

-- ----------------------------------------------------------------------------
Expand Down Expand Up @@ -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<x: int, y: string>)) 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<int>)) 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
Expand All @@ -341,5 +341,5 @@ query
SELECT cast(cast(null as map<string, int>) as string)

-- Map of map.
query expect_fallback(Cast from MapType)
query
SELECT cast(map('outer', map('inner', 1)) as string)
Original file line number Diff line number Diff line change
Expand Up @@ -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)
26 changes: 22 additions & 4 deletions spark/src/test/scala/org/apache/comet/CometCastSuite.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand All @@ -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")))
Comment thread
andygrove marked this conversation as resolved.
}
}
}
Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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))"
Expand All @@ -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
Expand Down
Loading