Include deletion vector when filtering active add entries in Delta

trinodb · Dec 19, 2024 · bb9d648 · bb9d648
1 parent d7aba15
commit bb9d648
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 7 deletions.
diff --git a/...elta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/DeletionVectorEntry.java b/...elta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/DeletionVectorEntry.java
@@ -34,6 +34,16 @@ public record DeletionVectorEntry(String storageType, String pathOrInlineDv, Opt
         requireNonNull(offset, "offset is null");
     }
 
+    // https://github.com/delta-io/delta/blob/34f02d8/kernel/kernel-api/src/main/java/io/delta/kernel/internal/actions/DeletionVectorDescriptor.java#L167-L174
+    public String uniqueId()
+    {
+        String uniqueFileId = storageType + pathOrInlineDv;
+        if (offset.isPresent()) {
+            return uniqueFileId + "@" + offset;
+        }
+        return uniqueFileId;
+    }
+
     public long getRetainedSizeInBytes()
     {
         return INSTANCE_SIZE

diff --git a/...lta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/TransactionLogAccess.java b/...lta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/TransactionLogAccess.java
@@ -429,21 +429,23 @@ public static ImmutableList<DeltaLakeColumnMetadata> columnsWithStats(List<Delta
 
     private Stream<AddFileEntry> activeAddEntries(Stream<DeltaLakeTransactionLogEntry> checkpointEntries, List<Transaction> transactions)
     {
-        Map<String, AddFileEntry> activeJsonEntries = new LinkedHashMap<>();
-        HashSet<String> removedFiles = new HashSet<>();
+        Map<FileEntryKey, AddFileEntry> activeJsonEntries = new LinkedHashMap<>();
+        HashSet<FileEntryKey> removedFiles = new HashSet<>();
 
         // The json entries containing the last few entries in the log need to be applied on top of the parquet snapshot:
         // - Any files which have been removed need to be excluded
         // - Any files with newer add actions need to be updated with the most recent metadata
         transactions.forEach(transaction -> {
-            Map<String, AddFileEntry> addFilesInTransaction = new LinkedHashMap<>();
-            Set<String> removedFilesInTransaction = new HashSet<>();
+            Map<FileEntryKey, AddFileEntry> addFilesInTransaction = new LinkedHashMap<>();
+            Set<FileEntryKey> removedFilesInTransaction = new HashSet<>();
             transaction.transactionEntries().forEach(deltaLakeTransactionLogEntry -> {
                 if (deltaLakeTransactionLogEntry.getAdd() != null) {
-                    addFilesInTransaction.put(deltaLakeTransactionLogEntry.getAdd().getPath(), deltaLakeTransactionLogEntry.getAdd());
+                    AddFileEntry add = deltaLakeTransactionLogEntry.getAdd();
+                    addFilesInTransaction.put(new FileEntryKey(add.getPath(), add.getDeletionVector().map(DeletionVectorEntry::uniqueId)), add);
                 }
                 else if (deltaLakeTransactionLogEntry.getRemove() != null) {
-                    removedFilesInTransaction.add(deltaLakeTransactionLogEntry.getRemove().path());
+                    RemoveFileEntry remove = deltaLakeTransactionLogEntry.getRemove();
+                    removedFilesInTransaction.add(new FileEntryKey(remove.path(), remove.deletionVector().map(DeletionVectorEntry::uniqueId)));
                 }
             });
 
@@ -456,11 +458,16 @@ else if (deltaLakeTransactionLogEntry.getRemove() != null) {
         Stream<AddFileEntry> filteredCheckpointEntries = checkpointEntries
                 .map(DeltaLakeTransactionLogEntry::getAdd)
                 .filter(Objects::nonNull)
-                .filter(addEntry -> !removedFiles.contains(addEntry.getPath()) && !activeJsonEntries.containsKey(addEntry.getPath()));
+                .filter(addEntry -> {
+                    FileEntryKey key = new FileEntryKey(addEntry.getPath(), addEntry.getDeletionVector().map(DeletionVectorEntry::uniqueId));
+                    return !removedFiles.contains(key) && !activeJsonEntries.containsKey(key);
+                });
 
         return Stream.concat(filteredCheckpointEntries, activeJsonEntries.values().stream());
     }
 
+    private record FileEntryKey(String path, Optional<String> deletionVectorId) {}
+
     public Stream<RemoveFileEntry> getRemoveEntries(ConnectorSession session, TableSnapshot tableSnapshot)
     {
         return getEntries(