diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index f3b531a18f..4db188f068 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/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala index 619b69912f..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,15 @@ 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 = @@ -38,11 +41,13 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { // When `spark.sql.legacy.castComplexTypesToString.enabled` is true, Spark wraps maps and // structs with `[]` (instead of `{}`) when casting to string, and omits NULL elements of - // structs/maps/arrays (instead of rendering them as the literal "null"). Comet only - // implements the default formatting, so fall back to Spark for any array/map/struct to-string - // cast when the flag is enabled. The flag is internal in Spark 4.0 and defaults to false. + // structs/maps/arrays (instead of rendering them as the literal "null"). Comet's native cast + // only implements the default formatting, so 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" + "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported natively" private def legacyCastComplexTypesToString: Boolean = SQLConf.get @@ -166,7 +171,7 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim { if (toType == DataTypes.StringType && legacyCastComplexTypesToString && (fromType .isInstanceOf[ArrayType] || fromType.isInstanceOf[StructType] || fromType.isInstanceOf[MapType])) { - return Unsupported(Some(legacyCastComplexTypesToStringReason)) + return Incompatible(Some(legacyCastComplexTypesToStringReason)) } (fromType, toType) match { 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 8eda097ce6..30a77543ee 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -449,9 +449,29 @@ 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, @@ -460,8 +480,12 @@ 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("spark.sql.legacy.negativeIndexInArrayInsert").toBoolean + SQLConf.get.getConfString(legacyNegativeIndexConfig, "false").toBoolean if (srcExprProto.isDefined && posExprProto.isDefined && itemExprProto.isDefined) { val arrayInsertBuilder = ExprOuterClass.ArrayInsert .newBuilder() 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..e5b431220a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/maps.scala +++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala @@ -140,12 +140,17 @@ 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, 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. + // 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 = 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..0f8c37350f 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,19 @@ 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 +255,19 @@ 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 +282,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/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 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/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")