diff --git a/dev/diffs/4.0.2.diff b/dev/diffs/4.0.2.diff
index d8e506b631..882f8128f4 100644
--- a/dev/diffs/4.0.2.diff
+++ b/dev/diffs/4.0.2.diff
@@ -1374,6 +1374,28 @@ index 2e33f6505ab..be967565303 100644
      }
  
      withTable("t1", "t2") {
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+index a40e34d94d0..abc1f035d15 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+@@ -21,6 +21,7 @@ import org.apache.spark.sql.QueryTest.sameRows
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+ import org.apache.spark.sql.catalyst.expressions.variant.{ToVariantObject, VariantExpressionEvalUtils}
++import org.apache.spark.sql.comet.CometNativeColumnarToRowExec
+ import org.apache.spark.sql.execution.WholeStageCodegenExec
+ import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+ import org.apache.spark.sql.functions._
+@@ -358,7 +359,8 @@ class VariantEndToEndSuite extends QueryTest with SharedSparkSession {
+             s"cast(to_variant_object(s) as ${schema(2).dataType.sql})")
+           checkAnswer(df, input)
+           val plan = df.queryExecution.executedPlan
+-          assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY"))
++          assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY")
++            || plan.isInstanceOf[CometNativeColumnarToRowExec])
+         }
+       }
+     }
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
 index 11e9547dfc5..637411056ae 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
diff --git a/dev/diffs/4.1.2.diff b/dev/diffs/4.1.2.diff
index 9a1b30b010..e56a3ed22d 100644
--- a/dev/diffs/4.1.2.diff
+++ b/dev/diffs/4.1.2.diff
@@ -1473,6 +1473,28 @@ index 3ba48da0e32..0313aaa9ec6 100644
      }
  
      withTable("t1", "t2") {
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+index 8a0e2c29653..d276a51cbc6 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+@@ -21,6 +21,7 @@ import org.apache.spark.sql.QueryTest.sameRows
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+ import org.apache.spark.sql.catalyst.expressions.variant.{ToVariantObject, VariantExpressionEvalUtils}
++import org.apache.spark.sql.comet.CometNativeColumnarToRowExec
+ import org.apache.spark.sql.execution.WholeStageCodegenExec
+ import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+ import org.apache.spark.sql.functions._
+@@ -358,7 +359,8 @@ class VariantEndToEndSuite extends QueryTest with SharedSparkSession {
+             s"cast(to_variant_object(s) as ${schema(2).dataType.sql})")
+           checkAnswer(df, input)
+           val plan = df.queryExecution.executedPlan
+-          assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY"))
++          assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode == "CODEGEN_ONLY")
++            || plan.isInstanceOf[CometNativeColumnarToRowExec])
+         }
+       }
+     }
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
 index fee375db10a..8c2c24e2c5f 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md
index b08d9a3bdb..c61f9cf7a0 100644
--- a/docs/source/user-guide/latest/compatibility/index.md
+++ b/docs/source/user-guide/latest/compatibility/index.md
@@ -102,3 +102,113 @@ string paths is tracked by
 Separately, Comet's native Parquet scan currently rejects string columns whose stored bytes are not
 valid UTF-8 rather than reading them like Spark
 ([#4121](https://github.com/apache/datafusion-comet/issues/4121)).
+
+## 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`, `spark.sql.legacy.timeParserPolicy`
+  for datetime parsing expressions, `spark.sql.legacy.sizeOfNull` for `size`/`cardinality`,
+  `spark.sql.legacy.followThreeValuedLogicInArrayExists` for `exists`), 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.decimal.retainFractionDigitsOnTruncate` | `false`                |
+| `spark.sql.legacy.literal.pickMinimumPrecision`           | `true`                 |
+
+Note: `spark.sql.legacy.allowNegativeScaleOfDecimal` is intentionally NOT in this list.
+Negative-scale decimals are handled per-expression — `CometCast.isSupported` returns
+`Incompatible` when the flag is `false` and `Compatible` when the user opts in — so enabling
+the legacy flag does not disable Comet for the whole session.
+
+**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.allowParameterlessCount`        | `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`                |
+
+**View resolution (Cast vs. UpCast injection)**
+
+| Config key                                | Comet-expected default |
+| ----------------------------------------- | ---------------------- |
+| `spark.sql.legacy.viewSchemaCompensation` | `true`                 |
+
+**Parquet reader semantics (per-scan, not session-wide)**
+
+Parquet legacy read configs affect how bytes on disk map to Spark rows, so a single non-default
+value would only ever change results for queries that read Parquet files. Instead of disabling
+Comet for the whole session, `CometScanRule` checks these keys per scan and falls back the
+individual scan to Spark. Non-Parquet queries in the same session continue to run on Comet.
+
+Both the primary name and the `spark.sql.legacy.*` alias are monitored (Spark's `SQLConf.contains`
+does NOT follow `withAlternative` links). Write-side rebase configs are intentionally not part
+of this check: they only affect writes and would not change scan output.
+
+| Config key                                          | Comet-expected default |
+| --------------------------------------------------- | ---------------------- |
+| `spark.sql.parquet.datetimeRebaseModeInRead`        | `CORRECTED`            |
+| `spark.sql.legacy.parquet.datetimeRebaseModeInRead` | `CORRECTED`            |
+| `spark.sql.parquet.int96RebaseModeInRead`           | `CORRECTED`            |
+| `spark.sql.legacy.parquet.int96RebaseModeInRead`    | `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 41c99d2724..3f57c4bd6d 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 202c52b158..1cb9df3f19 100644
--- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
+++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
@@ -150,6 +150,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.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..46e8d95619
--- /dev/null
+++ b/spark/src/main/scala/org/apache/comet/LegacyConfFallback.scala
@@ -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.
+ */
+
+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
+ * 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).
+ *
+ * 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 extends ShimLegacyConfFallback {
+
+  /**
+   * Keys in [[legacyConfDefaults]] whose value on `conf` differs (case-insensitive) from the
+   * recorded Spark default. Returned sorted so the warning message is deterministic.
+   */
+  def triggeredConfigs(conf: SQLConf): Seq[String] = {
+    legacyConfDefaults
+      .collect {
+        case (key, safeDefault)
+            if !conf.getConfString(key, safeDefault).equalsIgnoreCase(safeDefault) =>
+          key
+      }
+      .toSeq
+      .sorted
+  }
+}
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..3484651ece 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -25,30 +25,41 @@ import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType,
 
 import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, withFallbackReason}
-import org.apache.comet.serde.{CometExpressionSerde, Compatible, ExprOuterClass, Incompatible, SupportLevel, Unsupported}
+import org.apache.comet.DataTypeSupport.isComplexType
+import org.apache.comet.serde.{CodegenDispatchFallback, CometExpressionSerde, Compatible, ExprOuterClass, Incompatible, SupportLevel, Unsupported}
 import org.apache.comet.serde.ExprOuterClass.Expr
 import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProtoInternal, serializeDataType}
-import org.apache.comet.shims.CometExprShim
+import org.apache.comet.shims.{CometExprShim, CometTypeShim}
 
-object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
+object CometCast
+    extends CometExpressionSerde[Cast]
+    with CometExprShim
+    with CometTypeShim
+    with CodegenDispatchFallback {
 
   // Shared with CometCastSuite so the asserted reason cannot drift from production.
   private[comet] val negativeScaleDecimalToStringReason: String =
     "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"
+    "spark.sql.legacy.castComplexTypesToString.enabled=true is not supported natively"
+
+  private[comet] val nonDefaultTimeParserPolicyReason: String =
+    "spark.sql.legacy.timeParserPolicy is set to a non-CORRECTED value; the native " +
+      "string-to-datetime parser only implements CORRECTED semantics"
 
   private def legacyCastComplexTypesToString: Boolean =
     SQLConf.get
       .getConfString("spark.sql.legacy.castComplexTypesToString.enabled", "false")
       .toBoolean
 
+  // Non-CORRECTED policies (LEGACY, EXCEPTION) change string-to-date/timestamp parsing behavior in
+  // ways the native cast kernel does not replicate.
+  private def isNonDefaultTimeParserPolicy: Boolean =
+    !SQLConf.get
+      .getConfString("spark.sql.legacy.timeParserPolicy", "CORRECTED")
+      .equalsIgnoreCase("CORRECTED")
+
   def supportedTypes: Seq[DataType] =
     Seq(
       DataTypes.BooleanType,
@@ -72,6 +83,14 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
   // would only duplicate the matrix and risk drifting from it.
 
   override def getSupportLevel(cast: Cast): SupportLevel = {
+    // Reject `VariantType` before the Literal short-circuit below. Folding a Cast whose child or
+    // target is `VariantType` produces a `Literal[VariantType]` that no downstream Comet serde
+    // can serialize, and relying on `CometLiteral` to reject it after the fact leaves a native
+    // path that assumes the produced literal is safe. Guarding here (in addition to the
+    // recursive check in `isSupported`) forces Spark fallback for every VariantType cast shape.
+    if (isVariantType(cast.child.dataType) || isVariantType(cast.dataType)) {
+      return unsupported(cast.child.dataType, cast.dataType)
+    }
     if (cast.child.isInstanceOf[Literal]) {
       // A cast whose child is a literal is folded by Spark at planning time via `cast.eval()`
       // (see `convert`), so the cast never executes natively and the result matches Spark by
@@ -159,14 +178,22 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
       timeZoneId: Option[String],
       evalMode: CometEvalMode.Value): SupportLevel = {
 
+    // Spark 4's `VariantType` (SPARK-45827) has no native counterpart in Comet: serializing it
+    // into the DataFusion plan would fail in `serializeDataType`, and the codegen dispatcher
+    // cannot compile Variant read/write kernels either. Detect it via the version-shimmed
+    // `isVariantType` (which returns false on Spark 3.x where the class does not exist) and
+    // report `Unsupported` so the enclosing operator falls back to Spark.
+    if (isVariantType(fromType) || isVariantType(toType)) {
+      return unsupported(fromType, toType)
+    }
+
     if (fromType == toType) {
       return Compatible()
     }
 
-    if (toType == DataTypes.StringType && legacyCastComplexTypesToString && (fromType
-        .isInstanceOf[ArrayType] || fromType.isInstanceOf[StructType] ||
-        fromType.isInstanceOf[MapType])) {
-      return Unsupported(Some(legacyCastComplexTypesToStringReason))
+    if (toType == DataTypes.StringType && legacyCastComplexTypesToString && isComplexType(
+        fromType)) {
+      return Incompatible(Some(legacyCastComplexTypesToStringReason))
     }
 
     (fromType, toType) match {
@@ -248,6 +275,9 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
         Compatible()
       case _: DecimalType =>
         Compatible()
+      case DataTypes.DateType | DataTypes.TimestampType | _: TimestampNTZType
+          if isNonDefaultTimeParserPolicy =>
+        Incompatible(Some(nonDefaultTimeParserPolicyReason))
       case DataTypes.DateType =>
         // https://github.com/apache/datafusion-comet/issues/327
         Compatible(Some("Only supports years between 262143 BC and 262142 AD"))
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index 59331129d5..3557253a06 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -204,6 +204,12 @@ case class CometScanRule(session: SparkSession)
         s"Native Parquet scan requires ${COMET_EXEC_ENABLED.key} to be enabled")
       return None
     }
+    CometScanRule.parquetFallbackReason(conf) match {
+      case Some(reason) =>
+        withFallbackReason(scanExec, reason)
+        return None
+      case None =>
+    }
     // Comet's native readers go through object_store, which only understands a fixed set of URL
     // schemes. A custom Hadoop FileSystem (e.g. registered via spark.hadoop.fs.<scheme>.impl) would
     // surface at execution time as `Generic URL error: Unable to recognise URL "..."`. Decline here
@@ -844,6 +850,45 @@ object CometScanRule extends Logging {
   val SKIP_COMET_SCAN_TAG: org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit] =
     org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit]("comet.skipCometScan")
 
+  /**
+   * Parquet legacy configs that change how bytes on disk map to Spark rows/values. Set to a
+   * non-CORRECTED / non-false value, the native Parquet reader would silently return different
+   * results than Spark for the affected files. Comet monitors both the primary key
+   * (`spark.sql.parquet.*RebaseMode*`, Spark 3.2+) and its `spark.sql.legacy.*` alias since
+   * `SQLConf.contains` does NOT follow `withAlternative` links. Write-side rebase configs are
+   * intentionally omitted -- they only affect writes and shouldn't disqualify a scan.
+   */
+  private val parquetReadFallbackDefaults: Map[String, String] = Map(
+    "spark.sql.parquet.datetimeRebaseModeInRead" -> "CORRECTED",
+    "spark.sql.legacy.parquet.datetimeRebaseModeInRead" -> "CORRECTED",
+    "spark.sql.parquet.int96RebaseModeInRead" -> "CORRECTED",
+    "spark.sql.legacy.parquet.int96RebaseModeInRead" -> "CORRECTED",
+    "spark.sql.legacy.parquet.nanosAsLong" -> "false")
+
+  /**
+   * Returns a fallback reason when any monitored Parquet legacy config is explicitly set to a
+   * non-default value, or `None` otherwise. Called by [[nativeScan]] so the fallback only fires
+   * for the parquet scan it affects, not the whole session.
+   */
+  private[rules] def parquetFallbackReason(
+      conf: org.apache.spark.sql.internal.SQLConf): Option[String] = {
+    val triggered = parquetReadFallbackDefaults
+      .collect {
+        case (key, safeDefault)
+            if !conf.getConfString(key, safeDefault).equalsIgnoreCase(safeDefault) =>
+          key
+      }
+      .toSeq
+      .sorted
+    if (triggered.isEmpty) {
+      None
+    } else {
+      Some(
+        "Native Parquet scan does not implement the legacy semantics requested by " +
+          s"${triggered.mkString(", ")}; falling back to Spark for this scan.")
+    }
+  }
+
   /**
    * Single-pass validation of Iceberg FileScanTasks.
    *
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 8bc639021c..47e58332ee 100644
--- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
@@ -440,9 +440,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,
@@ -451,8 +471,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()
@@ -737,6 +761,10 @@ object CometArrayFilter extends CometExpressionSerde[ArrayFilter] {
   }
 }
 
+// `spark.sql.legacy.sizeOfNull` (default `true`, effective value `LEGACY_SIZE_OF_NULL &&
+// !ANSI_ENABLED`) is captured by Spark's `Size` constructor into the `legacySizeOfNull` field.
+// `convert` below reads that field and bakes the correct literal (`-1` or `null`) into the
+// CaseWhen's else branch, so both semantics run natively without a fallback.
 object CometSize extends CometExpressionSerde[Size] {
 
   override def getSupportLevel(expr: Size): SupportLevel = {
@@ -919,6 +947,10 @@ trait ArraysBase {
 
 object CometArrayTransform extends CometCodegenDispatch[ArrayTransform]
 
+// `spark.sql.legacy.followThreeValuedLogicInArrayExists` (default `true`) is captured by
+// Spark's `ArrayExists` constructor into the `followThreeValuedLogic` field. Routing through
+// codegen dispatch runs Spark's own `doGenCode`, which closes over that field, so both flag
+// values produce Spark-exact results without a per-serde gate.
 object CometArrayExists extends CometCodegenDispatch[ArrayExists]
 
 object CometArrayForAll extends CometCodegenDispatch[ArrayForAll]
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 c9a70a830f..b679e2c7fa 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, BinaryExpression, 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
@@ -243,8 +244,15 @@ object CometIsNaN extends CometExpressionSerde[IsNaN] {
 
 object CometIn extends CometExpressionSerde[In] with CodegenDispatchFallback {
 
+  override def getIncompatibleReasons(): Seq[String] = Seq(
+    LegacyConfHelpers.nullInEmptyListReason)
+
   override def getSupportLevel(expr: In): SupportLevel =
-    ComparisonUtils.collationSupportLevel("In", (expr.value +: expr.list): _*)
+    if (expr.list.isEmpty && LegacyConfHelpers.legacyNullInEmptyBehavior) {
+      Incompatible(Some(LegacyConfHelpers.nullInEmptyListReason))
+    } else {
+      ComparisonUtils.collationSupportLevel("In", (expr.value +: expr.list): _*)
+    }
 
   override def getUnsupportedReasons(): Seq[String] =
     Seq(ComparisonUtils.nonDefaultCollationDocReason)
@@ -259,8 +267,15 @@ object CometIn extends CometExpressionSerde[In] with CodegenDispatchFallback {
 
 object CometInSet extends CometExpressionSerde[InSet] with CodegenDispatchFallback {
 
+  override def getIncompatibleReasons(): Seq[String] = Seq(
+    LegacyConfHelpers.nullInEmptyListReason)
+
   override def getSupportLevel(expr: InSet): SupportLevel =
-    ComparisonUtils.collationSupportLevel("InSet", expr.child)
+    if (expr.hset.isEmpty && LegacyConfHelpers.legacyNullInEmptyBehavior) {
+      Incompatible(Some(LegacyConfHelpers.nullInEmptyListReason))
+    } else {
+      ComparisonUtils.collationSupportLevel("InSet", expr.child)
+    }
 
   override def getUnsupportedReasons(): Seq[String] =
     Seq(ComparisonUtils.nonDefaultCollationDocReason)
@@ -296,6 +311,25 @@ trait CollationAwareBinaryPredicate[T <: BinaryExpression]
     Seq(ComparisonUtils.nonDefaultCollationDocReason)
 }
 
+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 {
 
   // Comet's native equality/ordering/hashing compare raw bytes, so any predicate operand carrying
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 1e124dbd42..48ead53ad9 100644
--- a/spark/src/main/scala/org/apache/comet/serde/strings.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala
@@ -380,12 +380,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]) {
@@ -394,6 +397,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()
   }
 
@@ -413,7 +419,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]) {
@@ -422,6 +428,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/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala b/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala
index 97320be9e7..b71476c3dd 100644
--- a/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala
+++ b/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala
@@ -36,6 +36,9 @@ trait CometTypeShim {
   @nowarn // Spark 4 feature; Variant shredding doesn't exist in Spark 3.x.
   def isVariantStruct(s: StructType): Boolean = false
 
+  @nowarn // Spark 4 feature; VariantType doesn't exist in Spark 3.x.
+  def isVariantType(dt: DataType): Boolean = false
+
   @nowarn // Spark 4.1 feature; TimeType doesn't exist in Spark 3.x.
   def isTimeType(dt: DataType): Boolean = false
 }
diff --git a/spark/src/main/spark-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..568029df5d
--- /dev/null
+++ b/spark/src/main/spark-3.x/org/apache/comet/shims/ShimLegacyConfFallback.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.
+ *
+ * Parquet-specific legacy configs (`spark.sql.(legacy.)?parquet.*RebaseMode*`,
+ * `spark.sql.legacy.parquet.nanosAsLong`) are intentionally NOT in this session-wide set. They
+ * are checked per-scan in [[org.apache.comet.rules.CometScanRule.parquetFallbackReason]] and only
+ * fall back the scan they affect, so non-Parquet queries in the same session stay on Comet.
+ */
+trait ShimLegacyConfFallback {
+
+  protected def legacyConfDefaults: Map[String, String] = Map(
+    "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.readFileSourceTableCacheIgnoreOptions" -> "false")
+}
diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala
index 1d4a9f601e..f48955a7da 100644
--- a/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala
+++ b/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala
@@ -20,7 +20,7 @@
 package org.apache.comet.shims
 
 import org.apache.spark.sql.execution.datasources.VariantMetadata
-import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType, StructType}
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType, StructType, VariantType}
 
 trait CometTypeShim {
   // A `StringType` carries collation metadata in Spark 4.0. Only non-default (non-UTF8_BINARY)
@@ -54,6 +54,12 @@ trait CometTypeShim {
   // and force scan fallback.
   def isVariantStruct(s: StructType): Boolean = VariantMetadata.isVariantStruct(s)
 
+  // Comet has no native execution path for Spark 4's `VariantType` (introduced in
+  // SPARK-45827). Serdes call this to route casts/expressions touching the type back to Spark
+  // rather than serializing an unsupported datatype into the native plan. Stubbed to `false` in
+  // Spark 3.x where `VariantType` does not exist.
+  def isVariantType(dt: DataType): Boolean = dt.isInstanceOf[VariantType]
+
   def isTimeType(dt: DataType): Boolean =
     dt.getClass.getSimpleName.startsWith("TimeType")
 
diff --git a/spark/src/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..280b80721a
--- /dev/null
+++ b/spark/src/main/spark-4.x/org/apache/comet/shims/ShimLegacyConfFallback.scala
@@ -0,0 +1,72 @@
+/*
+ * 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.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, 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.
+ *
+ * Parquet-specific legacy configs (`spark.sql.(legacy.)?parquet.*RebaseMode*`,
+ * `spark.sql.legacy.parquet.nanosAsLong`) are intentionally NOT in this session-wide set. They
+ * are checked per-scan in [[org.apache.comet.rules.CometScanRule.parquetFallbackReason]] and only
+ * fall back the scan they affect, so non-Parquet queries in the same session stay on Comet.
+ *
+ * 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 {
+
+  protected def legacyConfDefaults: Map[String, String] = Map(
+    "spark.sql.legacy.decimal.retainFractionDigitsOnTruncate" ->
+      SQLConf.LEGACY_RETAIN_FRACTION_DIGITS_FIRST.defaultValueString,
+    "spark.sql.legacy.literal.pickMinimumPrecision" ->
+      SQLConf.LITERAL_PICK_MINIMUM_PRECISION.defaultValueString,
+    "spark.sql.legacy.charVarcharAsString" ->
+      SQLConf.LEGACY_CHAR_VARCHAR_AS_STRING.defaultValueString,
+    "spark.sql.legacy.allowParameterlessCount" ->
+      SQLConf.ALLOW_PARAMETERLESS_COUNT.defaultValueString,
+    "spark.sql.legacy.doLooseUpcast" ->
+      SQLConf.LEGACY_LOOSE_UPCAST.defaultValueString,
+    "spark.sql.legacy.typeCoercion.datetimeToString.enabled" ->
+      SQLConf.LEGACY_CAST_DATETIME_TO_STRING.defaultValueString,
+    "spark.sql.legacy.duplicateBetweenInput" ->
+      SQLConf.LEGACY_DUPLICATE_BETWEEN_INPUT.defaultValueString,
+    "spark.sql.legacy.inSubqueryNullability" ->
+      SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.defaultValueString,
+    "spark.sql.legacy.scalarSubqueryCountBugBehavior" ->
+      SQLConf.LEGACY_SCALAR_SUBQUERY_COUNT_BUG_HANDLING.defaultValueString,
+    "spark.sql.legacy.disableMapKeyNormalization" ->
+      SQLConf.DISABLE_MAP_KEY_NORMALIZATION.defaultValueString,
+    "spark.sql.legacy.setopsPrecedence.enabled" ->
+      SQLConf.LEGACY_SETOPS_PRECEDENCE_ENABLED.defaultValueString,
+    "spark.sql.legacy.viewSchemaCompensation" ->
+      SQLConf.VIEW_SCHEMA_COMPENSATION.defaultValueString,
+    "spark.sql.legacy.readFileSourceTableCacheIgnoreOptions" ->
+      SQLConf.READ_FILE_SOURCE_TABLE_CACHE_IGNORE_OPTIONS.defaultValueString)
+}
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<int>, 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/array/exists_legacy_three_valued.sql b/spark/src/test/resources/sql-tests/expressions/array/exists_legacy_three_valued.sql
new file mode 100644
index 0000000000..f9b1b6b110
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/array/exists_legacy_three_valued.sql
@@ -0,0 +1,72 @@
+-- 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.
+
+-- Regression coverage for `spark.sql.legacy.followThreeValuedLogicInArrayExists`. Spark's
+-- `ArrayExists` captures the flag into a boolean field at construction time
+-- (`ArrayExists.followThreeValuedLogic`). `CometArrayExists` is a `CometCodegenDispatch`, so
+-- Spark's own `doGenCode` runs inside the Comet kernel and closes over that field. Both flag
+-- values therefore produce Spark-exact results without any serde-level gate.
+--
+-- The config's default is `true` in all Comet-supported Spark versions (3.4.3, 3.5.8, 4.0.1,
+-- 4.1.1). Under `true`, a predicate that yields NULL for some element and no TRUE elsewhere
+-- makes `exists` return NULL. Under `false`, NULL predicate results are treated as FALSE.
+
+-- ConfigMatrix: spark.sql.legacy.followThreeValuedLogicInArrayExists=true,false
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+statement
+CREATE TABLE test_exists_3vl(a array<int>) USING parquet
+
+statement
+INSERT INTO test_exists_3vl VALUES
+  (array(1, 2, 3)),
+  (array(1, NULL, 3)),
+  (array(NULL, NULL)),
+  (array()),
+  (NULL)
+
+-- Non-null array with a matching element: always true regardless of flag.
+query
+SELECT exists(a, x -> x > 2) FROM test_exists_3vl
+
+-- Predicate never true for non-null elements. Under legacy=true a null element makes the
+-- result NULL; under legacy=false the null is treated as false so the result is false.
+query
+SELECT exists(a, x -> x > 10) FROM test_exists_3vl
+
+-- Predicate explicitly handles NULL (returns TRUE for null elements) — result is
+-- independent of the flag.
+query
+SELECT exists(a, x -> x IS NULL) FROM test_exists_3vl
+
+-- Predicate that is null-safe (COALESCE) — predicate never returns NULL, so both flag values
+-- produce the same TRUE/FALSE result.
+query
+SELECT exists(a, x -> coalesce(x, -1) > 0) FROM test_exists_3vl
+
+-- Literal-array cases pinning each corner:
+--   * a matching value is found -> true under both configs
+--   * no match, but a null element present -> null (legacy=true) / false (legacy=false)
+--   * empty array -> false under both configs
+--   * NULL array -> null under both configs
+query
+SELECT
+  exists(array(1, NULL, 3), x -> x > 2),
+  exists(array(1, NULL, 3), x -> x > 10),
+  exists(array(NULL, NULL), x -> x > 0),
+  exists(array(), x -> x > 0),
+  exists(cast(NULL as array<int>), x -> x > 0)
diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
index 8b1d989ae7..ac05b67675 100644
--- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
+++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex_types_to_string.sql
@@ -149,8 +149,7 @@ SELECT cast(named_struct('a', named_struct('b', named_struct('c', 1, 'd', 'leaf'
 query
 SELECT cast(named_struct('s1', '', 's2', ' ', 's3', cast(null as string)) as string)
 
--- Map-valued field: not supported, falls back to Spark.
-query expect_fallback(to StringType is not supported)
+query
 SELECT cast(named_struct('m', map('k', 1)) as string)
 
 -- ----------------------------------------------------------------------------
@@ -270,8 +269,6 @@ SELECT cast(array(cast(1.5 as double), cast('NaN' as double), cast('-Infinity' a
 query
 SELECT cast(array(array(array(1, 2), array(3)), array(array(cast(null as int)))) as string)
 
--- Array of map: not supported, falls back to Spark.
-query expect_fallback(to StringType is not supported)
 SELECT cast(array(map('k', 1)) as string)
 
 -- ----------------------------------------------------------------------------
@@ -282,57 +279,57 @@ SELECT cast(array(map('k', 1)) as string)
 -- tests use literal maps directly rather than reading from a parquet table.
 
 -- Map with string keys, int values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', 1, 'b', 2, 'c', 3) as string)
 
 -- Map with NULL values rendered as "null".
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', 1, 'b', cast(null as int), 'c', 3) as string)
 
 -- Map with int keys, string values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map(1, 'one', 2, 'two', 3, 'three') as string)
 
 -- Map with boolean values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('t', true, 'f', false, 'n', cast(null as boolean)) as string)
 
 -- Map with bigint values at min/max.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('max', 9223372036854775807, 'min', -9223372036854775808, 'zero', cast(0 as bigint)) as string)
 
 -- Map with decimal values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('pos', cast('1.234567890123456789' as decimal(38, 18)), 'neg', cast('-1.234567890123456789' as decimal(38, 18)), 'null', cast(null as decimal(38, 18))) as string)
 
 -- Map with date and timestamp values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', date '2024-01-15', 'b', date '1970-01-01', 'c', cast(null as date)) as string)
 
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', timestamp '2024-01-15 10:30:45', 'b', cast(null as timestamp)) as string)
 
 -- Map with binary values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', X'616263', 'b', X'', 'c', cast(null as binary)) as string)
 
 -- Map with float / double values: NaN / ±0 / ±Infinity / NULL.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('nan', cast('NaN' as float), 'neg0', cast(-0.0 as float), 'null', cast(null as float)) as string)
 
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('nan', cast('NaN' as double), 'inf', cast('Infinity' as double), 'ninf', cast('-Infinity' as double), 'null', cast(null as double)) as string)
 
 -- Map with struct values: each value rendered as `{f1, f2, ...}`.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', named_struct('x', 1, 'y', 'first'), 'b', cast(null as struct<x: int, y: string>)) as string)
 
 -- Map with array values.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('a', array(1, 2, 3), 'b', array(cast(null as int)), 'c', cast(null as array<int>)) as string)
 
 -- Empty map.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map() as string)
 
 -- NULL map: Spark constant-folds this to a literal NULL, so the cast never reaches Comet
@@ -341,5 +338,5 @@ query
 SELECT cast(cast(null as map<string, int>) as string)
 
 -- Map of map.
-query expect_fallback(Cast from MapType)
+query
 SELECT cast(map('outer', map('inner', 1)) as string)
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/scala/org/apache/comet/CometCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
index 99475a252f..d3d8472dd7 100644
--- a/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometCastSuite.scala
@@ -735,19 +735,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   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") {
@@ -1803,9 +1790,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
           assert(
             CometCast.isSupported(fromType, toType, None, CometEvalMode.LEGACY) ==
               Unsupported(Some(expectedMessage)))
-          checkSparkAnswerAndFallbackReason(
-            data.select(col("a").cast(toType).as("converted")),
-            expectedMessage)
+          checkSparkAnswerAndOperator(data.select(col("a").cast(toType).as("converted")))
         }
       }
     }
diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
index f3c442ad29..d335d5bbda 100644
--- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
@@ -2047,9 +2047,6 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
         sql(s"insert into $table values(0, 1, 100.000001)")
 
         Seq(
-          (
-            s"SELECT cast(make_interval(c0, c1, c0, c1, c0, c0, c2) as string) as C from $table",
-            Set("Cast from CalendarIntervalType to StringType is not supported")),
           (
             "SELECT "
               + "date_part('YEAR', make_interval(c0, c1, c0, c1, c0, c0, c2))"
@@ -2067,9 +2064,7 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
               + s"(SELECT c1, sum(c0) as sum_c0, sum(c2) as sum_c2 from $table group by c1) as A, "
               + s"(SELECT c1, cast(make_interval(c0, c1, c0, c1, c0, c0, c2) as string) as casted from $table) as B "
               + "where A.c1 = B.c1 ",
-            Set(
-              "Cast from CalendarIntervalType to StringType is not supported",
-              "Comet shuffle is not enabled: spark.comet.exec.shuffle.enabled is not enabled")),
+            Set("Comet shuffle is not enabled: spark.comet.exec.shuffle.enabled is not enabled")),
           (s"select * from $table LIMIT 10 OFFSET 3", Set("Comet shuffle is not enabled")))
           .foreach(test => {
             val qry = test._1
diff --git a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala
index 268fdf94eb..cf6c467dd0 100644
--- a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala
@@ -53,6 +53,58 @@ 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.charVarcharAsString", "true")
+    assert(!isCometLoaded(conf))
+
+    // Setting the config back to its Spark default (case-insensitive) clears the trigger.
+    conf.setConfString("spark.sql.legacy.charVarcharAsString", "FALSE")
+    assert(isCometLoaded(conf))
+
+    // Enum-default configs also trigger when set to a non-default value. Use viewSchemaCompensation
+    // as an enum-valued example (default `true`) since timeParserPolicy is handled per-expression.
+    conf.setConfString("spark.sql.legacy.viewSchemaCompensation", "false")
+    assert(!isCometLoaded(conf))
+    conf.setConfString("spark.sql.legacy.viewSchemaCompensation", "TRUE")
+    assert(isCometLoaded(conf))
+
+    // Legacy configs handled per-expression (e.g. castComplexTypesToString,
+    // allowNegativeScaleOfDecimal, timeParserPolicy) are NOT part of the fallback set and must not
+    // disable Comet on their own.
+    conf.setConfString("spark.sql.legacy.timeParserPolicy", "LEGACY")
+    assert(isCometLoaded(conf))
+    conf.unsetConf("spark.sql.legacy.timeParserPolicy")
+    conf.setConfString("spark.sql.legacy.castComplexTypesToString.enabled", "true")
+    assert(isCometLoaded(conf))
+    conf.unsetConf("spark.sql.legacy.castComplexTypesToString.enabled")
+    conf.setConfString("spark.sql.legacy.allowNegativeScaleOfDecimal", "true")
+    assert(isCometLoaded(conf))
+    conf.unsetConf("spark.sql.legacy.allowNegativeScaleOfDecimal")
+
+    // Parquet legacy configs are checked per-scan by CometScanRule, not session-wide, so setting
+    // them here must NOT disable Comet for the whole session.
+    conf.setConfString("spark.sql.parquet.datetimeRebaseModeInRead", "LEGACY")
+    assert(isCometLoaded(conf))
+    conf.unsetConf("spark.sql.parquet.datetimeRebaseModeInRead")
+    conf.setConfString("spark.sql.legacy.parquet.datetimeRebaseModeInRead", "LEGACY")
+    assert(isCometLoaded(conf))
+    conf.unsetConf("spark.sql.legacy.parquet.datetimeRebaseModeInRead")
+
+    // Opt-out: users can keep Comet enabled by disabling the fallback (compatibility not
+    // guaranteed).
+    conf.setConfString("spark.sql.legacy.charVarcharAsString", "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/rules/CometScanRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometScanRuleSuite.scala
index f444fe62c9..90ed75ed10 100644
--- a/spark/src/test/scala/org/apache/comet/rules/CometScanRuleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/rules/CometScanRuleSuite.scala
@@ -98,6 +98,41 @@ class CometScanRuleSuite extends CometTestBase {
     }
   }
 
+  test("parquetFallbackReason: default configs -> no fallback") {
+    val conf = new org.apache.spark.sql.internal.SQLConf
+    assert(CometScanRule.parquetFallbackReason(conf).isEmpty)
+  }
+
+  test("parquetFallbackReason: primary key set to LEGACY triggers per-scan fallback") {
+    val conf = new org.apache.spark.sql.internal.SQLConf
+    conf.setConfString("spark.sql.parquet.datetimeRebaseModeInRead", "LEGACY")
+    val reason = CometScanRule.parquetFallbackReason(conf)
+    assert(reason.isDefined)
+    assert(reason.get.contains("spark.sql.parquet.datetimeRebaseModeInRead"))
+  }
+
+  test("parquetFallbackReason: legacy alias set to LEGACY triggers per-scan fallback") {
+    val conf = new org.apache.spark.sql.internal.SQLConf
+    conf.setConfString("spark.sql.legacy.parquet.datetimeRebaseModeInRead", "LEGACY")
+    val reason = CometScanRule.parquetFallbackReason(conf)
+    assert(reason.isDefined)
+    assert(reason.get.contains("spark.sql.legacy.parquet.datetimeRebaseModeInRead"))
+  }
+
+  test("parquetFallbackReason: nanosAsLong=true triggers per-scan fallback") {
+    val conf = new org.apache.spark.sql.internal.SQLConf
+    conf.setConfString("spark.sql.legacy.parquet.nanosAsLong", "true")
+    val reason = CometScanRule.parquetFallbackReason(conf)
+    assert(reason.isDefined)
+    assert(reason.get.contains("spark.sql.legacy.parquet.nanosAsLong"))
+  }
+
+  test("parquetFallbackReason: write-side rebase config does not trigger scan fallback") {
+    val conf = new org.apache.spark.sql.internal.SQLConf
+    conf.setConfString("spark.sql.parquet.datetimeRebaseModeInWrite", "LEGACY")
+    assert(CometScanRule.parquetFallbackReason(conf).isEmpty)
+  }
+
   test("CometScanRule should fallback to Spark for ShortType when safety check enabled") {
     withTempPath { path =>
       // Create test data with ShortType which may be from unsigned UINT_8