Add missing changeDataFeed writer feature when table version supports reader/writer features in Delta Lake

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

@@ -253,6 +253,7 @@
 import static io.trino.plugin.deltalake.metastore.HiveMetastoreBackedDeltaLakeMetastore.verifyDeltaLakeTable;
 import static io.trino.plugin.deltalake.procedure.DeltaLakeTableProcedureId.OPTIMIZE;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.APPEND_ONLY_CONFIGURATION_KEY;
+import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.CHANGE_DATA_FEED_FEATURE_NAME;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.COLUMN_MAPPING_PHYSICAL_NAME_CONFIGURATION_KEY;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.ColumnMappingMode.ID;
 import static io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.ColumnMappingMode.NAME;
@@ -381,6 +382,9 @@ public class DeltaLakeMetadata
     // The highest reader and writer versions Trino supports
     private static final int MAX_READER_VERSION = 3;
     public static final int MAX_WRITER_VERSION = 7;
+    // The lowest versions Delta Lake supports reader and writer features
+    private static final int MIN_READER_FEATURE_VERSION = 3;
+    private static final int MIN_WRITER_FEATURE_VERSION = 7;
     private static final int CDF_SUPPORTED_WRITER_VERSION = 4;
     private static final int COLUMN_MAPPING_MODE_SUPPORTED_READER_VERSION = 2;
     private static final int COLUMN_MAPPING_MODE_SUPPORTED_WRITER_VERSION = 5;
@@ -2905,6 +2909,7 @@ private ProtocolEntry protocolEntry(int readerVersion, int writerVersion, boolea
         if (changeDataFeedEnabled.isPresent() && changeDataFeedEnabled.get()) {
             // Enabling cdf (change data feed) requires setting the writer version to 4
             writerVersion = CDF_SUPPORTED_WRITER_VERSION;
+            writerFeatures.add(CHANGE_DATA_FEED_FEATURE_NAME);
         }
         ColumnMappingMode columnMappingMode = getColumnMappingMode(properties);
         if (columnMappingMode == ID || columnMappingMode == NAME) {
@@ -2926,8 +2931,8 @@ private ProtocolEntry protocolEntry(int readerVersion, int writerVersion, boolea
         return new ProtocolEntry(
                 readerVersion,
                 writerVersion,
-                readerFeatures.isEmpty() ? Optional.empty() : Optional.of(readerFeatures),
-                writerFeatures.isEmpty() ? Optional.empty() : Optional.of(writerFeatures));
+                readerVersion < MIN_READER_FEATURE_VERSION || readerFeatures.isEmpty() ? Optional.empty() : Optional.of(readerFeatures),
+                writerVersion < MIN_WRITER_FEATURE_VERSION || writerFeatures.isEmpty() ? Optional.empty() : Optional.of(writerFeatures));
     }
 
     private void writeCheckpointIfNeeded(

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

@@ -108,7 +108,7 @@ private DeltaLakeSchemaSupport() {}
 
     // https://github.com/delta-io/delta/blob/master/PROTOCOL.md#valid-feature-names-in-table-features
     private static final String APPEND_ONLY_FEATURE_NAME = "appendOnly";
-    private static final String CHANGE_DATA_FEED_FEATURE_NAME = "changeDataFeed";
+    public static final String CHANGE_DATA_FEED_FEATURE_NAME = "changeDataFeed";
     private static final String CHECK_CONSTRAINTS_FEATURE_NAME = "checkConstraints";
     private static final String COLUMN_MAPPING_FEATURE_NAME = "columnMapping";
     public static final String DELETION_VECTORS_FEATURE_NAME = "deletionVectors";

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

@@ -1325,6 +1325,33 @@ public void testCreateTableWithChangeDataFeedColumnName()
         }
     }
 
+    @Test
+    public void testCreateTableWithChangeDataFeed()
+    {
+        try (TestTable table = new TestTable(getQueryRunner()::execute, "test_cdf", "(x int) WITH (change_data_feed_enabled = true)")) {
+            assertThat(query("SELECT * FROM \"" + table.getName() + "$properties\""))
+                    .skippingTypesCheck()
+                    .matches("VALUES " +
+                            "('delta.enableChangeDataFeed', 'true')," +
+                            "('delta.enableDeletionVectors', 'false')," +
+                            "('delta.minReaderVersion', '1')," +
+                            "('delta.minWriterVersion', '4')");
+        }
+
+        // timestamp type requires reader version 3 and writer version 7
+        try (TestTable table = new TestTable(getQueryRunner()::execute, "test_cdf", "(x timestamp) WITH (change_data_feed_enabled = true)")) {
+            assertThat(query("SELECT * FROM \"" + table.getName() + "$properties\""))
+                    .skippingTypesCheck()
+                    .matches("VALUES " +
+                            "('delta.enableChangeDataFeed', 'true')," +
+                            "('delta.enableDeletionVectors', 'false')," +
+                            "('delta.minReaderVersion', '3')," +
+                            "('delta.minWriterVersion', '7')," +
+                            "('delta.feature.timestampNtz', 'supported')," +
+                            "('delta.feature.changeDataFeed', 'supported')");
+        }
+    }
+
     @Test
     public void testUnsupportedCreateTableWithChangeDataFeed()
     {