chore: fallback to Spark if legacy sql configurations are set

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.

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

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

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
+  }
+}

spark/src/main/scala/org/apache/comet/expressions/CometCast.scala

@@ -25,24 +25,29 @@ 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 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 {

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,

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()