feat: Add support for delta.feature.X = supported table feature overrides

Add the ability to explicitly enable table features via table properties
using the delta.feature.<featureName> = supported syntax, matching the
Java Kernel's behavior.

Key Changes:
- Add SET_TABLE_FEATURE_SUPPORTED_PREFIX and SET_TABLE_FEATURE_SUPPORTED_VALUE
  constants to table_features module
- Add TableFeature::from_name() to parse feature names from strings
- Add TableFeature::is_reader_writer() to check feature type
- Add extract_feature_overrides() function that:
  - Extracts delta.feature.* properties from table properties
  - Validates the value is 'supported'
  - Returns features to add to protocol and cleaned properties
- Update CreateTableBuilder::build() to:
  - Call extract_feature_overrides() on table properties
  - Add extracted features to Protocol's reader/writer feature lists
  - Pass cleaned properties (without feature overrides) to Metadata

Usage:
  let result = TableManager::create_table(path, schema, "MyApp/1.0")
      .with_table_properties(HashMap::from([
          ("delta.feature.deletionVectors".to_string(), "supported".to_string()),
          ("delta.enableDeletionVectors".to_string(), "true".to_string()),
      ]))
      .build(engine, committer)?
      .commit(engine)?;

The delta.feature.* properties are consumed during build() and not stored
in the final Metadata configuration, matching Java Kernel behavior.

Author: sanujbasu

kernel/src/table_features/mod.rs

@@ -1,3 +1,6 @@
+use std::collections::HashMap;
+use std::str::FromStr;
+
 use serde::{Deserialize, Serialize};
 use strum::{AsRefStr, Display as StrumDisplay, EnumCount, EnumString};
 
@@ -15,6 +18,15 @@ pub(crate) use timestamp_ntz::validate_timestamp_ntz_feature_support;
 mod column_mapping;
 mod timestamp_ntz;
 
+/// Prefix for table feature override properties.
+/// Properties with this prefix (e.g., `delta.feature.deletionVectors`) are used to
+/// explicitly enable table features in the protocol.
+pub const SET_TABLE_FEATURE_SUPPORTED_PREFIX: &str = "delta.feature.";
+
+/// Value to enable a table feature when used with [`SET_TABLE_FEATURE_SUPPORTED_PREFIX`].
+/// Example: `"delta.feature.deletionVectors" -> "supported"`
+pub const SET_TABLE_FEATURE_SUPPORTED_VALUE: &str = "supported";
+
 /// Table features represent protocol capabilities required to correctly read or write a given table.
 /// - Readers must implement all features required for correct table reads.
 /// - Writers must implement all features required for correct table writes.
@@ -646,6 +658,91 @@ impl TableFeature {
             TableFeature::Unknown(_) => None,
         }
     }
+
+    /// Parse a feature name string into a TableFeature.
+    ///
+    /// This uses the strum `EnumString` derive to parse known feature names.
+    /// Unknown feature names are wrapped in `TableFeature::Unknown`.
+    pub(crate) fn from_name(name: &str) -> Self {
+        TableFeature::from_str(name).unwrap_or_else(|_| TableFeature::Unknown(name.to_string()))
+    }
+
+    /// Returns true if this is a ReaderWriter feature (appears in both reader and writer feature lists).
+    /// Returns false for Writer-only features and Unknown features.
+    pub(crate) fn is_reader_writer(&self) -> bool {
+        matches!(self.feature_type(), FeatureType::ReaderWriter)
+    }
+}
+
+/// Result of extracting feature overrides from table properties.
+///
+/// Contains the features to add to the protocol and the cleaned properties
+/// (with `delta.feature.*` entries removed).
+#[derive(Debug)]
+pub(crate) struct FeatureOverrides {
+    /// Features that should be added to the protocol
+    pub(crate) features: Vec<TableFeature>,
+    /// Table properties with `delta.feature.*` entries removed
+    pub(crate) cleaned_properties: HashMap<String, String>,
+}
+
+/// Extracts feature overrides from table properties.
+///
+/// Properties with the prefix `delta.feature.` are interpreted as feature override
+/// directives. For example, `delta.feature.deletionVectors = supported` adds the
+/// `deletionVectors` feature to the protocol.
+///
+/// # Arguments
+/// * `properties` - The table properties to process
+///
+/// # Returns
+/// A [`FeatureOverrides`] containing:
+/// - `features`: The list of features to add to the protocol
+/// - `cleaned_properties`: The table properties with feature override entries removed
+///
+/// # Errors
+/// Returns an error if:
+/// - A `delta.feature.*` property has a value other than "supported"
+///
+/// # Example
+/// ```ignore
+/// let props = HashMap::from([
+///     ("delta.feature.deletionVectors".to_string(), "supported".to_string()),
+///     ("delta.enableDeletionVectors".to_string(), "true".to_string()),
+/// ]);
+/// let overrides = extract_feature_overrides(props)?;
+/// // overrides.features = [TableFeature::DeletionVectors]
+/// // overrides.cleaned_properties = {"delta.enableDeletionVectors": "true"}
+/// ```
+pub(crate) fn extract_feature_overrides(
+    properties: HashMap<String, String>,
+) -> DeltaResult<FeatureOverrides> {
+    let mut features = Vec::new();
+    let mut cleaned_properties = HashMap::new();
+
+    for (key, value) in properties {
+        if let Some(feature_name) = key.strip_prefix(SET_TABLE_FEATURE_SUPPORTED_PREFIX) {
+            // Validate value is "supported"
+            if value != SET_TABLE_FEATURE_SUPPORTED_VALUE {
+                return Err(Error::generic(format!(
+                    "Invalid value '{}' for '{}'. Only '{}' is allowed.",
+                    value, key, SET_TABLE_FEATURE_SUPPORTED_VALUE
+                )));
+            }
+
+            // Parse feature name
+            let feature = TableFeature::from_name(feature_name);
+            features.push(feature);
+        } else {
+            // Keep non-feature-override properties
+            cleaned_properties.insert(key, value);
+        }
+    }
+
+    Ok(FeatureOverrides {
+        features,
+        cleaned_properties,
+    })
 }
 
 impl ToDataType for TableFeature {
@@ -786,4 +883,163 @@ mod tests {
             assert_eq!(from_str, feature);
         }
     }
+
+    #[test]
+    fn test_from_name() {
+        // Known features
+        assert_eq!(
+            TableFeature::from_name("deletionVectors"),
+            TableFeature::DeletionVectors
+        );
+        assert_eq!(
+            TableFeature::from_name("changeDataFeed"),
+            TableFeature::ChangeDataFeed
+        );
+        assert_eq!(
+            TableFeature::from_name("columnMapping"),
+            TableFeature::ColumnMapping
+        );
+        assert_eq!(
+            TableFeature::from_name("timestampNtz"),
+            TableFeature::TimestampWithoutTimezone
+        );
+
+        // Unknown features
+        assert_eq!(
+            TableFeature::from_name("unknownFeature"),
+            TableFeature::Unknown("unknownFeature".to_string())
+        );
+    }
+
+    #[test]
+    fn test_is_reader_writer() {
+        // ReaderWriter features
+        assert!(TableFeature::DeletionVectors.is_reader_writer());
+        assert!(TableFeature::ColumnMapping.is_reader_writer());
+        assert!(TableFeature::TimestampWithoutTimezone.is_reader_writer());
+        assert!(TableFeature::V2Checkpoint.is_reader_writer());
+
+        // Writer-only features
+        assert!(!TableFeature::ChangeDataFeed.is_reader_writer());
+        assert!(!TableFeature::AppendOnly.is_reader_writer());
+        assert!(!TableFeature::DomainMetadata.is_reader_writer());
+        assert!(!TableFeature::RowTracking.is_reader_writer());
+
+        // Unknown features
+        assert!(!TableFeature::unknown("something").is_reader_writer());
+    }
+
+    #[test]
+    fn test_extract_feature_overrides_basic() {
+        let props = HashMap::from([
+            (
+                "delta.feature.deletionVectors".to_string(),
+                "supported".to_string(),
+            ),
+            (
+                "delta.enableDeletionVectors".to_string(),
+                "true".to_string(),
+            ),
+        ]);
+
+        let result = extract_feature_overrides(props).unwrap();
+
+        assert_eq!(result.features.len(), 1);
+        assert_eq!(result.features[0], TableFeature::DeletionVectors);
+
+        // Feature override should be removed from cleaned properties
+        assert!(!result
+            .cleaned_properties
+            .contains_key("delta.feature.deletionVectors"));
+        // Regular property should be retained
+        assert_eq!(
+            result.cleaned_properties.get("delta.enableDeletionVectors"),
+            Some(&"true".to_string())
+        );
+    }
+
+    #[test]
+    fn test_extract_feature_overrides_multiple_features() {
+        let props = HashMap::from([
+            (
+                "delta.feature.deletionVectors".to_string(),
+                "supported".to_string(),
+            ),
+            (
+                "delta.feature.changeDataFeed".to_string(),
+                "supported".to_string(),
+            ),
+            (
+                "delta.feature.appendOnly".to_string(),
+                "supported".to_string(),
+            ),
+            (
+                "delta.enableDeletionVectors".to_string(),
+                "true".to_string(),
+            ),
+        ]);
+
+        let result = extract_feature_overrides(props).unwrap();
+
+        assert_eq!(result.features.len(), 3);
+        assert!(result.features.contains(&TableFeature::DeletionVectors));
+        assert!(result.features.contains(&TableFeature::ChangeDataFeed));
+        assert!(result.features.contains(&TableFeature::AppendOnly));
+
+        // Only the regular property should remain
+        assert_eq!(result.cleaned_properties.len(), 1);
+        assert_eq!(
+            result.cleaned_properties.get("delta.enableDeletionVectors"),
+            Some(&"true".to_string())
+        );
+    }
+
+    #[test]
+    fn test_extract_feature_overrides_invalid_value() {
+        let props = HashMap::from([(
+            "delta.feature.deletionVectors".to_string(),
+            "enabled".to_string(), // Wrong value - should be "supported"
+        )]);
+
+        let result = extract_feature_overrides(props);
+
+        assert!(result.is_err());
+        let err = result.unwrap_err().to_string();
+        assert!(err.contains("Invalid value"));
+        assert!(err.contains("enabled"));
+        assert!(err.contains("supported"));
+    }
+
+    #[test]
+    fn test_extract_feature_overrides_no_features() {
+        let props = HashMap::from([
+            (
+                "delta.enableDeletionVectors".to_string(),
+                "true".to_string(),
+            ),
+            ("delta.appendOnly".to_string(), "true".to_string()),
+            ("custom.property".to_string(), "value".to_string()),
+        ]);
+
+        let result = extract_feature_overrides(props).unwrap();
+
+        assert!(result.features.is_empty());
+        assert_eq!(result.cleaned_properties.len(), 3);
+    }
+
+    #[test]
+    fn test_extract_feature_overrides_unknown_feature() {
+        let props = HashMap::from([(
+            "delta.feature.futureFeature".to_string(),
+            "supported".to_string(),
+        )]);
+
+        let result = extract_feature_overrides(props).unwrap();
+
+        assert_eq!(result.features.len(), 1);
+        assert_eq!(
+            result.features[0],
+            TableFeature::Unknown("futureFeature".to_string())
+        );
+    }
 }

kernel/src/transaction/create_table.rs

@@ -36,6 +36,7 @@ use crate::actions::{
 };
 use crate::committer::Committer;
 use crate::schema::SchemaRef;
+use crate::table_features::extract_feature_overrides;
 use crate::transaction::Transaction;
 use crate::utils::{current_time_ms, try_parse_uri};
 use crate::{DeltaResult, Engine, Error};
@@ -172,25 +173,44 @@ impl CreateTableBuilder {
             }
         }
 
+        // Extract feature overrides (delta.feature.X = supported) from table properties
+        let feature_overrides = extract_feature_overrides(self.table_properties)?;
+
+        // Build reader/writer feature lists from explicit feature overrides
+        let mut reader_features: Vec<String> = Vec::new();
+        let mut writer_features: Vec<String> = Vec::new();
+
+        for feature in &feature_overrides.features {
+            let feature_name = feature.to_string();
+
+            // All features go into writer_features
+            writer_features.push(feature_name.clone());
+
+            // ReaderWriter features also go into reader_features
+            if feature.is_reader_writer() {
+                reader_features.push(feature_name);
+            }
+        }
+
         // Create Protocol action with table features support
         let protocol = Protocol::try_new(
             TABLE_FEATURES_MIN_READER_VERSION,
             TABLE_FEATURES_MIN_WRITER_VERSION,
-            Some(Vec::<String>::new()), // readerFeatures (empty for now)
-            Some(Vec::<String>::new()), // writerFeatures (empty for now)
+            Some(reader_features),
+            Some(writer_features),
         )?;
 
         // Get current timestamp
         let created_time = current_time_ms()?;
 
-        // Create Metadata action
+        // Create Metadata action with cleaned properties (feature overrides removed)
         let metadata = Metadata::try_new(
             None, // name
             None, // description
             (*self.schema).clone(),
             self.partition_columns,
             created_time,
-            self.table_properties,
+            feature_overrides.cleaned_properties,
         )?;
 
         // Create Transaction with cached Protocol and Metadata