Allow updating reader_version and writer_version properties for Delta tables

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

@@ -563,15 +563,21 @@ 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 four table properties::
+The following example uses all six 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
+    change_data_feed_enabled = true,
+    reader_version = 2,
+    writer_version = 4
   )
   AS SELECT name, comment, regionkey FROM tpch.tiny.nation;
 

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

@@ -17,6 +17,7 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
+import io.trino.plugin.deltalake.transactionlog.ProtocolEntry;
 import io.trino.spi.connector.ConnectorOutputTableHandle;
 
 import java.util.List;
@@ -37,6 +38,7 @@ public class DeltaLakeOutputTableHandle
     private final boolean external;
     private final Optional<String> comment;
     private final Optional<Boolean> changeDataFeedEnabled;
+    private final ProtocolEntry protocolEntry;
 
     @JsonCreator
     public DeltaLakeOutputTableHandle(
@@ -47,7 +49,8 @@ public DeltaLakeOutputTableHandle(
             @JsonProperty("checkpointInterval") Optional<Long> checkpointInterval,
             @JsonProperty("external") boolean external,
             @JsonProperty("comment") Optional<String> comment,
-            @JsonProperty("changeDataFeedEnabled") Optional<Boolean> changeDataFeedEnabled)
+            @JsonProperty("changeDataFeedEnabled") Optional<Boolean> changeDataFeedEnabled,
+            @JsonProperty("protocolEntry") ProtocolEntry protocolEntry)
     {
         this.schemaName = requireNonNull(schemaName, "schemaName is null");
         this.tableName = requireNonNull(tableName, "tableName is null");
@@ -57,6 +60,7 @@ public DeltaLakeOutputTableHandle(
         this.external = external;
         this.comment = requireNonNull(comment, "comment is null");
         this.changeDataFeedEnabled = requireNonNull(changeDataFeedEnabled, "changeDataFeedEnabled is null");
+        this.protocolEntry = requireNonNull(protocolEntry, "protocolEntry is null");
     }
 
     @JsonProperty
@@ -115,4 +119,10 @@ public Optional<Boolean> getChangeDataFeedEnabled()
     {
         return changeDataFeedEnabled;
     }
+
+    @JsonProperty
+    public ProtocolEntry getProtocolEntry()
+    {
+        return protocolEntry;
+    }
 }

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

@@ -26,8 +26,13 @@
 import java.util.Optional;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
+import static io.trino.plugin.deltalake.DeltaLakeMetadata.MAX_READER_VERSION;
+import static io.trino.plugin.deltalake.DeltaLakeMetadata.MAX_WRITER_VERSION;
+import static io.trino.plugin.deltalake.DeltaLakeMetadata.MIN_READER_VERSION;
+import static io.trino.plugin.deltalake.DeltaLakeMetadata.MIN_WRITER_VERSION;
 import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY;
 import static io.trino.spi.session.PropertyMetadata.booleanProperty;
+import static io.trino.spi.session.PropertyMetadata.integerProperty;
 import static io.trino.spi.session.PropertyMetadata.longProperty;
 import static io.trino.spi.session.PropertyMetadata.stringProperty;
 import static io.trino.spi.type.VarcharType.VARCHAR;
@@ -40,6 +45,8 @@ public class DeltaLakeTableProperties
     public static final String PARTITIONED_BY_PROPERTY = "partitioned_by";
     public static final String CHECKPOINT_INTERVAL_PROPERTY = "checkpoint_interval";
     public static final String CHANGE_DATA_FEED_ENABLED_PROPERTY = "change_data_feed_enabled";
+    public static final String READER_VERSION_PROPERTY = "reader_version";
+    public static final String WRITER_VERSION_PROPERTY = "writer_version";
 
     private final List<PropertyMetadata<?>> tableProperties;
 
@@ -74,6 +81,18 @@ public DeltaLakeTableProperties()
                         "Enables storing change data feed entries",
                         null,
                         false))
+                .add(integerProperty(
+                        READER_VERSION_PROPERTY,
+                        "Reader version",
+                        null,
+                        value -> validateVersion(READER_VERSION_PROPERTY, value, MIN_READER_VERSION, MAX_READER_VERSION),
+                        false))
+                .add(integerProperty(
+                        WRITER_VERSION_PROPERTY,
+                        "Writer version",
+                        null,
+                        value -> validateVersion(WRITER_VERSION_PROPERTY, value, MIN_WRITER_VERSION, MAX_WRITER_VERSION),
+                        false))
                 .build();
     }
 
@@ -109,4 +128,23 @@ public static Optional<Boolean> getChangeDataFeedEnabled(Map<String, Object> tab
     {
         return Optional.ofNullable((Boolean) tableProperties.get(CHANGE_DATA_FEED_ENABLED_PROPERTY));
     }
+
+    public static Optional<Integer> getReaderVersion(Map<String, Object> tableProperties)
+    {
+        return Optional.ofNullable((Integer) tableProperties.get(READER_VERSION_PROPERTY));
+    }
+
+    public static Optional<Integer> getWriterVersion(Map<String, Object> tableProperties)
+    {
+        return Optional.ofNullable((Integer) tableProperties.get(WRITER_VERSION_PROPERTY));
+    }
+
+    private static void validateVersion(String propertyName, Object value, int minSupportedVersion, int maxSupportedVersion)
+    {
+        int version = (int) value;
+        if (version < minSupportedVersion || version > maxSupportedVersion) {
+            throw new TrinoException(INVALID_TABLE_PROPERTY, format(
+                    "%s must be between %d and %d", propertyName, minSupportedVersion, maxSupportedVersion));
+        }
+    }
 }

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeConnectorSmokeTest.java

@@ -347,7 +347,9 @@ public void testShowCreateTable()
                                 ")\n" +
                                 "WITH (\n" +
                                 "   location = '%s',\n" +
-                                "   partitioned_by = ARRAY['age']\n" +
+                                "   partitioned_by = ARRAY['age'],\n" +
+                                "   reader_version = 1,\n" +
+                                "   writer_version = 2\n" +
                                 ")",
                         SCHEMA,
                         getLocationForTable(bucketName, "person")));
@@ -476,7 +478,9 @@ public void testCreatePartitionedTableAs()
                                 ")\n" +
                                 "WITH (\n" +
                                 "   location = '%s',\n" +
-                                "   partitioned_by = ARRAY['regionkey']\n" +
+                                "   partitioned_by = ARRAY['regionkey'],\n" +
+                                "   reader_version = 1,\n" +
+                                "   writer_version = 2\n" +
                                 ")",
                         DELTA_CATALOG, SCHEMA, tableName, getLocationForTable(bucketName, tableName)));
         assertQuery("SELECT * FROM " + tableName, "SELECT name, regionkey, comment FROM nation");
@@ -1347,7 +1351,9 @@ private void testDeltaLakeTableLocationChanged(boolean fewerEntries, boolean fir
                                 ")\n" +
                                 "WITH (\n" +
                                 "   location = '%s',\n" +
-                                "   partitioned_by = ARRAY[%s]\n" +
+                                "   partitioned_by = ARRAY[%s],\n" +
+                                "   reader_version = 1,\n" +
+                                "   writer_version = 2\n" +
                                 ")",
                         getSession().getCatalog().orElseThrow(),
                         SCHEMA,