[VL][Delta] Simplify DV scan payload handoff

Author: Mohammad Linjawi

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala

@@ -26,7 +26,6 @@ import org.apache.gluten.sql.shims.SparkShimLoader
 import org.apache.gluten.substrait.plan.PlanNode
 import org.apache.gluten.substrait.rel.{LocalFilesBuilder, LocalFilesNode, SplitInfo}
 import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
-import org.apache.gluten.utils.DeltaDeletionVectorRegistry
 import org.apache.gluten.vectorized._
 
 import org.apache.spark.{Partition, SparkConf, TaskContext}
@@ -41,17 +40,14 @@ import org.apache.spark.sql.utils.SparkInputMetricsUtil.InputMetricsWrapper
 import org.apache.spark.sql.vectorized.ColumnarBatch
 import org.apache.spark.util.SparkDirectoryUtil
 
-import org.apache.hadoop.fs.Path
-
 import java.lang.{Long => JLong}
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
 import java.time.ZoneOffset
-import java.util.{ArrayList => JArrayList, HashMap => JHashMap, UUID}
+import java.util.UUID
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
-import scala.util.Try
 
 class VeloxIteratorApi extends IteratorApi with Logging {
   private val deltaMetadataUtilsClassName =
@@ -102,8 +98,7 @@ class VeloxIteratorApi extends IteratorApi with Logging {
       .map(
         f => SparkShimLoader.getSparkShims.generateMetadataColumns(f, metadataColumnNames).asJava)
     val (otherMetadataColumns, deletionVectorPayloads) =
-      normalizeRegisteredDeltaSplitMetadata(partitionFiles, properties)
-        .orElse(normalizeDeltaSplitMetadata(partitionSchema.fields.length, partitionFiles))
+      normalizeDeltaSplitMetadata(partitionSchema.fields.length, partitionFiles)
         .getOrElse {
           (
             partitionFiles.map {
@@ -247,73 +242,6 @@ class VeloxIteratorApi extends IteratorApi with Logging {
     }
   }
 
-  private def normalizeRegisteredDeltaSplitMetadata(
-      partitionFiles: Seq[PartitionedFile],
-      properties: Map[String, String])
-      : Option[(Seq[java.util.Map[String, Object]], Array[Array[Byte]])] = {
-    properties
-      .get(DeltaDeletionVectorRegistry.RegistryIdProperty)
-      .flatMap(DeltaDeletionVectorRegistry.get)
-      .flatMap {
-        registeredEntries =>
-          val normalizedMetadataColumns = new JArrayList[java.util.Map[String, Object]]()
-          val deletionVectorPayloads = mutable.ArrayBuffer.empty[Array[Byte]]
-          var matchedDeletionVectors = 0
-          partitionFiles.foreach {
-            file =>
-              val metadata = new JHashMap[String, Object]()
-              val baseMetadata =
-                SparkShimLoader.getSparkShims.getOtherConstantMetadataColumnValues(file)
-              if (baseMetadata != null) {
-                metadata.putAll(baseMetadata)
-              }
-              lookupRegisteredDeltaDeletionVector(file, registeredEntries).foreach {
-                entry =>
-                  metadata.put("delta_dv_cardinality", Long.box(entry.cardinality))
-                  metadata.put("row_index_filter_type", entry.filterType)
-                  metadata.put("delta_dv_payload_index", Int.box(deletionVectorPayloads.length))
-                  deletionVectorPayloads += entry.payload
-                  matchedDeletionVectors += 1
-              }
-              normalizedMetadataColumns.add(metadata)
-          }
-          if (matchedDeletionVectors == 0) {
-            None
-          } else {
-            Some((normalizedMetadataColumns.asScala.toSeq, deletionVectorPayloads.toArray))
-          }
-      }
-  }
-
-  private def lookupRegisteredDeltaDeletionVector(
-      file: PartitionedFile,
-      registeredEntries: Map[String, DeltaDeletionVectorRegistry.Entry])
-      : Option[DeltaDeletionVectorRegistry.Entry] = {
-    deltaDeletionVectorPathCandidates(file).iterator
-      .map(registeredEntries.get)
-      .collectFirst { case Some(entry) => entry }
-  }
-
-  private def deltaDeletionVectorPathCandidates(file: PartitionedFile): Seq[String] = {
-    val rawPath = unescapePathName(file.filePath.toString)
-    val path = new Path(rawPath)
-    val pathUri = partitionedFilePathUri(file)
-    Seq(
-      pathUri.map(_.toASCIIString),
-      pathUri.map(_.getPath),
-      Some(rawPath),
-      Some(path.toUri.toASCIIString),
-      Some(path.toUri.getPath),
-      Some(rawPath.stripPrefix("/"))
-    ).flatten
-      .map(DeltaDeletionVectorRegistry.normalizePathKey(_))
-      .filter(_.nonEmpty)
-      .distinct
-  }
-
-  private def partitionedFilePathUri(file: PartitionedFile): Option[java.net.URI] =
-    Try(file.getClass.getMethod("pathUri").invoke(file).asInstanceOf[java.net.URI]).toOption
-
   /** Generate Iterator[ColumnarBatch] for first stage. */
   override def genFirstStageIterator(
       inputPartition: BaseGlutenPartition,