delta_kernel 0.30.0

//! Integration tests for the CreateTable API

mod clustering;
mod column_mapping;
mod ctas;
mod iceberg_compat;
mod ict;
mod interval;
mod partitioned;
mod row_tracking;
mod timestamp_ntz;
mod variant;

use std::sync::Arc;

use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::schema::{
    schema_ref, ColumnMetadataKey, DataType, MetadataValue, StructField, StructType,
};
use delta_kernel::snapshot::Snapshot;
use delta_kernel::table_features::{
    TableFeature, TABLE_FEATURES_MIN_READER_VERSION, TABLE_FEATURES_MIN_WRITER_VERSION,
};
use delta_kernel::table_properties::TableProperties;
use delta_kernel::transaction::create_table::{create_table, CreateTableTransaction};
use delta_kernel::transaction::data_layout::DataLayout;
use delta_kernel::{KernelError, KernelResult};
use rstest::rstest;
use serde_json::Value;
use test_utils::{assert_result_error_with_message, test_table_setup, test_table_setup_mt};

/// Helper to create a simple two-column schema for tests.
/// Shared with sub-modules.
pub(crate) fn simple_schema() -> KernelResult<Arc<StructType>> {
    Ok(schema_ref! {
        nullable "id": INTEGER,
        nullable "value": STRING,
    })
}

/// Helper to create a three-column schema for partition tests (id, date, value).
/// Shared with sub-modules.
pub(crate) fn partition_test_schema() -> KernelResult<Arc<StructType>> {
    Ok(schema_ref! {
        nullable "id": INTEGER,
        nullable "date": DATE,
        nullable "value": STRING,
    })
}

#[tokio::test]
async fn test_create_simple_table() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    // Create schema for an events table
    let schema = schema_ref! {
        nullable "event_id": LONG,
        nullable "user_id": LONG,
        nullable "event_type": STRING,
        nullable "timestamp": TIMESTAMP,
        nullable "properties": STRING,
    };

    // Create table using new API
    let _ = create_table(&table_path, schema.clone(), "DeltaKernel-RS/0.17.0", true)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
        .commit(engine.as_ref())?;

    // Verify table was created
    let table_url = delta_kernel::try_parse_uri(&table_path)?;
    let snapshot = Snapshot::builder_for(table_url).build(engine.as_ref())?;

    assert_eq!(snapshot.version(), 0);
    assert_eq!(snapshot.schema().fields().len(), 5);

    // Verify protocol versions via snapshot
    let protocol = snapshot.table_configuration().protocol();
    assert_eq!(
        protocol.min_reader_version(),
        TABLE_FEATURES_MIN_READER_VERSION
    );
    assert_eq!(
        protocol.min_writer_version(),
        TABLE_FEATURES_MIN_WRITER_VERSION
    );
    // Verify no reader/writer features are set (empty for table features mode)
    assert!(protocol.reader_features().is_some_and(|f| f.is_empty()));
    assert!(protocol.writer_features().is_some_and(|f| f.is_empty()));

    // Verify no table properties are set via public API
    assert_eq!(snapshot.table_properties(), &TableProperties::default());

    // Verify schema field names
    let field_names: Vec<_> = snapshot
        .schema()
        .fields()
        .map(|f| f.name().to_string())
        .collect();
    assert!(field_names.contains(&"event_id".to_string()));
    assert!(field_names.contains(&"user_id".to_string()));
    assert!(field_names.contains(&"event_type".to_string()));
    assert!(field_names.contains(&"timestamp".to_string()));
    assert!(field_names.contains(&"properties".to_string()));

    Ok(())
}

#[rstest]
#[case::cdf_enabled(Some("true"), Some("reserved for Change Data Feed"))]
#[case::cdf_supported_only(None, None)]
#[tokio::test]
async fn create_table_validates_cdf_column_names(
    #[case] cdf_enabled: Option<&str>,
    #[case] expected_error: Option<&str>,
    #[values(
        "_change_type",
        "_commit_version",
        "_commit_timestamp",
        "_CHANGE_TYPE",
        "_COMMIT_VERSION",
        "_COMMIT_TIMESTAMP"
    )]
    column_name: &str,
    #[values("none", "name", "id")] cm_mode: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let schema = schema_ref! { (StructField::nullable(column_name, DataType::STRING)) };
    let mut properties = vec![
        ("delta.feature.changeDataFeed", "supported"),
        ("delta.columnMapping.mode", cm_mode),
    ];
    if let Some(value) = cdf_enabled {
        properties.push(("delta.enableChangeDataFeed", value));
    }
    let result = create_table(&table_path, schema, "test", true)
        .with_table_properties(properties)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()));

    if let Some(expected_error) = expected_error {
        assert_result_error_with_message(result, expected_error);
    } else {
        let snapshot = result?
            .commit(engine.as_ref())?
            .unwrap_post_commit_snapshot();
        assert!(snapshot.schema().contains(column_name));
        assert_eq!(
            snapshot.table_properties().enable_change_data_feed,
            cdf_enabled.map(|v| v == "true")
        );
    }
    Ok(())
}

#[rstest]
#[case::name_cdf_enabled("name", true, Some("has physical name"))]
#[case::id_cdf_enabled("id", true, Some("has physical name"))]
#[case::none_cdf_enabled("none", true, None)]
#[case::name_cdf_supported_only("name", false, None)]
#[case::id_cdf_supported_only("id", false, None)]
#[case::none_cdf_supported_only("none", false, None)]
#[tokio::test]
async fn create_table_validates_cdf_reserved_physical_column_names(
    #[case] cm_mode: &str,
    #[case] cdf_enabled: bool,
    #[case] expected_error: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let schema = schema_ref! {
        (StructField::nullable("value", DataType::STRING).with_metadata([
            (ColumnMetadataKey::ColumnMappingPhysicalName.as_ref(), "_change_type"),
        ])),
    };
    let mut properties = vec![
        ("delta.feature.changeDataFeed", "supported"),
        ("delta.columnMapping.mode", cm_mode),
    ];
    if cdf_enabled {
        properties.push(("delta.enableChangeDataFeed", "true"));
    }
    let result = create_table(&table_path, schema, "test", true)
        .with_table_properties(properties)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()));

    if let Some(expected_error) = expected_error {
        assert_result_error_with_message(result, expected_error);
    } else {
        let snapshot = result?
            .commit(engine.as_ref())?
            .unwrap_post_commit_snapshot();
        let expected_physical_name =
            (cm_mode != "none").then(|| MetadataValue::String("_change_type".into()));
        assert_eq!(
            snapshot
                .schema()
                .field("value")
                .unwrap()
                .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName),
            expected_physical_name.as_ref()
        );
    }
    Ok(())
}

#[tokio::test]
async fn test_create_table_with_user_domain_metadata() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    let schema = simple_schema()?;

    // Create table with domainMetadata feature enabled
    let txn = create_table(&table_path, schema, "Test/1.0", true)
        .with_table_properties([("delta.feature.domainMetadata", "supported")])
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;

    // Add user domain metadata during table creation
    let domain = "app.settings";
    let config = r#"{"version": 1, "enabled": true}"#;

    let _ = txn
        .with_domain_metadata(domain.to_string(), config.to_string())
        .commit(engine.as_ref())?;

    // Load snapshot and verify domain metadata was persisted
    let table_url = delta_kernel::try_parse_uri(&table_path)?;
    let snapshot = Snapshot::builder_for(table_url).build(engine.as_ref())?;

    // Verify domainMetadata feature is enabled in protocol
    assert!(
        snapshot
            .table_configuration()
            .is_feature_supported(&TableFeature::DomainMetadata),
        "DomainMetadata feature should be enabled"
    );

    // Verify domain metadata string was persisted correctly
    let retrieved_config = snapshot.get_domain_metadata(domain, engine.as_ref())?;
    assert_eq!(
        retrieved_config,
        Some(config.to_string()),
        "Domain metadata should be persisted and retrievable"
    );

    // Parse and verify the JSON contents
    let parsed: Value = serde_json::from_str(retrieved_config.as_ref().unwrap())?;
    assert_eq!(parsed["version"], 1);
    assert_eq!(parsed["enabled"], true);

    // Verify non-existent domain returns None
    let missing = snapshot.get_domain_metadata("nonexistent.domain", engine.as_ref())?;
    assert!(missing.is_none(), "Non-existent domain should return None");

    Ok(())
}

#[tokio::test]
async fn test_create_table_already_exists() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    // Create schema for a user profiles table
    let schema = schema_ref! {
        nullable "user_id": LONG,
        nullable "username": STRING,
        nullable "email": STRING,
        nullable "created_at": TIMESTAMP,
        nullable "is_active": BOOLEAN,
    };

    // Create table first time
    let _ = create_table(
        &table_path,
        schema.clone(),
        "UserManagementService/1.2.0",
        true,
    )
    .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
    .commit(engine.as_ref())?;

    // Try to create again - should fail at build time (table already exists)
    let result = create_table(
        &table_path,
        schema.clone(),
        "UserManagementService/1.2.0",
        true,
    )
    .build(engine.as_ref(), Box::new(FileSystemCommitter::new()));

    assert!(matches!(
        result,
        Err(KernelError::TableAlreadyExistsAtPath(path)) if path == table_path
    ));

    Ok(())
}

#[tokio::test]
async fn test_create_table_empty_schema_succeeds() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    // CREATE TABLE with no columns is a valid Delta operation; users may add columns
    // later via ALTER TABLE ADD COLUMN.
    let schema = schema_ref! {};

    create_table(&table_path, schema, "EmptySchemaApp/0.1.0", true)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let table_url = delta_kernel::try_parse_uri(&table_path)?;
    let snapshot = Snapshot::builder_for(table_url).build(engine.as_ref())?;
    assert_eq!(snapshot.version(), 0);
    assert_eq!(snapshot.schema().num_fields(), 0);

    Ok(())
}

/// Checkpoints serialize the table schema into their parquet payload, so a zero-column
/// schema is a corner the writer can plausibly mishandle. This test creates an empty
/// table, checkpoints immediately, then reloads from scratch to confirm the schema
/// survives the round-trip across all column-mapping modes.
#[rstest]
#[case::no_cm(None)]
#[case::cm_name(Some("name"))]
#[case::cm_id(Some("id"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_create_table_empty_schema_checkpoint_round_trip(
    #[case] cm_mode: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup_mt()?;

    let schema = schema_ref! {};
    let mut builder = create_table(&table_path, schema, "EmptySchemaApp/0.1.0", true);
    if let Some(mode) = cm_mode {
        builder = builder.with_table_properties([("delta.columnMapping.mode", mode)]);
    }
    builder
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let table_url = delta_kernel::try_parse_uri(&table_path)?;
    let v0 = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
    let (_, _ckpt_snapshot) = v0.checkpoint(engine.as_ref(), None)?;

    let reloaded = Snapshot::builder_for(table_url).build(engine.as_ref())?;
    assert_eq!(reloaded.version(), 0);
    assert_eq!(reloaded.schema().num_fields(), 0);
    let max_id = reloaded
        .table_configuration()
        .metadata()
        .configuration()
        .get("delta.columnMapping.maxColumnId")
        .and_then(|v| v.parse::<i64>().ok());
    assert_eq!(max_id, cm_mode.map(|_| 0));

    Ok(())
}

#[rstest]
#[case::partitioned_empty(DataLayout::partitioned::<_, &str>([]), "at least one column")]
#[case::clustered_missing_column(DataLayout::clustered(["foo"]), "not found in schema")]
#[tokio::test]
async fn test_create_table_empty_schema_layout_errors(
    #[case] layout: DataLayout,
    #[case] expected_err: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    let schema = schema_ref! {};
    let result = create_table(&table_path, schema, "EmptySchemaApp/0.1.0", true)
        .with_data_layout(layout)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()));

    assert_result_error_with_message(result, expected_err);

    Ok(())
}

fn top_level_non_null_schema() -> Arc<StructType> {
    schema_ref! {
        not_null "id": INTEGER,
        nullable "value": STRING,
    }
}

fn nested_non_null_schema() -> Arc<StructType> {
    schema_ref! {
        nullable "nested": {
            not_null "child": INTEGER,
        },
    }
}

/// CREATE TABLE with non-null columns succeeds and auto-enables the `invariants`
/// writer feature in the protocol. Delta-Spark treats non-null schema columns as
/// implicit invariants and requires the feature to be declared.
#[rstest]
#[case::top_level_non_null(top_level_non_null_schema())]
#[case::nested_non_null(nested_non_null_schema())]
fn test_create_table_with_non_null_columns_auto_enables_invariants(
    #[case] schema: Arc<StructType>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    let _ = create_table(&table_path, schema, "Test/1.0", true)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
        .commit(engine.as_ref())?;

    let snapshot = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let protocol = snapshot.table_configuration().protocol();
    assert!(
        protocol
            .writer_features()
            .is_some_and(|f| f.contains(&TableFeature::Invariants)),
        "Invariants feature should be auto-added to writer features for non-null schema"
    );
    // Invariants is writer-only; reader features should not contain it.
    assert!(
        !protocol
            .reader_features()
            .is_some_and(|f| f.contains(&TableFeature::Invariants)),
        "Invariants should not appear in reader features"
    );

    Ok(())
}

/// CREATE TABLE with `delta.feature.invariants=supported` + a non-null schema succeeds
/// and lands the feature in writerFeatures exactly once. Non-null columns auto-enable
/// the feature and the explicit signal also tries to enable it; both paths firing
/// together must not duplicate the entry. This also verifies users can pre-enable the
/// feature so a later ALTER TABLE ADD COLUMN NOT NULL does not need a protocol upgrade.
#[tokio::test]
async fn test_create_table_with_invariants_feature_signal_allowed() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    // Non-null schema auto-enables Invariants; the feature signal also tries to add it.
    // Both code paths firing together must produce exactly one Invariants entry.
    let schema = top_level_non_null_schema();

    let _ = create_table(&table_path, schema, "Test/1.0", true)
        .with_table_properties([("delta.feature.invariants", "supported")])
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
        .commit(engine.as_ref())?;

    let snapshot = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let protocol = snapshot.table_configuration().protocol();
    let writer_features = protocol
        .writer_features()
        .expect("writer features should be present");
    let invariants_count = writer_features
        .iter()
        .filter(|f| **f == TableFeature::Invariants)
        .count();
    assert_eq!(
        invariants_count, 1,
        "Invariants should appear exactly once in writerFeatures; got {invariants_count}"
    );
    assert!(
        !protocol
            .reader_features()
            .is_some_and(|f| f.contains(&TableFeature::Invariants)),
        "Invariants should not appear in readerFeatures (writer-only feature)"
    );

    Ok(())
}

/// CREATE TABLE rejects any schema with `delta.invariants` metadata annotations
/// because kernel cannot evaluate SQL expression invariants.
#[tokio::test]
async fn test_create_table_rejects_delta_invariants_metadata() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    let mut field = StructField::nullable("x", DataType::INTEGER);
    field.metadata.insert(
        ColumnMetadataKey::Invariants.as_ref().to_string(),
        MetadataValue::String(r#"{"expression": {"expression": "x > 0"}}"#.to_string()),
    );
    let schema = schema_ref! { (field) };

    let result = create_table(&table_path, schema, "Test/1.0", true)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()));

    assert_result_error_with_message(result, "delta.invariants");

    Ok(())
}

#[rstest]
#[case::builder_metrics(false)]
#[case::transaction_metrics_replace_builder_metrics(true)]
#[tokio::test]
async fn test_create_table_log_actions(
    #[case] replace_metrics_after_build: bool,
    #[values("AuditService/2.1.0", "")] engine_info: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    // Create schema
    let schema = schema_ref! {
        nullable "user_id": LONG,
        nullable "action": STRING,
    };

    // Create table
    let mut txn = create_table(&table_path, schema, engine_info, true)
        .with_operation_parameters([("mode", Some("Create")), ("description", None)])
        .with_operation_metrics(if replace_metrics_after_build {
            [("staleMetric", Some("1"))]
        } else {
            [("numFiles", Some("0"))]
        })
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;
    if replace_metrics_after_build {
        txn = txn.with_operation_metrics([("numFiles", Some("0"))]);
    }
    let _ = txn.commit(engine.as_ref())?;

    // Read the actual Delta log file
    let log_file_path = format!("{table_path}/_delta_log/00000000000000000000.json");
    let log_contents = std::fs::read_to_string(&log_file_path).expect("Failed to read log file");

    // Parse each line (each line is a separate JSON action)
    let actions: Vec<Value> = log_contents
        .lines()
        .map(|line| serde_json::from_str(line).expect("Failed to parse JSON"))
        .collect();

    // Verify we have exactly 3 actions: CommitInfo, Protocol, Metadata
    // CommitInfo is first to comply with ICT (In-Commit Timestamps) protocol requirements
    assert_eq!(
        actions.len(),
        3,
        "Expected 3 actions (commitInfo, protocol, metaData), found {}",
        actions.len()
    );

    // Verify CommitInfo action (first for ICT compliance)
    let commit_info_action = &actions[0];
    assert!(
        commit_info_action.get("commitInfo").is_some(),
        "First action should be commitInfo"
    );
    let commit_info = commit_info_action.get("commitInfo").unwrap();
    assert!(
        commit_info.get("timestamp").is_some(),
        "CommitInfo should have timestamp"
    );
    assert!(
        commit_info.get("engineInfo").is_some(),
        "CommitInfo should have engineInfo"
    );
    assert!(
        commit_info.get("operation").is_some(),
        "CommitInfo should have operation"
    );
    assert_eq!(
        commit_info["operation"], "CREATE TABLE",
        "Operation should be CREATE TABLE"
    );
    assert_eq!(
        commit_info["operationParameters"],
        serde_json::json!({"description": null, "mode": "Create"})
    );
    assert_eq!(
        commit_info["operationMetrics"],
        serde_json::json!({"numFiles": "0"})
    );

    // Verify Protocol action
    let protocol_action = &actions[1];
    assert!(
        protocol_action.get("protocol").is_some(),
        "Second action should be protocol"
    );
    let protocol = protocol_action.get("protocol").unwrap();
    assert_eq!(
        protocol["minReaderVersion"],
        TABLE_FEATURES_MIN_READER_VERSION
    );
    assert_eq!(
        protocol["minWriterVersion"],
        TABLE_FEATURES_MIN_WRITER_VERSION
    );

    // Verify Metadata action
    let metadata_action = &actions[2];
    assert!(
        metadata_action.get("metaData").is_some(),
        "Third action should be metaData"
    );
    let metadata = metadata_action.get("metaData").unwrap();
    assert!(metadata.get("id").is_some(), "Metadata should have id");
    assert!(
        metadata.get("schemaString").is_some(),
        "Metadata should have schemaString"
    );
    assert!(
        metadata.get("createdTime").is_some(),
        "Metadata should have createdTime"
    );

    // Additional CommitInfo verification (commit_info was already extracted from actions[0] above)
    assert_eq!(
        commit_info["engineInfo"], engine_info,
        "CommitInfo should contain the engine info we provided"
    );

    assert!(
        commit_info.get("txnId").is_some(),
        "CommitInfo should have txnId"
    );

    // Verify kernelVersion is present
    let kernel_version = commit_info.get("kernelVersion");
    assert!(
        kernel_version.is_some(),
        "CommitInfo should have kernelVersion"
    );
    assert!(
        kernel_version.unwrap().as_str().unwrap().starts_with("v"),
        "Kernel version should start with 'v'"
    );

    Ok(())
}

/// Helper to create a `CreateTableTransaction` for tests.
fn create_test_create_table_txn() -> KernelResult<(
    Arc<impl delta_kernel::Engine>,
    CreateTableTransaction,
    tempfile::TempDir,
)> {
    let (tempdir, table_path, engine) = test_table_setup()?;
    let schema = schema_ref! {
        nullable "id": INTEGER,
        nullable "name": STRING,
    };
    let txn = create_table(&table_path, schema, "test_engine", true)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;
    Ok((engine, txn, tempdir))
}

#[tokio::test]
async fn test_create_table_txn_debug() -> KernelResult<()> {
    let (_engine, txn, _tempdir) = create_test_create_table_txn()?;
    let debug_str = format!("{txn:?}");
    assert!(
        debug_str.contains("Transaction") && debug_str.contains("create_table"),
        "Debug output should contain Transaction info: {debug_str}"
    );
    Ok(())
}

#[rstest]
// ReaderWriter features (AlwaysIfSupported)
#[case(
    "vacuumProtocolCheck",
    TableFeature::VacuumProtocolCheck,
    true,
    true,
    false
)]
#[case("v2Checkpoint", TableFeature::V2Checkpoint, true, true, false)]
#[case("variantType", TableFeature::VariantType, true, true, false)]
#[case("variantShredding", TableFeature::VariantShredding, true, true, false)]
// ReaderWriter features (EnabledIf -- feature signal alone does not enable)
#[case("deletionVectors", TableFeature::DeletionVectors, true, false, false)]
#[case("typeWidening", TableFeature::TypeWidening, true, false, false)]
// WriterOnly features (EnabledIf -- feature signal alone does not enable)
#[case("appendOnly", TableFeature::AppendOnly, false, false, false)]
#[case("changeDataFeed", TableFeature::ChangeDataFeed, false, false, false)]
#[case("rowTracking", TableFeature::RowTracking, false, false, true)]
#[case("domainMetadata", TableFeature::DomainMetadata, false, true, true)]
// WriterOnly features (AlwaysIfSupported)
#[case(
    "materializePartitionColumns",
    TableFeature::MaterializePartitionColumns,
    false,
    true,
    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,
    #[case] expects_domain_metadata: bool,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    let property_key = format!("delta.feature.{feature_name}");
    let _ = create_table(&table_path, simple_schema()?, "Test/1.0", true)
        .with_table_properties([(property_key.as_str(), "supported")])
        .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!(
        table_config.is_feature_supported(&TableFeature::DomainMetadata),
        expects_domain_metadata,
    );

    assert!(
        table_config.is_feature_supported(&feature),
        "{feature_name} should be supported"
    );
    assert_eq!(
        table_config.is_feature_enabled(&feature),
        enabled_when_supported,
        "{feature_name}: is_feature_enabled should be {enabled_when_supported}"
    );
    let protocol = table_config.protocol();
    assert!(
        protocol
            .writer_features()
            .is_some_and(|f| f.contains(&feature)),
        "{feature_name} should be in writer features"
    );
    if is_reader_writer {
        assert!(
            protocol
                .reader_features()
                .is_some_and(|f| f.contains(&feature)),
            "{feature_name} should be in reader features"
        );
    }

    Ok(())
}

#[rstest]
#[case::plain(&[], DataLayout::None, false)]
#[case::row_tracking_property(&[("delta.enableRowTracking", "true")], DataLayout::None, true)]
#[case::row_tracking_dependency(&[("delta.enableIcebergCompatV3", "true")], DataLayout::None, true)]
#[case::clustered(&[], DataLayout::clustered(["id"]), true)]
#[case::deduplicated(
    &[("delta.feature.domainMetadata", "supported"), ("delta.enableRowTracking", "true")],
    DataLayout::clustered(["id"]),
    true
)]
fn create_resolves_domain_metadata_after_all_feature_selection(
    #[case] properties: &[(&str, &str)],
    #[case] layout: DataLayout,
    #[case] expects_domain_metadata: bool,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table(&table_path, simple_schema()?, "test", true)
        .with_table_properties(properties.iter().copied())
        .with_data_layout(layout)
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
        .commit(engine.as_ref())?
        .unwrap_post_commit_snapshot();
    let features = snapshot
        .table_configuration()
        .protocol()
        .writer_features()
        .unwrap();
    assert_eq!(
        features
            .iter()
            .filter(|feature| **feature == TableFeature::DomainMetadata)
            .count(),
        usize::from(expects_domain_metadata),
    );
    Ok(())
}

#[test]
fn test_create_table_with_variant_shredding_has_variant_feature() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let schema = schema_ref! {
        nullable "id": INTEGER,
    };

    let _ = create_table(&table_path, schema, "Test/1.0", true)
        .with_table_properties([("delta.feature.variantShredding", "supported")])
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
        .commit(engine.as_ref())?;

    let snapshot = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let protocol = snapshot.table_configuration().protocol();
    let reader_features = protocol
        .reader_features()
        .expect("reader features must be present");
    let writer_features = protocol
        .writer_features()
        .expect("writer features must be present");

    assert!(reader_features.contains(&TableFeature::VariantType));
    assert!(reader_features.contains(&TableFeature::VariantShredding));
    assert!(writer_features.contains(&TableFeature::VariantType));
    assert!(writer_features.contains(&TableFeature::VariantShredding));

    Ok(())
}

#[rstest]
#[case::v2("v2", true)]
#[case::classic("classic", false)]
fn test_create_table_checkpoint_policy_auto_enables_v2_checkpoint(
    #[case] policy: &str,
    #[case] expect_v2_checkpoint: bool,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    let _ = create_table(&table_path, simple_schema()?, "Test/1.0", true)
        .with_table_properties([("delta.checkpointPolicy", policy)])
        .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();
    let protocol = table_config.protocol();

    assert_eq!(
        table_config.is_feature_supported(&TableFeature::V2Checkpoint),
        expect_v2_checkpoint,
        "checkpointPolicy={policy}: v2Checkpoint supported should be {expect_v2_checkpoint}"
    );
    assert_eq!(
        protocol
            .reader_features()
            .is_some_and(|f| f.contains(&TableFeature::V2Checkpoint)),
        expect_v2_checkpoint,
    );
    assert_eq!(
        protocol
            .writer_features()
            .is_some_and(|f| f.contains(&TableFeature::V2Checkpoint)),
        expect_v2_checkpoint,
    );

    Ok(())
}

#[rstest]
fn test_create_table_with_checkpoint_stats_properties(
    #[values(true, false)] write_stats_as_json: bool,
    #[values(true, false)] write_stats_as_struct: bool,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    let json_val = write_stats_as_json.to_string();
    let struct_val = write_stats_as_struct.to_string();

    let _ = create_table(&table_path, simple_schema()?, "Test/1.0", true)
        .with_table_properties([
            ("delta.checkpoint.writeStatsAsJson", json_val.as_str()),
            ("delta.checkpoint.writeStatsAsStruct", struct_val.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 tp = snapshot.table_properties();
    assert_eq!(tp.checkpoint_write_stats_as_json, Some(write_stats_as_json));
    assert_eq!(
        tp.checkpoint_write_stats_as_struct,
        Some(write_stats_as_struct)
    );

    Ok(())
}

#[rstest]
// 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)]
#[case("delta.enableRowTracking", TableFeature::RowTracking, 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,
) -> KernelResult<()> {
    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", true)
        .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!(
        table_config.is_feature_supported(&feature),
        expect_enabled,
        "{property}={value}: feature supported should be {expect_enabled}"
    );
    assert_eq!(
        table_config.is_feature_enabled(&feature),
        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(&feature)),
        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(())
}

#[rstest]
#[case::without_cm(false)]
#[case::with_cm(true)]
fn test_create_table_special_char_column_name(#[case] cm_enabled: bool) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    let schema = schema_ref! {
        nullable "valid_col": INTEGER,
        nullable "bad column": STRING,
    };

    let mut builder = create_table(&table_path, schema, "Test/1.0", true);
    if cm_enabled {
        builder = builder.with_table_properties([("delta.columnMapping.mode", "name")]);
    }
    let result = builder.build(engine.as_ref(), Box::new(FileSystemCommitter::new()));

    if cm_enabled {
        let txn = result?;
        let _ = txn.commit(engine.as_ref())?;

        let snapshot = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
        assert_eq!(snapshot.version(), 0);
        let field_names: Vec<_> = snapshot
            .schema()
            .fields()
            .map(|f| f.name().clone())
            .collect();
        assert!(
            field_names.contains(&"bad column".to_string()),
            "Schema should contain field 'bad column', got: {field_names:?}"
        );
    } else {
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("invalid character"),
            "Expected invalid character error, got: {err}"
        );
    }

    Ok(())
}