feat: allow AppendOnly, ChangeDataFeed, and TypeWidening in CREATE TABLE (delta-io#2279)

## What changes are proposed in this pull request?

Add three simple protocol-only features to the CREATE TABLE allow lists:
AppendOnly, ChangeDataFeed, TypeWidening

No additional auto-enablement code is needed.
`maybe_auto_enable_property_driven_features` (from delta-io#2245) derives
enablement from `FeatureInfo::enablement_check`.

## How was this change tested?

Parameterized integration tests in kernel/tests/create_table/main.rs:
- `test_create_table_with_feature_signal` extended with
`is_reader_writer` parameter and cases for all three features
(WriterOnly features only check `writer_features()`)
- `test_create_table_with_enablement_property` generalized to cover all
property-driven features (DV, TypeWidening, CDF, AppendOnly) with
true/false cases via `#[values]`

Co-authored-by: Sanuj Basu <sanujbasu@users.noreply.github.com>

Author: sanujbasu

kernel/src/transaction/builder/create_table.rs

@@ -28,9 +28,10 @@ use crate::table_features::{
     SET_TABLE_FEATURE_SUPPORTED_VALUE,
 };
 use crate::table_properties::{
-    TableProperties, CHECKPOINT_WRITE_STATS_AS_JSON, CHECKPOINT_WRITE_STATS_AS_STRUCT,
+    TableProperties, APPEND_ONLY, CHECKPOINT_WRITE_STATS_AS_JSON, CHECKPOINT_WRITE_STATS_AS_STRUCT,
     COLUMN_MAPPING_MAX_COLUMN_ID, COLUMN_MAPPING_MODE, DELTA_PROPERTY_PREFIX,
-    ENABLE_DELETION_VECTORS, ENABLE_IN_COMMIT_TIMESTAMPS, SET_TRANSACTION_RETENTION_DURATION,
+    ENABLE_CHANGE_DATA_FEED, ENABLE_DELETION_VECTORS, ENABLE_IN_COMMIT_TIMESTAMPS,
+    ENABLE_TYPE_WIDENING, SET_TRANSACTION_RETENTION_DURATION,
 };
 use crate::transaction::create_table::CreateTableTransaction;
 use crate::transaction::data_layout::DataLayout;
@@ -56,6 +57,12 @@ const ALLOWED_DELTA_FEATURES: &[TableFeature] = &[
     // specifying clustering columns via `with_data_layout()`.
     TableFeature::DeletionVectors,
     TableFeature::V2Checkpoint,
+    // Simple protocol-only features: enabling these only updates the protocol action.
+    // They can also be auto-enabled via their enablement properties (e.g. delta.appendOnly=true)
+    // through `maybe_auto_enable_property_driven_features`.
+    TableFeature::AppendOnly,
+    TableFeature::ChangeDataFeed,
+    TableFeature::TypeWidening,
 ];
 
 /// Delta properties allowed to be set during CREATE TABLE.
@@ -71,8 +78,11 @@ const ALLOWED_DELTA_PROPERTIES: &[&str] = &[
     // Checkpoint stats format properties
     CHECKPOINT_WRITE_STATS_AS_JSON,
     CHECKPOINT_WRITE_STATS_AS_STRUCT,
-    // DeletionVectors enablement property: auto-enables the DeletionVectors feature
+    // Property-driven feature enablement properties
     ENABLE_DELETION_VECTORS,
+    ENABLE_CHANGE_DATA_FEED,
+    ENABLE_TYPE_WIDENING,
+    APPEND_ONLY,
     // Set transaction retention duration: controls expiration of txn identifiers
     SET_TRANSACTION_RETENTION_DURATION,
 ];
@@ -719,6 +729,7 @@ mod tests {
     use super::*;
     use crate::expressions::ColumnName;
     use crate::schema::{DataType, StructField, StructType};
+    use crate::table_properties::ENABLE_ICEBERG_COMPAT_V1;
     use crate::utils::test_utils::assert_result_error_with_message;
 
     fn test_schema() -> SchemaRef {
@@ -830,14 +841,12 @@ mod tests {
 
     #[test]
     fn test_validate_unsupported_properties() {
-        use crate::table_properties::{APPEND_ONLY, ENABLE_CHANGE_DATA_FEED};
-
         // Delta properties not on allow list are rejected
         let mut properties = HashMap::new();
-        properties.insert(ENABLE_CHANGE_DATA_FEED.to_string(), "true".to_string());
+        properties.insert(ENABLE_ICEBERG_COMPAT_V1.to_string(), "true".to_string());
         assert_result_error_with_message(
             validate_extract_table_features_and_properties(properties),
-            "Setting delta property 'delta.enableChangeDataFeed' is not supported",
+            "Setting delta property 'delta.enableIcebergCompatV1' is not supported",
         );
 
         // Feature signals for features not in ALLOWED_DELTA_FEATURES are rejected
@@ -863,10 +872,10 @@ mod tests {
         // Mixed properties with unsupported delta property are rejected
         let mut properties = HashMap::new();
         properties.insert("myapp.version".to_string(), "1.0".to_string());
-        properties.insert(APPEND_ONLY.to_string(), "true".to_string());
+        properties.insert(ENABLE_ICEBERG_COMPAT_V1.to_string(), "true".to_string());
         assert_result_error_with_message(
             validate_extract_table_features_and_properties(properties),
-            "Setting delta property 'delta.appendOnly' is not supported",
+            "Setting delta property 'delta.enableIcebergCompatV1' is not supported",
         );
     }
 

kernel/tests/create_table/main.rs

@@ -342,14 +342,19 @@ async fn test_create_table_txn_debug() -> DeltaResult<()> {
 }
 
 #[rstest]
-#[case("vacuumProtocolCheck", TableFeature::VacuumProtocolCheck, true)]
-#[case("v2Checkpoint", TableFeature::V2Checkpoint, true)]
-// DV uses EnabledIf: the feature signal alone makes it supported but not enabled
-// (requires delta.enableDeletionVectors=true property to be enabled)
-#[case("deletionVectors", TableFeature::DeletionVectors, false)]
+// ReaderWriter features (AlwaysIfSupported)
+#[case("vacuumProtocolCheck", TableFeature::VacuumProtocolCheck, true, true)]
+#[case("v2Checkpoint", TableFeature::V2Checkpoint, true, true)]
+// ReaderWriter features (EnabledIf -- feature signal alone does not enable)
+#[case("deletionVectors", TableFeature::DeletionVectors, true, false)]
+#[case("typeWidening", TableFeature::TypeWidening, true, false)]
+// WriterOnly features (EnabledIf -- feature signal alone does not enable)
+#[case("appendOnly", TableFeature::AppendOnly, false, false)]
+#[case("changeDataFeed", TableFeature::ChangeDataFeed, false, false)]
 fn test_create_table_with_feature_signal(
     #[case] feature_name: &str,
     #[case] feature: TableFeature,
+    #[case] is_reader_writer: bool,
     #[case] enabled_when_supported: bool,
 ) -> DeltaResult<()> {
     let (_temp_dir, table_path, engine) = test_table_setup()?;
@@ -379,12 +384,14 @@ fn test_create_table_with_feature_signal(
             .is_some_and(|f| f.contains(&feature)),
         "{feature_name} should be in writer features"
     );
-    assert!(
-        protocol
-            .reader_features()
-            .is_some_and(|f| f.contains(&feature)),
-        "{feature_name} should be in reader features"
-    );
+    if is_reader_writer {
+        assert!(
+            protocol
+                .reader_features()
+                .is_some_and(|f| f.contains(&feature)),
+            "{feature_name} should be in reader features"
+        );
+    }
 
     Ok(())
 }
@@ -419,51 +426,56 @@ fn test_create_table_with_checkpoint_stats_properties(
 }
 
 #[rstest]
-#[case("true", true)]
-#[case("false", false)]
-fn test_create_table_with_deletion_vectors_property(
-    #[case] value: &str,
-    #[case] expect_enabled: bool,
+// ReaderWriter features
+#[case("delta.enableDeletionVectors", TableFeature::DeletionVectors, true)]
+#[case("delta.enableTypeWidening", TableFeature::TypeWidening, true)]
+// WriterOnly features
+#[case("delta.enableChangeDataFeed", TableFeature::ChangeDataFeed, false)]
+#[case("delta.appendOnly", TableFeature::AppendOnly, false)]
+fn test_create_table_with_enablement_property(
+    #[case] property: &str,
+    #[case] feature: TableFeature,
+    #[case] is_reader_writer: bool,
+    #[values(true, false)] expect_enabled: bool,
 ) -> DeltaResult<()> {
     let (_temp_dir, table_path, engine) = test_table_setup()?;
+    let value = expect_enabled.to_string();
 
     let _ = create_table(&table_path, simple_schema()?, "Test/1.0")
-        .with_table_properties([("delta.enableDeletionVectors", value)])
+        .with_table_properties([(property, value.as_str())])
         .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
         .commit(engine.as_ref())?;
 
     let snapshot = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
     let table_config = snapshot.table_configuration();
 
     assert_eq!(
-        snapshot.table_properties().enable_deletion_vectors,
-        Some(value.parse::<bool>().unwrap())
-    );
-    assert_eq!(
-        table_config.is_feature_supported(&TableFeature::DeletionVectors),
+        table_config.is_feature_supported(&feature),
         expect_enabled,
-        "DeletionVectors supported should be {expect_enabled}"
+        "{property}={value}: feature supported should be {expect_enabled}"
     );
     assert_eq!(
-        table_config.is_feature_enabled(&TableFeature::DeletionVectors),
+        table_config.is_feature_enabled(&feature),
         expect_enabled,
-        "DeletionVectors enabled should be {expect_enabled}"
+        "{property}={value}: feature enabled should be {expect_enabled}"
     );
     let protocol = table_config.protocol();
     assert_eq!(
         protocol
             .writer_features()
-            .is_some_and(|f| f.contains(&TableFeature::DeletionVectors)),
-        expect_enabled,
-        "DeletionVectors in writer features should be {expect_enabled}"
-    );
-    assert_eq!(
-        protocol
-            .reader_features()
-            .is_some_and(|f| f.contains(&TableFeature::DeletionVectors)),
+            .is_some_and(|f| f.contains(&feature)),
         expect_enabled,
-        "DeletionVectors in reader features should be {expect_enabled}"
-    );
+        "{property}={value}: in writer features should be {expect_enabled}"
+    );
+    if is_reader_writer {
+        assert_eq!(
+            protocol
+                .reader_features()
+                .is_some_and(|f| f.contains(&feature)),
+            expect_enabled,
+            "{property}={value}: in reader features should be {expect_enabled}"
+        );
+    }
 
     Ok(())
 }