From b02b44f87848b3227f5b16ad2d02cf544450bcd7 Mon Sep 17 00:00:00 2001 From: comphead Date: Thu, 2 Jul 2026 13:25:08 -0700 Subject: [PATCH 01/11] chore: fallback to Spark if legacy sql configurations are set --- .../user-guide/latest/compatibility/index.md | 21 +++++++++++++ .../scala/org/apache/comet/CometConf.scala | 11 +++++++ .../comet/CometSparkSessionExtensions.scala | 18 +++++++++++ .../CometSparkSessionExtensionsSuite.scala | 30 +++++++++++++++++++ 4 files changed, 80 insertions(+) diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index f3b531a18f..ce1da965c5 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -70,3 +70,24 @@ This is distinct from expressions that have **no** codegen-dispatch path: there, incompatible cases fall back to Spark by default, and `allowIncompatible=true` runs the native (incompatible) path instead. `cast` is the main example; see the [expression reference](../expressions.md) for which expressions have incompatible cases. + +## Spark legacy configs + +Spark exposes a family of `spark.sql.legacy.*` configs that opt a query into pre-modern Spark +semantics (for example, `spark.sql.legacy.castComplexTypesToString.enabled` or +`spark.sql.legacy.timeParserPolicy`). Comet implements current Spark semantics and does **not** +reproduce these legacy behaviors. + +By default, when Comet detects that any `spark.sql.legacy.*` config is set to `true`, it disables +itself for that session so the query runs on vanilla Spark and gets the expected legacy results. +The fallback is logged as a warning listing the offending config keys. + +If you would rather keep Comet enabled even when a legacy config is set, set: + +``` +spark.comet.legacyConfFallback.enabled=false +``` + +In that mode Comet will accelerate the query as usual, but it **cannot guarantee Spark +compatibility** — the legacy config will be silently ignored by Comet-executed operators, so +results may diverge from what Spark would produce with the same legacy config. diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index 8e47151358..f3aa031a4e 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -93,6 +93,17 @@ object CometConf extends ShimCometConf { .booleanConf .createWithEnvVarOrDefault("ENABLE_COMET", true) + val COMET_LEGACY_CONF_FALLBACK_ENABLED: ConfigEntry[Boolean] = + conf("spark.comet.legacyConfFallback.enabled") + .category(CATEGORY_EXEC) + .doc( + "When true (default), Comet falls back to Spark whenever any spark.sql.legacy.* " + + "config is enabled, because Comet does not implement Spark's legacy semantics. Set " + + "to false to keep Comet enabled even when legacy configs are set; Comet cannot " + + "guarantee Spark compatibility in that case.") + .booleanConf + .createWithDefault(true) + val COMET_NATIVE_SCAN_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.scan.enabled") .category(CATEGORY_TESTING) .doc("Whether to enable native scans. Intended for use in Comet's own test suites to " + diff --git a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala index 22c3c9c93e..db65e71abc 100644 --- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala +++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala @@ -145,6 +145,24 @@ object CometSparkSessionExtensions extends Logging { return false } + // Fall back to Spark when any spark.sql.legacy.* config is enabled, since these opt into + // pre-modern Spark semantics that Comet does not replicate. Users can disable this fallback + // via COMET_LEGACY_CONF_FALLBACK_ENABLED if they accept the loss of Spark compatibility. + if (COMET_LEGACY_CONF_FALLBACK_ENABLED.get(conf)) { + val enabledLegacyConfs = conf.getAllConfs.collect { + case (k, v) if k.startsWith("spark.sql.legacy.") && v.equalsIgnoreCase("true") => k + } + if (enabledLegacyConfs.nonEmpty) { + logWarning( + "Comet extension is disabled because the following spark.sql.legacy.* configs are " + + s"enabled: ${enabledLegacyConfs.toSeq.sorted.mkString(", ")}. " + + "Comet does not support Spark legacy semantics. Set " + + s"${COMET_LEGACY_CONF_FALLBACK_ENABLED.key}=false to keep Comet enabled anyway " + + "(Spark compatibility is not guaranteed in that case).") + return false + } + } + try { // This will load the Comet native lib on demand, and if success, should set // `NativeBase.loaded` to true diff --git a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala index 268fdf94eb..4a6ef0699f 100644 --- a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala @@ -53,6 +53,36 @@ class CometSparkSessionExtensionsSuite extends CometTestBase { NativeBase.setLoaded(true) } + test("isCometLoaded falls back to Spark when spark.sql.legacy.* is enabled") { + val conf = new SQLConf + conf.setConfString(CometConf.COMET_ENABLED.key, "true") + conf.setConfString(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "false") + + // Baseline: no legacy configs set, Comet should load. + assert(isCometLoaded(conf)) + + // Any spark.sql.legacy.* set to true should disable Comet. + conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "true") + assert(!isCometLoaded(conf)) + + // Case-insensitive true value is also honored. + conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "TRUE") + assert(!isCometLoaded(conf)) + + // Setting the config to false should re-enable Comet. + conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "false") + assert(isCometLoaded(conf)) + + // Non-legacy spark.sql.* configs must not trigger fallback. + conf.setConfString("spark.sql.shuffle.partitions", "10") + assert(isCometLoaded(conf)) + + // Users can opt out of the legacy-config fallback and keep Comet enabled. + conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "true") + conf.setConfString(CometConf.COMET_LEGACY_CONF_FALLBACK_ENABLED.key, "false") + assert(isCometLoaded(conf)) + } + test("isCometLoaded requires CometShuffleManager when shuffle.enabled=true") { val conf = new SQLConf conf.setConfString(CometConf.COMET_ENABLED.key, "true") From dde00d1e5b617fabe01b7f016388c464712232c3 Mon Sep 17 00:00:00 2001 From: comphead Date: Thu, 2 Jul 2026 14:24:46 -0700 Subject: [PATCH 02/11] chore: fallback to Spark if legacy sql configurations are set --- .../apache/comet/expressions/CometCast.scala | 38 ++-------- .../scala/org/apache/comet/serde/arrays.scala | 8 +- .../scala/org/apache/comet/serde/maps.scala | 19 ++--- .../serde/operator/CometNativeScan.scala | 17 ++--- .../org/apache/comet/CometCastSuite.scala | 73 ------------------- .../apache/comet/CometExpressionSuite.scala | 33 --------- .../comet/exec/CometNativeReaderSuite.scala | 4 - 7 files changed, 21 insertions(+), 171 deletions(-) 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..9873809518 100644 --- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala +++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala @@ -32,23 +32,6 @@ import org.apache.comet.shims.CometExprShim object CometCast extends CometExpressionSerde[Cast] with CometExprShim { - // 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. - private[comet] val legacyCastComplexTypesToStringReason: String = - "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported" - - private def legacyCastComplexTypesToString: Boolean = - SQLConf.get - .getConfString("spark.sql.legacy.castComplexTypesToString.enabled", "false") - .toBoolean - def supportedTypes: Seq[DataType] = Seq( DataTypes.BooleanType, @@ -163,12 +146,6 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { return Compatible() } - if (toType == DataTypes.StringType && legacyCastComplexTypesToString && (fromType - .isInstanceOf[ArrayType] || fromType.isInstanceOf[StructType] || - fromType.isInstanceOf[MapType])) { - return Unsupported(Some(legacyCastComplexTypesToStringReason)) - } - (fromType, toType) match { case (dt: ArrayType, _: ArrayType) if dt.elementType == NullType => Compatible() case (ArrayType(DataTypes.DateType, _), ArrayType(toElementType, _)) @@ -277,16 +254,11 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { "String formatting can differ for floating-point values near precision limits " + "or when scientific notation is used")) case d: DecimalType if d.scale < 0 => - // Negative-scale decimals require spark.sql.legacy.allowNegativeScaleOfDecimal=true. - // When that config is enabled, Spark formats them using Java BigDecimal.toString() - // which produces scientific notation (e.g. "1.23E+4"). Comet matches this behavior. - // When the config is disabled, negative-scale decimals cannot be created in Spark, - // so we mark this as incompatible to avoid native execution on unexpected inputs. - val allowNegativeScale = SQLConf.get - .getConfString("spark.sql.legacy.allowNegativeScaleOfDecimal", "false") - .toBoolean - if (allowNegativeScale) Compatible() - else Incompatible(Some(negativeScaleDecimalToStringReason)) + // Negative-scale decimals require spark.sql.legacy.allowNegativeScaleOfDecimal=true, + // which the blanket legacy-conf fallback in CometSparkSessionExtensions.isCometLoaded + // already disables Comet for. If a user opts out of that fallback, Spark formats these + // via Java BigDecimal.toString() (scientific notation) and Comet matches that behavior. + Compatible() case _: DecimalType => // Compatible across all eval modes: LEGACY uses cast_decimal128_to_utf8 which // replicates Java BigDecimal.toString() (scientific notation when adj_exp < -6); diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala b/spark/src/main/scala/org/apache/comet/serde/arrays.scala index 8eda097ce6..8aa21e53f6 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -24,7 +24,6 @@ import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.expressions.{And, ArrayAggregate, ArrayAppend, ArrayContains, ArrayExcept, ArrayExists, ArrayFilter, ArrayForAll, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayPosition, ArrayRemove, ArrayRepeat, ArraySort, ArraysOverlap, ArraysZip, ArrayTransform, ArrayUnion, Attribute, Cast, CreateArray, ElementAt, EmptyRow, Expression, Flatten, GetArrayItem, IsNotNull, Literal, Reverse, Sequence, Size, Slice, SortArray, ZipWith} import org.apache.spark.sql.catalyst.util.GenericArrayData -import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.comet.CometConf @@ -460,15 +459,16 @@ object CometArrayInsert extends CometExpressionSerde[ArrayInsert] { val srcExprProto = exprToProtoInternal(expr.children.head, inputs, binding) val posExprProto = exprToProtoInternal(expr.children(1), inputs, binding) val itemExprProto = exprToProtoInternal(expr.children(2), inputs, binding) - val legacyNegativeIndex = - SQLConf.get.getConfString("spark.sql.legacy.negativeIndexInArrayInsert").toBoolean if (srcExprProto.isDefined && posExprProto.isDefined && itemExprProto.isDefined) { val arrayInsertBuilder = ExprOuterClass.ArrayInsert .newBuilder() .setSrcArrayExpr(srcExprProto.get) .setPosExpr(posExprProto.get) .setItemExpr(itemExprProto.get) - .setLegacyNegativeIndex(legacyNegativeIndex) + // spark.sql.legacy.negativeIndexInArrayInsert=true is handled by the blanket + // legacy-conf fallback in CometSparkSessionExtensions.isCometLoaded, so from + // Comet's perspective this always runs with the non-legacy semantics. + .setLegacyNegativeIndex(false) Some( ExprOuterClass.Expr diff --git a/spark/src/main/scala/org/apache/comet/serde/maps.scala b/spark/src/main/scala/org/apache/comet/serde/maps.scala index d663941c51..52334154fa 100644 --- a/spark/src/main/scala/org/apache/comet/serde/maps.scala +++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala @@ -20,7 +20,6 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto} @@ -142,25 +141,17 @@ object CometMapFromEntries object CometStrToMap extends CometScalarFunction[StringToMap]("str_to_map") with CometTypeShim { - // Spark 4.1.1+ honours spark.sql.legacy.truncateForEmptyRegexSplit by truncating trailing - // empty entries from the split result. Comet's native str_to_map always behaves as if the flag - // were false, so it is incompatible when legacy truncation is enabled. Read by string key so it - // resolves on older Spark versions where the config is not registered. - private val legacyTruncateConfig = "spark.sql.legacy.truncateForEmptyRegexSplit" - - private val legacyTruncateReason = - s"`$legacyTruncateConfig` is enabled, so trailing empty split entries may differ from Spark." - + // spark.sql.legacy.truncateForEmptyRegexSplit=true (Spark 4.1.1+) is handled by the blanket + // legacy-conf fallback in CometSparkSessionExtensions.isCometLoaded. Comet's native str_to_map + // always behaves as if the flag were false. private val collationReason = "`str_to_map` does not support non-UTF8_BINARY collations on the input string or delimiters." override def getIncompatibleReasons(): Seq[String] = - Seq(legacyTruncateReason, collationReason) + Seq(collationReason) override def getSupportLevel(expr: StringToMap): SupportLevel = { - if (SQLConf.get.getConfString(legacyTruncateConfig, "false").toBoolean) { - Incompatible(Some(legacyTruncateReason)) - } else if (expr.children.exists(child => hasNonDefaultStringCollation(child.dataType))) { + if (expr.children.exists(child => hasNonDefaultStringCollation(child.dataType))) { Incompatible(Some(collationReason)) } else { Compatible(None) diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala index 763602bb7d..e56d1c7fa2 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala @@ -190,16 +190,13 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging { commonBuilder.setSessionTimezone(scan.conf.getConfString("spark.sql.session.timeZone")) commonBuilder.setCaseSensitive(scan.conf.getConf[Boolean](SQLConf.CASE_SENSITIVE)) - // SPARK-53535 (Spark 4.1+): when reading a struct whose requested fields are all - // missing in the Parquet file, the new default preserves the parent struct's - // nullness from the file (so non-null parents materialize as a struct of all-null - // fields). Pre-4.1 Spark hardcodes the legacy behavior (whole struct null), which - // matches the Comet default we use as fallback. - val returnNullStructConfKey = - "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" - val returnNullStructDefault = if (isSpark41Plus) "false" else "true" - commonBuilder.setReturnNullStructIfAllFieldsMissing( - scan.conf.getConfString(returnNullStructConfKey, returnNullStructDefault).toBoolean) + // SPARK-53535 (Spark 4.1+): reading a struct whose requested fields are all missing in + // the Parquet file preserves the parent struct's nullness. The legacy behavior is toggled + // by spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing, which is handled by the + // blanket legacy-conf fallback in CometSparkSessionExtensions.isCometLoaded. Comet always + // runs the modern (non-legacy) behavior on Spark 4.1+; on older Spark versions the legacy + // behavior is the hardcoded default, which matches Comet's fallback. + commonBuilder.setReturnNullStructIfAllFieldsMissing(!isSpark41Plus) // Field-ID matching: only ask the native side to do extra work when the conf is on AND // the requested schema actually carries IDs. Spark's ParquetReadSupport applies the same diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala index 7245f72842..928ef12399 100644 --- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala @@ -733,66 +733,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { castTest(generateDecimalsPrecision38Scale18(), DataTypes.StringType) } - test("cast DecimalType with negative scale to StringType") { - // Negative-scale decimals are a legacy Spark feature gated on - // spark.sql.legacy.allowNegativeScaleOfDecimal=true. Spark LEGACY cast uses Java's - // BigDecimal.toString() which produces scientific notation for negative-scale values - // (e.g. 12300 stored as Decimal(7,-2) with unscaled=123 → "1.23E+4"). - // CometCast.canCastToString checks the - // config and returns Incompatible when it is false. - // - // Parquet does not support negative-scale decimals so we use checkSparkAnswer directly - // (no parquet round-trip) to avoid schema coercion. - - // With config enabled, enable localTableScan so Comet can take over the full plan - // and execute the cast natively. Parquet does not support negative-scale decimals so - // the data is kept in-memory; localTableScan.enabled bridges that gap. - withSQLConf( - "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true", - "spark.comet.exec.localTableScan.enabled" -> "true") { - val dfNeg2 = Seq( - Some(BigDecimal("0")), - Some(BigDecimal("100")), - Some(BigDecimal("12300")), - Some(BigDecimal("-99900")), - Some(BigDecimal("9999900")), - None) - .toDF("b") - .withColumn("a", col("b").cast(DecimalType(7, -2))) - .drop("b") - .select(col("a").cast(DataTypes.StringType).as("result")) - checkSparkAnswerAndOperator(dfNeg2) - - val dfNeg4 = Seq( - Some(BigDecimal("0")), - Some(BigDecimal("10000")), - Some(BigDecimal("120000")), - Some(BigDecimal("-9990000")), - None) - .toDF("b") - .withColumn("a", col("b").cast(DecimalType(7, -4))) - .drop("b") - .select(col("a").cast(DataTypes.StringType).as("result")) - checkSparkAnswerAndOperator(dfNeg4) - } - - // With config disabled (default): the SQL parser rejects negative scale, so - // negative-scale decimals cannot be created through normal SQL paths. - // CometCast.isSupported returns Incompatible for this case, ensuring Comet does - // not attempt native execution if such a value ever reaches the planner. - // Note: DecimalType(7, -2) must be constructed while config=true, because the - // constructor itself checks the config and throws if negative scale is disallowed. - var negScaleType: DecimalType = null - withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true") { - negScaleType = DecimalType(7, -2) - } - withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "false") { - assert( - CometCast.isSupported(negScaleType, DataTypes.StringType, None, CometEvalMode.LEGACY) == - Incompatible(Some(CometCast.negativeScaleDecimalToStringReason))) - } - } - test("cast DecimalType(10,2) to TimestampType") { castTest(generateDecimalsPrecision10Scale2(), DataTypes.TimestampType) } @@ -1565,19 +1505,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { castTest(generateDecimalsPrecision10Scale2(), DataTypes.createDecimalType(10, 4)) } - test("cast StringType to DecimalType with negative scale (allowNegativeScaleOfDecimal)") { - // With allowNegativeScaleOfDecimal=true, Spark allows DECIMAL(p, s) where s < 0. - // The value is rounded to the nearest 10^|s| — e.g. DECIMAL(10,-4) rounds to - // the nearest 10000. This requires the legacy SQL parser config to be enabled. - withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true") { - val values = - Seq("12500", "15000", "99990000", "-12500", "0", "0.001", "abc", null).toDF("a") - // testTry=false: try_cast uses SQL string interpolation (toType.sql → "DECIMAL(10,-4)") - // which the SQL parser rejects regardless of allowNegativeScaleOfDecimal. - castTest(values, DataTypes.createDecimalType(10, -4), testTry = false) - } - } - test("cast between decimals with negative precision") { // cast to negative scale checkSparkAnswerMaybeThrows( diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index 33bd58c55e..efc49d46b9 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -3154,39 +3154,6 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } - test("vectorized reader: missing all struct fields") { - Seq(true, false).foreach { offheapEnabled => - withSQLConf( - SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", - CometConf.COMET_EXEC_ENABLED.key -> "true", - CometConf.COMET_ENABLED.key -> "true", - CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "false", - SQLConf.PARQUET_VECTORIZED_READER_NESTED_COLUMN_ENABLED.key -> "true", - SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offheapEnabled.toString, - // SPARK-53535 (Spark 4.1+) flipped the default to "false", which preserves the parent - // struct's nullness so non-null parents materialise as Row(Row(null, null)). This test - // asserts the legacy "all missing fields => null struct" answer, so pin the conf to - // "true" to keep the expectation valid on both 3.x/4.0 and 4.1+. The non-legacy - // behaviour is covered separately by `issue #4136` in CometNativeReaderSuite. - "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" -> "true") { - val data = Seq(Tuple1((1, "a")), Tuple1((2, null)), Tuple1(null)) - - val readSchema = new StructType().add( - "_1", - new StructType() - .add("_3", IntegerType, nullable = false) - .add("_4", StringType, nullable = false), - nullable = false) - - withParquetFile(data) { file => - checkAnswer( - spark.read.schema(readSchema).parquet(file), - Row(null) :: Row(null) :: Row(null) :: Nil) - } - } - } - } - test("test length function") { // cast(id as binary) is rejected by Spark 4 ANSI analyzer withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") { diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala index 876565c5e5..86353c1926 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -726,10 +726,8 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper // reader flag. We've seen CI fail on the off-heap branch when the on-heap branch passes. for { offheapEnabled <- Seq("true", "false") - legacy <- Seq("true", "false") } withSQLConf( "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true", - "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" -> legacy, "spark.sql.columnVector.offheap.enabled" -> offheapEnabled) { val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) checkSparkAnswer(df) @@ -773,10 +771,8 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper for { offheapEnabled <- Seq("true", "false") - legacy <- Seq("true", "false") } withSQLConf( "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true", - "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" -> legacy, "spark.sql.columnVector.offheap.enabled" -> offheapEnabled) { val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) checkSparkAnswer(df) From 6e11122b8c0b6a23b3c57bcd3c246b22cbb8b5f1 Mon Sep 17 00:00:00 2001 From: comphead Date: Thu, 2 Jul 2026 17:40:55 -0700 Subject: [PATCH 03/11] chore: fallback to Spark if legacy sql configurations are set --- .../user-guide/latest/compatibility/index.md | 21 ------ .../scala/org/apache/comet/CometConf.scala | 11 --- .../comet/CometSparkSessionExtensions.scala | 18 ----- .../apache/comet/expressions/CometCast.scala | 47 ++++++++++-- .../org/apache/comet/serde/aggregates.scala | 22 ++++++ .../scala/org/apache/comet/serde/arrays.scala | 35 +++++++-- .../scala/org/apache/comet/serde/maps.scala | 26 +++++-- .../serde/operator/CometNativeScan.scala | 17 +++-- .../cast_complex_types_to_string_legacy.sql | 22 +++--- .../map/str_to_map_legacy_truncate.sql | 19 +++-- .../org/apache/comet/CometCastSuite.scala | 73 +++++++++++++++++++ .../apache/comet/CometExpressionSuite.scala | 33 +++++++++ .../CometSparkSessionExtensionsSuite.scala | 30 -------- .../comet/exec/CometNativeReaderSuite.scala | 4 + 14 files changed, 254 insertions(+), 124 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index ce1da965c5..f3b531a18f 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -70,24 +70,3 @@ This is distinct from expressions that have **no** codegen-dispatch path: there, incompatible cases fall back to Spark by default, and `allowIncompatible=true` runs the native (incompatible) path instead. `cast` is the main example; see the [expression reference](../expressions.md) for which expressions have incompatible cases. - -## Spark legacy configs - -Spark exposes a family of `spark.sql.legacy.*` configs that opt a query into pre-modern Spark -semantics (for example, `spark.sql.legacy.castComplexTypesToString.enabled` or -`spark.sql.legacy.timeParserPolicy`). Comet implements current Spark semantics and does **not** -reproduce these legacy behaviors. - -By default, when Comet detects that any `spark.sql.legacy.*` config is set to `true`, it disables -itself for that session so the query runs on vanilla Spark and gets the expected legacy results. -The fallback is logged as a warning listing the offending config keys. - -If you would rather keep Comet enabled even when a legacy config is set, set: - -``` -spark.comet.legacyConfFallback.enabled=false -``` - -In that mode Comet will accelerate the query as usual, but it **cannot guarantee Spark -compatibility** — the legacy config will be silently ignored by Comet-executed operators, so -results may diverge from what Spark would produce with the same legacy config. diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index f3aa031a4e..8e47151358 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -93,17 +93,6 @@ object CometConf extends ShimCometConf { .booleanConf .createWithEnvVarOrDefault("ENABLE_COMET", true) - val COMET_LEGACY_CONF_FALLBACK_ENABLED: ConfigEntry[Boolean] = - conf("spark.comet.legacyConfFallback.enabled") - .category(CATEGORY_EXEC) - .doc( - "When true (default), Comet falls back to Spark whenever any spark.sql.legacy.* " + - "config is enabled, because Comet does not implement Spark's legacy semantics. Set " + - "to false to keep Comet enabled even when legacy configs are set; Comet cannot " + - "guarantee Spark compatibility in that case.") - .booleanConf - .createWithDefault(true) - val COMET_NATIVE_SCAN_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.scan.enabled") .category(CATEGORY_TESTING) .doc("Whether to enable native scans. Intended for use in Comet's own test suites to " + diff --git a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala index db65e71abc..22c3c9c93e 100644 --- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala +++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala @@ -145,24 +145,6 @@ object CometSparkSessionExtensions extends Logging { return false } - // Fall back to Spark when any spark.sql.legacy.* config is enabled, since these opt into - // pre-modern Spark semantics that Comet does not replicate. Users can disable this fallback - // via COMET_LEGACY_CONF_FALLBACK_ENABLED if they accept the loss of Spark compatibility. - if (COMET_LEGACY_CONF_FALLBACK_ENABLED.get(conf)) { - val enabledLegacyConfs = conf.getAllConfs.collect { - case (k, v) if k.startsWith("spark.sql.legacy.") && v.equalsIgnoreCase("true") => k - } - if (enabledLegacyConfs.nonEmpty) { - logWarning( - "Comet extension is disabled because the following spark.sql.legacy.* configs are " + - s"enabled: ${enabledLegacyConfs.toSeq.sorted.mkString(", ")}. " + - "Comet does not support Spark legacy semantics. Set " + - s"${COMET_LEGACY_CONF_FALLBACK_ENABLED.key}=false to keep Comet enabled anyway " + - "(Spark compatibility is not guaranteed in that case).") - return false - } - } - try { // This will load the Comet native lib on demand, and if success, should set // `NativeBase.loaded` to true 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 9873809518..6e9767228e 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,34 @@ 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.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 -object CometCast extends CometExpressionSerde[Cast] with CometExprShim { +object CometCast + extends CometExpressionSerde[Cast] + with CometExprShim + with CodegenDispatchFallback { + + // 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's native cast + // only implements the default formatting, so when the flag is on we mark the cast Incompatible + // and let the [[CodegenDispatchFallback]] trait route it through the JVM codegen dispatcher + // (Spark's own `doGenCode` inside the Comet kernel) so results still match Spark exactly. 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 natively" + + private def legacyCastComplexTypesToString: Boolean = + SQLConf.get + .getConfString("spark.sql.legacy.castComplexTypesToString.enabled", "false") + .toBoolean def supportedTypes: Seq[DataType] = Seq( @@ -146,6 +168,12 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { return Compatible() } + if (toType == DataTypes.StringType && legacyCastComplexTypesToString && (fromType + .isInstanceOf[ArrayType] || fromType.isInstanceOf[StructType] || + fromType.isInstanceOf[MapType])) { + return Incompatible(Some(legacyCastComplexTypesToStringReason)) + } + (fromType, toType) match { case (dt: ArrayType, _: ArrayType) if dt.elementType == NullType => Compatible() case (ArrayType(DataTypes.DateType, _), ArrayType(toElementType, _)) @@ -254,11 +282,16 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { "String formatting can differ for floating-point values near precision limits " + "or when scientific notation is used")) case d: DecimalType if d.scale < 0 => - // Negative-scale decimals require spark.sql.legacy.allowNegativeScaleOfDecimal=true, - // which the blanket legacy-conf fallback in CometSparkSessionExtensions.isCometLoaded - // already disables Comet for. If a user opts out of that fallback, Spark formats these - // via Java BigDecimal.toString() (scientific notation) and Comet matches that behavior. - Compatible() + // Negative-scale decimals require spark.sql.legacy.allowNegativeScaleOfDecimal=true. + // When that config is enabled, Spark formats them using Java BigDecimal.toString() + // which produces scientific notation (e.g. "1.23E+4"). Comet matches this behavior. + // When the config is disabled, negative-scale decimals cannot be created in Spark, + // so we mark this as incompatible to avoid native execution on unexpected inputs. + val allowNegativeScale = SQLConf.get + .getConfString("spark.sql.legacy.allowNegativeScaleOfDecimal", "false") + .toBoolean + if (allowNegativeScale) Compatible() + else Incompatible(Some(negativeScaleDecimalToStringReason)) case _: DecimalType => // Compatible across all eval modes: LEGACY uses cast_decimal128_to_utf8 which // replicates Java BigDecimal.toString() (scientific notation when adj_exp < -6); diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala index 5710232cb4..f15e0f0319 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -106,6 +106,28 @@ object CometMax extends CometAggregateExpressionSerde[Max] { } object CometCount extends CometAggregateExpressionSerde[Count] { + + // When `spark.sql.legacy.allowParameterlessCount=true`, Spark allows `count()` with no + // arguments and treats it as `count(*)`. Comet's native planner asserts on non-empty children + // and would panic on such an expression, so mark it Unsupported here and let the aggregate fall + // back to Spark. Aggregate serdes have no [[CodegenDispatchFallback]] path (aggregates cannot + // be routed through the JVM codegen dispatcher), so a clean Spark fallback is the appropriate + // outcome. Under the default config value, Spark's analyzer rejects parameterless `count()` so + // this branch is unreachable. + private val legacyAllowParameterlessCountReason: String = + "`spark.sql.legacy.allowParameterlessCount=true` produces `count()` with no children, which " + + "the native planner does not support" + + override def getUnsupportedReasons(): Seq[String] = Seq(legacyAllowParameterlessCountReason) + + override def getSupportLevel(expr: Count): SupportLevel = { + if (expr.children.isEmpty) { + Unsupported(Some(legacyAllowParameterlessCountReason)) + } else { + Compatible() + } + } + override def convert( aggExpr: AggregateExpression, expr: Count, diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala b/spark/src/main/scala/org/apache/comet/serde/arrays.scala index 8aa21e53f6..b2f3df1ed4 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -24,6 +24,7 @@ import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.expressions.{And, ArrayAggregate, ArrayAppend, ArrayContains, ArrayExcept, ArrayExists, ArrayFilter, ArrayForAll, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayPosition, ArrayRemove, ArrayRepeat, ArraySort, ArraysOverlap, ArraysZip, ArrayTransform, ArrayUnion, Attribute, Cast, CreateArray, ElementAt, EmptyRow, Expression, Flatten, GetArrayItem, IsNotNull, Literal, Reverse, Sequence, Size, Slice, SortArray, ZipWith} import org.apache.spark.sql.catalyst.util.GenericArrayData +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.comet.CometConf @@ -448,9 +449,28 @@ object CometArrayJoin } } -object CometArrayInsert extends CometExpressionSerde[ArrayInsert] { +object CometArrayInsert extends CometExpressionSerde[ArrayInsert] with CodegenDispatchFallback { - override def getSupportLevel(expr: ArrayInsert): SupportLevel = Compatible() + // Spark's `spark.sql.legacy.negativeIndexInArrayInsert=true` changes how a 0-based/negative + // position is interpreted. Rather than maintain a parallel native code path for the legacy + // semantics, mark `array_insert` Incompatible when the flag is on so + // [[CodegenDispatchFallback]] routes the expression through the JVM codegen dispatcher + // (Spark's own `doGenCode` inside the Comet kernel) — that gives Spark-exact results + // without duplicating the legacy branch natively. + private val legacyNegativeIndexConfig = "spark.sql.legacy.negativeIndexInArrayInsert" + + private val legacyNegativeIndexReason = + s"`$legacyNegativeIndexConfig=true` legacy negative-index semantics are not implemented natively" + + override def getIncompatibleReasons(): Seq[String] = Seq(legacyNegativeIndexReason) + + override def getSupportLevel(expr: ArrayInsert): SupportLevel = { + if (SQLConf.get.getConfString(legacyNegativeIndexConfig, "false").toBoolean) { + Incompatible(Some(legacyNegativeIndexReason)) + } else { + Compatible() + } + } override def convert( expr: ArrayInsert, @@ -459,16 +479,19 @@ object CometArrayInsert extends CometExpressionSerde[ArrayInsert] { val srcExprProto = exprToProtoInternal(expr.children.head, inputs, binding) val posExprProto = exprToProtoInternal(expr.children(1), inputs, binding) val itemExprProto = exprToProtoInternal(expr.children(2), inputs, binding) + // Reached in two cases: + // 1. Legacy conf is false → getSupportLevel returned Compatible → run native. + // 2. Legacy conf is true AND user set allowIncompatible=true → opt in to native. + // In case (2) the native impl honors the legacy semantics directly so we forward the flag. + val legacyNegativeIndex = + SQLConf.get.getConfString(legacyNegativeIndexConfig, "false").toBoolean if (srcExprProto.isDefined && posExprProto.isDefined && itemExprProto.isDefined) { val arrayInsertBuilder = ExprOuterClass.ArrayInsert .newBuilder() .setSrcArrayExpr(srcExprProto.get) .setPosExpr(posExprProto.get) .setItemExpr(itemExprProto.get) - // spark.sql.legacy.negativeIndexInArrayInsert=true is handled by the blanket - // legacy-conf fallback in CometSparkSessionExtensions.isCometLoaded, so from - // Comet's perspective this always runs with the non-legacy semantics. - .setLegacyNegativeIndex(false) + .setLegacyNegativeIndex(legacyNegativeIndex) Some( ExprOuterClass.Expr diff --git a/spark/src/main/scala/org/apache/comet/serde/maps.scala b/spark/src/main/scala/org/apache/comet/serde/maps.scala index 52334154fa..e5b431220a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/maps.scala +++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala @@ -20,6 +20,7 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto} @@ -139,19 +140,32 @@ object CometMapFromEntries } } -object CometStrToMap extends CometScalarFunction[StringToMap]("str_to_map") with CometTypeShim { +object CometStrToMap + extends CometScalarFunction[StringToMap]("str_to_map") + with CometTypeShim + with CodegenDispatchFallback { + + // Spark 4.1.1+ honours spark.sql.legacy.truncateForEmptyRegexSplit by truncating trailing + // empty entries from the split result. Comet's native str_to_map always behaves as if the flag + // were false. When the flag is true, mark this Incompatible so the CodegenDispatchFallback + // trait routes the expression through the JVM codegen dispatcher (Spark's own doGenCode inside + // the Comet kernel) rather than falling the entire projection back to Spark. Read by string + // key so it resolves on older Spark versions where the config is not registered. + private val legacyTruncateConfig = "spark.sql.legacy.truncateForEmptyRegexSplit" + + private val legacyTruncateReason = + s"`$legacyTruncateConfig` is enabled, so trailing empty split entries may differ from Spark." - // spark.sql.legacy.truncateForEmptyRegexSplit=true (Spark 4.1.1+) is handled by the blanket - // legacy-conf fallback in CometSparkSessionExtensions.isCometLoaded. Comet's native str_to_map - // always behaves as if the flag were false. private val collationReason = "`str_to_map` does not support non-UTF8_BINARY collations on the input string or delimiters." override def getIncompatibleReasons(): Seq[String] = - Seq(collationReason) + Seq(legacyTruncateReason, collationReason) override def getSupportLevel(expr: StringToMap): SupportLevel = { - if (expr.children.exists(child => hasNonDefaultStringCollation(child.dataType))) { + if (SQLConf.get.getConfString(legacyTruncateConfig, "false").toBoolean) { + Incompatible(Some(legacyTruncateReason)) + } else if (expr.children.exists(child => hasNonDefaultStringCollation(child.dataType))) { Incompatible(Some(collationReason)) } else { Compatible(None) diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala index e56d1c7fa2..763602bb7d 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala @@ -190,13 +190,16 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging { commonBuilder.setSessionTimezone(scan.conf.getConfString("spark.sql.session.timeZone")) commonBuilder.setCaseSensitive(scan.conf.getConf[Boolean](SQLConf.CASE_SENSITIVE)) - // SPARK-53535 (Spark 4.1+): reading a struct whose requested fields are all missing in - // the Parquet file preserves the parent struct's nullness. The legacy behavior is toggled - // by spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing, which is handled by the - // blanket legacy-conf fallback in CometSparkSessionExtensions.isCometLoaded. Comet always - // runs the modern (non-legacy) behavior on Spark 4.1+; on older Spark versions the legacy - // behavior is the hardcoded default, which matches Comet's fallback. - commonBuilder.setReturnNullStructIfAllFieldsMissing(!isSpark41Plus) + // SPARK-53535 (Spark 4.1+): when reading a struct whose requested fields are all + // missing in the Parquet file, the new default preserves the parent struct's + // nullness from the file (so non-null parents materialize as a struct of all-null + // fields). Pre-4.1 Spark hardcodes the legacy behavior (whole struct null), which + // matches the Comet default we use as fallback. + val returnNullStructConfKey = + "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" + val returnNullStructDefault = if (isSpark41Plus) "false" else "true" + commonBuilder.setReturnNullStructIfAllFieldsMissing( + scan.conf.getConfString(returnNullStructConfKey, returnNullStructDefault).toBoolean) // Field-ID matching: only ask the native side to do extra work when the conf is on AND // the requested schema actually carries IDs. Spark's ParquetReadSupport applies the same 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..92010a7c89 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 @@ -17,24 +17,26 @@ -- 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. +-- (instead of rendering them as the literal "null"). Comet's native cast does not implement +-- the legacy formatting; the [[CodegenDispatchFallback]] mixin on `CometCast` routes these +-- casts through the JVM codegen dispatcher (Spark's own `doGenCode` inside the Comet kernel) +-- so results match Spark exactly without a Spark fallback. -- 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 routed through the codegen dispatcher. +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 routed through the codegen dispatcher. +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 routed through the codegen dispatcher. +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 also routed through the codegen dispatcher via the outer type. +query SELECT CAST(struct(array(1, null), map('k', null)) AS STRING) diff --git a/spark/src/test/resources/sql-tests/expressions/map/str_to_map_legacy_truncate.sql b/spark/src/test/resources/sql-tests/expressions/map/str_to_map_legacy_truncate.sql index a5b5eb9f16..8ab8058920 100644 --- a/spark/src/test/resources/sql-tests/expressions/map/str_to_map_legacy_truncate.sql +++ b/spark/src/test/resources/sql-tests/expressions/map/str_to_map_legacy_truncate.sql @@ -15,19 +15,22 @@ -- specific language governing permissions and limitations -- under the License. --- Tests that str_to_map falls back to Spark when --- spark.sql.legacy.truncateForEmptyRegexSplit is enabled. In legacy mode Spark truncates trailing --- empty entries from the split result, which Comet's native str_to_map does not honour. +-- Tests that str_to_map routes through the JVM codegen dispatcher when +-- spark.sql.legacy.truncateForEmptyRegexSplit is enabled. In legacy mode Spark truncates +-- trailing empty entries from the split result, which Comet's native str_to_map does not +-- honour. `CometStrToMap` marks the expression Incompatible when the flag is on and mixes in +-- [[CodegenDispatchFallback]], so the projection stays native (Spark's own `doGenCode` runs +-- inside the Comet kernel) while producing Spark-exact results. -- See https://github.com/apache/datafusion-comet/issues/4477 -- Config: spark.sql.legacy.truncateForEmptyRegexSplit=true --- trailing pair delimiter: legacy mode truncates the trailing empty entry, so Comet must fall --- back to Spark -query expect_fallback(truncateForEmptyRegexSplit) +-- trailing pair delimiter: legacy mode truncates the trailing empty entry; Comet delegates to +-- the codegen dispatcher. +query SELECT str_to_map('a:1,b:2,', ',', ':') --- column input also falls back +-- column input is also handled via the codegen dispatcher statement CREATE TABLE test_str_to_map_legacy(s STRING, pair_delim STRING, key_value_delim STRING) USING parquet @@ -37,5 +40,5 @@ INSERT INTO test_str_to_map_legacy VALUES ('x:1;y:2;', ';', ':'), (NULL, ',', ':') -query expect_fallback(truncateForEmptyRegexSplit) +query SELECT str_to_map(s, pair_delim, key_value_delim) FROM test_str_to_map_legacy diff --git a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala index 928ef12399..7245f72842 100644 --- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala @@ -733,6 +733,66 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { castTest(generateDecimalsPrecision38Scale18(), DataTypes.StringType) } + test("cast DecimalType with negative scale to StringType") { + // Negative-scale decimals are a legacy Spark feature gated on + // spark.sql.legacy.allowNegativeScaleOfDecimal=true. Spark LEGACY cast uses Java's + // BigDecimal.toString() which produces scientific notation for negative-scale values + // (e.g. 12300 stored as Decimal(7,-2) with unscaled=123 → "1.23E+4"). + // CometCast.canCastToString checks the + // config and returns Incompatible when it is false. + // + // Parquet does not support negative-scale decimals so we use checkSparkAnswer directly + // (no parquet round-trip) to avoid schema coercion. + + // With config enabled, enable localTableScan so Comet can take over the full plan + // and execute the cast natively. Parquet does not support negative-scale decimals so + // the data is kept in-memory; localTableScan.enabled bridges that gap. + withSQLConf( + "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true", + "spark.comet.exec.localTableScan.enabled" -> "true") { + val dfNeg2 = Seq( + Some(BigDecimal("0")), + Some(BigDecimal("100")), + Some(BigDecimal("12300")), + Some(BigDecimal("-99900")), + Some(BigDecimal("9999900")), + None) + .toDF("b") + .withColumn("a", col("b").cast(DecimalType(7, -2))) + .drop("b") + .select(col("a").cast(DataTypes.StringType).as("result")) + checkSparkAnswerAndOperator(dfNeg2) + + val dfNeg4 = Seq( + Some(BigDecimal("0")), + Some(BigDecimal("10000")), + Some(BigDecimal("120000")), + Some(BigDecimal("-9990000")), + None) + .toDF("b") + .withColumn("a", col("b").cast(DecimalType(7, -4))) + .drop("b") + .select(col("a").cast(DataTypes.StringType).as("result")) + checkSparkAnswerAndOperator(dfNeg4) + } + + // With config disabled (default): the SQL parser rejects negative scale, so + // negative-scale decimals cannot be created through normal SQL paths. + // CometCast.isSupported returns Incompatible for this case, ensuring Comet does + // not attempt native execution if such a value ever reaches the planner. + // Note: DecimalType(7, -2) must be constructed while config=true, because the + // constructor itself checks the config and throws if negative scale is disallowed. + var negScaleType: DecimalType = null + withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true") { + negScaleType = DecimalType(7, -2) + } + withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "false") { + assert( + CometCast.isSupported(negScaleType, DataTypes.StringType, None, CometEvalMode.LEGACY) == + Incompatible(Some(CometCast.negativeScaleDecimalToStringReason))) + } + } + test("cast DecimalType(10,2) to TimestampType") { castTest(generateDecimalsPrecision10Scale2(), DataTypes.TimestampType) } @@ -1505,6 +1565,19 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { castTest(generateDecimalsPrecision10Scale2(), DataTypes.createDecimalType(10, 4)) } + test("cast StringType to DecimalType with negative scale (allowNegativeScaleOfDecimal)") { + // With allowNegativeScaleOfDecimal=true, Spark allows DECIMAL(p, s) where s < 0. + // The value is rounded to the nearest 10^|s| — e.g. DECIMAL(10,-4) rounds to + // the nearest 10000. This requires the legacy SQL parser config to be enabled. + withSQLConf("spark.sql.legacy.allowNegativeScaleOfDecimal" -> "true") { + val values = + Seq("12500", "15000", "99990000", "-12500", "0", "0.001", "abc", null).toDF("a") + // testTry=false: try_cast uses SQL string interpolation (toType.sql → "DECIMAL(10,-4)") + // which the SQL parser rejects regardless of allowNegativeScaleOfDecimal. + castTest(values, DataTypes.createDecimalType(10, -4), testTry = false) + } + } + test("cast between decimals with negative precision") { // cast to negative scale checkSparkAnswerMaybeThrows( diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index efc49d46b9..33bd58c55e 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -3154,6 +3154,39 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + test("vectorized reader: missing all struct fields") { + Seq(true, false).foreach { offheapEnabled => + withSQLConf( + SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "false", + SQLConf.PARQUET_VECTORIZED_READER_NESTED_COLUMN_ENABLED.key -> "true", + SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offheapEnabled.toString, + // SPARK-53535 (Spark 4.1+) flipped the default to "false", which preserves the parent + // struct's nullness so non-null parents materialise as Row(Row(null, null)). This test + // asserts the legacy "all missing fields => null struct" answer, so pin the conf to + // "true" to keep the expectation valid on both 3.x/4.0 and 4.1+. The non-legacy + // behaviour is covered separately by `issue #4136` in CometNativeReaderSuite. + "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" -> "true") { + val data = Seq(Tuple1((1, "a")), Tuple1((2, null)), Tuple1(null)) + + val readSchema = new StructType().add( + "_1", + new StructType() + .add("_3", IntegerType, nullable = false) + .add("_4", StringType, nullable = false), + nullable = false) + + withParquetFile(data) { file => + checkAnswer( + spark.read.schema(readSchema).parquet(file), + Row(null) :: Row(null) :: Row(null) :: Nil) + } + } + } + } + test("test length function") { // cast(id as binary) is rejected by Spark 4 ANSI analyzer withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") { diff --git a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala index 4a6ef0699f..268fdf94eb 100644 --- a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala @@ -53,36 +53,6 @@ class CometSparkSessionExtensionsSuite extends CometTestBase { NativeBase.setLoaded(true) } - test("isCometLoaded falls back to Spark when spark.sql.legacy.* is enabled") { - val conf = new SQLConf - conf.setConfString(CometConf.COMET_ENABLED.key, "true") - conf.setConfString(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "false") - - // Baseline: no legacy configs set, Comet should load. - assert(isCometLoaded(conf)) - - // Any spark.sql.legacy.* set to true should disable Comet. - conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "true") - assert(!isCometLoaded(conf)) - - // Case-insensitive true value is also honored. - conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "TRUE") - assert(!isCometLoaded(conf)) - - // Setting the config to false should re-enable Comet. - conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "false") - assert(isCometLoaded(conf)) - - // Non-legacy spark.sql.* configs must not trigger fallback. - conf.setConfString("spark.sql.shuffle.partitions", "10") - assert(isCometLoaded(conf)) - - // Users can opt out of the legacy-config fallback and keep Comet enabled. - conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "true") - conf.setConfString(CometConf.COMET_LEGACY_CONF_FALLBACK_ENABLED.key, "false") - assert(isCometLoaded(conf)) - } - test("isCometLoaded requires CometShuffleManager when shuffle.enabled=true") { val conf = new SQLConf conf.setConfString(CometConf.COMET_ENABLED.key, "true") diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala index 86353c1926..876565c5e5 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -726,8 +726,10 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper // reader flag. We've seen CI fail on the off-heap branch when the on-heap branch passes. for { offheapEnabled <- Seq("true", "false") + legacy <- Seq("true", "false") } withSQLConf( "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true", + "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" -> legacy, "spark.sql.columnVector.offheap.enabled" -> offheapEnabled) { val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) checkSparkAnswer(df) @@ -771,8 +773,10 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper for { offheapEnabled <- Seq("true", "false") + legacy <- Seq("true", "false") } withSQLConf( "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true", + "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" -> legacy, "spark.sql.columnVector.offheap.enabled" -> offheapEnabled) { val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) checkSparkAnswer(df) From 80242c194efd7a816fb9fc8c0c134ec43ee7f55a Mon Sep 17 00:00:00 2001 From: comphead Date: Thu, 2 Jul 2026 17:42:09 -0700 Subject: [PATCH 04/11] chore: fallback to Spark if legacy sql configurations are set --- .../aggregate/count_parameterless_legacy.sql | 45 ++++++++++++++ .../array/array_insert_legacy_dispatch.sql | 60 +++++++++++++++++++ 2 files changed, 105 insertions(+) create mode 100644 spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_insert_legacy_dispatch.sql diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql new file mode 100644 index 0000000000..b47476b0f5 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql @@ -0,0 +1,45 @@ +-- 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. + +-- When `spark.sql.legacy.allowParameterlessCount=true`, Spark accepts `count()` (no arguments) +-- and treats it as `count(*)`. Comet's native planner asserts non-empty children on Count, so +-- `CometCount.getSupportLevel` marks parameterless Count `Unsupported` and lets the aggregate +-- fall back to Spark. Aggregate serdes do not have a JVM codegen dispatcher path, so the +-- Spark fallback is the correct outcome. + +-- Config: spark.sql.legacy.allowParameterlessCount=true + +statement +CREATE TABLE test_count_parameterless(i int, grp string) USING parquet + +statement +INSERT INTO test_count_parameterless VALUES (1, 'x'), (2, 'x'), (NULL, 'y'), (3, 'y'), (NULL, 'y') + +-- Parameterless count() falls back to Spark; the aggregate result must still match Spark. +query expect_fallback(spark.sql.legacy.allowParameterlessCount=true) +SELECT count() FROM test_count_parameterless + +-- Parameterless count() with GROUP BY. +query expect_fallback(spark.sql.legacy.allowParameterlessCount=true) +SELECT grp, count() FROM test_count_parameterless GROUP BY grp ORDER BY grp + +-- Parameterless count() on empty table. +statement +CREATE TABLE test_count_empty(i int) USING parquet + +query expect_fallback(spark.sql.legacy.allowParameterlessCount=true) +SELECT count() FROM test_count_empty diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_insert_legacy_dispatch.sql b/spark/src/test/resources/sql-tests/expressions/array/array_insert_legacy_dispatch.sql new file mode 100644 index 0000000000..4229312c3e --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_insert_legacy_dispatch.sql @@ -0,0 +1,60 @@ +-- 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. + +-- Tests array_insert with legacy negative index mode enabled but without opting into the +-- native (incompatible) path. `CometArrayInsert` mixes in [[CodegenDispatchFallback]] so with +-- spark.sql.legacy.negativeIndexInArrayInsert=true the expression is routed through the JVM +-- codegen dispatcher (Spark's own `doGenCode` inside the Comet kernel), producing Spark-exact +-- results without a Spark fallback and without touching the native legacy branch. +-- The companion file array_insert_legacy.sql covers the allowIncompatible=true opt-in path. + +-- ConfigMatrix: parquet.enable.dictionary=false,true +-- Config: spark.sql.legacy.negativeIndexInArrayInsert=true + +-- -1 inserts before last element in legacy mode +query +SELECT array_insert(array(1, 2, 3), -1, 10) + +-- -2 inserts before second-to-last +query +SELECT array_insert(array(1, 2, 3), -2, 10) + +-- -3 inserts before first element +query +SELECT array_insert(array(1, 2, 3), -3, 10) + +-- negative beyond start with null padding (legacy mode pads differently) +query +SELECT array_insert(array(1, 2, 3), -5, 10) + +-- far negative beyond start +query +SELECT array_insert(array(1, 3, 4), -2, 2) + +-- column-based test +statement +CREATE TABLE test_ai_legacy_dispatch(arr array, pos int, val int) USING parquet + +statement +INSERT INTO test_ai_legacy_dispatch VALUES + (array(1, 2, 3), -1, 10), + (array(4, 5), -1, 20), + (array(1, 2, 3), -4, 10), + (NULL, -1, 10) + +query +SELECT array_insert(arr, pos, val) FROM test_ai_legacy_dispatch From 50cb062517db0ee9747ba5fc47665f434dd4f00e Mon Sep 17 00:00:00 2001 From: comphead Date: Mon, 6 Jul 2026 09:11:42 -0700 Subject: [PATCH 05/11] chore: fallback to Spark if legacy sql configurations are set --- .../user-guide/latest/compatibility/index.md | 99 +++++++++++++++++++ .../scala/org/apache/comet/CometConf.scala | 14 +++ .../comet/CometSparkSessionExtensions.scala | 21 ++++ .../org/apache/comet/LegacyConfFallback.scala | 85 ++++++++++++++++ .../scala/org/apache/comet/serde/arrays.scala | 9 +- .../org/apache/comet/serde/predicates.scala | 46 ++++++++- .../org/apache/comet/serde/strings.scala | 13 ++- .../org/apache/comet/serde/structs.scala | 17 +++- .../CometSparkSessionExtensionsSuite.scala | 35 +++++++ 9 files changed, 329 insertions(+), 10 deletions(-) create mode 100644 spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index f3b531a18f..be54e9f1eb 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -70,3 +70,102 @@ This is distinct from expressions that have **no** codegen-dispatch path: there, incompatible cases fall back to Spark by default, and `allowIncompatible=true` runs the native (incompatible) path instead. `cast` is the main example; see the [expression reference](../expressions.md) for which expressions have incompatible cases. + +## Spark legacy configs + +Spark exposes a family of `spark.sql.legacy.*` configs that opt a query into pre-modern Spark +semantics. Comet handles these in two ways: + +- **Per-expression**: when a legacy config affects a specific Spark expression that Comet + supports (for example `spark.sql.legacy.castComplexTypesToString.enabled` for `Cast`, + `spark.sql.legacy.negativeIndexInArrayInsert` for `array_insert`, + `spark.sql.legacy.nullInEmptyListBehavior` for `IN`), Comet's serde routes the expression + through the JVM codegen dispatcher (Spark's own `doGenCode` inside the Comet kernel) or + through a native code path that honors the flag. No session-wide fallback is triggered. +- **Session-wide execution fallback**: when a legacy config affects execution semantics but + is consumed by an analyzer/optimizer rule, a data-source reader/writer, or a type-system + utility (rather than a specific Comet-supported expression), Comet cannot fix the divergence + in a single serde. Instead, when + [`spark.comet.legacyConfFallback.enabled`](../configs.md) is `true` (default) and any config + in the curated list is set to a non-default value, Comet disables itself for the session so + Spark's own execution provides the legacy semantics. The warning names the offending config + keys. + +### Curated legacy configs that trigger the session-wide fallback + +The list below is the exact set checked by +`spark.comet.legacyConfFallback.enabled`. Each entry names the Spark config key and the value +Comet compares against. The comparison is case-insensitive, and the fallback only fires when +the key is explicitly set in the session AND its value differs from the recorded default. Keys +absent from the session conf never trigger the fallback, regardless of their runtime resolution +in Spark. The defaults recorded here are Spark 4.0's static defaults; when a Spark 4.0 default +depends on another config (for example ANSI mode), the value used is what Spark 4.0 itself +resolves to under its own defaults. + +**Decimal type-system / analyzer rules** + +| Config key | Comet-expected default | +| --- | --- | +| `spark.sql.legacy.allowNegativeScaleOfDecimal` | `false` | +| `spark.sql.legacy.decimal.retainFractionDigitsOnTruncate` | `false` | +| `spark.sql.legacy.literal.pickMinimumPrecision` | `true` | + +**Char/varchar padding and analyzer-inserted write-side validation** + +| Config key | Comet-expected default | +| --- | --- | +| `spark.sql.legacy.charVarcharAsString` | `false` | + +**Type coercion and upcast rules** + +| Config key | Comet-expected default | +| --- | --- | +| `spark.sql.legacy.doLooseUpcast` | `false` | +| `spark.sql.legacy.typeCoercion.datetimeToString.enabled` | `false` | + +**Optimizer rules that reshape plans handed to Comet** + +| Config key | Comet-expected default | +| --- | --- | +| `spark.sql.legacy.duplicateBetweenInput` | `false` | +| `spark.sql.legacy.inSubqueryNullability` | `false` | +| `spark.sql.legacy.scalarSubqueryCountBugBehavior` | `false` | +| `spark.sql.legacy.disableMapKeyNormalization` | `false` | +| `spark.sql.legacy.setopsPrecedence.enabled` | `false` | + +**View resolution (Cast vs. UpCast injection)** + +| Config key | Comet-expected default | +| --- | --- | +| `spark.sql.legacy.viewSchemaCompensation` | `true` | + +**Datetime parser policy** + +| Config key | Comet-expected default | +| --- | --- | +| `spark.sql.legacy.timeParserPolicy` | `CORRECTED` | + +**Parquet reader/writer semantics** + +| Config key | Comet-expected default | +| --- | --- | +| `spark.sql.legacy.parquet.datetimeRebaseModeInRead` | `CORRECTED` | +| `spark.sql.legacy.parquet.datetimeRebaseModeInWrite` | `CORRECTED` | +| `spark.sql.legacy.parquet.int96RebaseModeInRead` | `CORRECTED` | +| `spark.sql.legacy.parquet.int96RebaseModeInWrite` | `CORRECTED` | +| `spark.sql.legacy.parquet.nanosAsLong` | `false` | + +**Cached-plan behavior on file-source scans** + +| Config key | Comet-expected default | +| --- | --- | +| `spark.sql.legacy.readFileSourceTableCacheIgnoreOptions` | `false` | + +### Opting out of the session-wide fallback + +The fallback is on by default (`spark.comet.legacyConfFallback.enabled=true`). To keep Comet +enabled even when one of the configs above is set to a non-default value, set +`spark.comet.legacyConfFallback.enabled=false`. In that mode Comet's native operators do not +implement the legacy semantics the flag requests: results may silently diverge from Spark for +queries that touch the affected code paths. Spark compatibility is not guaranteed while the +opt-out is in effect. diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index 8e47151358..86553b0ead 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -93,6 +93,20 @@ object CometConf extends ShimCometConf { .booleanConf .createWithEnvVarOrDefault("ENABLE_COMET", true) + val COMET_LEGACY_CONF_FALLBACK_ENABLED: ConfigEntry[Boolean] = + conf("spark.comet.legacyConfFallback.enabled") + .category(CATEGORY_EXEC) + .doc( + "When true (default), Comet disables itself for the session if any spark.sql.legacy.* " + + "config that Comet does NOT already handle per-expression is set to a non-default " + + "value. Legacy configs consumed by specific Spark expressions are already routed " + + "through the JVM codegen dispatcher (or an explicit incompat check) inside Comet " + + "and do not trigger this fallback. Set this config to false to keep Comet enabled " + + "when other legacy configs are set; Spark compatibility is not guaranteed in that " + + "case.") + .booleanConf + .createWithDefault(true) + val COMET_NATIVE_SCAN_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.scan.enabled") .category(CATEGORY_TESTING) .doc("Whether to enable native scans. Intended for use in Comet's own test suites to " + diff --git a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala index 22c3c9c93e..66cf7013af 100644 --- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala +++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala @@ -145,6 +145,27 @@ object CometSparkSessionExtensions extends Logging { return false } + // Some spark.sql.legacy.* configs affect execution semantics for queries Comet accelerates + // but are not tied to a specific expression that Comet's serdes can gate on (parquet + // datetime rebase modes, decimal-precision analyzer rules, type-coercion policies, etc.). + // When any such config is set to a non-default value we disable Comet for the session so + // Spark's own execution provides the legacy semantics. The list is intentionally narrow -- + // legacy configs whose consumers ARE Comet-supported expressions (Cast, ArrayInsert, In, + // etc.) are handled per-expression via [[CodegenDispatchFallback]] and are NOT in this set. + if (COMET_LEGACY_CONF_FALLBACK_ENABLED.get(conf)) { + val triggered = LegacyConfFallback.triggeredConfigs(conf) + if (triggered.nonEmpty) { + val keys = triggered.toSeq.sorted.mkString(", ") + logWarning( + "Comet extension is disabled because the following execution-affecting " + + s"spark.sql.legacy.* configs are set to non-default values: $keys. Comet does not " + + "implement these legacy execution semantics. To keep Comet enabled anyway, set " + + s"${COMET_LEGACY_CONF_FALLBACK_ENABLED.key}=false (Spark compatibility is not " + + "guaranteed in that case).") + return false + } + } + try { // This will load the Comet native lib on demand, and if success, should set // `NativeBase.loaded` to true diff --git a/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala b/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala new file mode 100644 index 0000000000..1e92213179 --- /dev/null +++ b/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala @@ -0,0 +1,85 @@ +/* + * 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. + */ + +package org.apache.comet + +import org.apache.spark.sql.internal.SQLConf + +/** + * Curated set of Spark `spark.sql.legacy.*` configs whose behavior is NOT tied to a specific + * Comet-supported expression (built-in functions with a legacy dependency are handled + * per-expression via [[org.apache.comet.serde.CodegenDispatchFallback]] or a native passthrough + * in the serde). The keys in this list are consumed by analyzer/optimizer rules, data-source + * readers/writers, or type-system utilities, and Comet's native execution does not replicate + * their legacy semantics. + * + * When [[CometConf.COMET_LEGACY_CONF_FALLBACK_ENABLED]] is true (default), Comet disables itself + * for the session if any of these keys is set to its non-default value, so Spark's own execution + * path is used instead. Users can set `spark.comet.legacyConfFallback.enabled=false` to override + * the fallback and keep Comet enabled (Spark compatibility is not guaranteed in that case). + */ +private[comet] object LegacyConfFallback { + + /** + * Map of legacy config key -> case-insensitive Spark default value. A config triggers the + * fallback when it is present in the session conf AND its value is not equal (case-insensitive) + * to the default recorded here. + */ + val executionAffectingDefaults: Map[String, String] = Map( + // Decimal type-system / analyzer rules that reshape plans reaching Comet. + "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "false", + "spark.sql.legacy.decimal.retainFractionDigitsOnTruncate" -> "false", + "spark.sql.legacy.literal.pickMinimumPrecision" -> "false", + // Char/varchar padding + write-side validation inserted by the analyzer. + "spark.sql.legacy.charVarcharAsString" -> "false", + // Type-coercion / upcast rules. + "spark.sql.legacy.doLooseUpcast" -> "false", + "spark.sql.legacy.typeCoercion.datetimeToString.enabled" -> "false", + // Optimizer rules that reshape plans (subqueries, Between, empty-list IN nullability). + "spark.sql.legacy.duplicateBetweenInput" -> "false", + "spark.sql.legacy.inSubqueryNullability" -> "false", + "spark.sql.legacy.scalarSubqueryCountBugBehavior" -> "false", + // Map-key normalization used by CreateMap and friends inside ArrayBasedMapBuilder. + "spark.sql.legacy.disableMapKeyNormalization" -> "false", + // Set-op precedence changes the plan topology handed to Comet operators. + "spark.sql.legacy.setopsPrecedence.enabled" -> "false", + // View schema compensation controls whether Cast (Comet-supported) or UpCast (Comet + // unsupported) is injected during view resolution. + "spark.sql.legacy.viewSchemaCompensation" -> "true", + // Datetime parser policy affects CSV/JSON scan options and datetime formatters. + "spark.sql.legacy.timeParserPolicy" -> "CORRECTED", + // Datasource readers/writers Comet may accelerate. + "spark.sql.legacy.parquet.datetimeRebaseModeInRead" -> "CORRECTED", + "spark.sql.legacy.parquet.datetimeRebaseModeInWrite" -> "CORRECTED", + "spark.sql.legacy.parquet.int96RebaseModeInRead" -> "CORRECTED", + "spark.sql.legacy.parquet.int96RebaseModeInWrite" -> "CORRECTED", + "spark.sql.legacy.parquet.nanosAsLong" -> "false", + // Cached-plan behavior that leaves stale options on a Comet-accelerated file scan. + "spark.sql.legacy.readFileSourceTableCacheIgnoreOptions" -> "false") + + /** Keys in [[executionAffectingDefaults]] that are set to a non-default value on `conf`. */ + def triggeredConfigs(conf: SQLConf): Iterable[String] = { + executionAffectingDefaults.iterator.collect { + case (key, safeDefault) + if conf.contains(key) && + !conf.getConfString(key).equalsIgnoreCase(safeDefault) => + key + }.toSeq + } +} diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala b/spark/src/main/scala/org/apache/comet/serde/arrays.scala index b2f3df1ed4..30a77543ee 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -455,12 +455,13 @@ object CometArrayInsert extends CometExpressionSerde[ArrayInsert] with CodegenDi // position is interpreted. Rather than maintain a parallel native code path for the legacy // semantics, mark `array_insert` Incompatible when the flag is on so // [[CodegenDispatchFallback]] routes the expression through the JVM codegen dispatcher - // (Spark's own `doGenCode` inside the Comet kernel) — that gives Spark-exact results + // (Spark's own `doGenCode` inside the Comet kernel) -- that gives Spark-exact results // without duplicating the legacy branch natively. private val legacyNegativeIndexConfig = "spark.sql.legacy.negativeIndexInArrayInsert" private val legacyNegativeIndexReason = - s"`$legacyNegativeIndexConfig=true` legacy negative-index semantics are not implemented natively" + s"`$legacyNegativeIndexConfig=true` legacy negative-index semantics are not implemented" + + " natively" override def getIncompatibleReasons(): Seq[String] = Seq(legacyNegativeIndexReason) @@ -480,8 +481,8 @@ object CometArrayInsert extends CometExpressionSerde[ArrayInsert] with CodegenDi val posExprProto = exprToProtoInternal(expr.children(1), inputs, binding) val itemExprProto = exprToProtoInternal(expr.children(2), inputs, binding) // Reached in two cases: - // 1. Legacy conf is false → getSupportLevel returned Compatible → run native. - // 2. Legacy conf is true AND user set allowIncompatible=true → opt in to native. + // 1. Legacy conf is false -> getSupportLevel returned Compatible -> run native. + // 2. Legacy conf is true AND user set allowIncompatible=true -> opt in to native. // In case (2) the native impl honors the legacy semantics directly so we forward the flag. val legacyNegativeIndex = SQLConf.get.getConfString(legacyNegativeIndexConfig, "false").toBoolean diff --git a/spark/src/main/scala/org/apache/comet/serde/predicates.scala b/spark/src/main/scala/org/apache/comet/serde/predicates.scala index 63b64fbcf2..3aedae1851 100644 --- a/spark/src/main/scala/org/apache/comet/serde/predicates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/predicates.scala @@ -22,6 +22,7 @@ package org.apache.comet.serde import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualNullSafe, EqualTo, Expression, GreaterThan, GreaterThanOrEqual, In, InSet, IsNaN, IsNotNull, IsNull, LessThan, LessThanOrEqual, Literal, Not, Or} +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.BooleanType import org.apache.comet.CometSparkSessionExtensions.withFallbackReason @@ -233,7 +234,18 @@ object CometIsNaN extends CometExpressionSerde[IsNaN] { } } -object CometIn extends CometExpressionSerde[In] { +object CometIn extends CometExpressionSerde[In] with CodegenDispatchFallback { + + override def getIncompatibleReasons(): Seq[String] = Seq(LegacyConfHelpers.nullInEmptyListReason) + + override def getSupportLevel(expr: In): SupportLevel = { + if (expr.list.isEmpty && LegacyConfHelpers.legacyNullInEmptyBehavior) { + Incompatible(Some(LegacyConfHelpers.nullInEmptyListReason)) + } else { + Compatible() + } + } + override def convert( expr: In, inputs: Seq[Attribute], @@ -242,7 +254,18 @@ object CometIn extends CometExpressionSerde[In] { } } -object CometInSet extends CometExpressionSerde[InSet] { +object CometInSet extends CometExpressionSerde[InSet] with CodegenDispatchFallback { + + override def getIncompatibleReasons(): Seq[String] = Seq(LegacyConfHelpers.nullInEmptyListReason) + + override def getSupportLevel(expr: InSet): SupportLevel = { + if (expr.hset.isEmpty && LegacyConfHelpers.legacyNullInEmptyBehavior) { + Incompatible(Some(LegacyConfHelpers.nullInEmptyListReason)) + } else { + Compatible() + } + } + override def convert( expr: InSet, inputs: Seq[Attribute], @@ -257,6 +280,25 @@ object CometInSet extends CometExpressionSerde[InSet] { } } +private[serde] object LegacyConfHelpers { + + // Reason string shared with CometIn/CometInSet for the `null IN (empty)` divergence. + val nullInEmptyListReason: String = + "`spark.sql.legacy.nullInEmptyListBehavior=true` (or its effective default `!ansiEnabled`)" + + " changes `null IN (empty list)` from false to null; the native in-list path only" + + " implements the non-legacy semantics." + + // Resolve `spark.sql.legacy.nullInEmptyListBehavior` the same way Spark does: use the explicit + // value if set, otherwise fall back to `!ansiEnabled`. Read by string key to stay compatible + // with Spark versions where the accessor is not available. + def legacyNullInEmptyBehavior: Boolean = { + val conf = SQLConf.get + Option(conf.getConfString("spark.sql.legacy.nullInEmptyListBehavior", null)) + .map(_.equalsIgnoreCase("true")) + .getOrElse(!conf.ansiEnabled) + } +} + object ComparisonUtils { def in( 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 3186818c9c..f40b899b55 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -443,12 +443,15 @@ object CometRLike extends CometExpressionSerde[RLike] with NativeOptInAvailable private object PadReasons { val literalStrReason = "Scalar values are not supported for the `str` argument." val nonLiteralPadReason = "Only scalar values are supported for the `pad` argument." + val binaryStrReason: String = + "`spark.sql.legacy.lpadRpadAlwaysReturnString=true` allows lpad/rpad to run with a" + + " BinaryType `str` argument; Comet's native `lpad`/`rpad` only support string inputs." } object CometStringRPad extends CometExpressionSerde[StringRPad] { override def getUnsupportedReasons(): Seq[String] = - Seq(PadReasons.literalStrReason, PadReasons.nonLiteralPadReason) + Seq(PadReasons.literalStrReason, PadReasons.nonLiteralPadReason, PadReasons.binaryStrReason) override def getSupportLevel(expr: StringRPad): SupportLevel = { if (expr.str.isInstanceOf[Literal]) { @@ -457,6 +460,9 @@ object CometStringRPad extends CometExpressionSerde[StringRPad] { if (!expr.pad.isInstanceOf[Literal]) { return Unsupported(Some(PadReasons.nonLiteralPadReason)) } + if (expr.str.dataType == BinaryType) { + return Unsupported(Some(PadReasons.binaryStrReason)) + } Compatible() } @@ -476,7 +482,7 @@ object CometStringRPad extends CometExpressionSerde[StringRPad] { object CometStringLPad extends CometExpressionSerde[StringLPad] { override def getUnsupportedReasons(): Seq[String] = - Seq(PadReasons.literalStrReason, PadReasons.nonLiteralPadReason) + Seq(PadReasons.literalStrReason, PadReasons.nonLiteralPadReason, PadReasons.binaryStrReason) override def getSupportLevel(expr: StringLPad): SupportLevel = { if (expr.str.isInstanceOf[Literal]) { @@ -485,6 +491,9 @@ object CometStringLPad extends CometExpressionSerde[StringLPad] { if (!expr.pad.isInstanceOf[Literal]) { return Unsupported(Some(PadReasons.nonLiteralPadReason)) } + if (expr.str.dataType == BinaryType) { + return Unsupported(Some(PadReasons.binaryStrReason)) + } Compatible() } 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..c30fe1d67a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/structs.scala +++ b/spark/src/main/scala/org/apache/comet/serde/structs.scala @@ -259,13 +259,23 @@ object CometJsonToStructs extends CometCodegenDispatch[JsonToStructs] with Nativ } } -object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] { +object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] with CodegenDispatchFallback { private val incompatibleDataTypes = Seq(DateType, TimestampType, TimestampNTZType, BinaryType) + // When true, Spark's UnivocityGenerator wraps null values as quoted empty strings; Comet's + // native to_csv writer emits unquoted empty strings. Mark Incompatible so the + // CodegenDispatchFallback trait routes the expression through the JVM codegen dispatcher. + private val legacyNullValueConfKey = + "spark.sql.legacy.nullValueWrittenAsQuotedEmptyStringCsv" + private val legacyNullValueReason = + s"`$legacyNullValueConfKey=true` quotes NULLs as an empty quoted string in the CSV output;" + + " Comet's native `to_csv` writer does not implement that legacy behavior." + override def getIncompatibleReasons(): Seq[String] = Seq( "Date, Timestamp, TimestampNTZ, and Binary data types may produce different results" + - " (https://github.com/apache/datafusion-comet/issues/3232)") + " (https://github.com/apache/datafusion-comet/issues/3232)", + legacyNullValueReason) override def getUnsupportedReasons(): Seq[String] = Seq( "Complex types (arrays, maps, structs) in the schema are not supported") @@ -285,6 +295,9 @@ object CometStructsToCsv extends CometExpressionSerde[StructsToCsv] { s"The schema ${expr.inputSchema} is not supported because " + s"it includes a incompatible data types: $incompatibleDataTypes")) } + if (SQLConf.get.getConfString(legacyNullValueConfKey, "false").toBoolean) { + return Incompatible(Some(legacyNullValueReason)) + } // https://github.com/apache/datafusion-comet/issues/3232 Incompatible() } diff --git a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala index 268fdf94eb..e3358c6f77 100644 --- a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala @@ -53,6 +53,41 @@ class CometSparkSessionExtensionsSuite extends CometTestBase { NativeBase.setLoaded(true) } + test("isCometLoaded falls back when execution-affecting spark.sql.legacy.* config is set") { + val conf = new SQLConf + conf.setConfString(CometConf.COMET_ENABLED.key, "true") + conf.setConfString(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "false") + + // Baseline: no legacy configs set, Comet should load. + assert(isCometLoaded(conf)) + + // A single boolean-false-default execution-affecting legacy config triggers the fallback. + conf.setConfString("spark.sql.legacy.allowNegativeScaleOfDecimal", "true") + assert(!isCometLoaded(conf)) + + // Setting the config back to its Spark default (case-insensitive) clears the trigger. + conf.setConfString("spark.sql.legacy.allowNegativeScaleOfDecimal", "FALSE") + assert(isCometLoaded(conf)) + + // Enum-default configs also trigger when set to a non-default value. + conf.setConfString("spark.sql.legacy.timeParserPolicy", "LEGACY") + assert(!isCometLoaded(conf)) + conf.setConfString("spark.sql.legacy.timeParserPolicy", "CORRECTED") + assert(isCometLoaded(conf)) + + // Legacy configs handled per-expression (e.g. castComplexTypesToString) are NOT part of the + // fallback set and must not disable Comet on their own. + conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "true") + assert(isCometLoaded(conf)) + conf.unsetConf("spark.sql.legacy.castComplexTypesToString.enabled") + + // Opt-out: users can keep Comet enabled by disabling the fallback (compatibility not + // guaranteed). + conf.setConfString("spark.sql.legacy.allowNegativeScaleOfDecimal", "true") + conf.setConfString(CometConf.COMET_LEGACY_CONF_FALLBACK_ENABLED.key, "false") + assert(isCometLoaded(conf)) + } + test("isCometLoaded requires CometShuffleManager when shuffle.enabled=true") { val conf = new SQLConf conf.setConfString(CometConf.COMET_ENABLED.key, "true") From 4922a8af612a4f09af3556365c62d3858556bb1f Mon Sep 17 00:00:00 2001 From: comphead Date: Mon, 6 Jul 2026 09:43:32 -0700 Subject: [PATCH 06/11] chore: fallback to Spark if legacy sql configurations are set --- .../user-guide/latest/compatibility/index.md | 70 +++++++++---------- 1 file changed, 35 insertions(+), 35 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index be54e9f1eb..4db188f068 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -104,62 +104,62 @@ resolves to under its own defaults. **Decimal type-system / analyzer rules** -| Config key | Comet-expected default | -| --- | --- | -| `spark.sql.legacy.allowNegativeScaleOfDecimal` | `false` | -| `spark.sql.legacy.decimal.retainFractionDigitsOnTruncate` | `false` | -| `spark.sql.legacy.literal.pickMinimumPrecision` | `true` | +| Config key | Comet-expected default | +| --------------------------------------------------------- | ---------------------- | +| `spark.sql.legacy.allowNegativeScaleOfDecimal` | `false` | +| `spark.sql.legacy.decimal.retainFractionDigitsOnTruncate` | `false` | +| `spark.sql.legacy.literal.pickMinimumPrecision` | `true` | **Char/varchar padding and analyzer-inserted write-side validation** -| Config key | Comet-expected default | -| --- | --- | -| `spark.sql.legacy.charVarcharAsString` | `false` | +| Config key | Comet-expected default | +| -------------------------------------- | ---------------------- | +| `spark.sql.legacy.charVarcharAsString` | `false` | **Type coercion and upcast rules** -| Config key | Comet-expected default | -| --- | --- | -| `spark.sql.legacy.doLooseUpcast` | `false` | -| `spark.sql.legacy.typeCoercion.datetimeToString.enabled` | `false` | +| Config key | Comet-expected default | +| -------------------------------------------------------- | ---------------------- | +| `spark.sql.legacy.doLooseUpcast` | `false` | +| `spark.sql.legacy.typeCoercion.datetimeToString.enabled` | `false` | **Optimizer rules that reshape plans handed to Comet** -| Config key | Comet-expected default | -| --- | --- | -| `spark.sql.legacy.duplicateBetweenInput` | `false` | -| `spark.sql.legacy.inSubqueryNullability` | `false` | -| `spark.sql.legacy.scalarSubqueryCountBugBehavior` | `false` | -| `spark.sql.legacy.disableMapKeyNormalization` | `false` | -| `spark.sql.legacy.setopsPrecedence.enabled` | `false` | +| Config key | Comet-expected default | +| ------------------------------------------------- | ---------------------- | +| `spark.sql.legacy.duplicateBetweenInput` | `false` | +| `spark.sql.legacy.inSubqueryNullability` | `false` | +| `spark.sql.legacy.scalarSubqueryCountBugBehavior` | `false` | +| `spark.sql.legacy.disableMapKeyNormalization` | `false` | +| `spark.sql.legacy.setopsPrecedence.enabled` | `false` | **View resolution (Cast vs. UpCast injection)** -| Config key | Comet-expected default | -| --- | --- | -| `spark.sql.legacy.viewSchemaCompensation` | `true` | +| Config key | Comet-expected default | +| ----------------------------------------- | ---------------------- | +| `spark.sql.legacy.viewSchemaCompensation` | `true` | **Datetime parser policy** -| Config key | Comet-expected default | -| --- | --- | -| `spark.sql.legacy.timeParserPolicy` | `CORRECTED` | +| Config key | Comet-expected default | +| ----------------------------------- | ---------------------- | +| `spark.sql.legacy.timeParserPolicy` | `CORRECTED` | **Parquet reader/writer semantics** -| Config key | Comet-expected default | -| --- | --- | -| `spark.sql.legacy.parquet.datetimeRebaseModeInRead` | `CORRECTED` | -| `spark.sql.legacy.parquet.datetimeRebaseModeInWrite` | `CORRECTED` | -| `spark.sql.legacy.parquet.int96RebaseModeInRead` | `CORRECTED` | -| `spark.sql.legacy.parquet.int96RebaseModeInWrite` | `CORRECTED` | -| `spark.sql.legacy.parquet.nanosAsLong` | `false` | +| Config key | Comet-expected default | +| ---------------------------------------------------- | ---------------------- | +| `spark.sql.legacy.parquet.datetimeRebaseModeInRead` | `CORRECTED` | +| `spark.sql.legacy.parquet.datetimeRebaseModeInWrite` | `CORRECTED` | +| `spark.sql.legacy.parquet.int96RebaseModeInRead` | `CORRECTED` | +| `spark.sql.legacy.parquet.int96RebaseModeInWrite` | `CORRECTED` | +| `spark.sql.legacy.parquet.nanosAsLong` | `false` | **Cached-plan behavior on file-source scans** -| Config key | Comet-expected default | -| --- | --- | -| `spark.sql.legacy.readFileSourceTableCacheIgnoreOptions` | `false` | +| Config key | Comet-expected default | +| -------------------------------------------------------- | ---------------------- | +| `spark.sql.legacy.readFileSourceTableCacheIgnoreOptions` | `false` | ### Opting out of the session-wide fallback From d93e1c097c2d81d832cd3318218e3fda85efb795 Mon Sep 17 00:00:00 2001 From: comphead Date: Mon, 6 Jul 2026 10:13:45 -0700 Subject: [PATCH 07/11] chore: fallback to Spark if legacy sql configurations are set --- .../src/main/scala/org/apache/comet/serde/predicates.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/predicates.scala b/spark/src/main/scala/org/apache/comet/serde/predicates.scala index 3aedae1851..0f8c37350f 100644 --- a/spark/src/main/scala/org/apache/comet/serde/predicates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/predicates.scala @@ -236,7 +236,8 @@ object CometIsNaN extends CometExpressionSerde[IsNaN] { object CometIn extends CometExpressionSerde[In] with CodegenDispatchFallback { - override def getIncompatibleReasons(): Seq[String] = Seq(LegacyConfHelpers.nullInEmptyListReason) + override def getIncompatibleReasons(): Seq[String] = Seq( + LegacyConfHelpers.nullInEmptyListReason) override def getSupportLevel(expr: In): SupportLevel = { if (expr.list.isEmpty && LegacyConfHelpers.legacyNullInEmptyBehavior) { @@ -256,7 +257,8 @@ object CometIn extends CometExpressionSerde[In] with CodegenDispatchFallback { object CometInSet extends CometExpressionSerde[InSet] with CodegenDispatchFallback { - override def getIncompatibleReasons(): Seq[String] = Seq(LegacyConfHelpers.nullInEmptyListReason) + override def getIncompatibleReasons(): Seq[String] = Seq( + LegacyConfHelpers.nullInEmptyListReason) override def getSupportLevel(expr: InSet): SupportLevel = { if (expr.hset.isEmpty && LegacyConfHelpers.legacyNullInEmptyBehavior) { From 3da742cdd4fe8c5bd0db5aaedb653362e4d0c539 Mon Sep 17 00:00:00 2001 From: comphead Date: Mon, 6 Jul 2026 12:51:13 -0700 Subject: [PATCH 08/11] chore: fallback to Spark if legacy sql configurations are set --- docs/source/user-guide/latest/compatibility/index.md | 1 + .../main/scala/org/apache/comet/LegacyConfFallback.scala | 3 +++ .../expressions/aggregate/count_parameterless_legacy.sql | 6 +++--- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index 4db188f068..85dc0e411b 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -127,6 +127,7 @@ resolves to under its own defaults. | Config key | Comet-expected default | | ------------------------------------------------- | ---------------------- | +| `spark.sql.legacy.allowParameterlessCount` | `false` | | `spark.sql.legacy.duplicateBetweenInput` | `false` | | `spark.sql.legacy.inSubqueryNullability` | `false` | | `spark.sql.legacy.scalarSubqueryCountBugBehavior` | `false` | diff --git a/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala b/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala index 1e92213179..a8c3962311 100644 --- a/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala +++ b/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala @@ -48,6 +48,9 @@ private[comet] object LegacyConfFallback { "spark.sql.legacy.literal.pickMinimumPrecision" -> "false", // Char/varchar padding + write-side validation inserted by the analyzer. "spark.sql.legacy.charVarcharAsString" -> "false", + // Analyzer rule that lets `count()` be treated as `count(*)`; the native planner has no + // representation for a Count with zero children. + "spark.sql.legacy.allowParameterlessCount" -> "false", // Type-coercion / upcast rules. "spark.sql.legacy.doLooseUpcast" -> "false", "spark.sql.legacy.typeCoercion.datetimeToString.enabled" -> "false", diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql index b47476b0f5..018e190754 100644 --- a/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql @@ -30,16 +30,16 @@ statement INSERT INTO test_count_parameterless VALUES (1, 'x'), (2, 'x'), (NULL, 'y'), (3, 'y'), (NULL, 'y') -- Parameterless count() falls back to Spark; the aggregate result must still match Spark. -query expect_fallback(spark.sql.legacy.allowParameterlessCount=true) +query spark_answer_only SELECT count() FROM test_count_parameterless -- Parameterless count() with GROUP BY. -query expect_fallback(spark.sql.legacy.allowParameterlessCount=true) +query spark_answer_only SELECT grp, count() FROM test_count_parameterless GROUP BY grp ORDER BY grp -- Parameterless count() on empty table. statement CREATE TABLE test_count_empty(i int) USING parquet -query expect_fallback(spark.sql.legacy.allowParameterlessCount=true) +query spark_answer_only SELECT count() FROM test_count_empty From b739a6704519d80a8c535e0dee0c04d32b241cdf Mon Sep 17 00:00:00 2001 From: comphead Date: Mon, 6 Jul 2026 13:30:41 -0700 Subject: [PATCH 09/11] chore: fallback to Spark if legacy sql configurations are set --- .../org/apache/comet/LegacyConfFallback.scala | 44 +++------- .../org/apache/comet/serde/aggregates.scala | 21 ----- .../comet/shims/ShimLegacyConfFallback.scala | 54 +++++++++++++ .../comet/shims/ShimLegacyConfFallback.scala | 80 +++++++++++++++++++ .../aggregate/count_parameterless_legacy.sql | 45 ----------- 5 files changed, 143 insertions(+), 101 deletions(-) create mode 100644 spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala create mode 100644 spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala delete mode 100644 spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql diff --git a/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala b/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala index a8c3962311..9c2788c271 100644 --- a/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala +++ b/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala @@ -21,6 +21,8 @@ package org.apache.comet import org.apache.spark.sql.internal.SQLConf +import org.apache.comet.shims.ShimLegacyConfFallback + /** * Curated set of Spark `spark.sql.legacy.*` configs whose behavior is NOT tied to a specific * Comet-supported expression (built-in functions with a legacy dependency are handled @@ -33,48 +35,20 @@ import org.apache.spark.sql.internal.SQLConf * for the session if any of these keys is set to its non-default value, so Spark's own execution * path is used instead. Users can set `spark.comet.legacyConfFallback.enabled=false` to override * the fallback and keep Comet enabled (Spark compatibility is not guaranteed in that case). + * + * The map of legacy key -> Spark 4 default value comes from [[ShimLegacyConfFallback]]. The 4.x + * shim derives defaults from live [[org.apache.spark.internal.config.ConfigEntry]] references so + * additions/removals in Spark 4 are picked up automatically; the 3.x shim hardcodes the same + * defaults because several of these keys do not exist as ConfigEntry instances in Spark 3. */ -private[comet] object LegacyConfFallback { +private[comet] object LegacyConfFallback extends ShimLegacyConfFallback { /** * Map of legacy config key -> case-insensitive Spark default value. A config triggers the * fallback when it is present in the session conf AND its value is not equal (case-insensitive) * to the default recorded here. */ - val executionAffectingDefaults: Map[String, String] = Map( - // Decimal type-system / analyzer rules that reshape plans reaching Comet. - "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "false", - "spark.sql.legacy.decimal.retainFractionDigitsOnTruncate" -> "false", - "spark.sql.legacy.literal.pickMinimumPrecision" -> "false", - // Char/varchar padding + write-side validation inserted by the analyzer. - "spark.sql.legacy.charVarcharAsString" -> "false", - // Analyzer rule that lets `count()` be treated as `count(*)`; the native planner has no - // representation for a Count with zero children. - "spark.sql.legacy.allowParameterlessCount" -> "false", - // Type-coercion / upcast rules. - "spark.sql.legacy.doLooseUpcast" -> "false", - "spark.sql.legacy.typeCoercion.datetimeToString.enabled" -> "false", - // Optimizer rules that reshape plans (subqueries, Between, empty-list IN nullability). - "spark.sql.legacy.duplicateBetweenInput" -> "false", - "spark.sql.legacy.inSubqueryNullability" -> "false", - "spark.sql.legacy.scalarSubqueryCountBugBehavior" -> "false", - // Map-key normalization used by CreateMap and friends inside ArrayBasedMapBuilder. - "spark.sql.legacy.disableMapKeyNormalization" -> "false", - // Set-op precedence changes the plan topology handed to Comet operators. - "spark.sql.legacy.setopsPrecedence.enabled" -> "false", - // View schema compensation controls whether Cast (Comet-supported) or UpCast (Comet - // unsupported) is injected during view resolution. - "spark.sql.legacy.viewSchemaCompensation" -> "true", - // Datetime parser policy affects CSV/JSON scan options and datetime formatters. - "spark.sql.legacy.timeParserPolicy" -> "CORRECTED", - // Datasource readers/writers Comet may accelerate. - "spark.sql.legacy.parquet.datetimeRebaseModeInRead" -> "CORRECTED", - "spark.sql.legacy.parquet.datetimeRebaseModeInWrite" -> "CORRECTED", - "spark.sql.legacy.parquet.int96RebaseModeInRead" -> "CORRECTED", - "spark.sql.legacy.parquet.int96RebaseModeInWrite" -> "CORRECTED", - "spark.sql.legacy.parquet.nanosAsLong" -> "false", - // Cached-plan behavior that leaves stale options on a Comet-accelerated file scan. - "spark.sql.legacy.readFileSourceTableCacheIgnoreOptions" -> "false") + val executionAffectingDefaults: Map[String, String] = legacyConfDefaults /** Keys in [[executionAffectingDefaults]] that are set to a non-default value on `conf`. */ def triggeredConfigs(conf: SQLConf): Iterable[String] = { diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala index f15e0f0319..78cf9e36ee 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -107,27 +107,6 @@ object CometMax extends CometAggregateExpressionSerde[Max] { object CometCount extends CometAggregateExpressionSerde[Count] { - // When `spark.sql.legacy.allowParameterlessCount=true`, Spark allows `count()` with no - // arguments and treats it as `count(*)`. Comet's native planner asserts on non-empty children - // and would panic on such an expression, so mark it Unsupported here and let the aggregate fall - // back to Spark. Aggregate serdes have no [[CodegenDispatchFallback]] path (aggregates cannot - // be routed through the JVM codegen dispatcher), so a clean Spark fallback is the appropriate - // outcome. Under the default config value, Spark's analyzer rejects parameterless `count()` so - // this branch is unreachable. - private val legacyAllowParameterlessCountReason: String = - "`spark.sql.legacy.allowParameterlessCount=true` produces `count()` with no children, which " + - "the native planner does not support" - - override def getUnsupportedReasons(): Seq[String] = Seq(legacyAllowParameterlessCountReason) - - override def getSupportLevel(expr: Count): SupportLevel = { - if (expr.children.isEmpty) { - Unsupported(Some(legacyAllowParameterlessCountReason)) - } else { - Compatible() - } - } - override def convert( aggExpr: AggregateExpression, expr: Count, diff --git a/spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala b/spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala new file mode 100644 index 0000000000..f67c53294b --- /dev/null +++ b/spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala @@ -0,0 +1,54 @@ +/* + * 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. + */ + +package org.apache.comet.shims + +/** + * Spark 3.x variant: several entries in this map (view-schema compensation, decimal truncate, + * duplicate-between, scalar-subquery count-bug, disable-map-key-normalization, cache-ignore- + * options) do not exist as [[org.apache.spark.internal.config.ConfigEntry]] instances in Spark + * 3.5 or earlier, so the defaults are hardcoded. They match the Spark 4 defaults on purpose: + * the fallback rule is "when a legacy key is set to something other than the Spark 4 default, + * disable Comet". This keeps 3.x and 4.x behaviourally aligned even though 3.x can't derive + * the defaults live. See the 4.x shim for the reference-derived variant. + */ +trait ShimLegacyConfFallback { + + protected def legacyConfDefaults: Map[String, String] = Map( + "spark.sql.legacy.allowNegativeScaleOfDecimal" -> "false", + "spark.sql.legacy.decimal.retainFractionDigitsOnTruncate" -> "false", + "spark.sql.legacy.literal.pickMinimumPrecision" -> "true", + "spark.sql.legacy.charVarcharAsString" -> "false", + "spark.sql.legacy.allowParameterlessCount" -> "false", + "spark.sql.legacy.doLooseUpcast" -> "false", + "spark.sql.legacy.typeCoercion.datetimeToString.enabled" -> "false", + "spark.sql.legacy.duplicateBetweenInput" -> "false", + "spark.sql.legacy.inSubqueryNullability" -> "false", + "spark.sql.legacy.scalarSubqueryCountBugBehavior" -> "false", + "spark.sql.legacy.disableMapKeyNormalization" -> "false", + "spark.sql.legacy.setopsPrecedence.enabled" -> "false", + "spark.sql.legacy.viewSchemaCompensation" -> "true", + "spark.sql.legacy.timeParserPolicy" -> "CORRECTED", + "spark.sql.legacy.parquet.datetimeRebaseModeInRead" -> "CORRECTED", + "spark.sql.legacy.parquet.datetimeRebaseModeInWrite" -> "CORRECTED", + "spark.sql.legacy.parquet.int96RebaseModeInRead" -> "CORRECTED", + "spark.sql.legacy.parquet.int96RebaseModeInWrite" -> "CORRECTED", + "spark.sql.legacy.parquet.nanosAsLong" -> "false", + "spark.sql.legacy.readFileSourceTableCacheIgnoreOptions" -> "false") +} diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala new file mode 100644 index 0000000000..ec74b94ac3 --- /dev/null +++ b/spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala @@ -0,0 +1,80 @@ +/* + * 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. + */ + +package org.apache.comet.shims + +import org.apache.spark.internal.config.ConfigEntry +import org.apache.spark.sql.internal.SQLConf + +/** + * Spark 4.x variant: defaults are read from live `ConfigEntry` references so this file stays in + * sync with Spark 4 without duplicated string literals. See [[ShimLegacyConfFallback]] on 3.x for + * the hardcoded counterpart. + * + * Every key returned here uses the `spark.sql.legacy.*` name that the check compares against. + * When a legacy key was removed in Spark 4 (the four parquet rebase aliases), we still ship the + * legacy key -> Spark 4 non-legacy default; on Spark 4 the check is effectively inert because + * `SQLConf.set` rejects removed keys, but the entry is kept so 3.x behaviour is consistent. + */ +trait ShimLegacyConfFallback { + + private def entryDefault(entry: ConfigEntry[_]): String = entry.defaultValueString + + protected def legacyConfDefaults: Map[String, String] = Map( + "spark.sql.legacy.allowNegativeScaleOfDecimal" -> + entryDefault(SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED), + "spark.sql.legacy.decimal.retainFractionDigitsOnTruncate" -> + entryDefault(SQLConf.LEGACY_RETAIN_FRACTION_DIGITS_FIRST), + "spark.sql.legacy.literal.pickMinimumPrecision" -> + entryDefault(SQLConf.LITERAL_PICK_MINIMUM_PRECISION), + "spark.sql.legacy.charVarcharAsString" -> + entryDefault(SQLConf.LEGACY_CHAR_VARCHAR_AS_STRING), + "spark.sql.legacy.allowParameterlessCount" -> + entryDefault(SQLConf.ALLOW_PARAMETERLESS_COUNT), + "spark.sql.legacy.doLooseUpcast" -> + entryDefault(SQLConf.LEGACY_LOOSE_UPCAST), + "spark.sql.legacy.typeCoercion.datetimeToString.enabled" -> + entryDefault(SQLConf.LEGACY_CAST_DATETIME_TO_STRING), + "spark.sql.legacy.duplicateBetweenInput" -> + entryDefault(SQLConf.LEGACY_DUPLICATE_BETWEEN_INPUT), + "spark.sql.legacy.inSubqueryNullability" -> + entryDefault(SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY), + "spark.sql.legacy.scalarSubqueryCountBugBehavior" -> + entryDefault(SQLConf.LEGACY_SCALAR_SUBQUERY_COUNT_BUG_HANDLING), + "spark.sql.legacy.disableMapKeyNormalization" -> + entryDefault(SQLConf.DISABLE_MAP_KEY_NORMALIZATION), + "spark.sql.legacy.setopsPrecedence.enabled" -> + entryDefault(SQLConf.LEGACY_SETOPS_PRECEDENCE_ENABLED), + "spark.sql.legacy.viewSchemaCompensation" -> + entryDefault(SQLConf.VIEW_SCHEMA_COMPENSATION), + "spark.sql.legacy.timeParserPolicy" -> + entryDefault(SQLConf.LEGACY_TIME_PARSER_POLICY), + "spark.sql.legacy.parquet.datetimeRebaseModeInRead" -> + entryDefault(SQLConf.PARQUET_REBASE_MODE_IN_READ), + "spark.sql.legacy.parquet.datetimeRebaseModeInWrite" -> + entryDefault(SQLConf.PARQUET_REBASE_MODE_IN_WRITE), + "spark.sql.legacy.parquet.int96RebaseModeInRead" -> + entryDefault(SQLConf.PARQUET_INT96_REBASE_MODE_IN_READ), + "spark.sql.legacy.parquet.int96RebaseModeInWrite" -> + entryDefault(SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE), + "spark.sql.legacy.parquet.nanosAsLong" -> + entryDefault(SQLConf.LEGACY_PARQUET_NANOS_AS_LONG), + "spark.sql.legacy.readFileSourceTableCacheIgnoreOptions" -> + entryDefault(SQLConf.READ_FILE_SOURCE_TABLE_CACHE_IGNORE_OPTIONS)) +} diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql deleted file mode 100644 index 018e190754..0000000000 --- a/spark/src/test/resources/sql-tests/expressions/aggregate/count_parameterless_legacy.sql +++ /dev/null @@ -1,45 +0,0 @@ --- 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. - --- When `spark.sql.legacy.allowParameterlessCount=true`, Spark accepts `count()` (no arguments) --- and treats it as `count(*)`. Comet's native planner asserts non-empty children on Count, so --- `CometCount.getSupportLevel` marks parameterless Count `Unsupported` and lets the aggregate --- fall back to Spark. Aggregate serdes do not have a JVM codegen dispatcher path, so the --- Spark fallback is the correct outcome. - --- Config: spark.sql.legacy.allowParameterlessCount=true - -statement -CREATE TABLE test_count_parameterless(i int, grp string) USING parquet - -statement -INSERT INTO test_count_parameterless VALUES (1, 'x'), (2, 'x'), (NULL, 'y'), (3, 'y'), (NULL, 'y') - --- Parameterless count() falls back to Spark; the aggregate result must still match Spark. -query spark_answer_only -SELECT count() FROM test_count_parameterless - --- Parameterless count() with GROUP BY. -query spark_answer_only -SELECT grp, count() FROM test_count_parameterless GROUP BY grp ORDER BY grp - --- Parameterless count() on empty table. -statement -CREATE TABLE test_count_empty(i int) USING parquet - -query spark_answer_only -SELECT count() FROM test_count_empty From e0a75bf433a8df413b28108c9943f735cf8322f1 Mon Sep 17 00:00:00 2001 From: comphead Date: Mon, 6 Jul 2026 13:41:25 -0700 Subject: [PATCH 10/11] chore: fallback to Spark if legacy sql configurations are set --- .../org/apache/comet/shims/ShimLegacyConfFallback.scala | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala b/spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala index f67c53294b..5cc466389b 100644 --- a/spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala +++ b/spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala @@ -23,10 +23,10 @@ package org.apache.comet.shims * Spark 3.x variant: several entries in this map (view-schema compensation, decimal truncate, * duplicate-between, scalar-subquery count-bug, disable-map-key-normalization, cache-ignore- * options) do not exist as [[org.apache.spark.internal.config.ConfigEntry]] instances in Spark - * 3.5 or earlier, so the defaults are hardcoded. They match the Spark 4 defaults on purpose: - * the fallback rule is "when a legacy key is set to something other than the Spark 4 default, - * disable Comet". This keeps 3.x and 4.x behaviourally aligned even though 3.x can't derive - * the defaults live. See the 4.x shim for the reference-derived variant. + * 3.5 or earlier, so the defaults are hardcoded. They match the Spark 4 defaults on purpose: the + * fallback rule is "when a legacy key is set to something other than the Spark 4 default, disable + * Comet". This keeps 3.x and 4.x behaviourally aligned even though 3.x can't derive the defaults + * live. See the 4.x shim for the reference-derived variant. */ trait ShimLegacyConfFallback { From b3345d2c3e804d53098d959ee700020b8364c58d Mon Sep 17 00:00:00 2001 From: comphead Date: Mon, 6 Jul 2026 15:24:44 -0700 Subject: [PATCH 11/11] chore: fallback to Spark if legacy sql configurations are set --- .../comet/shims/ShimLegacyConfFallback.scala | 47 ++++++++++--------- 1 file changed, 24 insertions(+), 23 deletions(-) diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala index ec74b94ac3..6fbf91d070 100644 --- a/spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala +++ b/spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala @@ -19,7 +19,6 @@ package org.apache.comet.shims -import org.apache.spark.internal.config.ConfigEntry import org.apache.spark.sql.internal.SQLConf /** @@ -31,50 +30,52 @@ import org.apache.spark.sql.internal.SQLConf * When a legacy key was removed in Spark 4 (the four parquet rebase aliases), we still ship the * legacy key -> Spark 4 non-legacy default; on Spark 4 the check is effectively inert because * `SQLConf.set` rejects removed keys, but the entry is kept so 3.x behaviour is consistent. + * + * Note: `ConfigEntry` itself is `private[spark]`, so we never spell the type — every value below + * is a direct method call on a `SQLConf` val, and the compiler resolves `defaultValueString` + * without exposing the type name to this compilation unit. */ trait ShimLegacyConfFallback { - private def entryDefault(entry: ConfigEntry[_]): String = entry.defaultValueString - protected def legacyConfDefaults: Map[String, String] = Map( "spark.sql.legacy.allowNegativeScaleOfDecimal" -> - entryDefault(SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED), + SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED.defaultValueString, "spark.sql.legacy.decimal.retainFractionDigitsOnTruncate" -> - entryDefault(SQLConf.LEGACY_RETAIN_FRACTION_DIGITS_FIRST), + SQLConf.LEGACY_RETAIN_FRACTION_DIGITS_FIRST.defaultValueString, "spark.sql.legacy.literal.pickMinimumPrecision" -> - entryDefault(SQLConf.LITERAL_PICK_MINIMUM_PRECISION), + SQLConf.LITERAL_PICK_MINIMUM_PRECISION.defaultValueString, "spark.sql.legacy.charVarcharAsString" -> - entryDefault(SQLConf.LEGACY_CHAR_VARCHAR_AS_STRING), + SQLConf.LEGACY_CHAR_VARCHAR_AS_STRING.defaultValueString, "spark.sql.legacy.allowParameterlessCount" -> - entryDefault(SQLConf.ALLOW_PARAMETERLESS_COUNT), + SQLConf.ALLOW_PARAMETERLESS_COUNT.defaultValueString, "spark.sql.legacy.doLooseUpcast" -> - entryDefault(SQLConf.LEGACY_LOOSE_UPCAST), + SQLConf.LEGACY_LOOSE_UPCAST.defaultValueString, "spark.sql.legacy.typeCoercion.datetimeToString.enabled" -> - entryDefault(SQLConf.LEGACY_CAST_DATETIME_TO_STRING), + SQLConf.LEGACY_CAST_DATETIME_TO_STRING.defaultValueString, "spark.sql.legacy.duplicateBetweenInput" -> - entryDefault(SQLConf.LEGACY_DUPLICATE_BETWEEN_INPUT), + SQLConf.LEGACY_DUPLICATE_BETWEEN_INPUT.defaultValueString, "spark.sql.legacy.inSubqueryNullability" -> - entryDefault(SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY), + SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.defaultValueString, "spark.sql.legacy.scalarSubqueryCountBugBehavior" -> - entryDefault(SQLConf.LEGACY_SCALAR_SUBQUERY_COUNT_BUG_HANDLING), + SQLConf.LEGACY_SCALAR_SUBQUERY_COUNT_BUG_HANDLING.defaultValueString, "spark.sql.legacy.disableMapKeyNormalization" -> - entryDefault(SQLConf.DISABLE_MAP_KEY_NORMALIZATION), + SQLConf.DISABLE_MAP_KEY_NORMALIZATION.defaultValueString, "spark.sql.legacy.setopsPrecedence.enabled" -> - entryDefault(SQLConf.LEGACY_SETOPS_PRECEDENCE_ENABLED), + SQLConf.LEGACY_SETOPS_PRECEDENCE_ENABLED.defaultValueString, "spark.sql.legacy.viewSchemaCompensation" -> - entryDefault(SQLConf.VIEW_SCHEMA_COMPENSATION), + SQLConf.VIEW_SCHEMA_COMPENSATION.defaultValueString, "spark.sql.legacy.timeParserPolicy" -> - entryDefault(SQLConf.LEGACY_TIME_PARSER_POLICY), + SQLConf.LEGACY_TIME_PARSER_POLICY.defaultValueString, "spark.sql.legacy.parquet.datetimeRebaseModeInRead" -> - entryDefault(SQLConf.PARQUET_REBASE_MODE_IN_READ), + SQLConf.PARQUET_REBASE_MODE_IN_READ.defaultValueString, "spark.sql.legacy.parquet.datetimeRebaseModeInWrite" -> - entryDefault(SQLConf.PARQUET_REBASE_MODE_IN_WRITE), + SQLConf.PARQUET_REBASE_MODE_IN_WRITE.defaultValueString, "spark.sql.legacy.parquet.int96RebaseModeInRead" -> - entryDefault(SQLConf.PARQUET_INT96_REBASE_MODE_IN_READ), + SQLConf.PARQUET_INT96_REBASE_MODE_IN_READ.defaultValueString, "spark.sql.legacy.parquet.int96RebaseModeInWrite" -> - entryDefault(SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE), + SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.defaultValueString, "spark.sql.legacy.parquet.nanosAsLong" -> - entryDefault(SQLConf.LEGACY_PARQUET_NANOS_AS_LONG), + SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.defaultValueString, "spark.sql.legacy.readFileSourceTableCacheIgnoreOptions" -> - entryDefault(SQLConf.READ_FILE_SOURCE_TABLE_CACHE_IGNORE_OPTIONS)) + SQLConf.READ_FILE_SOURCE_TABLE_CACHE_IGNORE_OPTIONS.defaultValueString) }