feat: auto-detect and configure Delta DV read strategy

Users no longer need to manually set
`spark.databricks.delta.deletionVectors.useMetadataRowIndex=false`.

Comet now registers a Catalyst optimizer rule (`CometDeltaDvConfigRule`)
via `CometSparkSessionExtensions.injectOptimizerRule` that automatically
sets the config when `COMET_DELTA_NATIVE_ENABLED=true`. The rule runs
in the optimizer phase which executes BEFORE Delta's planner strategies,
so the config takes effect for the SAME query where it's first needed.

Delta's default `useMetadataRowIndex=true` strategy applies DVs opaquely
inside `DeletionVectorBoundFileFormat` at read time — no plan-level
Filter is inserted, making it impossible for Comet's physical-plan
rewrite to detect and handle. Setting it to `false` forces Delta's
older strategy (Project + Filter + __delta_internal_is_row_deleted)
which our `stripDeltaDvWrappers` can recognize and route through
DeltaDvFilterExec.

Removed the manual config from CometDeltaNativeSuite's sparkConf.

Tests: succeeded 34, failed 0, canceled 0, ignored 0, pending 0

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: schenksj

spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala

@@ -49,6 +49,40 @@ class CometSparkSessionExtensions
     extensions.injectColumnar { session => CometExecColumnar(session) }
     extensions.injectQueryStagePrepRule { session => CometScanRule(session) }
     extensions.injectQueryStagePrepRule { session => CometExecRule(session) }
+
+    // Phase 3 auto-config: tell Delta to use its older DV read strategy
+    // (Project + Filter with __delta_internal_is_row_deleted) instead of
+    // the default metadata-row-index strategy. The older strategy inserts
+    // a plan-level Filter that CometScanRule.stripDeltaDvWrappers can
+    // detect and remove, routing DVs through our DeltaDvFilterExec.
+    //
+    // The metadata-row-index strategy applies DVs opaquely inside
+    // DeletionVectorBoundFileFormat at read time, which Comet's native
+    // reader can't intercept. By setting this config in an optimizer rule,
+    // it takes effect BEFORE Delta's PreprocessTableWithDVsStrategy runs
+    // (optimizer -> strategies ordering in Catalyst).
+    extensions.injectOptimizerRule { session =>
+      new org.apache.spark.sql.catalyst.rules.Rule[
+        org.apache.spark.sql.catalyst.plans.logical.LogicalPlan] {
+        @volatile private var configured = false
+        override def apply(plan: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan)
+            : org.apache.spark.sql.catalyst.plans.logical.LogicalPlan = {
+          if (!configured) {
+            try {
+              if (CometConf.COMET_DELTA_NATIVE_ENABLED.get(session.sessionState.conf)) {
+                session.conf
+                  .set("spark.databricks.delta.deletionVectors.useMetadataRowIndex", "false")
+              }
+            } catch {
+              case _: Throwable => // delta-spark not on classpath; ignore
+            }
+            configured = true
+          }
+          plan
+        }
+        override val ruleName: String = "CometDeltaDvConfigRule"
+      }
+    }
   }
 
   case class CometScanColumnar(session: SparkSession) extends ColumnarRule {

spark/src/test/scala/org/apache/comet/CometDeltaNativeSuite.scala

@@ -62,15 +62,9 @@ class CometDeltaNativeSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     // Explicitly clear the prefixes so production-shaped filenames are used.
     conf.set("spark.databricks.delta.testOnly.dataFileNamePrefix", "")
     conf.set("spark.databricks.delta.testOnly.dvFileNamePrefix", "")
-    // Delta 3.x's default `useMetadataRowIndex=true` strategy rewrites DV-in-use
-    // scans to read the parquet file with `_metadata.row_index` + other metadata
-    // columns, and applies the DV *inside* `DeletionVectorBoundFileFormat` at read
-    // time - no Filter is inserted in the plan. That makes the DV completely
-    // opaque to any physical-plan rewrite: there's nothing to detect.
-    // Setting this to false falls Delta back to its older strategy that DOES
-    // insert `Project -> Filter(__delta_internal_is_row_deleted = 0) -> scan`,
-    // which our `stripDeltaDvWrappers` rewrite can recognize and unwind.
-    conf.set("spark.databricks.delta.deletionVectors.useMetadataRowIndex", "false")
+    // useMetadataRowIndex is now set automatically by CometDeltaDvConfigRule
+    // (injected via CometSparkSessionExtensions.injectOptimizerRule) when
+    // COMET_DELTA_NATIVE_ENABLED=true. No manual config needed.
     conf
   }