Preserve table features on CREATE OR REPLACE TABLE

According to
https://github.com/delta-io/delta/blob/master/PROTOCOL.md#table-features-for-new-and-existing-tables,
retain existing read and write features when executing
`CREATE OR REPLACE TABLE`, as previous table data may
still depend on them.

Author: chenjian2664

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

@@ -1355,12 +1355,12 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
                             transactionLogWriter.appendRemoveFileEntry(new RemoveFileEntry(addFileEntry.getPath(), addFileEntry.getPartitionValues(), writeTimestamp, true, Optional.empty()));
                         }
                     }
-                    protocolEntry = protocolEntryForTable(tableHandle.getProtocolEntry().minReaderVersion(), tableHandle.getProtocolEntry().minWriterVersion(), containsTimestampType, tableMetadata.getProperties());
+                    protocolEntry = protocolEntryForTable(ProtocolEntry.builder(tableHandle.getProtocolEntry()), containsTimestampType, tableMetadata.getProperties());
                     statisticsAccess.deleteExtendedStatistics(session, schemaTableName, location, tableHandle.toCredentialsHandle());
                 }
                 else {
                     setRollback(() -> deleteRecursivelyIfExists(fileSystem, deltaLogDirectory));
-                    protocolEntry = protocolEntryForTable(DEFAULT_READER_VERSION, DEFAULT_WRITER_VERSION, containsTimestampType, tableMetadata.getProperties());
+                    protocolEntry = protocolEntryForTable(ProtocolEntry.builder(DEFAULT_READER_VERSION, DEFAULT_WRITER_VERSION), containsTimestampType, tableMetadata.getProperties());
                 }
 
                 MetadataEntry metadataEntry = MetadataEntry.builder()
@@ -1545,15 +1545,15 @@ public DeltaLakeOutputTableHandle beginCreateTable(
 
         boolean isNewCheckpointFileRequired = false;
         if (replaceExistingTable) {
-            protocolEntry = protocolEntryForTable(handle.getProtocolEntry().minReaderVersion(), handle.getProtocolEntry().minWriterVersion(), containsTimestampType, tableMetadata.getProperties());
+            protocolEntry = protocolEntryForTable(ProtocolEntry.builder(handle.getProtocolEntry()), containsTimestampType, tableMetadata.getProperties());
             readVersion = OptionalLong.of(handle.getReadVersion());
             isNewCheckpointFileRequired = isNewCheckpointFileRequired(getColumns(handle.getMetadataEntry(), handle.getProtocolEntry()), columnHandles.build());
         }
         else {
             TrinoFileSystem fileSystem = fileSystemFactory.create(session, location);
             checkPathContainsNoFiles(fileSystem, finalLocation);
             setRollback(() -> deleteRecursivelyIfExists(fileSystem, finalLocation));
-            protocolEntry = protocolEntryForTable(DEFAULT_READER_VERSION, DEFAULT_WRITER_VERSION, containsTimestampType, tableMetadata.getProperties());
+            protocolEntry = protocolEntryForTable(ProtocolEntry.builder(DEFAULT_READER_VERSION, DEFAULT_WRITER_VERSION), containsTimestampType, tableMetadata.getProperties());
         }
 
         return new DeltaLakeOutputTableHandle(
@@ -3133,10 +3133,10 @@ private void checkSupportedWriterVersion(DeltaLakeTableHandle handle)
         }
     }
 
-    private ProtocolEntry protocolEntryForTable(int readerVersion, int writerVersion, boolean containsTimestampType, Map<String, Object> properties)
+    private ProtocolEntry protocolEntryForTable(ProtocolEntry.Builder builder, boolean containsTimestampType, Map<String, Object> properties)
     {
         return protocolEntry(
-                ProtocolEntry.builder(readerVersion, writerVersion),
+                builder,
                 containsTimestampType,
                 getChangeDataFeedEnabled(properties),
                 getColumnMappingMode(properties),

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

@@ -61,7 +61,7 @@ public record ProtocolEntry(
         // Spark done the similar logic here:
         // https://github.com/delta-io/delta/blob/f5ebfbb10bda87b6b2b7c9032e86edaf40d446bc/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala#L155-L157
         if (minReaderVersion >= MIN_VERSION_SUPPORTS_READER_FEATURES && readerFeatures.isEmpty()) {
-            log.warn("readerFeatures must exist when minReaderVersion is greater than or equal to " + MIN_VERSION_SUPPORTS_READER_FEATURES);
+            log.debug("readerFeatures must exist when minReaderVersion is greater than or equal to %s", MIN_VERSION_SUPPORTS_READER_FEATURES);
         }
         if (minWriterVersion < MIN_VERSION_SUPPORTS_WRITER_FEATURES && writerFeatures.isPresent()) {
             throw new IllegalArgumentException("writerFeatures must not exist when minWriterVersion is less than " + MIN_VERSION_SUPPORTS_WRITER_FEATURES);
@@ -72,13 +72,13 @@ public record ProtocolEntry(
         // Spark done the similar logic here:
         // https://github.com/delta-io/delta/blob/f5ebfbb10bda87b6b2b7c9032e86edaf40d446bc/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala#L158-L160
         if (minWriterVersion >= MIN_VERSION_SUPPORTS_WRITER_FEATURES && writerFeatures.isEmpty()) {
-            log.warn("writerFeatures must exist when minWriterVersion is greater than or equal to " + MIN_VERSION_SUPPORTS_WRITER_FEATURES);
+            log.debug("writerFeatures must exist when minWriterVersion is greater than or equal to %s", MIN_VERSION_SUPPORTS_WRITER_FEATURES);
         }
 
         // Spark done the similar logic here:
         // https://github.com/delta-io/delta/blob/f5ebfbb10bda87b6b2b7c9032e86edaf40d446bc/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala#L162-L166
         if (minReaderVersion >= MIN_VERSION_SUPPORTS_READER_FEATURES && minWriterVersion < MIN_VERSION_SUPPORTS_WRITER_FEATURES) {
-            log.warn("When reader is on table features, writer must be on table features too");
+            log.debug("When reader is on table features, writer must be on table features too");
         }
 
         readerFeatures = requireNonNull(readerFeatures, "readerFeatures is null").map(ImmutableSet::copyOf);

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

@@ -494,6 +494,27 @@ void testWriteCheckpointOnSchemaChangeCTAS()
         }
     }
 
+    @Test
+    void testCreateOrReplacePreservesFeatures()
+            throws Exception
+    {
+        try (TestTable table = newTrinoTable("test_create_or_replace_preserves_features_", "(x int) WITH (deletion_vectors_enabled = true)")) {
+            assertThat((String) computeScalar("SHOW CREATE TABLE " + table.getName()))
+                    .contains("deletion_vectors_enabled = true");
+
+            assertUpdate("CREATE OR REPLACE TABLE " + table.getName() + " (x int)");
+            // verify deletion_vector_enabled is preserved after CREATE OR REPLACE TABLE
+            Path tableLocation = Path.of(getTableLocation(table.getName()).replace("file://", ""));
+            ProtocolEntry protocolEntry = loadProtocolEntry(1, tableLocation);
+            assertThat(protocolEntry.readerFeaturesContains("deletionVectors")).isTrue();
+            assertThat(protocolEntry.writerFeaturesContains("deletionVectors")).isTrue();
+
+            // deletion_vectors_enabled is not enabled, since we created the table without it
+            assertThat((String) computeScalar("SHOW CREATE TABLE " + table.getName()))
+                    .doesNotContain("deletion_vectors_enabled = true");
+        }
+    }
+
     @Test // regression test for https://github.com/trinodb/trino/issues/24121
     void testPartitionValuesParsedCheckpoint()
             throws Exception

testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeCreateOrReplaceTableAsSelectCompatibility.java

@@ -140,6 +140,24 @@ public void testCreateOrReplaceTableWithSchemaChangeOnCheckpoint()
         }
     }
 
+    @Test(groups = {DELTA_LAKE_OSS, PROFILE_SPECIFIC_TESTS})
+    public void testCreateOrReplaceTableWithOnlyFeaturesChangedOnTrino()
+    {
+        String tableName = "test_create_or_replace_table_discard_features_" + randomNameSuffix();
+
+        onTrino().executeQuery("CREATE TABLE delta.default." + tableName + " (x INT) " +
+                "WITH (location = 's3://" + bucketName + "/delta-lake-test-" + tableName + "', deletion_vectors_enabled = true)");
+        try {
+            onTrino().executeQuery("INSERT INTO delta.default." + tableName + " VALUES 1, 2, 3");
+
+            onTrino().executeQuery("CREATE OR REPLACE TABLE delta.default." + tableName + " (x INT)");
+            assertThat(onDelta().executeQuery("SELECT * FROM " + tableName)).hasNoRows();
+        }
+        finally {
+            onTrino().executeQuery("DROP TABLE delta.default." + tableName);
+        }
+    }
+
     private static List<Row> performInsert(QueryExecutor queryExecutor, String tableName, int numberOfRows)
     {
         ImmutableList.Builder<Row> expectedRowBuilder = ImmutableList.builder();