Infer required Delta Lake version from table features

docs/src/main/sphinx/connector/delta-lake.rst

@@ -157,16 +157,6 @@ values. Typical usage does not require you to configure them.
     * - ``delta.register-table-procedure.enabled``
       - Enable to allow users to call the ``register_table`` procedure
       - ``false``
-    * - ``delta.default-reader-version``
-      - The default reader version used by new tables.
-        The value can be overridden for a specific table with the
-        ``reader_version`` table property.
-      - ``1``
-    * - ``delta.default-writer-version``
-      - The default writer version used by new tables.
-        The value can be overridden for a specific table with the
-        ``writer_version`` table property.
-      - ``2``
 
 The following table describes performance tuning catalog properties for the
 connector.
@@ -573,21 +563,15 @@ The following properties are available for use:
     - Set the checkpoint interval in seconds.
   * - ``change_data_feed_enabled``
     - Enables storing change data feed entries.
-  * - ``reader_version``
-    - Set reader version.
-  * - ``writer_version``
-    - Set writer version.
 
-The following example uses all six table properties::
+The following example uses all available table properties::
 
   CREATE TABLE example.default.example_partitioned_table
   WITH (
     location = 's3://my-bucket/a/path',
     partitioned_by = ARRAY['regionkey'],
     checkpoint_interval = 5,
-    change_data_feed_enabled = true,
-    reader_version = 2,
-    writer_version = 4
+    change_data_feed_enabled = true
   )
   AS SELECT name, comment, regionkey FROM tpch.tiny.nation;
 

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

@@ -25,7 +25,6 @@
 
 import javax.validation.constraints.DecimalMax;
 import javax.validation.constraints.DecimalMin;
-import javax.validation.constraints.Max;
 import javax.validation.constraints.Min;
 import javax.validation.constraints.NotNull;
 
@@ -49,12 +48,6 @@ public class DeltaLakeConfig
     @VisibleForTesting
     static final DataSize DEFAULT_DATA_FILE_CACHE_SIZE = DataSize.succinctBytes(Math.floorDiv(Runtime.getRuntime().maxMemory(), 10L));
 
-    public static final int MIN_READER_VERSION = 1;
-    public static final int MIN_WRITER_VERSION = 2;
-    // The highest reader and writer versions Trino supports writing to
-    public static final int MAX_READER_VERSION = 2;
-    public static final int MAX_WRITER_VERSION = 4;
-
     private Duration metadataCacheTtl = new Duration(5, TimeUnit.MINUTES);
     private long metadataCacheMaxSize = 1000;
     private DataSize dataFileCacheSize = DEFAULT_DATA_FILE_CACHE_SIZE;
@@ -84,8 +77,6 @@ public class DeltaLakeConfig
     private boolean uniqueTableLocation = true;
     private boolean legacyCreateTableWithExistingLocationEnabled;
     private boolean registerTableProcedureEnabled;
-    private int defaultReaderVersion = MIN_READER_VERSION;
-    private int defaultWriterVersion = MIN_WRITER_VERSION;
 
     public Duration getMetadataCacheTtl()
     {
@@ -484,34 +475,4 @@ public DeltaLakeConfig setRegisterTableProcedureEnabled(boolean registerTablePro
         this.registerTableProcedureEnabled = registerTableProcedureEnabled;
         return this;
     }
-
-    @Min(value = MIN_READER_VERSION, message = "Must be in between " + MIN_READER_VERSION + " and " + MAX_READER_VERSION)
-    @Max(value = MAX_READER_VERSION, message = "Must be in between " + MIN_READER_VERSION + " and " + MAX_READER_VERSION)
-    public int getDefaultReaderVersion()
-    {
-        return defaultReaderVersion;
-    }
-
-    @Config("delta.default-reader-version")
-    @ConfigDescription("The default reader version used by new tables")
-    public DeltaLakeConfig setDefaultReaderVersion(int defaultReaderVersion)
-    {
-        this.defaultReaderVersion = defaultReaderVersion;
-        return this;
-    }
-
-    @Min(value = MIN_WRITER_VERSION, message = "Must be in between " + MIN_WRITER_VERSION + " and " + MAX_WRITER_VERSION)
-    @Max(value = MAX_WRITER_VERSION, message = "Must be in between " + MIN_WRITER_VERSION + " and " + MAX_WRITER_VERSION)
-    public int getDefaultWriterVersion()
-    {
-        return defaultWriterVersion;
-    }
-
-    @Config("delta.default-writer-version")
-    @ConfigDescription("The default writer version used by new tables")
-    public DeltaLakeConfig setDefaultWriterVersion(int defaultWriterVersion)
-    {
-        this.defaultWriterVersion = defaultWriterVersion;
-        return this;
-    }
 }

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

@@ -159,6 +159,7 @@
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static com.google.common.collect.MoreCollectors.toOptional;
 import static com.google.common.collect.Sets.difference;
+import static com.google.common.primitives.Ints.max;
 import static io.trino.plugin.deltalake.DataFileInfo.DataFileType.DATA;
 import static io.trino.plugin.deltalake.DeltaLakeAnalyzeProperties.getColumnNames;
 import static io.trino.plugin.deltalake.DeltaLakeAnalyzeProperties.getFilesModifiedAfterProperty;
@@ -171,7 +172,6 @@
 import static io.trino.plugin.deltalake.DeltaLakeColumnType.PARTITION_KEY;
 import static io.trino.plugin.deltalake.DeltaLakeColumnType.REGULAR;
 import static io.trino.plugin.deltalake.DeltaLakeColumnType.SYNTHESIZED;
-import static io.trino.plugin.deltalake.DeltaLakeConfig.MAX_WRITER_VERSION;
 import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_BAD_WRITE;
 import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_INVALID_SCHEMA;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getHiveCatalogName;
@@ -183,14 +183,10 @@
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.CHECKPOINT_INTERVAL_PROPERTY;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.LOCATION_PROPERTY;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.PARTITIONED_BY_PROPERTY;
-import static io.trino.plugin.deltalake.DeltaLakeTableProperties.READER_VERSION_PROPERTY;
-import static io.trino.plugin.deltalake.DeltaLakeTableProperties.WRITER_VERSION_PROPERTY;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.getChangeDataFeedEnabled;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.getCheckpointInterval;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.getLocation;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.getPartitionedBy;
-import static io.trino.plugin.deltalake.DeltaLakeTableProperties.getReaderVersion;
-import static io.trino.plugin.deltalake.DeltaLakeTableProperties.getWriterVersion;
 import static io.trino.plugin.deltalake.metastore.HiveMetastoreBackedDeltaLakeMetastore.TABLE_PROVIDER_PROPERTY;
 import static io.trino.plugin.deltalake.metastore.HiveMetastoreBackedDeltaLakeMetastore.TABLE_PROVIDER_VALUE;
 import static io.trino.plugin.deltalake.procedure.DeltaLakeTableProcedureId.OPTIMIZE;
@@ -210,6 +206,7 @@
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.serializeStatsAsJson;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.validateType;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.verifySupportedColumnMapping;
+import static io.trino.plugin.deltalake.transactionlog.MetadataEntry.DELTA_CHANGE_DATA_FEED_ENABLED_PROPERTY;
 import static io.trino.plugin.deltalake.transactionlog.MetadataEntry.configurationForNewTable;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogParser.getMandatoryCurrentVersion;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogUtil.getTransactionLogDir;
@@ -287,6 +284,10 @@ public class DeltaLakeMetadata
     public static final String CHANGE_COLUMN_OPERATION = "CHANGE COLUMN";
     public static final String ISOLATION_LEVEL = "WriteSerializable";
 
+    public static final int DEFAULT_READER_VERSION = 1;
+    public static final int DEFAULT_WRITER_VERSION = 2;
+    // The highest reader and writer versions Trino supports writing to
+    public static final int MAX_WRITER_VERSION = 4;
     private static final int CDF_SUPPORTED_WRITER_VERSION = 4;
 
     // Matches the dummy column Databricks stores in the metastore
@@ -297,7 +298,7 @@ public class DeltaLakeMetadata
             .add(NUMBER_OF_DISTINCT_VALUES_SUMMARY)
             .build();
     private static final String ENABLE_NON_CONCURRENT_WRITES_CONFIGURATION_KEY = "delta.enable-non-concurrent-writes";
-    public static final Set<String> UPDATABLE_TABLE_PROPERTIES = ImmutableSet.of(READER_VERSION_PROPERTY, WRITER_VERSION_PROPERTY);
+    public static final Set<String> UPDATABLE_TABLE_PROPERTIES = ImmutableSet.of(CHANGE_DATA_FEED_ENABLED_PROPERTY);
 
     private final DeltaLakeMetastore metastore;
     private final TrinoFileSystemFactory fileSystemFactory;
@@ -320,8 +321,6 @@ public class DeltaLakeMetadata
     private final boolean deleteSchemaLocationsFallback;
     private final boolean useUniqueTableLocation;
     private final boolean allowManagedTableRename;
-    private final int defaultReaderVersion;
-    private final int defaultWriterVersion;
 
     public DeltaLakeMetadata(
             DeltaLakeMetastore metastore,
@@ -342,9 +341,7 @@ public DeltaLakeMetadata(
             DeltaLakeRedirectionsProvider deltaLakeRedirectionsProvider,
             ExtendedStatisticsAccess statisticsAccess,
             boolean useUniqueTableLocation,
-            boolean allowManagedTableRename,
-            int defaultReaderVersion,
-            int defaultWriterVersion)
+            boolean allowManagedTableRename)
     {
         this.metastore = requireNonNull(metastore, "metastore is null");
         this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null");
@@ -366,8 +363,6 @@ public DeltaLakeMetadata(
         this.deleteSchemaLocationsFallback = deleteSchemaLocationsFallback;
         this.useUniqueTableLocation = useUniqueTableLocation;
         this.allowManagedTableRename = allowManagedTableRename;
-        this.defaultReaderVersion = defaultReaderVersion;
-        this.defaultWriterVersion = defaultWriterVersion;
     }
 
     @Override
@@ -488,10 +483,6 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
         Optional<Boolean> changeDataFeedEnabled = tableHandle.getMetadataEntry().isChangeDataFeedEnabled();
         changeDataFeedEnabled.ifPresent(value -> properties.put(CHANGE_DATA_FEED_ENABLED_PROPERTY, value));
 
-        ProtocolEntry protocolEntry = getProtocolEntry(session, tableHandle.getSchemaTableName());
-        properties.put(READER_VERSION_PROPERTY, protocolEntry.getMinReaderVersion());
-        properties.put(WRITER_VERSION_PROPERTY, protocolEntry.getMinWriterVersion());
-
         return new ConnectorTableMetadata(
                 tableHandle.getSchemaTableName(),
                 columns,
@@ -758,7 +749,7 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
                         CREATE_TABLE_OPERATION,
                         session,
                         tableMetadata.getComment(),
-                        getProtocolEntry(tableMetadata.getProperties()));
+                        protocolEntryForNewTable(tableMetadata.getProperties()));
 
                 setRollback(() -> deleteRecursivelyIfExists(fileSystem, deltaLogDirectory));
                 transactionLogWriter.flush();
@@ -883,7 +874,7 @@ public DeltaLakeOutputTableHandle beginCreateTable(ConnectorSession session, Con
                 external,
                 tableMetadata.getComment(),
                 getChangeDataFeedEnabled(tableMetadata.getProperties()),
-                getProtocolEntry(tableMetadata.getProperties()));
+                protocolEntryForNewTable(tableMetadata.getProperties()));
     }
 
     private Optional<String> getSchemaLocation(Database database)
@@ -1756,28 +1747,15 @@ private ProtocolEntry getProtocolEntry(ConnectorSession session, SchemaTableName
         return metastore.getProtocol(session, metastore.getSnapshot(schemaTableName, session));
     }
 
-    private ProtocolEntry getProtocolEntry(Map<String, Object> properties)
+    private ProtocolEntry protocolEntryForNewTable(Map<String, Object> properties)
     {
-        Optional<Integer> readerVersion = getReaderVersion(properties);
-        Optional<Integer> writerVersion = getWriterVersion(properties);
+        int writerVersion = DEFAULT_WRITER_VERSION;
         Optional<Boolean> changeDataFeedEnabled = getChangeDataFeedEnabled(properties);
-
         if (changeDataFeedEnabled.isPresent() && changeDataFeedEnabled.get()) {
-            if (writerVersion.isPresent()) {
-                if (writerVersion.get() < CDF_SUPPORTED_WRITER_VERSION) {
-                    throw new TrinoException(
-                            INVALID_TABLE_PROPERTY,
-                            WRITER_VERSION_PROPERTY + " cannot be set less than " + CDF_SUPPORTED_WRITER_VERSION + " when cdf is enabled");
-                }
-            }
-            else {
-                // Enabling cdf (change data feed) requires setting the writer version to 4
-                writerVersion = Optional.of(CDF_SUPPORTED_WRITER_VERSION);
-            }
+            // Enabling cdf (change data feed) requires setting the writer version to 4
+            writerVersion = CDF_SUPPORTED_WRITER_VERSION;
         }
-        return new ProtocolEntry(
-                readerVersion.orElse(defaultReaderVersion),
-                writerVersion.orElse(defaultWriterVersion));
+        return new ProtocolEntry(DEFAULT_READER_VERSION, writerVersion);
     }
 
     private void writeCheckpointIfNeeded(ConnectorSession session, SchemaTableName table, Optional<Long> checkpointInterval, long newVersion)
@@ -1889,39 +1867,35 @@ public void setTableProperties(ConnectorSession session, ConnectorTableHandle ta
         DeltaLakeTableHandle handle = (DeltaLakeTableHandle) tableHandle;
         ProtocolEntry currentProtocolEntry = getProtocolEntry(session, handle.getSchemaTableName());
 
-        Optional<Integer> readerVersion = Optional.empty();
-        if (properties.containsKey(READER_VERSION_PROPERTY)) {
-            readerVersion = Optional.of((int) properties.get(READER_VERSION_PROPERTY)
-                    .orElseThrow(() -> new IllegalArgumentException("The " + READER_VERSION_PROPERTY + " property cannot be empty")));
-            if (readerVersion.get() < currentProtocolEntry.getMinReaderVersion()) {
-                throw new TrinoException(INVALID_TABLE_PROPERTY, format(
-                        "%s cannot be downgraded from %d to %d", READER_VERSION_PROPERTY, currentProtocolEntry.getMinReaderVersion(), readerVersion.get()));
-            }
-        }
+        long createdTime = Instant.now().toEpochMilli();
 
-        Optional<Integer> writerVersion = Optional.empty();
-        if (properties.containsKey(WRITER_VERSION_PROPERTY)) {
-            writerVersion = Optional.of((int) properties.get(WRITER_VERSION_PROPERTY)
-                    .orElseThrow(() -> new IllegalArgumentException("The " + WRITER_VERSION_PROPERTY + " property cannot be empty")));
-            if (writerVersion.get() < currentProtocolEntry.getMinWriterVersion()) {
-                throw new TrinoException(INVALID_TABLE_PROPERTY, format(
-                        "%s cannot be downgraded from %d to %d", WRITER_VERSION_PROPERTY, currentProtocolEntry.getMinWriterVersion(), writerVersion.get()));
+        int requiredWriterVersion = currentProtocolEntry.getMinWriterVersion();
+        Optional<MetadataEntry> metadataEntry = Optional.empty();
+        if (properties.containsKey(CHANGE_DATA_FEED_ENABLED_PROPERTY)) {
+            boolean changeDataFeedEnabled = (Boolean) properties.get(CHANGE_DATA_FEED_ENABLED_PROPERTY)
+                    .orElseThrow(() -> new IllegalArgumentException("The change_data_feed_enabled  property cannot be empty"));
+            if (changeDataFeedEnabled) {
+                requiredWriterVersion = max(requiredWriterVersion, CDF_SUPPORTED_WRITER_VERSION);
             }
+            Map<String, String> configuration = new HashMap<>(handle.getMetadataEntry().getConfiguration());
+            configuration.put(DELTA_CHANGE_DATA_FEED_ENABLED_PROPERTY, String.valueOf(changeDataFeedEnabled));
+            metadataEntry = Optional.of(buildMetadataEntry(handle.getMetadataEntry(), configuration, createdTime));
         }
 
         long readVersion = handle.getReadVersion();
         long commitVersion = readVersion + 1;
 
-        ProtocolEntry protocolEntry = new ProtocolEntry(
-                readerVersion.orElse(currentProtocolEntry.getMinReaderVersion()),
-                writerVersion.orElse(currentProtocolEntry.getMinWriterVersion()));
+        Optional<ProtocolEntry> protocolEntry = Optional.empty();
+        if (requiredWriterVersion != currentProtocolEntry.getMinWriterVersion()) {
+            protocolEntry = Optional.of(new ProtocolEntry(currentProtocolEntry.getMinReaderVersion(), requiredWriterVersion));
+        }
 
         try {
             TransactionLogWriter transactionLogWriter = transactionLogWriterFactory.newWriter(session, handle.getLocation());
-
-            long createdTime = Instant.now().toEpochMilli();
             transactionLogWriter.appendCommitInfoEntry(getCommitInfoEntry(session, commitVersion, createdTime, SET_TBLPROPERTIES_OPERATION, readVersion));
-            transactionLogWriter.appendProtocolEntry(protocolEntry);
+            protocolEntry.ifPresent(transactionLogWriter::appendProtocolEntry);
+
+            metadataEntry.ifPresent(transactionLogWriter::appendMetadataEntry);
 
             transactionLogWriter.flush();
         }
@@ -1930,6 +1904,19 @@ public void setTableProperties(ConnectorSession session, ConnectorTableHandle ta
         }
     }
 
+    private MetadataEntry buildMetadataEntry(MetadataEntry metadataEntry, Map<String, String> configuration, long createdTime)
+    {
+        return new MetadataEntry(
+                metadataEntry.getId(),
+                metadataEntry.getName(),
+                metadataEntry.getDescription(),
+                metadataEntry.getFormat(),
+                metadataEntry.getSchemaString(),
+                metadataEntry.getOriginalPartitionColumns(),
+                configuration,
+                createdTime);
+    }
+
     @Override
     public Map<String, Object> getSchemaProperties(ConnectorSession session, CatalogSchemaName schemaName)
     {

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

@@ -58,8 +58,6 @@ public class DeltaLakeMetadataFactory
     private final long perTransactionMetastoreCacheMaximumSize;
     private final boolean deleteSchemaLocationsFallback;
     private final boolean useUniqueTableLocation;
-    private final int defaultReaderVersion;
-    private final int defaultWriterVersion;
 
     private final boolean allowManagedTableRename;
     private final String trinoVersion;
@@ -102,8 +100,6 @@ public DeltaLakeMetadataFactory(
         this.perTransactionMetastoreCacheMaximumSize = deltaLakeConfig.getPerTransactionMetastoreCacheMaximumSize();
         this.deleteSchemaLocationsFallback = deltaLakeConfig.isDeleteSchemaLocationsFallback();
         this.useUniqueTableLocation = deltaLakeConfig.isUniqueTableLocation();
-        this.defaultReaderVersion = deltaLakeConfig.getDefaultReaderVersion();
-        this.defaultWriterVersion = deltaLakeConfig.getDefaultWriterVersion();
         this.allowManagedTableRename = allowManagedTableRename;
         this.trinoVersion = requireNonNull(nodeVersion, "nodeVersion is null").toString();
     }
@@ -145,8 +141,6 @@ public DeltaLakeMetadata create(ConnectorIdentity identity)
                 deltaLakeRedirectionsProvider,
                 statisticsAccess,
                 useUniqueTableLocation,
-                allowManagedTableRename,
-                defaultReaderVersion,
-                defaultWriterVersion);
+                allowManagedTableRename);
     }
 }