Merge remote-tracking branch 'upstream/branch-25.10' into ray/13471

Author: liurenjie1024

delta-lake/delta-33x/src/main/scala/com/nvidia/spark/rapids/delta/delta33x/Delta33xProvider.scala

@@ -23,11 +23,11 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.connector.catalog.SupportsWrite
 import org.apache.spark.sql.delta.{DeltaLog, DeltaParquetFileFormat}
-import org.apache.spark.sql.delta.DeltaParquetFileFormat.{IS_ROW_DELETED_COLUMN_NAME, ROW_INDEX_COLUMN_NAME}
+import org.apache.spark.sql.delta.DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME
 import org.apache.spark.sql.delta.catalog.{DeltaCatalog, DeltaTableV2}
 import org.apache.spark.sql.delta.commands.{DeleteCommand, MergeIntoCommand, OptimizeTableCommand, UpdateCommand}
 import org.apache.spark.sql.delta.rapids.DeltaRuntimeShim
-import org.apache.spark.sql.delta.sources.DeltaDataSource
+import org.apache.spark.sql.delta.sources.{DeltaDataSource, DeltaSQLConf}
 import org.apache.spark.sql.execution.FileSourceScanExec
 import org.apache.spark.sql.execution.command.RunnableCommand
 import org.apache.spark.sql.execution.datasources.{FileFormat, HadoopFsRelation, SaveIntoDataSourceCommand}
@@ -54,6 +54,10 @@ object Delta33xProvider extends DeltaIOProvider {
     write == classOf[DeltaTableV2] || write == classOf[GpuDeltaCatalog#GpuStagedDeltaTableV2]
   }
 
+  override def isSupportedFormat(format: Class[_ <: FileFormat]): Boolean = {
+    super.isSupportedFormat(format) || format == classOf[GpuDelta33xParquetFileFormat]
+  }
+
   override def tagForGpu(
       cpuExec: AppendDataExecV1,
       meta: AppendDataExecV1Meta): Unit = {
@@ -90,20 +94,14 @@ object Delta33xProvider extends DeltaIOProvider {
   override def tagSupportForGpuFileSourceScan(meta: SparkPlanMeta[FileSourceScanExec]): Unit = {
     val format = meta.wrapped.relation.fileFormat
     if (format.getClass == classOf[DeltaParquetFileFormat]) {
+      val session = meta.wrapped.session
+      val useMetadataRowIndex =
+        session.sessionState.conf.getConf(DeltaSQLConf.DELETION_VECTORS_USE_METADATA_ROW_INDEX)
       val requiredSchema = meta.wrapped.requiredSchema
-      if (!meta.conf.isParquetPerFileReadEnabled) {
-        if (requiredSchema.exists(_.name == IS_ROW_DELETED_COLUMN_NAME)) {
-          meta.willNotWorkOnGpu(
-            s"reading metadata column $IS_ROW_DELETED_COLUMN_NAME is supported for PERFILE and " +
-              " not supported for " +
-              s"${RapidsReaderType.withName(meta.conf.get(RapidsConf.PARQUET_READER_TYPE))}")
-        }
-        if (requiredSchema.exists(_.name == ROW_INDEX_COLUMN_NAME)) {
-          meta.willNotWorkOnGpu(
-            s"reading metadata column $ROW_INDEX_COLUMN_NAME is supported for PERFILE and " +
-              " not supported for " +
-              s"${RapidsReaderType.withName(meta.conf.get(RapidsConf.PARQUET_READER_TYPE))}")
-        }
+      val isRowDeletedCol =  requiredSchema.exists(_.name == IS_ROW_DELETED_COLUMN_NAME)
+      if (useMetadataRowIndex && isRowDeletedCol) {
+        meta.willNotWorkOnGpu("we don't support generating metadata row index for " +
+          s"${meta.wrapped.getClass.getSimpleName}")
       }
       GpuReadParquetFileFormat.tagSupport(meta)
     } else {