feat: Add with_table_properties method to CreateTableTransactionBuilder

Add the ability to set table properties when creating a Delta table.

Changes:
- Add with_table_properties() method that accepts any iterable of (K, V) pairs
- Supports only custom properties. Delta properties and signal flags to
  support features are deny listed. They will be selectively enabled
  for supported functionality.
- Method can be called multiple times (properties merge, duplicates overwrite)

Example:
  create_table(path, schema, engine_info)
      .with_table_properties([
          ("myapp.version", "1.0"),
      ])
      .build(engine, committer)?
      .commit(engine)?;

Tests
Unit tests added to validate create table building with supported
and unsupported properties.

Author: sanujbasu

kernel/src/table_features/mod.rs

@@ -24,6 +24,15 @@ pub const TABLE_FEATURES_MIN_READER_VERSION: i32 = 3;
 /// When set to 7, the protocol requires an explicit `writerFeatures` array.
 pub const TABLE_FEATURES_MIN_WRITER_VERSION: i32 = 7;
 
+/// Prefix for table feature override properties.
+/// Properties with this prefix (e.g., `delta.feature.deletionVectors`) are used to
+/// explicitly turn on support for the feature in the protocol.
+pub const SET_TABLE_FEATURE_SUPPORTED_PREFIX: &str = "delta.feature.";
+
+/// Value to add support for 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.

kernel/src/table_properties.rs

@@ -23,6 +23,9 @@ use strum::EnumString;
 mod deserialize;
 pub use deserialize::ParseIntervalError;
 
+/// Prefix for delta table properties (e.g., `delta.enableChangeDataFeed`, `delta.appendOnly`).
+pub const DELTA_PROPERTY_PREFIX: &str = "delta.";
+
 /// Delta table properties. These are parsed from the 'configuration' map in the most recent
 /// 'Metadata' action of a table.
 ///

kernel/src/transaction/create_table.rs

@@ -2,7 +2,29 @@
 //!
 //! This module provides a type-safe API for creating Delta tables.
 //! Use the [`create_table`] function to get a [`CreateTableTransactionBuilder`] that can be
-//! configured before building the [`Transaction`].
+//! configured with table properties and other options before building the [`Transaction`].
+//!
+//! # Example
+//!
+//! ```rust,no_run
+//! use delta_kernel::transaction::create_table::create_table;
+//! use delta_kernel::schema::{StructType, StructField, DataType};
+//! use delta_kernel::committer::FileSystemCommitter;
+//! use std::sync::Arc;
+//! # use delta_kernel::Engine;
+//! # fn example(engine: &dyn Engine) -> delta_kernel::DeltaResult<()> {
+//!
+//! let schema = Arc::new(StructType::try_new(vec![
+//!     StructField::new("id", DataType::INTEGER, false),
+//! ])?);
+//!
+//! let result = create_table("/path/to/table", schema, "MyApp/1.0")
+//!     .with_table_properties([("myapp.version", "1.0")])
+//!     .build(engine, Box::new(FileSystemCommitter::new()))?
+//!     .commit(engine)?;
+//! # Ok(())
+//! # }
+//! ```
 
 // Allow `pub` items in this module even though the module itself may be `pub(crate)`.
 // The module visibility controls external access; items are `pub` for use within the crate
@@ -20,11 +42,25 @@ use crate::log_segment::LogSegment;
 use crate::schema::SchemaRef;
 use crate::snapshot::Snapshot;
 use crate::table_configuration::TableConfiguration;
-use crate::table_features::{TABLE_FEATURES_MIN_READER_VERSION, TABLE_FEATURES_MIN_WRITER_VERSION};
+use crate::table_features::{
+    SET_TABLE_FEATURE_SUPPORTED_PREFIX, TABLE_FEATURES_MIN_READER_VERSION,
+    TABLE_FEATURES_MIN_WRITER_VERSION,
+};
+use crate::table_properties::DELTA_PROPERTY_PREFIX;
 use crate::transaction::Transaction;
 use crate::utils::{current_time_ms, try_parse_uri};
 use crate::{DeltaResult, Engine, Error, StorageHandler, PRE_COMMIT_VERSION};
 
+/// Properties that are allowed to be set during create table.
+/// This list will expand as more features are supported (e.g., column mapping, clustering).
+/// The allow list will be deprecated once auto feature enablement is implemented
+/// like the Java Kernel.
+const ALLOWED_DELTA_PROPERTIES: &[&str] = &[
+    // Empty for now - will add properties as features are implemented:
+    // - "delta.columnMapping.mode" (for column mapping)
+    // - etc.
+];
+
 /// Ensures that no Delta table exists at the given path.
 ///
 /// This function checks the `_delta_log` directory to determine if a table already exists.
@@ -78,6 +114,35 @@ fn ensure_table_does_not_exist(
     }
 }
 
+/// Validates that table properties are allowed during CREATE TABLE.
+///
+/// This function enforces an allow list for delta properties:
+/// - Feature override properties (`delta.feature.*`) are never allowed
+/// - Delta properties (`delta.*`) must be on the allow list
+/// - Non-delta properties (user/application properties) are always allowed
+fn validate_table_properties(properties: &HashMap<String, String>) -> DeltaResult<()> {
+    for key in properties.keys() {
+        // Block all delta.feature.* properties (feature override properties)
+        if key.starts_with(SET_TABLE_FEATURE_SUPPORTED_PREFIX) {
+            return Err(Error::generic(format!(
+                "Setting feature override property '{}' is not supported during CREATE TABLE",
+                key
+            )));
+        }
+        // For delta.* properties, check against allow list
+        if key.starts_with(DELTA_PROPERTY_PREFIX)
+            && !ALLOWED_DELTA_PROPERTIES.contains(&key.as_str())
+        {
+            return Err(Error::generic(format!(
+                "Setting delta property '{}' is not supported during CREATE TABLE",
+                key
+            )));
+        }
+        // Non-delta properties (user/application properties) are always allowed
+    }
+    Ok(())
+}
+
 /// Creates a builder for creating a new Delta table.
 ///
 /// This function returns a [`CreateTableTransactionBuilder`] that can be configured with table
@@ -150,12 +215,55 @@ impl CreateTableTransactionBuilder {
         }
     }
 
+    /// Sets table properties for the new Delta table.
+    ///
+    /// Custom application properties (those not starting with `delta.`) are always allowed.
+    /// Delta properties (`delta.*`) are validated against an allow list during [`build()`].
+    /// Feature flags (`delta.feature.*`) are not supported during CREATE TABLE.
+    ///
+    /// This method can be called multiple times. If a property key already exists from a
+    /// previous call, the new value will overwrite the old one.
+    ///
+    /// # Arguments
+    ///
+    /// * `properties` - A map of table property names to their values
+    ///
+    /// # Example
+    ///
+    /// ```rust,no_run
+    /// # use delta_kernel::transaction::create_table::create_table;
+    /// # use delta_kernel::schema::{StructType, DataType, StructField};
+    /// # use std::sync::Arc;
+    /// # fn example() -> delta_kernel::DeltaResult<()> {
+    /// # let schema = Arc::new(StructType::try_new(vec![StructField::new("id", DataType::INTEGER, false)])?);
+    /// let builder = create_table("/path/to/table", schema, "MyApp/1.0")
+    ///     .with_table_properties([
+    ///         ("myapp.version", "1.0"),
+    ///         ("myapp.author", "test"),
+    ///     ]);
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// [`build()`]: CreateTableTransactionBuilder::build
+    pub fn with_table_properties<I, K, V>(mut self, properties: I) -> Self
+    where
+        I: IntoIterator<Item = (K, V)>,
+        K: Into<String>,
+        V: Into<String>,
+    {
+        self.table_properties
+            .extend(properties.into_iter().map(|(k, v)| (k.into(), v.into())));
+        self
+    }
+
     /// Builds a [`Transaction`] that can be committed to create the table.
     ///
     /// This method performs validation:
     /// - Checks that the table path is valid
     /// - Verifies the table doesn't already exist
     /// - Validates the schema is non-empty
+    /// - Validates table properties against the allow list
     ///
     /// # Arguments
     ///
@@ -168,6 +276,7 @@ impl CreateTableTransactionBuilder {
     /// - The table path is invalid
     /// - A table already exists at the given path
     /// - The schema is empty
+    /// - Unsupported delta properties or feature flags are specified
     pub fn build(
         self,
         engine: &dyn Engine,
@@ -181,6 +290,9 @@ impl CreateTableTransactionBuilder {
             return Err(Error::generic("Schema cannot be empty"));
         }
 
+        // Validate table properties against allow list
+        validate_table_properties(&self.table_properties)?;
+
         // Check if table already exists by looking for _delta_log directory
         let delta_log_url = table_url.join("_delta_log/")?;
         let storage = engine.storage_handler();
@@ -207,13 +319,6 @@ impl CreateTableTransactionBuilder {
         // Create pre-commit snapshot from protocol/metadata
         let log_root = table_url.join("_delta_log/")?;
         let log_segment = LogSegment::for_pre_commit(log_root);
-
-        // We validate that the table properties are empty. Supported
-        // table properties for create table will be allowlisted in the future.
-        assert!(
-            metadata.configuration().is_empty(),
-            "Base create table API does not support table properties"
-        );
         let table_configuration =
             TableConfiguration::try_new(metadata, protocol, table_url, PRE_COMMIT_VERSION)?;
 
@@ -231,6 +336,7 @@ impl CreateTableTransactionBuilder {
 mod tests {
     use super::*;
     use crate::schema::{DataType, StructField, StructType};
+    use crate::utils::test_utils::assert_result_error_with_message;
     use std::sync::Arc;
 
     fn test_schema() -> SchemaRef {
@@ -263,4 +369,79 @@ mod tests {
 
         assert_eq!(builder.path, "/path/to/table/nested");
     }
+
+    #[test]
+    fn test_with_table_properties() {
+        let schema = test_schema();
+
+        let builder = CreateTableTransactionBuilder::new("/path/to/table", schema, "TestApp/1.0")
+            .with_table_properties([("key1", "value1")]);
+
+        assert_eq!(
+            builder.table_properties.get("key1"),
+            Some(&"value1".to_string())
+        );
+    }
+
+    #[test]
+    fn test_with_multiple_table_properties() {
+        let schema = test_schema();
+
+        let builder = CreateTableTransactionBuilder::new("/path/to/table", schema, "TestApp/1.0")
+            .with_table_properties([("key1", "value1")])
+            .with_table_properties([("key2", "value2")]);
+
+        assert_eq!(
+            builder.table_properties.get("key1"),
+            Some(&"value1".to_string())
+        );
+        assert_eq!(
+            builder.table_properties.get("key2"),
+            Some(&"value2".to_string())
+        );
+    }
+
+    #[test]
+    fn test_validate_supported_properties() {
+        // Empty properties are allowed
+        let properties = HashMap::new();
+        assert!(validate_table_properties(&properties).is_ok());
+
+        // User/application properties are allowed
+        let mut properties = HashMap::new();
+        properties.insert("myapp.version".to_string(), "1.0".to_string());
+        properties.insert("custom.setting".to_string(), "value".to_string());
+        assert!(validate_table_properties(&properties).is_ok());
+    }
+
+    #[test]
+    fn test_validate_unsupported_properties() {
+        // Delta properties not on allow list are rejected
+        let mut properties = HashMap::new();
+        properties.insert("delta.enableChangeDataFeed".to_string(), "true".to_string());
+        assert_result_error_with_message(
+            validate_table_properties(&properties),
+            "Setting delta property 'delta.enableChangeDataFeed' is not supported",
+        );
+
+        // Feature override properties are rejected
+        let mut properties = HashMap::new();
+        properties.insert(
+            "delta.feature.domainMetadata".to_string(),
+            "supported".to_string(),
+        );
+        assert_result_error_with_message(
+            validate_table_properties(&properties),
+            "Setting feature override property 'delta.feature.domainMetadata' is not supported",
+        );
+
+        // 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("delta.appendOnly".to_string(), "true".to_string());
+        assert_result_error_with_message(
+            validate_table_properties(&properties),
+            "Setting delta property 'delta.appendOnly' is not supported",
+        );
+    }
 }