feat: Unify TableFeatures followups (delta-io#1404)

## What changes are proposed in this pull request?
<!--
Please clarify what changes you are proposing and why the changes are
needed.
The purpose of this section is to outline the changes, why they are
needed, and how this PR fixes the issue.
If the reason for the change is already explained clearly in an issue,
then it does not need to be restated here.
1. If you propose a new API or feature, clarify the use case for a new
API or feature.
  2. If you fix a bug, you can clarify why it is a bug.
-->

This PR does a followup to unifying TableFeatures followups.

1. refactors some functions around for testing
2. Failing on Unknown Table features
3. Enabled ignored test by removing column mapping

## How was this change tested?
<!--
Please make sure to add test cases that check the changes thoroughly
including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please
clarify how you tested, ideally via a reproducible test documented in
the PR description.
-->

Fix all tests to use the unified table features

Author: DrakeLin

kernel/src/actions/mod.rs

@@ -402,23 +402,6 @@ impl Protocol {
         reader_features: Option<impl IntoIterator<Item = impl ToString>>,
         writer_features: Option<impl IntoIterator<Item = impl ToString>>,
     ) -> DeltaResult<Self> {
-        if min_reader_version == 3 {
-            require!(
-                reader_features.is_some(),
-                Error::invalid_protocol(
-                    "Reader features must be present when minimum reader version = 3"
-                )
-            );
-        }
-        if min_writer_version == 7 {
-            require!(
-                writer_features.is_some(),
-                Error::invalid_protocol(
-                    "Writer features must be present when minimum writer version = 7"
-                )
-            );
-        }
-
         let reader_features = parse_features(reader_features);
         let writer_features = parse_features(writer_features);
 
@@ -474,9 +457,44 @@ impl Protocol {
 
     /// Validates the relationship between reader features and writer features in the protocol.
     pub(crate) fn validate_table_features(&self) -> DeltaResult<()> {
+        // The protocol states that Reader features may be present if and only if the min_reader_version is 3
+        if self.min_reader_version == 3 {
+            require!(
+                self.reader_features.is_some(),
+                Error::invalid_protocol(
+                    "Reader features must be present when minimum reader version = 3"
+                )
+            );
+        } else {
+            require!(
+                self.reader_features.is_none(),
+                Error::invalid_protocol(
+                    "Reader features must not be present when minimum reader version != 3"
+                )
+            );
+        }
+
+        // The protocol states that Writer features may be present if and only if the min_writer_version is 7
+        if self.min_writer_version == 7 {
+            require!(
+                self.writer_features.is_some(),
+                Error::invalid_protocol(
+                    "Writer features must be present when minimum writer version = 7"
+                )
+            );
+        } else {
+            require!(
+                self.writer_features.is_none(),
+                Error::invalid_protocol(
+                    "Writer features must not be present when minimum writer version != 7"
+                )
+            );
+        }
+
         match (&self.reader_features, &self.writer_features) {
             (Some(reader_features), Some(writer_features)) => {
                 // Check all reader features are ReaderWriter and present in writer features.
+                // Unknown features are treated as potentially ReaderWriter for forward compatibility.
                 let check_r = reader_features.iter().all(|feature| {
                     matches!(
                         feature.feature_type(),
@@ -490,13 +508,14 @@ impl Protocol {
                     )
                 );
 
-                // Check all writer features are either Writer or also in reader features (ReaderWriter).
-                let check_w = writer_features.iter().all(|feature| {
-                    matches!(
-                        feature.feature_type(),
-                        FeatureType::Writer | FeatureType::Unknown
-                    ) || reader_features.contains(feature)
-                });
+                // Check all writer features that are ReaderWriter must also be in reader features
+                // Unknown features are treated as potentially Writer-only for forward compatibility.
+                let check_w = writer_features
+                    .iter()
+                    .all(|feature| match feature.feature_type() {
+                        FeatureType::Writer | FeatureType::Unknown => true,
+                        FeatureType::ReaderWriter => reader_features.contains(feature),
+                    });
                 require!(
                     check_w,
                     Error::invalid_protocol(
@@ -507,6 +526,7 @@ impl Protocol {
             }
             (None, None) => Ok(()),
             (None, Some(writer_features)) => {
+                // Unknown features are treated as potentially Writer-only for forward compatibility.
                 require!(
                     writer_features.iter().all(|feature| matches!(
                         feature.feature_type(),
@@ -1373,9 +1393,9 @@ mod tests {
                 Some(vec![TableFeature::AppendOnly]),
                 "Reader features must contain only ReaderWriter features that are also listed in writer features",
             ),
-            (Some(vec![TableFeature::DeletionVectors]), None, "Reader features should be present in writer features"),
+            (Some(vec![TableFeature::DeletionVectors]), Some(vec![]), "Reader features must contain only ReaderWriter features that are also listed in writer features"),
             // ReaderWriter feature not present in reader features
-            (None, Some(vec![TableFeature::DeletionVectors]), "Writer features must be Writer-only or also listed in reader features"),
+            (Some(vec![]), Some(vec![TableFeature::DeletionVectors]), "Writer features must be Writer-only or also listed in reader features"),
             (
                 Some(vec![TableFeature::VariantType]),
                 Some(vec![
@@ -1390,12 +1410,6 @@ mod tests {
                 Some(vec![TableFeature::AppendOnly]),
                 "Reader features must contain only ReaderWriter features that are also listed in writer features",
             ),
-            // Reader only feature is not allowed
-            (
-                Some(vec![TableFeature::Unknown("r".to_string())]),
-                Some(vec![TableFeature::Unknown("w".to_string())]),
-                "Reader features must contain only ReaderWriter features that are also listed in writer features",
-            ),
         ];
 
         for (reader_features, writer_features, error_msg) in invalid_features {
@@ -1417,6 +1431,40 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_validate_table_features_unknown() {
+        // Unknown features are allowed during validation for forward compatibility,
+        // but will be rejected when trying to use the protocol (ensure_read_supported/ensure_write_supported)
+
+        // Test unknown features in reader - validation passes
+        let protocol = Protocol {
+            min_reader_version: 3,
+            min_writer_version: 7,
+            reader_features: Some(vec![TableFeature::Unknown("unknown_reader".to_string())]),
+            writer_features: Some(vec![TableFeature::Unknown("unknown_reader".to_string())]),
+        };
+        assert!(protocol.validate_table_features().is_ok());
+        // But ensure_read_supported should fail
+        assert!(matches!(
+            protocol.ensure_read_supported(),
+            Err(Error::Unsupported(_))
+        ));
+
+        // Test unknown features in writer - validation passes
+        let protocol = Protocol {
+            min_reader_version: 3,
+            min_writer_version: 7,
+            reader_features: Some(vec![]),
+            writer_features: Some(vec![TableFeature::Unknown("unknown_writer".to_string())]),
+        };
+        assert!(protocol.validate_table_features().is_ok());
+        // But ensure_write_supported should fail
+        assert!(matches!(
+            protocol.ensure_write_supported(),
+            Err(Error::Unsupported(_))
+        ));
+    }
+
     #[test]
     fn test_validate_table_features_valid() {
         // (reader_feature, writer_feature)
@@ -1427,12 +1475,12 @@ mod tests {
                 Some(vec![TableFeature::DeletionVectors]),
                 Some(vec![TableFeature::DeletionVectors]),
             ),
-            (None, Some(vec![TableFeature::AppendOnly])),
+            (Some(vec![]), Some(vec![TableFeature::AppendOnly])),
             (
                 Some(vec![TableFeature::VariantType]),
                 Some(vec![TableFeature::VariantType, TableFeature::AppendOnly]),
             ),
-            // Unknwon feature may be ReaderWriter or Writer only
+            // Unknown feature may be ReaderWriter or Writer only (for forward compatibility)
             (
                 Some(vec![TableFeature::Unknown("rw".to_string())]),
                 Some(vec![
@@ -1441,7 +1489,7 @@ mod tests {
                 ]),
             ),
             // Empty feature set is valid
-            (None, None),
+            (Some(vec![]), Some(vec![])),
         ];
 
         for (reader_features, writer_features) in valid_features {
@@ -1534,7 +1582,8 @@ mod tests {
             r#"Unsupported: Unknown TableFeatures: "identityColumns". Supported TableFeatures: "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
         );
 
-        // Unknown writer features should cause an error
+        // Unknown writer features are allowed during creation for forward compatibility,
+        // but will fail when trying to write
         let protocol = Protocol::try_new(
             3,
             7,

kernel/src/table_changes/log_replay/tests.rs

@@ -566,7 +566,7 @@ async fn failing_protocol() {
 
     let protocol = Protocol::try_new(
         3,
-        1,
+        7,
         ["fake_feature".to_string()].into(),
         ["fake_feature".to_string()].into(),
     )

kernel/src/table_features/mod.rs

@@ -41,12 +41,9 @@ mod timestamp_ntz;
 #[serde(rename_all = "camelCase")]
 #[internal_api]
 pub(crate) enum TableFeature {
-    /// CatalogManaged tables:
-    /// <https://github.com/delta-io/delta/blob/master/protocol_rfcs/catalog-managed.md>
-    CatalogManaged,
-    #[strum(serialize = "catalogOwned-preview")]
-    #[serde(rename = "catalogOwned-preview")]
-    CatalogOwnedPreview,
+    //////////////////////////
+    // Writer-only features //
+    //////////////////////////
     /// Append Only Tables
     AppendOnly,
     /// Table invariants
@@ -57,16 +54,37 @@ pub(crate) enum TableFeature {
     ChangeDataFeed,
     /// Columns with generated values
     GeneratedColumns,
-    /// Mapping of one column to another
-    ColumnMapping,
     /// ID Columns
     IdentityColumns,
     /// Monotonically increasing timestamps in the CommitInfo
     InCommitTimestamp,
-    /// Deletion vectors for merge, update, delete
-    DeletionVectors,
     /// Row tracking on tables
     RowTracking,
+    /// domain specific metadata
+    DomainMetadata,
+    /// Iceberg compatibility support
+    IcebergCompatV1,
+    /// Iceberg compatibility support
+    IcebergCompatV2,
+    /// The Clustered Table feature facilitates the physical clustering of rows
+    /// that share similar values on a predefined set of clustering columns.
+    #[strum(serialize = "clustering")]
+    #[serde(rename = "clustering")]
+    ClusteredTable,
+
+    ///////////////////////////
+    // ReaderWriter features //
+    ///////////////////////////
+    /// CatalogManaged tables:
+    /// <https://github.com/delta-io/delta/blob/master/protocol_rfcs/catalog-managed.md>
+    CatalogManaged,
+    #[strum(serialize = "catalogOwned-preview")]
+    #[serde(rename = "catalogOwned-preview")]
+    CatalogOwnedPreview,
+    /// Mapping of one column to another
+    ColumnMapping,
+    /// Deletion vectors for merge, update, delete
+    DeletionVectors,
     /// timestamps without timezone support
     #[strum(serialize = "timestampNtz")]
     #[serde(rename = "timestampNtz")]
@@ -76,22 +94,11 @@ pub(crate) enum TableFeature {
     #[strum(serialize = "typeWidening-preview")]
     #[serde(rename = "typeWidening-preview")]
     TypeWideningPreview,
-    /// domain specific metadata
-    DomainMetadata,
     /// version 2 of checkpointing
     V2Checkpoint,
-    /// Iceberg compatibility support
-    IcebergCompatV1,
-    /// Iceberg compatibility support
-    IcebergCompatV2,
     /// vacuumProtocolCheck ReaderWriter feature ensures consistent application of reader and writer
     /// protocol checks during VACUUM operations
     VacuumProtocolCheck,
-    /// The Clustered Table feature facilitates the physical clustering of rows
-    /// that share similar values on a predefined set of clustering columns.
-    #[strum(serialize = "clustering")]
-    #[serde(rename = "clustering")]
-    ClusteredTable,
     /// This feature enables support for the variant data type, which stores semi-structured data.
     VariantType,
     #[strum(serialize = "variantType-preview")]
@@ -100,15 +107,20 @@ pub(crate) enum TableFeature {
     #[strum(serialize = "variantShredding-preview")]
     #[serde(rename = "variantShredding-preview")]
     VariantShreddingPreview,
+
     #[serde(untagged)]
     #[strum(default)]
     Unknown(String),
 }
 
-#[derive(Eq, PartialEq)]
+/// Classifies table features by their type
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum FeatureType {
+    /// Feature only affects write operations
     Writer,
+    /// Feature affects both read and write operations (must appear in both feature lists)
     ReaderWriter,
+    /// Unknown feature type (for forward compatibility)
     Unknown,
 }
 
@@ -247,7 +259,7 @@ mod tests {
     }
 
     #[test]
-    fn test_roundtrip_reader_features() {
+    fn test_roundtrip_table_features() {
         let cases = [
             (TableFeature::CatalogManaged, "catalogManaged"),
             (TableFeature::CatalogOwnedPreview, "catalogOwned-preview"),

kernel/tests/write.rs

@@ -859,7 +859,6 @@ async fn test_append_timestamp_ntz() -> Result<(), Box<dyn std::error::Error>> {
     Ok(())
 }
 
-#[ignore]
 #[tokio::test]
 async fn test_append_variant() -> Result<(), Box<dyn std::error::Error>> {
     // setup tracing
@@ -880,38 +879,20 @@ async fn test_append_variant() -> Result<(), Box<dyn std::error::Error>> {
 
     // create a table with VARIANT column
     let table_schema = Arc::new(StructType::try_new(vec![
-        StructField::nullable("v", DataType::unshredded_variant())
-            .with_metadata([("delta.columnMapping.physicalName", "col1")])
-            .add_metadata([("delta.columnMapping.id", 1)]),
-        StructField::nullable("i", DataType::INTEGER)
-            .with_metadata([("delta.columnMapping.physicalName", "col2")])
-            .add_metadata([("delta.columnMapping.id", 2)]),
+        StructField::nullable("v", DataType::unshredded_variant()),
+        StructField::nullable("i", DataType::INTEGER),
         StructField::nullable(
             "nested",
             // We flip the value and metadata fields in the actual parquet file for the test
             StructType::try_new(vec![StructField::nullable(
                 "nested_v",
                 unshredded_variant_schema_flipped(),
-            )
-            .with_metadata([("delta.columnMapping.physicalName", "col21")])
-            .add_metadata([("delta.columnMapping.id", 3)])])?,
-        )
-        .with_metadata([("delta.columnMapping.physicalName", "col3")])
-        .add_metadata([("delta.columnMapping.id", 4)]),
-    ])?);
-
-    let write_schema = Arc::new(StructType::try_new(vec![
-        StructField::nullable("col1", DataType::unshredded_variant()),
-        StructField::nullable("col2", DataType::INTEGER),
-        StructField::nullable(
-            "col3",
-            StructType::try_new(vec![StructField::nullable(
-                "col21",
-                unshredded_variant_schema_flipped(),
             )])?,
         ),
     ])?);
 
+    let write_schema = table_schema.clone();
+
     let tmp_test_dir = tempdir()?;
     let tmp_test_dir_url = Url::from_directory_path(tmp_test_dir.path()).unwrap();
 
@@ -921,16 +902,13 @@ async fn test_append_variant() -> Result<(), Box<dyn std::error::Error>> {
     // We can add shredding features as well as we are allowed to write unshredded variants
     // into shredded tables and shredded reads are explicitly blocked in the default
     // engine's parquet reader.
-    // TODO: (#1124) we don't actually support column mapping writes yet, but have some
-    // tests that do column mapping on writes. For now omit the writer feature to let tests
-    // run, but after actual support this should be enabled.
     let table_url = create_table(
         store.clone(),
         table_location,
         table_schema.clone(),
         &[],
         true,
-        vec!["variantType", "variantShredding-preview", "columnMapping"],
+        vec!["variantType", "variantShredding-preview"],
         vec!["variantType", "variantShredding-preview"],
     )
     .await?;
@@ -1016,7 +994,7 @@ async fn test_append_variant() -> Result<(), Box<dyn std::error::Error>> {
             Arc::new(Int32Array::from(i_values.clone())),
             // nested struct<nested_v variant>
             Arc::new(StructArray::try_new(
-                vec![Field::new("col21", variant_arrow_type_flipped(), true)].into(),
+                vec![Field::new("nested_v", variant_arrow_type_flipped(), true)].into(),
                 vec![variant_nested_v_array.clone()],
                 None,
             )?),