diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/DeltaParquetFileFormat.scala b/spark/src/main/scala/org/apache/spark/sql/delta/DeltaParquetFileFormat.scala index d92a07fba5f..77b75c10921 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/DeltaParquetFileFormat.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/DeltaParquetFileFormat.scala @@ -55,6 +55,18 @@ 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, @@ -62,14 +74,13 @@ abstract class DeltaParquetFileFormatBase( 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 diff --git a/spark/v2/src/main/java/io/delta/spark/internal/v2/read/DeltaParquetFileFormatV2.java b/spark/v2/src/main/java/io/delta/spark/internal/v2/read/DeltaParquetFileFormatV2.java index 417713c5614..8af6722c623 100644 --- a/spark/v2/src/main/java/io/delta/spark/internal/v2/read/DeltaParquetFileFormatV2.java +++ b/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 tablePath, - boolean isCDCRead) { + boolean isCDCRead, + Option useMetadataRowIndex) { super( new ProtocolMetadataAdapterV2(protocol, metadata), nullableRowTrackingConstantFields, nullableRowTrackingGeneratedFields, optimizationsEnabled, tablePath, - isCDCRead); + isCDCRead, + // Java's Option can't directly pass to Scala's Option[Boolean] parameter, + // because Scala compiles Option[Boolean] to Option in bytecode for primitive + // handling. + useMetadataRowIndex.map(x -> x)); } @Override diff --git a/spark/v2/src/main/java/io/delta/spark/internal/v2/utils/PartitionUtils.java b/spark/v2/src/main/java/io/delta/spark/internal/v2/utils/PartitionUtils.java index e6434cc1166..6f2ed698cbd 100644 --- a/spark/v2/src/main/java/io/delta/spark/internal/v2/utils/PartitionUtils.java +++ b/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> readFunc = deltaFormat.buildReaderWithPartitionValues(