[Kernel-Spark] Refactor DeltaParquetFileFormatBase to support explicit useMetadataRowIndex control

spark/src/main/scala/org/apache/spark/sql/delta/DeltaParquetFileFormat.scala

@@ -55,21 +55,32 @@ import org.apache.spark.util.SerializableConfiguration
  *  - populated a column from the deletion vector of this file (if exists) to indicate
  *    whether the row is deleted or not according to the deletion vector. Consumers
  *    of this scan can use the column values to filter out the deleted rows.
+ *
+ * @param protocolMetadataAdapter Adapter providing protocol and metadata info for the table
+ * @param nullableRowTrackingConstantFields If true, row tracking constant fields (e.g., base row
+ *                                          ID, default row commit version) are nullable in schema
+ * @param nullableRowTrackingGeneratedFields If true, row tracking generated fields are nullable
+ * @param optimizationsEnabled Whether to enable optimizations (file splitting, predicate pushdown)
+ * @param tablePath Table path for deletion vector support; None disables DV processing
+ * @param isCDCRead Whether this is a CDC (Change Data Capture) read
+ * @param useMetadataRowIndexOpt Controls row index source for DV filtering. When provided,
+ *                               must match optimizationsEnabled (true enables _metadata.row_index
+ *                               and file splitting; false uses internal counter, no splitting).
+ *                               When None, reads from session config.
  */
 abstract class DeltaParquetFileFormatBase(
     protected val protocolMetadataAdapter: ProtocolMetadataAdapter,
     protected val nullableRowTrackingConstantFields: Boolean = false,
     protected val nullableRowTrackingGeneratedFields: Boolean = false,
     protected val optimizationsEnabled: Boolean = true,
     protected val tablePath: Option[String] = None,
-    protected val isCDCRead: Boolean = false)
+    protected val isCDCRead: Boolean = false,
+    protected val useMetadataRowIndexOpt: Option[Boolean] = None)
   extends ParquetFileFormat with Logging {
 
   // Validate either we have all arguments for DV enabled read or none of them.
   if (hasTablePath) {
-    SparkSession.getActiveSession.map { session =>
-      val useMetadataRowIndex =
-        session.sessionState.conf.getConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX)
+    useMetadataRowIndexOpt.foreach { useMetadataRowIndex =>
       require(useMetadataRowIndex == optimizationsEnabled,
         "Wrong arguments for Delta table scan with deletion vectors")
     }
@@ -149,8 +160,9 @@ abstract class DeltaParquetFileFormatBase(
       options: Map[String, String],
       hadoopConf: Configuration): PartitionedFile => Iterator[InternalRow] = {
 
-    val useMetadataRowIndexConf = DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX
-    val useMetadataRowIndex = sparkSession.sessionState.conf.getConf(useMetadataRowIndexConf)
+    // Use explicitly provided value if available, otherwise read from config
+    val useMetadataRowIndex = useMetadataRowIndexOpt.getOrElse(
+      sparkSession.sessionState.conf.getConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX))
 
     val parquetDataReader: PartitionedFile => Iterator[InternalRow] =
       super.buildReaderWithPartitionValues(
@@ -532,7 +544,10 @@ case class DeltaParquetFileFormat(
     nullableRowTrackingGeneratedFields = nullableRowTrackingGeneratedFields,
     optimizationsEnabled = optimizationsEnabled,
     tablePath = tablePath,
-    isCDCRead = isCDCRead) {
+    isCDCRead = isCDCRead,
+    // V1: capture config at construction, used in buildReaderWithPartitionValues
+    useMetadataRowIndexOpt = SparkSession.getActiveSession.map(
+      _.sessionState.conf.getConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX))) {
 
   /**
    * We sometimes need to replace FileFormat within LogicalPlans, so we have to override

spark/v2/src/main/java/io/delta/spark/internal/v2/read/DeltaParquetFileFormatV2.java

@@ -42,6 +42,7 @@ public class DeltaParquetFileFormatV2 extends DeltaParquetFileFormatBase {
    * @param optimizationsEnabled whether to enable optimizations (splits, predicate pushdown)
    * @param tablePath table path for deletion vector support
    * @param isCDCRead whether this is a CDC read
+   * @param useMetadataRowIndex V2: explicit control over _metadata.row_index usage for DV filtering
    */
   public DeltaParquetFileFormatV2(
       Protocol protocol,
@@ -50,14 +51,19 @@ public DeltaParquetFileFormatV2(
       boolean nullableRowTrackingGeneratedFields,
       boolean optimizationsEnabled,
       Option<String> tablePath,
-      boolean isCDCRead) {
+      boolean isCDCRead,
+      Option<Boolean> useMetadataRowIndex) {
     super(
         new ProtocolMetadataAdapterV2(protocol, metadata),
         nullableRowTrackingConstantFields,
         nullableRowTrackingGeneratedFields,
         optimizationsEnabled,
         tablePath,
-        isCDCRead);
+        isCDCRead,
+        // Java's Option<Boolean> can't directly pass to Scala's Option[Boolean] parameter,
+        // because Scala compiles Option[Boolean] to Option<Object> in bytecode for primitive
+        // handling.
+        useMetadataRowIndex.map(x -> x));
   }
 
   @Override

spark/v2/src/main/java/io/delta/spark/internal/v2/utils/PartitionUtils.java

@@ -194,7 +194,8 @@ public static PartitionReaderFactory createDeltaParquetReaderFactory(
             /* nullableRowTrackingGeneratedFields */ false,
             /* optimizationsEnabled */ true,
             Option.apply(tablePath),
-            /* isCDCRead */ false);
+            /* isCDCRead */ false,
+            /* useMetadataRowIndexOpt */ Option.empty());
 
     Function1<PartitionedFile, Iterator<InternalRow>> readFunc =
         deltaFormat.buildReaderWithPartitionValues(