Support reading deletion vectors in Delta Lake

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeColumnHandle.java

@@ -44,6 +44,7 @@ public class DeltaLakeColumnHandle
 {
     private static final int INSTANCE_SIZE = instanceSize(DeltaLakeColumnHandle.class);
 
+    public static final String ROW_POSITION_COLUMN_NAME = "$row_position";
     public static final String ROW_ID_COLUMN_NAME = "$row_id";
 
     public static final Type MERGE_ROW_ID_TYPE = rowType(
@@ -218,6 +219,11 @@ public HiveColumnHandle toHiveColumnHandle()
                 Optional.empty());
     }
 
+    public static DeltaLakeColumnHandle rowPositionColumnHandle()
+    {
+        return new DeltaLakeColumnHandle(ROW_POSITION_COLUMN_NAME, BIGINT, OptionalInt.empty(), ROW_POSITION_COLUMN_NAME, BIGINT, SYNTHESIZED, Optional.empty());
+    }
+
     public static DeltaLakeColumnHandle pathColumnHandle()
     {
         return new DeltaLakeColumnHandle(PATH_COLUMN_NAME, PATH_TYPE, OptionalInt.empty(), PATH_COLUMN_NAME, PATH_TYPE, SYNTHESIZED, Optional.empty());

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

@@ -1654,7 +1654,8 @@ private static void appendAddFileEntries(TransactionLogWriter transactionLogWrit
                             dataChange,
                             Optional.of(serializeStatsAsJson(statisticsWithExactNames)),
                             Optional.empty(),
-                            ImmutableMap.of()));
+                            ImmutableMap.of(),
+                            Optional.empty()));
         }
     }
 
@@ -3084,7 +3085,8 @@ private AddFileEntry prepareUpdatedAddFileEntry(ComputedStatistics stats, AddFil
                     false,
                     Optional.of(serializeStatsAsJson(deltaLakeJsonFileStatistics)),
                     Optional.empty(),
-                    addFileEntry.getTags());
+                    addFileEntry.getTags(),
+                    addFileEntry.getDeletionVector());
         }
         catch (JsonProcessingException e) {
             throw new TrinoException(GENERIC_INTERNAL_ERROR, "Statistics serialization error", e);

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakePageSource.java

@@ -15,6 +15,7 @@
 
 import io.airlift.json.JsonCodec;
 import io.airlift.json.JsonCodecFactory;
+import io.trino.plugin.deltalake.delete.PageFilter;
 import io.trino.plugin.hive.ReaderProjectionsAdapter;
 import io.trino.spi.Page;
 import io.trino.spi.TrinoException;
@@ -33,6 +34,7 @@
 import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.Set;
+import java.util.function.Supplier;
 
 import static com.google.common.base.Throwables.throwIfInstanceOf;
 import static io.airlift.slice.Slices.utf8Slice;
@@ -63,6 +65,7 @@ public class DeltaLakePageSource
     private final Block partitionsBlock;
     private final ConnectorPageSource delegate;
     private final Optional<ReaderProjectionsAdapter> projectionsAdapter;
+    private final Supplier<Optional<PageFilter>> deletePredicate;
 
     public DeltaLakePageSource(
             List<DeltaLakeColumnHandle> columns,
@@ -73,7 +76,8 @@ public DeltaLakePageSource(
             Optional<ReaderProjectionsAdapter> projectionsAdapter,
             String path,
             long fileSize,
-            long fileModifiedTime)
+            long fileModifiedTime,
+            Supplier<Optional<PageFilter>> deletePredicate)
     {
         int size = columns.size();
         requireNonNull(partitionKeys, "partitionKeys is null");
@@ -131,6 +135,7 @@ else if (missingColumnNames.contains(column.getBaseColumnName())) {
         this.rowIdIndex = rowIdIndex;
         this.pathBlock = pathBlock;
         this.partitionsBlock = partitionsBlock;
+        this.deletePredicate = requireNonNull(deletePredicate, "deletePredicate is null");
     }
 
     @Override
@@ -168,6 +173,11 @@ public Page getNextPage()
             if (projectionsAdapter.isPresent()) {
                 dataPage = projectionsAdapter.get().adaptPage(dataPage);
             }
+            Optional<PageFilter> deleteFilterPredicate = deletePredicate.get();
+            if (deleteFilterPredicate.isPresent()) {
+                dataPage = deleteFilterPredicate.get().apply(dataPage);
+            }
+
             int batchSize = dataPage.getPositionCount();
             Block[] blocks = new Block[prefilledBlocks.length];
             for (int i = 0; i < prefilledBlocks.length; i++) {

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakePageSourceProvider.java

@@ -13,17 +13,22 @@
  */
 package io.trino.plugin.deltalake;
 
+import com.google.common.base.Suppliers;
 import com.google.common.collect.AbstractIterator;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
 import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
 import io.trino.filesystem.TrinoFileSystemFactory;
 import io.trino.filesystem.TrinoInputFile;
 import io.trino.parquet.ParquetDataSource;
 import io.trino.parquet.ParquetReaderOptions;
 import io.trino.parquet.reader.MetadataReader;
+import io.trino.plugin.deltalake.delete.PageFilter;
+import io.trino.plugin.deltalake.delete.PositionDeleteFilter;
+import io.trino.plugin.deltalake.transactionlog.DeletionVectorEntry;
 import io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.ColumnMappingMode;
 import io.trino.plugin.hive.FileFormatDataSourceStats;
 import io.trino.plugin.hive.HiveColumnHandle;
@@ -35,6 +40,7 @@
 import io.trino.plugin.hive.parquet.ParquetReaderConfig;
 import io.trino.plugin.hive.parquet.TrinoParquetDataSource;
 import io.trino.spi.Page;
+import io.trino.spi.TrinoException;
 import io.trino.spi.block.Block;
 import io.trino.spi.block.LongArrayBlock;
 import io.trino.spi.connector.ColumnHandle;
@@ -56,13 +62,15 @@
 import org.apache.parquet.schema.MessageType;
 import org.apache.parquet.schema.Type;
 import org.joda.time.DateTimeZone;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Supplier;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -71,11 +79,13 @@
 import static io.airlift.slice.SizeOf.SIZE_OF_LONG;
 import static io.trino.plugin.deltalake.DeltaHiveTypeTranslator.toHiveType;
 import static io.trino.plugin.deltalake.DeltaLakeColumnHandle.ROW_ID_COLUMN_NAME;
+import static io.trino.plugin.deltalake.DeltaLakeColumnHandle.rowPositionColumnHandle;
 import static io.trino.plugin.deltalake.DeltaLakeColumnType.REGULAR;
+import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_INVALID_SCHEMA;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getParquetMaxReadBlockRowCount;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getParquetMaxReadBlockSize;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isParquetUseColumnIndex;
-import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.ColumnMappingMode.NONE;
+import static io.trino.plugin.deltalake.delete.DeletionVectors.readDeletionVectors;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.extractSchema;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.getColumnMappingMode;
 import static io.trino.plugin.hive.parquet.ParquetPageSourceFactory.PARQUET_ROW_INDEX_COLUMN;
@@ -172,6 +182,7 @@ public ConnectorPageSource createPageSource(
         if (filteredSplitPredicate.isAll() &&
                 split.getStart() == 0 && split.getLength() == split.getFileSize() &&
                 split.getFileRowCount().isPresent() &&
+                split.getDeletionVector().isEmpty() &&
                 (regularColumns.isEmpty() || onlyRowIdColumn(regularColumns))) {
             return new DeltaLakePageSource(
                     deltaLakeColumns,
@@ -182,11 +193,13 @@ public ConnectorPageSource createPageSource(
                     Optional.empty(),
                     split.getPath(),
                     split.getFileSize(),
-                    split.getFileModifiedTime());
+                    split.getFileModifiedTime(),
+                    Optional::empty);
         }
 
         Location location = Location.of(split.getPath());
-        TrinoInputFile inputFile = fileSystemFactory.create(session).newInputFile(location, split.getFileSize());
+        TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+        TrinoInputFile inputFile = fileSystem.newInputFile(location, split.getFileSize());
         ParquetReaderOptions options = parquetReaderOptions.withMaxReadBlockSize(getParquetMaxReadBlockSize(session))
                 .withMaxReadBlockRowCount(getParquetMaxReadBlockRowCount(session))
                 .withUseColumnIndex(isParquetUseColumnIndex(session));
@@ -204,6 +217,9 @@ public ConnectorPageSource createPageSource(
                     hiveColumnHandles::add,
                     () -> missingColumnNames.add(column.getBaseColumnName()));
         }
+        if (split.getDeletionVector().isPresent() && !regularColumns.contains(rowPositionColumnHandle())) {
+            hiveColumnHandles.add(PARQUET_ROW_INDEX_COLUMN);
+        }
 
         TupleDomain<HiveColumnHandle> parquetPredicate = getParquetTupleDomain(filteredSplitPredicate.simplify(domainCompactionThreshold), columnMappingMode, parquetFieldIdToName);
 
@@ -227,6 +243,19 @@ public ConnectorPageSource createPageSource(
                         column -> ((HiveColumnHandle) column).getType(),
                         HivePageSourceProvider::getProjection));
 
+        Supplier<Optional<PageFilter>> deletePredicate = Suppliers.memoize(() -> {
+            if (split.getDeletionVector().isEmpty()) {
+                return Optional.empty();
+            }
+
+            List<DeltaLakeColumnHandle> requiredColumns = ImmutableList.<DeltaLakeColumnHandle>builderWithExpectedSize(deltaLakeColumns.size() + 1)
+                    .addAll(deltaLakeColumns)
+                    .add(rowPositionColumnHandle())
+                    .build();
+            PositionDeleteFilter deleteFilter = readDeletes(fileSystem, Location.of(table.location()), split.getDeletionVector().get());
+            return Optional.of(deleteFilter.createPredicate(requiredColumns));
+        });
+
         return new DeltaLakePageSource(
                 deltaLakeColumns,
                 missingColumnNames.build(),
@@ -236,7 +265,22 @@ public ConnectorPageSource createPageSource(
                 projectionsAdapter,
                 split.getPath(),
                 split.getFileSize(),
-                split.getFileModifiedTime());
+                split.getFileModifiedTime(),
+                deletePredicate);
+    }
+
+    private PositionDeleteFilter readDeletes(
+            TrinoFileSystem fileSystem,
+            Location tableLocation,
+            DeletionVectorEntry deletionVector)
+    {
+        try {
+            Roaring64NavigableMap deletedRows = readDeletionVectors(fileSystem, tableLocation, deletionVector);
+            return new PositionDeleteFilter(deletedRows);
+        }
+        catch (IOException e) {
+            throw new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Failed to read deletion vectors", e);
+        }
     }
 
     public Map<Integer, String> loadParquetIdAndNameMapping(TrinoInputFile inputFile, ParquetReaderOptions options)

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeSplit.java

@@ -19,6 +19,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import io.airlift.slice.SizeOf;
+import io.trino.plugin.deltalake.transactionlog.DeletionVectorEntry;
 import io.trino.spi.HostAddress;
 import io.trino.spi.SplitWeight;
 import io.trino.spi.connector.ConnectorSplit;
@@ -47,6 +48,7 @@ public class DeltaLakeSplit
     private final long fileSize;
     private final Optional<Long> fileRowCount;
     private final long fileModifiedTime;
+    private final Optional<DeletionVectorEntry> deletionVector;
     private final SplitWeight splitWeight;
     private final TupleDomain<DeltaLakeColumnHandle> statisticsPredicate;
     private final Map<String, Optional<String>> partitionKeys;
@@ -59,6 +61,7 @@ public DeltaLakeSplit(
             @JsonProperty("fileSize") long fileSize,
             @JsonProperty("rowCount") Optional<Long> fileRowCount,
             @JsonProperty("fileModifiedTime") long fileModifiedTime,
+            @JsonProperty("deletionVector") Optional<DeletionVectorEntry> deletionVector,
             @JsonProperty("splitWeight") SplitWeight splitWeight,
             @JsonProperty("statisticsPredicate") TupleDomain<DeltaLakeColumnHandle> statisticsPredicate,
             @JsonProperty("partitionKeys") Map<String, Optional<String>> partitionKeys)
@@ -69,6 +72,7 @@ public DeltaLakeSplit(
         this.fileSize = fileSize;
         this.fileRowCount = requireNonNull(fileRowCount, "rowCount is null");
         this.fileModifiedTime = fileModifiedTime;
+        this.deletionVector = requireNonNull(deletionVector, "deletionVector is null");
         this.splitWeight = requireNonNull(splitWeight, "splitWeight is null");
         this.statisticsPredicate = requireNonNull(statisticsPredicate, "statisticsPredicate is null");
         this.partitionKeys = requireNonNull(partitionKeys, "partitionKeys is null");
@@ -130,6 +134,12 @@ public long getFileModifiedTime()
         return fileModifiedTime;
     }
 
+    @JsonProperty
+    public Optional<DeletionVectorEntry> getDeletionVector()
+    {
+        return deletionVector;
+    }
+
     /**
      * A TupleDomain representing the min/max statistics from the file this split was generated from. This does not contain any partitioning information.
      */
@@ -151,6 +161,7 @@ public long getRetainedSizeInBytes()
         return INSTANCE_SIZE
                 + estimatedSizeOf(path)
                 + sizeOf(fileRowCount, value -> LONG_INSTANCE_SIZE)
+                + sizeOf(deletionVector, DeletionVectorEntry::sizeInBytes)
                 + splitWeight.getRetainedSizeInBytes()
                 + statisticsPredicate.getRetainedSizeInBytes(DeltaLakeColumnHandle::getRetainedSizeInBytes)
                 + estimatedSizeOf(partitionKeys, SizeOf::estimatedSizeOf, value -> sizeOf(value, SizeOf::estimatedSizeOf));
@@ -175,6 +186,7 @@ public String toString()
                 .add("length", length)
                 .add("fileSize", fileSize)
                 .add("rowCount", fileRowCount)
+                .add("deletionVector", deletionVector)
                 .add("statisticsPredicate", statisticsPredicate)
                 .add("partitionKeys", partitionKeys)
                 .toString();
@@ -195,13 +207,14 @@ public boolean equals(Object o)
                 fileSize == that.fileSize &&
                 path.equals(that.path) &&
                 fileRowCount.equals(that.fileRowCount) &&
+                deletionVector.equals(that.deletionVector) &&
                 Objects.equals(statisticsPredicate, that.statisticsPredicate) &&
                 Objects.equals(partitionKeys, that.partitionKeys);
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hash(path, start, length, fileSize, fileRowCount, statisticsPredicate, partitionKeys);
+        return Objects.hash(path, start, length, fileSize, fileRowCount, deletionVector, statisticsPredicate, partitionKeys);
     }
 }

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeSplitManager.java

@@ -290,6 +290,7 @@ private List<DeltaLakeSplit> splitsForFile(
                     fileSize,
                     addFileEntry.getStats().flatMap(DeltaLakeFileStatistics::getNumRecords),
                     addFileEntry.getModificationTime(),
+                    addFileEntry.getDeletionVector(),
                     SplitWeight.standard(),
                     statisticsPredicate,
                     partitionKeys));
@@ -314,6 +315,7 @@ private List<DeltaLakeSplit> splitsForFile(
                     fileSize,
                     Optional.empty(),
                     addFileEntry.getModificationTime(),
+                    addFileEntry.getDeletionVector(),
                     SplitWeight.fromProportion(Math.min(Math.max((double) splitSize / maxSplitSize, minimumAssignedSplitWeight), 1.0)),
                     statisticsPredicate,
                     partitionKeys));

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/delete/PageFilter.java

@@ -0,0 +1,21 @@
+/*
+ * Licensed 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 io.trino.plugin.deltalake.delete;
+
+import io.trino.spi.Page;
+
+import java.util.function.Function;
+
+public interface PageFilter
+        extends Function<Page, Page> {}