Add support for writer version 7 in Delta Lake

Support the following writer features:
* appendOnly
* invariants
* checkConstraints
* changeDataFeed
* columnMapping

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

@@ -17,6 +17,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
 import io.trino.plugin.deltalake.transactionlog.MetadataEntry;
+import io.trino.plugin.deltalake.transactionlog.ProtocolEntry;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.SchemaTableName;
 
@@ -30,6 +31,7 @@ public class DeltaLakeInsertTableHandle
     private final SchemaTableName tableName;
     private final String location;
     private final MetadataEntry metadataEntry;
+    private final ProtocolEntry protocolEntry;
     private final List<DeltaLakeColumnHandle> inputColumns;
     private final long readVersion;
     private final boolean retriesEnabled;
@@ -39,12 +41,14 @@ public DeltaLakeInsertTableHandle(
             @JsonProperty("tableName") SchemaTableName tableName,
             @JsonProperty("location") String location,
             @JsonProperty("metadataEntry") MetadataEntry metadataEntry,
+            @JsonProperty("protocolEntry") ProtocolEntry protocolEntry,
             @JsonProperty("inputColumns") List<DeltaLakeColumnHandle> inputColumns,
             @JsonProperty("readVersion") long readVersion,
             @JsonProperty("retriesEnabled") boolean retriesEnabled)
     {
         this.tableName = requireNonNull(tableName, "tableName is null");
         this.metadataEntry = requireNonNull(metadataEntry, "metadataEntry is null");
+        this.protocolEntry = requireNonNull(protocolEntry, "protocolEntry is null");
         this.inputColumns = ImmutableList.copyOf(inputColumns);
         this.location = requireNonNull(location, "location is null");
         this.readVersion = readVersion;
@@ -69,6 +73,12 @@ public MetadataEntry getMetadataEntry()
         return metadataEntry;
     }
 
+    @JsonProperty
+    public ProtocolEntry getProtocolEntry()
+    {
+        return protocolEntry;
+    }
+
     @JsonProperty
     public List<DeltaLakeColumnHandle> getInputColumns()
     {

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

@@ -119,7 +119,7 @@ public ConnectorPageSink createPageSink(ConnectorTransactionHandle transactionHa
     {
         DeltaLakeInsertTableHandle tableHandle = (DeltaLakeInsertTableHandle) insertTableHandle;
         MetadataEntry metadataEntry = tableHandle.getMetadataEntry();
-        DeltaLakeParquetSchemaMapping parquetSchemaMapping = createParquetSchemaMapping(metadataEntry, typeManager);
+        DeltaLakeParquetSchemaMapping parquetSchemaMapping = createParquetSchemaMapping(metadataEntry, tableHandle.getProtocolEntry(), typeManager);
         return new DeltaLakePageSink(
                 typeManager.getTypeOperators(),
                 tableHandle.getInputColumns(),
@@ -142,7 +142,7 @@ public ConnectorPageSink createPageSink(ConnectorTransactionHandle transactionHa
         switch (executeHandle.getProcedureId()) {
             case OPTIMIZE:
                 DeltaTableOptimizeHandle optimizeHandle = (DeltaTableOptimizeHandle) executeHandle.getProcedureHandle();
-                DeltaLakeParquetSchemaMapping parquetSchemaMapping = createParquetSchemaMapping(optimizeHandle.getMetadataEntry(), typeManager);
+                DeltaLakeParquetSchemaMapping parquetSchemaMapping = createParquetSchemaMapping(optimizeHandle.getMetadataEntry(), optimizeHandle.getProtocolEntry(), typeManager);
                 return new DeltaLakePageSink(
                         typeManager.getTypeOperators(),
                         optimizeHandle.getTableColumns(),
@@ -167,7 +167,7 @@ public ConnectorMergeSink createMergeSink(ConnectorTransactionHandle transaction
         DeltaLakeMergeTableHandle merge = (DeltaLakeMergeTableHandle) mergeHandle;
         DeltaLakeInsertTableHandle tableHandle = merge.getInsertTableHandle();
         ConnectorPageSink pageSink = createPageSink(transactionHandle, session, tableHandle, pageSinkId);
-        DeltaLakeParquetSchemaMapping parquetSchemaMapping = createParquetSchemaMapping(tableHandle.getMetadataEntry(), typeManager);
+        DeltaLakeParquetSchemaMapping parquetSchemaMapping = createParquetSchemaMapping(tableHandle.getMetadataEntry(), tableHandle.getProtocolEntry(), typeManager);
 
         return new DeltaLakeMergeSink(
                 typeManager.getTypeOperators(),
@@ -183,7 +183,7 @@ public ConnectorMergeSink createMergeSink(ConnectorTransactionHandle transaction
                 tableHandle.getInputColumns(),
                 domainCompactionThreshold,
                 () -> createCdfPageSink(merge, session),
-                changeDataFeedEnabled(tableHandle.getMetadataEntry()),
+                changeDataFeedEnabled(tableHandle.getMetadataEntry(), tableHandle.getProtocolEntry()).orElse(false),
                 parquetSchemaMapping);
     }
 
@@ -193,7 +193,7 @@ private DeltaLakeCdfPageSink createCdfPageSink(
     {
         MetadataEntry metadataEntry = mergeTableHandle.getTableHandle().getMetadataEntry();
         Set<String> partitionKeys = mergeTableHandle.getTableHandle().getMetadataEntry().getOriginalPartitionColumns().stream().collect(toImmutableSet());
-        List<DeltaLakeColumnHandle> tableColumns = extractSchema(metadataEntry, typeManager).stream()
+        List<DeltaLakeColumnHandle> tableColumns = extractSchema(metadataEntry, mergeTableHandle.getTableHandle().getProtocolEntry(), typeManager).stream()
                 .map(metadata -> new DeltaLakeColumnHandle(
                         metadata.getName(),
                         metadata.getType(),
@@ -216,7 +216,7 @@ private DeltaLakeCdfPageSink createCdfPageSink(
                 .build();
         Location tableLocation = Location.of(mergeTableHandle.getTableHandle().getLocation());
 
-        DeltaLakeParquetSchemaMapping parquetSchemaMapping = createParquetSchemaMapping(metadataEntry, typeManager, true);
+        DeltaLakeParquetSchemaMapping parquetSchemaMapping = createParquetSchemaMapping(metadataEntry, mergeTableHandle.getTableHandle().getProtocolEntry(), typeManager, true);
 
         return new DeltaLakeCdfPageSink(
                 typeManager.getTypeOperators(),

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

@@ -141,7 +141,7 @@ public ConnectorPageSource createPageSource(
         Optional<List<String>> partitionValues = Optional.empty();
         if (deltaLakeColumns.stream().anyMatch(column -> column.getBaseColumnName().equals(ROW_ID_COLUMN_NAME))) {
             partitionValues = Optional.of(new ArrayList<>());
-            for (DeltaLakeColumnMetadata column : extractSchema(table.getMetadataEntry(), typeManager)) {
+            for (DeltaLakeColumnMetadata column : extractSchema(table.getMetadataEntry(), table.getProtocolEntry(), typeManager)) {
                 Optional<String> value = partitionKeys.get(column.getName());
                 if (value != null) {
                     partitionValues.get().add(value.orElse(null));
@@ -183,7 +183,7 @@ public ConnectorPageSource createPageSource(
                 .withMaxReadBlockRowCount(getParquetMaxReadBlockRowCount(session))
                 .withUseColumnIndex(isParquetUseColumnIndex(session));
 
-        ColumnMappingMode columnMappingMode = getColumnMappingMode(table.getMetadataEntry());
+        ColumnMappingMode columnMappingMode = getColumnMappingMode(table.getMetadataEntry(), table.getProtocolEntry());
         Map<Integer, String> parquetFieldIdToName = columnMappingMode == ColumnMappingMode.ID ? loadParquetIdAndNameMapping(inputFile, options) : ImmutableMap.of();
 
         ImmutableSet.Builder<String> missingColumnNames = ImmutableSet.builder();

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

@@ -21,6 +21,7 @@
 import io.airlift.json.ObjectMapperProvider;
 import io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport;
 import io.trino.plugin.deltalake.transactionlog.MetadataEntry;
+import io.trino.plugin.deltalake.transactionlog.ProtocolEntry;
 import io.trino.spi.Location;
 import io.trino.spi.TrinoException;
 import io.trino.spi.type.DecimalType;
@@ -86,14 +87,14 @@ public final class DeltaLakeParquetSchemas
 
     private DeltaLakeParquetSchemas() {}
 
-    public static DeltaLakeParquetSchemaMapping createParquetSchemaMapping(MetadataEntry metadataEntry, TypeManager typeManager)
+    public static DeltaLakeParquetSchemaMapping createParquetSchemaMapping(MetadataEntry metadataEntry, ProtocolEntry protocolEntry, TypeManager typeManager)
     {
-        return createParquetSchemaMapping(metadataEntry, typeManager, false);
+        return createParquetSchemaMapping(metadataEntry, protocolEntry, typeManager, false);
     }
 
-    public static DeltaLakeParquetSchemaMapping createParquetSchemaMapping(MetadataEntry metadataEntry, TypeManager typeManager, boolean addChangeDataFeedFields)
+    public static DeltaLakeParquetSchemaMapping createParquetSchemaMapping(MetadataEntry metadataEntry, ProtocolEntry protocolEntry, TypeManager typeManager, boolean addChangeDataFeedFields)
     {
-        DeltaLakeSchemaSupport.ColumnMappingMode columnMappingMode = getColumnMappingMode(metadataEntry);
+        DeltaLakeSchemaSupport.ColumnMappingMode columnMappingMode = getColumnMappingMode(metadataEntry, protocolEntry);
         return createParquetSchemaMapping(
                 metadataEntry.getSchemaString(),
                 typeManager,

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

@@ -173,7 +173,7 @@ private Stream<DeltaLakeSplit> getSplits(
                         .map(DeltaLakeColumnHandle.class::cast))
                 .map(DeltaLakeColumnHandle::getBaseColumnName)
                 .collect(toImmutableSet());
-        List<DeltaLakeColumnMetadata> schema = extractSchema(tableHandle.getMetadataEntry(), typeManager);
+        List<DeltaLakeColumnMetadata> schema = extractSchema(tableHandle.getMetadataEntry(), tableHandle.getProtocolEntry(), typeManager);
         List<DeltaLakeColumnMetadata> predicatedColumns = schema.stream()
                 .filter(column -> predicatedColumnNames.contains(column.getName()))
                 .collect(toImmutableList());

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

@@ -19,6 +19,7 @@
 import com.google.common.collect.ImmutableSet;
 import io.airlift.units.DataSize;
 import io.trino.plugin.deltalake.transactionlog.MetadataEntry;
+import io.trino.plugin.deltalake.transactionlog.ProtocolEntry;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.predicate.TupleDomain;
 
@@ -46,6 +47,7 @@ public enum WriteType
     private final boolean managed;
     private final String location;
     private final MetadataEntry metadataEntry;
+    private final ProtocolEntry protocolEntry;
     private final TupleDomain<DeltaLakeColumnHandle> enforcedPartitionConstraint;
     private final TupleDomain<DeltaLakeColumnHandle> nonPartitionConstraint;
     private final Optional<WriteType> writeType;
@@ -73,6 +75,7 @@ public DeltaLakeTableHandle(
             @JsonProperty("managed") boolean managed,
             @JsonProperty("location") String location,
             @JsonProperty("metadataEntry") MetadataEntry metadataEntry,
+            @JsonProperty("protocolEntry") ProtocolEntry protocolEntry,
             @JsonProperty("enforcedPartitionConstraint") TupleDomain<DeltaLakeColumnHandle> enforcedPartitionConstraint,
             @JsonProperty("nonPartitionConstraint") TupleDomain<DeltaLakeColumnHandle> nonPartitionConstraint,
             @JsonProperty("writeType") Optional<WriteType> writeType,
@@ -88,6 +91,7 @@ public DeltaLakeTableHandle(
                 managed,
                 location,
                 metadataEntry,
+                protocolEntry,
                 enforcedPartitionConstraint,
                 nonPartitionConstraint,
                 ImmutableSet.of(),
@@ -107,6 +111,7 @@ public DeltaLakeTableHandle(
             boolean managed,
             String location,
             MetadataEntry metadataEntry,
+            ProtocolEntry protocolEntry,
             TupleDomain<DeltaLakeColumnHandle> enforcedPartitionConstraint,
             TupleDomain<DeltaLakeColumnHandle> nonPartitionConstraint,
             Set<DeltaLakeColumnHandle> constraintColumns,
@@ -124,6 +129,7 @@ public DeltaLakeTableHandle(
         this.managed = managed;
         this.location = requireNonNull(location, "location is null");
         this.metadataEntry = requireNonNull(metadataEntry, "metadataEntry is null");
+        this.protocolEntry = requireNonNull(protocolEntry, "protocolEntry is null");
         this.enforcedPartitionConstraint = requireNonNull(enforcedPartitionConstraint, "enforcedPartitionConstraint is null");
         this.nonPartitionConstraint = requireNonNull(nonPartitionConstraint, "nonPartitionConstraint is null");
         this.writeType = requireNonNull(writeType, "writeType is null");
@@ -147,6 +153,7 @@ public DeltaLakeTableHandle withProjectedColumns(Set<DeltaLakeColumnHandle> proj
                 managed,
                 location,
                 metadataEntry,
+                protocolEntry,
                 enforcedPartitionConstraint,
                 nonPartitionConstraint,
                 constraintColumns,
@@ -168,6 +175,7 @@ public DeltaLakeTableHandle forOptimize(boolean recordScannedFiles, DataSize max
                 managed,
                 location,
                 metadataEntry,
+                protocolEntry,
                 enforcedPartitionConstraint,
                 nonPartitionConstraint,
                 constraintColumns,
@@ -234,6 +242,12 @@ public MetadataEntry getMetadataEntry()
         return metadataEntry;
     }
 
+    @JsonProperty
+    public ProtocolEntry getProtocolEntry()
+    {
+        return protocolEntry;
+    }
+
     @JsonProperty
     public TupleDomain<DeltaLakeColumnHandle> getEnforcedPartitionConstraint()
     {
@@ -324,6 +338,7 @@ public boolean equals(Object o)
                 managed == that.managed &&
                 Objects.equals(location, that.location) &&
                 Objects.equals(metadataEntry, that.metadataEntry) &&
+                Objects.equals(protocolEntry, that.protocolEntry) &&
                 Objects.equals(enforcedPartitionConstraint, that.enforcedPartitionConstraint) &&
                 Objects.equals(nonPartitionConstraint, that.nonPartitionConstraint) &&
                 Objects.equals(writeType, that.writeType) &&
@@ -344,6 +359,7 @@ public int hashCode()
                 managed,
                 location,
                 metadataEntry,
+                protocolEntry,
                 enforcedPartitionConstraint,
                 nonPartitionConstraint,
                 writeType,

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

@@ -19,6 +19,7 @@
 import io.airlift.units.DataSize;
 import io.trino.plugin.deltalake.DeltaLakeColumnHandle;
 import io.trino.plugin.deltalake.transactionlog.MetadataEntry;
+import io.trino.plugin.deltalake.transactionlog.ProtocolEntry;
 
 import java.util.List;
 import java.util.Optional;
@@ -30,6 +31,7 @@ public class DeltaTableOptimizeHandle
         extends DeltaTableProcedureHandle
 {
     private final MetadataEntry metadataEntry;
+    private final ProtocolEntry protocolEntry;
     private final List<DeltaLakeColumnHandle> tableColumns;
     private final List<String> originalPartitionColumns;
     private final DataSize maxScannedFileSize;
@@ -39,13 +41,15 @@ public class DeltaTableOptimizeHandle
     @JsonCreator
     public DeltaTableOptimizeHandle(
             MetadataEntry metadataEntry,
+            ProtocolEntry protocolEntry,
             List<DeltaLakeColumnHandle> tableColumns,
             List<String> originalPartitionColumns,
             DataSize maxScannedFileSize,
             Optional<Long> currentVersion,
             boolean retriesEnabled)
     {
         this.metadataEntry = requireNonNull(metadataEntry, "metadataEntry is null");
+        this.protocolEntry = requireNonNull(protocolEntry, "protocolEntry is null");
         this.tableColumns = ImmutableList.copyOf(requireNonNull(tableColumns, "tableColumns is null"));
         this.originalPartitionColumns = ImmutableList.copyOf(requireNonNull(originalPartitionColumns, "originalPartitionColumns is null"));
         this.maxScannedFileSize = requireNonNull(maxScannedFileSize, "maxScannedFileSize is null");
@@ -58,6 +62,7 @@ public DeltaTableOptimizeHandle withCurrentVersion(long currentVersion)
         checkState(this.currentVersion.isEmpty(), "currentVersion already set");
         return new DeltaTableOptimizeHandle(
                 metadataEntry,
+                protocolEntry,
                 tableColumns,
                 originalPartitionColumns,
                 maxScannedFileSize,
@@ -71,6 +76,12 @@ public MetadataEntry getMetadataEntry()
         return metadataEntry;
     }
 
+    @JsonProperty
+    public ProtocolEntry getProtocolEntry()
+    {
+        return protocolEntry;
+    }
+
     @JsonProperty
     public List<DeltaLakeColumnHandle> getTableColumns()
     {

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

@@ -14,6 +14,7 @@
 package io.trino.plugin.deltalake.procedure;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
 import io.airlift.log.Logger;
@@ -60,6 +61,7 @@
 import static io.trino.plugin.deltalake.DeltaLakeMetadata.MAX_WRITER_VERSION;
 import static io.trino.plugin.deltalake.DeltaLakeMetadata.checkValidTableHandle;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getVacuumMinRetention;
+import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.unsupportedWriterFeatures;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogUtil.TRANSACTION_LOG_DIRECTORY;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogUtil.getTransactionLogDir;
 import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
@@ -175,11 +177,15 @@ private void doVacuum(
         accessControl.checkCanDeleteFromTable(null, tableName);
 
         TableSnapshot tableSnapshot = transactionLogAccess.loadSnapshot(tableName, handle.getLocation(), session);
-        // TODO https://github.com/trinodb/trino/issues/15873 Check writer features when supporting writer version 7
         ProtocolEntry protocolEntry = transactionLogAccess.getProtocolEntry(session, tableSnapshot);
         if (protocolEntry.getMinWriterVersion() > MAX_WRITER_VERSION) {
             throw new TrinoException(NOT_SUPPORTED, "Cannot execute vacuum procedure with %d writer version".formatted(protocolEntry.getMinWriterVersion()));
         }
+        Set<String> unsupportedWriterFeatures = unsupportedWriterFeatures(protocolEntry.getWriterFeatures().orElse(ImmutableSet.of()));
+        if (!unsupportedWriterFeatures.isEmpty()) {
+            throw new TrinoException(NOT_SUPPORTED, "Cannot execute vacuum procedure with %s writer features".formatted(unsupportedWriterFeatures));
+        }
+
         String tableLocation = tableSnapshot.getTableLocation();
         String transactionLogDir = getTransactionLogDir(tableLocation);
         TrinoFileSystem fileSystem = fileSystemFactory.create(session);

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

@@ -76,7 +76,7 @@ public TableStatistics getTableStatistics(ConnectorSession session, DeltaLakeTab
         double numRecords = 0L;
 
         MetadataEntry metadata = tableHandle.getMetadataEntry();
-        List<DeltaLakeColumnMetadata> columnMetadata = DeltaLakeSchemaSupport.extractSchema(metadata, typeManager);
+        List<DeltaLakeColumnMetadata> columnMetadata = DeltaLakeSchemaSupport.extractSchema(metadata, tableHandle.getProtocolEntry(), typeManager);
         List<DeltaLakeColumnHandle> columns = columnMetadata.stream()
                 .map(columnMeta -> new DeltaLakeColumnHandle(
                         columnMeta.getName(),