feat: Add CreateTable API with simplified single-stage flow

Implement create-table functionality where CreateTableTransactionBuilder::build()
returns a Transaction with stored actions to be used for a commit.

API Usage:
  let result = create_table(path, schema, engine_info)
      .build(engine, committer)?
      .commit(engine)?;

This specific change doesn't allow table properties and features to be set and
has validations in the transaction module which error if unsupported features or
properties such as row tracking and ICT are set in the table configuration being
pushed down.

Key Changes:
- CreateTableTransactionBuilder::build() takes committer and returns Transaction
  with commit info, protocol and metadata actions.
- Transaction struct holds optional protocol/metadata actions for
  create-table
- Adds try_new_create_table() constructor alongside
  try_new_existing_table()
- commit() handles both existing-table and create-table flows
- get_write_context() now returns DeltaResult<WriteContext> for
  proper error handling

This aligns the Rust Kernel's create-table flow with the Java Kernel's
approach where Transaction is the single unit for all commit operations.

Testing:
Unit tests

Author: sanujbasu

.gitignore

@@ -7,6 +7,8 @@
 .vscode/
 .vim
 .zed
+.cache/
+.clangd
 
 # Rust
 target/

ffi/src/transaction/mod.rs

@@ -288,7 +288,9 @@ mod tests {
                 ))
             };
 
-            let write_context = unsafe { get_write_context(txn_with_engine_info.shallow_copy()) };
+            let write_context = ok_or_panic(unsafe {
+                get_write_context(txn_with_engine_info.shallow_copy(), engine.shallow_copy())
+            });
 
             // Ensure we get the correct schema
             let write_schema = unsafe { get_write_schema(write_context.shallow_copy()) };

ffi/src/transaction/write_context.rs

@@ -1,5 +1,8 @@
+use crate::error::{ExternResult, IntoExternResult};
 use crate::handle::Handle;
-use crate::{kernel_string_slice, AllocateStringFn, NullableCvoid, SharedSchema};
+use crate::{
+    kernel_string_slice, AllocateStringFn, NullableCvoid, SharedExternEngine, SharedSchema,
+};
 use delta_kernel::transaction::WriteContext;
 use delta_kernel_ffi_macros::handle_descriptor;
 
@@ -19,13 +22,18 @@ pub struct SharedWriteContext;
 ///
 /// # Safety
 ///
-/// Caller is responsible for passing a [valid][Handle#Validity] transaction handle.
+/// Caller is responsible for passing valid transaction and engine handles.
 #[no_mangle]
 pub unsafe extern "C" fn get_write_context(
     txn: Handle<ExclusiveTransaction>,
-) -> Handle<SharedWriteContext> {
+    engine: Handle<SharedExternEngine>,
+) -> ExternResult<Handle<SharedWriteContext>> {
     let txn = unsafe { txn.as_ref() };
-    Arc::new(txn.get_write_context()).into()
+    let engine = unsafe { engine.as_ref() };
+    let write_context = txn.get_write_context();
+    write_context
+        .map(|wc| Arc::new(wc).into())
+        .into_extern_result(&engine)
 }
 
 #[no_mangle]

kernel/examples/write-table/src/main.rs

@@ -1,5 +1,5 @@
 use std::collections::HashMap;
-use std::fs::{create_dir_all, write};
+use std::fs::create_dir_all;
 use std::path::Path;
 use std::process::ExitCode;
 use std::sync::Arc;
@@ -9,9 +9,7 @@ use arrow::util::pretty::print_batches;
 use clap::Parser;
 use common::{LocationArgs, ParseWithExamples};
 use itertools::Itertools;
-use serde_json::{json, to_vec};
 use url::Url;
-use uuid::Uuid;
 
 use delta_kernel::arrow::array::TimestampMicrosecondArray;
 use delta_kernel::committer::FileSystemCommitter;
@@ -20,6 +18,7 @@ use delta_kernel::engine::arrow_data::{ArrowEngineData, EngineDataArrowExt};
 use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
 use delta_kernel::engine::default::{DefaultEngine, DefaultEngineBuilder};
 use delta_kernel::schema::{DataType, SchemaRef, StructField, StructType};
+use delta_kernel::transaction::create_table::create_table as create_delta_table;
 use delta_kernel::transaction::{CommitResult, RetryableTransaction};
 use delta_kernel::{DeltaResult, Engine, Error, Snapshot, SnapshotRef};
 
@@ -94,7 +93,7 @@ async fn try_main() -> DeltaResult<()> {
         .with_data_change(true);
 
     // Write the data using the engine
-    let write_context = Arc::new(txn.get_write_context());
+    let write_context = Arc::new(txn.get_write_context()?);
     let file_metadata = engine
         .write_parquet(&sample_data, write_context.as_ref(), HashMap::new())
         .await?;
@@ -152,7 +151,7 @@ async fn create_or_get_base_snapshot(
             // Create new table
             println!("Creating new Delta table...");
             let schema = parse_schema(schema_str)?;
-            create_table(url, &schema).await?;
+            create_table(url, &schema, engine).await?;
             Snapshot::builder_for(url.clone()).build(engine)
         }
     }
@@ -192,66 +191,13 @@ fn parse_schema(schema_str: &str) -> DeltaResult<SchemaRef> {
     Ok(Arc::new(StructType::try_new(fields)?))
 }
 
-/// Create a new Delta table with the given schema.
-///
-/// Creating a Delta table is not officially supported by kernel-rs yet, so we manually create the
-/// initial transaction log.
-async fn create_table(table_url: &Url, schema: &SchemaRef) -> DeltaResult<()> {
-    let table_id = Uuid::new_v4().to_string();
-    let schema_str = serde_json::to_string(&schema)?;
-
-    let (reader_features, writer_features) = {
-        let reader_features: Vec<&'static str> = vec![];
-        let writer_features: Vec<&'static str> = vec![];
-
-        // TODO: Support adding specific table features
-        (reader_features, writer_features)
-    };
-
-    let protocol = json!({
-        "protocol": {
-            "minReaderVersion": 3,
-            "minWriterVersion": 7,
-            "readerFeatures": reader_features,
-            "writerFeatures": writer_features,
-        }
-    });
-    let partition_columns: Vec<String> = vec![];
-    let metadata = json!({
-        "metaData": {
-            "id": table_id,
-            "format": {
-                "provider": "parquet",
-                "options": {}
-            },
-            "schemaString": schema_str,
-            "partitionColumns": partition_columns,
-            "configuration": {},
-            "createdTime": 1677811175819u64
-        }
-    });
-
-    let data = [
-        to_vec(&protocol).unwrap(),
-        b"\n".to_vec(),
-        to_vec(&metadata).unwrap(),
-    ]
-    .concat();
-
-    // Write the initial transaction with protocol and metadata to 0.json
-    let delta_log_path = table_url
-        .join("_delta_log/")?
-        .to_file_path()
-        .map_err(|_e| Error::generic("URL cannot be converted to local file path"))?;
-    let file_path = delta_log_path.join("00000000000000000000.json");
-
-    // Create the _delta_log directory if it doesn't exist
-    create_dir_all(&delta_log_path)
-        .map_err(|e| Error::generic(format!("Failed to create _delta_log directory: {e}")))?;
-
-    // Write the file using standard filesystem operations
-    write(&file_path, data)
-        .map_err(|e| Error::generic(format!("Failed to write initial transaction log: {e}")))?;
+/// Create a new Delta table with the given schema using the official CreateTable API.
+async fn create_table(table_url: &Url, schema: &SchemaRef, engine: &dyn Engine) -> DeltaResult<()> {
+    // Use the create_table API to create the table
+    let table_path = table_url.as_str();
+    let _result = create_delta_table(table_path, schema.clone(), "write-table-example/1.0")
+        .build(engine, Box::new(FileSystemCommitter::new()))?
+        .commit(engine)?;
 
     println!("✓ Created Delta table with schema: {schema:#?}");
     Ok(())

kernel/src/lib.rs

@@ -175,6 +175,11 @@ pub mod engine;
 
 /// Delta table version is 8 byte unsigned int
 pub type Version = u64;
+
+/// Sentinel version indicating a pre-commit state (table does not exist yet).
+/// Used for create-table transactions before the first commit.
+pub const PRE_COMMIT_VERSION: Version = u64::MAX;
+
 pub type FileSize = u64;
 pub type FileIndex = u64;
 

kernel/src/log_segment.rs

@@ -77,6 +77,27 @@ pub(crate) struct LogSegment {
 }
 
 impl LogSegment {
+    /// Creates a synthetic LogSegment for pre-commit transactions (e.g., create-table).
+    /// The sentinel version PRE_COMMIT_VERSION indicates no version exists yet on disk.
+    /// This is used to construct a pre-commit snapshot that provides table configuration
+    /// (protocol, metadata, schema) for operations like CTAS.
+    #[allow(dead_code)] // Used by create_table module
+    pub(crate) fn for_pre_commit(log_root: Url) -> Self {
+        use crate::PRE_COMMIT_VERSION;
+        Self {
+            end_version: PRE_COMMIT_VERSION,
+            checkpoint_version: None,
+            log_root,
+            ascending_commit_files: vec![],
+            ascending_compaction_files: vec![],
+            checkpoint_parts: vec![],
+            latest_crc_file: None,
+            latest_commit_file: None,
+            checkpoint_schema: None,
+            max_published_version: None,
+        }
+    }
+
     #[internal_api]
     pub(crate) fn try_new(
         listed_files: ListedLogFiles,

kernel/src/snapshot.rs

@@ -443,7 +443,7 @@ impl Snapshot {
 
     /// Create a [`Transaction`] for this `SnapshotRef`. With the specified [`Committer`].
     pub fn transaction(self: Arc<Self>, committer: Box<dyn Committer>) -> DeltaResult<Transaction> {
-        Transaction::try_new(self, committer)
+        Transaction::try_new_existing_table(self, committer)
     }
 
     /// Fetch the latest version of the provided `application_id` for this snapshot. Filters the txn based on the SetTransactionRetentionDuration property and lastUpdated

kernel/src/table_features/mod.rs

@@ -15,6 +15,18 @@ pub(crate) use timestamp_ntz::validate_timestamp_ntz_feature_support;
 mod column_mapping;
 mod timestamp_ntz;
 
+/// Minimum reader version for tables that use table features.
+/// When set to 3, the protocol requires an explicit `readerFeatures` array.
+#[internal_api]
+#[allow(dead_code)] // Used by create_table module
+pub(crate) const TABLE_FEATURES_MIN_READER_VERSION: i32 = 3;
+
+/// Minimum writer version for tables that use table features.
+/// When set to 7, the protocol requires an explicit `writerFeatures` array.
+#[internal_api]
+#[allow(dead_code)] // Used by create_table module
+pub(crate) const TABLE_FEATURES_MIN_WRITER_VERSION: i32 = 7;
+
 /// 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.