Extract Delta Lake deletion vectors

This change extracts deletion vectors represented as roaring bitmaps in delta lake files and
converts them into the XTable intermediate representation.

Previously, XTable only detected tables changes that included adding or removing of data files. Now
the detected table change also includes any deletion vectors files added in the commit.

Note that, in Delta Lake, the Deletion vectors are represented in a compressed binary format.
However, once extracted by Xtable, the offset are currently extracted into a list of long offsets.
This representation is not the most efficient for large datasets. Optimization is pending to
prioritize end-to-end conversion completion.

Author: ashvina

xtable-api/src/main/java/org/apache/xtable/model/TableChange.java

@@ -30,7 +30,7 @@
  * @since 0.1
  */
 @Value
-@Builder(toBuilder = true)
+@Builder(toBuilder = true, builderClassName = "Builder")
 public class TableChange {
   // Change in files at the specified instant
   InternalFilesDiff filesDiff;

xtable-api/src/main/java/org/apache/xtable/model/storage/InternalDeletionVector.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package org.apache.xtable.model.storage;
+
+import java.util.Iterator;
+import java.util.function.Supplier;
+
+import lombok.AccessLevel;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NonNull;
+import lombok.ToString;
+import lombok.experimental.Accessors;
+import lombok.experimental.FieldDefaults;
+import lombok.experimental.SuperBuilder;
+
+@Accessors(fluent = true)
+@SuperBuilder(toBuilder = true)
+@FieldDefaults(makeFinal = true, level = lombok.AccessLevel.PRIVATE)
+@Getter
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
+public class InternalDeletionVector extends InternalFile {
+  // path (absolute with scheme) of data file to which this deletion vector belongs
+  @NonNull String dataFilePath;
+
+  // super.getFileSizeBytes() is the size of the deletion vector file
+  // super.getPhysicalPath() is the absolute path (with scheme) of the deletion vector file
+  // super.getRecordCount() is the count of records in the deletion vector file
+
+  // offset of deletion vector start in a deletion vector file
+  int offset;
+
+  /**
+   * binary representation of the deletion vector. The consumer can use the {@link
+   * #ordinalsIterator()} to extract the ordinals represented in the binary format.
+   */
+  byte[] binaryRepresentation;
+
+  /**
+   * Supplier for an iterator that returns the ordinals of records deleted by this deletion vector
+   * in the linked data file, identified by {@link #dataFilePath}.
+   *
+   * <p>The {@link InternalDeletionVector} instance does not guarantee that a new or distinct result
+   * will be returned each time the supplier is invoked. However, the supplier is expected to return
+   * a new iterator for each call.
+   */
+  @Getter(AccessLevel.NONE)
+  Supplier<Iterator<Long>> ordinalsSupplier;
+
+  /**
+   * @return An iterator that returns the ordinals of records deleted by this deletion vector in the
+   *     linked data file. There is no guarantee that a new or distinct iterator will be returned
+   *     each time the iterator is invoked.
+   */
+  public Iterator<Long> ordinalsIterator() {
+    return ordinalsSupplier.get();
+  }
+}

xtable-core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java

@@ -18,18 +18,26 @@
 
 package org.apache.xtable.delta;
 
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 
 import org.apache.spark.sql.delta.Snapshot;
 import org.apache.spark.sql.delta.actions.AddFile;
 import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor;
 import org.apache.spark.sql.delta.actions.RemoveFile;
+import org.apache.spark.sql.delta.deletionvectors.RoaringBitmapArray;
+import org.apache.spark.sql.delta.storage.dv.DeletionVectorStore;
+import org.apache.spark.sql.delta.storage.dv.HadoopFileSystemDVStore;
+
+import com.google.common.annotations.VisibleForTesting;
 
 import org.apache.xtable.exception.NotSupportedException;
 import org.apache.xtable.model.schema.InternalField;
@@ -38,6 +46,7 @@
 import org.apache.xtable.model.stat.FileStats;
 import org.apache.xtable.model.storage.FileFormat;
 import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.InternalDeletionVector;
 
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class DeltaActionsConverter {
@@ -113,16 +122,66 @@ static String getFullPathToFile(Snapshot snapshot, String dataFilePath) {
    *
    * @param snapshot the commit snapshot
    * @param addFile the add file action
-   * @return the deletion vector representation (path of data file), or null if no deletion vector
-   *     is present
+   * @return the deletion vector representation, or null if no deletion vector is present
    */
-  public String extractDeletionVectorFile(Snapshot snapshot, AddFile addFile) {
+  public InternalDeletionVector extractDeletionVector(Snapshot snapshot, AddFile addFile) {
     DeletionVectorDescriptor deletionVector = addFile.deletionVector();
     if (deletionVector == null) {
       return null;
     }
 
     String dataFilePath = addFile.path();
-    return getFullPathToFile(snapshot, dataFilePath);
+    dataFilePath = getFullPathToFile(snapshot, dataFilePath);
+
+    InternalDeletionVector.InternalDeletionVectorBuilder<?, ?> deleteVectorBuilder =
+        InternalDeletionVector.builder()
+            .recordCount(deletionVector.cardinality())
+            .fileSizeBytes(deletionVector.sizeInBytes())
+            .dataFilePath(dataFilePath);
+
+    if (deletionVector.isInline()) {
+      deleteVectorBuilder
+          .binaryRepresentation(deletionVector.inlineData())
+          .physicalPath("")
+          .ordinalsSupplier(() -> ordinalsIterator(deletionVector.inlineData()));
+    } else {
+      Path deletionVectorFilePath = deletionVector.absolutePath(snapshot.deltaLog().dataPath());
+      deleteVectorBuilder
+          .offset(getOffset(deletionVector))
+          .physicalPath(deletionVectorFilePath.toString())
+          .ordinalsSupplier(() -> ordinalsIterator(snapshot, deletionVector));
+    }
+
+    return deleteVectorBuilder.build();
+  }
+
+  private Iterator<Long> ordinalsIterator(byte[] bytes) {
+    RoaringBitmapArray rbm = RoaringBitmapArray.readFrom(bytes);
+    long[] ordinals = rbm.values();
+    return Arrays.stream(ordinals).iterator();
+  }
+
+  private Iterator<Long> ordinalsIterator(
+      Snapshot snapshot, DeletionVectorDescriptor deleteVector) {
+    Path deletionVectorFilePath = deleteVector.absolutePath(snapshot.deltaLog().dataPath());
+    int offset = getOffset(deleteVector);
+    long[] ordinals =
+        parseOrdinalFile(
+            snapshot.deltaLog().newDeltaHadoopConf(),
+            deletionVectorFilePath,
+            deleteVector.sizeInBytes(),
+            offset);
+    return Arrays.stream(ordinals).iterator();
+  }
+
+  private static int getOffset(DeletionVectorDescriptor deleteVector) {
+    return deleteVector.offset().isDefined() ? (int) deleteVector.offset().get() : 1;
+  }
+
+  @VisibleForTesting
+  long[] parseOrdinalFile(Configuration conf, Path filePath, int size, int offset) {
+    DeletionVectorStore dvStore = new HadoopFileSystemDVStore(conf);
+    RoaringBitmapArray rbm = dvStore.read(filePath, offset, size);
+    return rbm.values();
   }
 }

xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java

@@ -22,11 +22,11 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import lombok.Builder;
 import lombok.extern.log4j.Log4j2;
@@ -53,6 +53,8 @@
 import org.apache.xtable.model.schema.InternalSchema;
 import org.apache.xtable.model.storage.FileFormat;
 import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.InternalDeletionVector;
+import org.apache.xtable.model.storage.InternalFile;
 import org.apache.xtable.model.storage.InternalFilesDiff;
 import org.apache.xtable.model.storage.PartitionFileGroup;
 import org.apache.xtable.spi.extractor.ConversionSource;
@@ -113,8 +115,8 @@ public TableChange getTableChangeForCommit(Long versionNumber) {
     // All 3 of the following data structures use data file's absolute path as the key
     Map<String, InternalDataFile> addedFiles = new HashMap<>();
     Map<String, InternalDataFile> removedFiles = new HashMap<>();
-    // Set of data file paths for which deletion vectors exists.
-    Set<String> deletionVectors = new HashSet<>();
+    // Map of data file paths for which deletion vectors exists.
+    Map<String, InternalDeletionVector> deletionVectors = new HashMap<>();
 
     for (Action action : actionsForVersion) {
       if (action instanceof AddFile) {
@@ -129,10 +131,10 @@ public TableChange getTableChangeForCommit(Long versionNumber) {
                 DeltaPartitionExtractor.getInstance(),
                 DeltaStatsExtractor.getInstance());
         addedFiles.put(dataFile.getPhysicalPath(), dataFile);
-        String deleteVectorPath =
-            actionsConverter.extractDeletionVectorFile(snapshotAtVersion, (AddFile) action);
-        if (deleteVectorPath != null) {
-          deletionVectors.add(deleteVectorPath);
+        InternalDeletionVector deletionVector =
+            actionsConverter.extractDeletionVector(snapshotAtVersion, (AddFile) action);
+        if (deletionVector != null) {
+          deletionVectors.put(deletionVector.dataFilePath(), deletionVector);
         }
       } else if (action instanceof RemoveFile) {
         InternalDataFile dataFile =
@@ -151,7 +153,7 @@ public TableChange getTableChangeForCommit(Long versionNumber) {
     // entry which is replaced by a new entry, AddFile with delete vector information. Since the
     // same data file is removed and added, we need to remove it from the added and removed file
     // maps which are used to track actual added and removed data files.
-    for (String deletionVector : deletionVectors) {
+    for (String deletionVector : deletionVectors.keySet()) {
       // validate that a Remove action is also added for the data file
       if (removedFiles.containsKey(deletionVector)) {
         addedFiles.remove(deletionVector);
@@ -163,11 +165,15 @@ public TableChange getTableChangeForCommit(Long versionNumber) {
       }
     }
 
+    List<InternalFile> allAddedFiles =
+        Stream.concat(addedFiles.values().stream(), deletionVectors.values().stream())
+            .collect(Collectors.toList());
     InternalFilesDiff internalFilesDiff =
         InternalFilesDiff.builder()
-            .filesAdded(addedFiles.values())
+            .filesAdded(allAddedFiles)
             .filesRemoved(removedFiles.values())
             .build();
+
     return TableChange.builder()
         .tableAsOfChange(tableAtVersion)
         .filesDiff(internalFilesDiff)

xtable-core/src/test/java/org/apache/xtable/TestSparkDeltaTable.java

@@ -39,6 +39,7 @@
 import org.apache.spark.sql.functions;
 
 import org.apache.spark.sql.delta.DeltaLog;
+import org.apache.spark.sql.delta.actions.AddFile;
 
 import com.google.common.base.Preconditions;
 
@@ -212,11 +213,15 @@ private String initBasePath(Path tempDir, String tableName) throws IOException {
   }
 
   public List<String> getAllActiveFiles() {
-    return deltaLog.snapshot().allFiles().collectAsList().stream()
+    return getAllActiveFilesInfo().stream()
         .map(addFile -> addSlashToBasePath(basePath) + addFile.path())
         .collect(Collectors.toList());
   }
 
+  public List<AddFile> getAllActiveFilesInfo() {
+    return deltaLog.snapshot().allFiles().collectAsList();
+  }
+
   private String addSlashToBasePath(String basePath) {
     if (basePath.endsWith("/")) {
       return basePath;

xtable-core/src/test/java/org/apache/xtable/ValidationTestHelper.java

@@ -30,6 +30,7 @@
 import org.apache.xtable.model.InternalSnapshot;
 import org.apache.xtable.model.TableChange;
 import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.InternalFile;
 
 public class ValidationTestHelper {
 
@@ -96,7 +97,7 @@ public static List<String> getAllFilePaths(InternalSnapshot internalSnapshot) {
   }
 
   private static Set<String> extractPathsFromDataFile(Set<InternalDataFile> dataFiles) {
-    return dataFiles.stream().map(InternalDataFile::getPhysicalPath).collect(Collectors.toSet());
+    return dataFiles.stream().map(InternalFile::getPhysicalPath).collect(Collectors.toSet());
   }
 
   private static void replaceFileScheme(List<String> filePaths) {