delta_kernel 0.30.0

//! Integration tests for schema changes.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use delta_kernel::arrow::array::{new_null_array, Array, Int32Array, StringArray};
use delta_kernel::arrow::datatypes::DataType as ArrowDataType;
use delta_kernel::arrow::record_batch::RecordBatch;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::arrow_conversion::TryIntoArrow as _;
use delta_kernel::engine::arrow_data::ArrowEngineData;
use delta_kernel::expressions::{column_name, ColumnName, Scalar};
use delta_kernel::schema::{
    schema, schema_ref, try_schema, ArrayType, ColumnMetadataKey, DataType, MapType, MetadataValue,
    SchemaRef, StructField, StructType,
};
use delta_kernel::snapshot::Snapshot;
use delta_kernel::table_features::ColumnMappingMode;
use delta_kernel::transaction::create_table::create_table;
use delta_kernel::transaction::data_layout::DataLayout;
use delta_kernel::transaction::UpdateTableOperation;
use delta_kernel::{KernelError, KernelResult};
use rstest::rstest;
use serde_json::json;
use test_utils::{
    add_commit, assert_result_error_with_message, column_mapping_fixtures as fixtures,
    create_table as create_test_table, create_table_and_load_snapshot, engine_store_setup,
    read_actions_from_commit, test_read, test_table_setup, test_table_setup_mt,
    write_batch_to_table,
};

fn simple_schema() -> SchemaRef {
    schema_ref! {
        nullable "id": INTEGER,
        nullable "name": STRING,
    }
}

fn committer() -> Box<FileSystemCommitter> {
    Box::new(FileSystemCommitter::new())
}

#[rstest]
#[tokio::test]
async fn schema_evolution_while_writing_round_trips_new_column(
    #[values("none", "name", "id")] cm_mode: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[("delta.columnMapping.mode", cm_mode)],
    )?;
    let old_batch = RecordBatch::try_new(
        Arc::new(snapshot.schema().as_ref().try_into_arrow()?),
        vec![
            Arc::new(Int32Array::from(vec![1, 2])),
            Arc::new(StringArray::from(vec!["old-a", "old-b"])),
        ],
    )?;
    let before =
        write_batch_to_table(&snapshot, engine.as_ref(), old_batch, HashMap::new()).await?;
    let old_max_id = max_column_id(&before);
    assert_eq!(old_max_id.is_some(), cm_mode != "none");
    let mut transaction = before
        .clone()
        .update_table(UpdateTableOperation::Write, "test-engine", true)
        .add_column(StructField::nullable("country", DataType::STRING))
        .build(engine.as_ref(), committer())?;

    let write_context = transaction.write_state()?.write_context_builder().build()?;
    assert!(write_context.logical_data_schema().contains("country"));
    let batch = RecordBatch::try_new(
        Arc::new(
            write_context
                .logical_data_schema()
                .as_ref()
                .try_into_arrow()?,
        ),
        vec![
            Arc::new(Int32Array::from(vec![3, 4])),
            Arc::new(StringArray::from(vec!["new-a", "new-b"])),
            Arc::new(StringArray::from(vec!["US", "CA"])),
        ],
    )?;
    let add_metadata = engine
        .write_parquet(&ArrowEngineData::new(batch), &write_context)
        .await?;
    transaction.add_files(add_metadata);

    let post = transaction
        .commit(engine.as_ref())?
        .unwrap_post_commit_snapshot();
    let fresh = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let adds = read_actions_from_commit(post.table_root(), 2, "add")?;
    assert_eq!(adds.len(), 1);
    assert_eq!(adds[0]["dataChange"], true);
    assert!(!before.schema().contains("country"));
    for snapshot in [post, fresh] {
        assert_eq!(snapshot.version(), 2);
        let schema = snapshot.schema();
        for name in ["id", "name"] {
            assert_eq!(schema.field(name), before.schema().field(name));
        }
        let country = schema.field("country").expect("added column");
        if let Some(old_max_id) = old_max_id {
            let new_id = country.column_mapping_id().expect("new column ID");
            assert!(new_id > old_max_id);
            assert_eq!(max_column_id(&snapshot), Some(new_id));
            let physical_name = country
                .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName)
                .expect("new physical name");
            assert!(
                matches!(physical_name, MetadataValue::String(name) if name.starts_with("col-"))
            );
            for old_field in before.schema().fields() {
                assert_ne!(
                    Some(physical_name),
                    old_field.get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName),
                );
            }
        } else {
            assert_eq!(country.column_mapping_id(), None);
            assert_eq!(max_column_id(&snapshot), None);
        }
        let scan = snapshot.clone().scan_builder().build()?;
        let batches = test_utils::read_scan(&scan, engine.clone())?;
        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 4);
        let mut rows = HashMap::new();
        for batch in batches {
            let ids = batch
                .column_by_name("id")
                .unwrap()
                .as_any()
                .downcast_ref::<Int32Array>()
                .unwrap();
            let names = batch
                .column_by_name("name")
                .unwrap()
                .as_any()
                .downcast_ref::<StringArray>()
                .unwrap();
            let countries = batch
                .column_by_name("country")
                .unwrap()
                .as_any()
                .downcast_ref::<StringArray>()
                .unwrap();
            for row in 0..batch.num_rows() {
                let country = (!countries.is_null(row)).then(|| countries.value(row).to_string());
                assert!(rows
                    .insert(ids.value(row), (names.value(row).to_string(), country))
                    .is_none());
            }
        }
        assert_eq!(
            rows,
            HashMap::from([
                (1, ("old-a".to_string(), None)),
                (2, ("old-b".to_string(), None)),
                (3, ("new-a".to_string(), Some("US".to_string()))),
                (4, ("new-b".to_string(), Some("CA".to_string()))),
            ]),
        );
    }
    Ok(())
}

#[rstest]
#[tokio::test]
async fn alter_table_rejects_data_change_true_at_build(
    #[values("none", "name", "id")] cm_mode: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[("delta.columnMapping.mode", cm_mode)],
    )?;
    let old_batch = RecordBatch::try_new(
        Arc::new(snapshot.schema().as_ref().try_into_arrow()?),
        vec![
            Arc::new(Int32Array::from(vec![1, 2])),
            Arc::new(StringArray::from(vec!["old-a", "old-b"])),
        ],
    )?;
    let before =
        write_batch_to_table(&snapshot, engine.as_ref(), old_batch, HashMap::new()).await?;
    let result = before
        .clone()
        .update_table(UpdateTableOperation::AlterTable, "test-engine", true)
        .add_column(StructField::nullable("country", DataType::STRING))
        .build(engine.as_ref(), committer());

    assert_result_error_with_message(result, "ALTER TABLE requires data_change to be false");
    let fresh = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    assert_eq!(fresh.version(), before.version());
    assert_eq!(fresh.schema(), before.schema());
    Ok(())
}

#[rstest]
#[tokio::test]
async fn alter_table_commit_rejects_staged_files(
    #[values("none", "name", "id")] cm_mode: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let (temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[("delta.columnMapping.mode", cm_mode)],
    )?;
    let old_batch = RecordBatch::try_new(
        Arc::new(snapshot.schema().as_ref().try_into_arrow()?),
        vec![
            Arc::new(Int32Array::from(vec![1, 2])),
            Arc::new(StringArray::from(vec!["old-a", "old-b"])),
        ],
    )?;
    let before = write_batch_to_table(
        &snapshot,
        engine.as_ref(),
        old_batch.clone(),
        HashMap::new(),
    )
    .await?;
    let mut transaction = before
        .clone()
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(StructField::nullable("country", DataType::STRING))
        .build(engine.as_ref(), committer())?;
    let write_context = transaction.write_state()?.write_context_builder().build()?;
    let batch = RecordBatch::try_new(
        Arc::new(
            write_context
                .logical_data_schema()
                .as_ref()
                .try_into_arrow()?,
        ),
        vec![
            Arc::new(Int32Array::from(vec![3, 4])),
            Arc::new(StringArray::from(vec!["new-a", "new-b"])),
            Arc::new(StringArray::from(vec!["US", "CA"])),
        ],
    )?;
    transaction.add_files(
        engine
            .write_parquet(&ArrowEngineData::new(batch), &write_context)
            .await?,
    );

    let error = transaction.commit(engine.as_ref()).unwrap_err();
    assert!(
        matches!(&error, KernelError::InvalidTransactionState(message)
        if message == "ALTER TABLE cannot include data-file actions")
    );
    let fresh = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    assert_eq!(fresh.version(), before.version());
    assert_eq!(fresh.schema(), before.schema());
    assert!(!temp_dir
        .path()
        .join("_delta_log")
        .join(format!("{:020}.json", before.version() + 1))
        .exists());
    test_read(&ArrowEngineData::new(old_batch), fresh.table_root(), engine)?;
    Ok(())
}

#[rstest]
#[tokio::test]
async fn schema_evolution_during_data_preserving_rewrite(
    #[values("none", "name", "id")] cm_mode: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[("delta.columnMapping.mode", cm_mode)],
    )?;
    let old_batch = RecordBatch::try_new(
        Arc::new(snapshot.schema().as_ref().try_into_arrow()?),
        vec![
            Arc::new(Int32Array::from(vec![1, 2])),
            Arc::new(StringArray::from(vec!["old-a", "old-b"])),
        ],
    )?;
    let before = write_batch_to_table(
        &snapshot,
        engine.as_ref(),
        old_batch.clone(),
        HashMap::new(),
    )
    .await?;
    let mut transaction = before
        .clone()
        .update_table(UpdateTableOperation::Optimize, "test-engine", false)
        .add_column(StructField::nullable("country", DataType::STRING))
        .build(engine.as_ref(), committer())?;
    let write_context = transaction.write_state()?.write_context_builder().build()?;
    let mut columns = old_batch.columns().to_vec();
    columns.push(new_null_array(&ArrowDataType::Utf8, old_batch.num_rows()));
    let batch = RecordBatch::try_new(
        Arc::new(
            write_context
                .logical_data_schema()
                .as_ref()
                .try_into_arrow()?,
        ),
        columns,
    )?;
    transaction.add_files(
        engine
            .write_parquet(&ArrowEngineData::new(batch.clone()), &write_context)
            .await?,
    );
    let scan = before.clone().scan_builder().build()?;
    for metadata in scan.scan_metadata(engine.as_ref())? {
        transaction.remove_files(metadata?.scan_files);
    }

    let post = transaction
        .commit(engine.as_ref())?
        .unwrap_post_commit_snapshot();
    let fresh = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    for snapshot in [post, fresh] {
        assert_eq!(snapshot.version(), before.version() + 1);
        assert!(snapshot.schema().contains("country"));
        for field in before.schema().fields() {
            assert_eq!(snapshot.schema().field(field.name()), Some(field));
        }
        for action in ["add", "remove"] {
            let actions =
                read_actions_from_commit(snapshot.table_root(), snapshot.version(), action)?;
            assert_eq!(actions.len(), 1);
            assert_eq!(actions[0]["dataChange"], false);
        }
    }
    test_read(&ArrowEngineData::new(batch), before.table_root(), engine)?;
    Ok(())
}

/// Reads `delta.columnMapping.maxColumnId` from the snapshot's metadata. Returns
/// `None` when the property is absent (e.g. non-CM tables) so callers can compare it
/// directly against `cm_mode.map(...)`-style expectations. CM-only callers should
/// `.expect(...)` at the use site to surface the protocol violation explicitly.
fn max_column_id(snap: &Snapshot) -> Option<i64> {
    snap.table_configuration()
        .metadata()
        .configuration()
        .get("delta.columnMapping.maxColumnId")
        .and_then(|v| v.parse().ok())
}

// ============================================================================
// Add column tests
// ============================================================================

#[rstest]
#[case::cdf_enabled(Some("true"), Some("reserved for Change Data Feed"))]
#[case::cdf_supported_only(None, None)]
#[tokio::test]
async fn add_column_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 mut properties = vec![
        ("delta.feature.changeDataFeed", "supported"),
        ("delta.columnMapping.mode", cm_mode),
    ];
    if let Some(value) = cdf_enabled {
        properties.push(("delta.enableChangeDataFeed", value));
    }
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &properties)?;
    let result = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(StructField::nullable(column_name, DataType::STRING))
        .build(engine.as_ref(), committer());

    if let Some(expected_error) = expected_error {
        assert_result_error_with_message(result, expected_error);
        let snapshot = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
        assert_eq!(snapshot.version(), 0);
        assert!(!snapshot.schema().contains(column_name));
    } else {
        let snapshot = result?
            .commit(engine.as_ref())?
            .unwrap_post_commit_snapshot();
        assert_eq!(snapshot.version(), 1);
        assert!(snapshot.schema().contains(column_name));
    }
    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 add_column_validates_cdf_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 mut properties = vec![
        ("delta.feature.changeDataFeed", "supported"),
        ("delta.columnMapping.mode", cm_mode),
    ];
    if cdf_enabled {
        properties.push(("delta.enableChangeDataFeed", "true"));
    }
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &properties)?;
    let result = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(
            StructField::nullable("value", DataType::STRING).with_metadata([(
                ColumnMetadataKey::ColumnMappingPhysicalName.as_ref(),
                "_change_type",
            )]),
        )
        .build(engine.as_ref(), committer());

    if let Some(expected_error) = expected_error {
        assert_result_error_with_message(result, expected_error);
        let snapshot = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
        assert_eq!(snapshot.version(), 0);
        assert!(!snapshot.schema().contains("value"));
    } else {
        let snapshot = result?
            .commit(engine.as_ref())?
            .unwrap_post_commit_snapshot();
        assert_eq!(snapshot.version(), 1);
        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 alter_table_commit_info_includes_operation_maps() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &[])?;
    let table_url = snapshot.table_root().clone();

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(StructField::nullable("added", DataType::STRING))
        .with_operation_parameters([("columns", Some(r#"["added"]"#))])
        .with_operation_metrics([("numAddedColumns", Some("1"))])
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let commit_infos = read_actions_from_commit(&table_url, 1, "commitInfo")?;
    assert_eq!(commit_infos.len(), 1);
    assert_eq!(
        commit_infos[0]["operationParameters"],
        json!({"columns": "[\"added\"]"})
    );
    assert_eq!(
        commit_infos[0]["operationMetrics"],
        json!({"numAddedColumns": "1"})
    );
    Ok(())
}

/// End-to-end lifecycle: write, ALTER to add columns, scan, write populated rows, scan again.
/// Each column is added in its own alter commit with a checkpoint after, exercising
/// "do some ops -> checkpoint -> do more ops -> checkpoint". Under CM, also verifies fresh
/// ids/physical names and that `maxColumnId` advanced.
#[rstest]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn add_columns_lifecycle(
    #[values(None, Some("name"), Some("id"))] cm_mode: Option<&str>,
    #[values(1, 3)] num_columns: usize,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup_mt()?;
    let properties: Vec<(&str, &str)> = cm_mode
        .map(|m| vec![("delta.columnMapping.mode", m)])
        .unwrap_or_default();
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &properties)?;
    let original_max_id =
        cm_mode.map(|_| max_column_id(&snapshot).expect("CM table must have maxColumnId"));

    // Write two rows with only the original columns populated.
    let batch = RecordBatch::try_new(
        Arc::new(simple_schema().as_ref().try_into_arrow().unwrap()),
        vec![
            Arc::new(Int32Array::from(vec![1, 2])),
            Arc::new(StringArray::from(vec!["a", "b"])),
        ],
    )
    .unwrap();
    let snapshot = write_batch_to_table(&snapshot, engine.as_ref(), batch, HashMap::new()).await?;

    let new_col_names: Vec<String> = (0..num_columns).map(|i| format!("col_{i}")).collect();

    // One alter+checkpoint cycle per column.
    let mut current = snapshot;
    for name in &new_col_names {
        let committed = current
            .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
            .add_column(StructField::nullable(name, DataType::STRING))
            .build(engine.as_ref(), committer())?
            .commit(engine.as_ref())?
            .unwrap_committed();
        let post = committed
            .post_commit_snapshot()
            .expect("post-commit snapshot");
        let (_, ckpt) = post.clone().checkpoint(engine.as_ref(), None)?;
        current = ckpt;
    }

    // Reload from storage to verify persistence. v0 = create, v1 = write, then `num_columns`
    // alter commits.
    let alter_end_version = 1 + num_columns as u64;
    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    assert_eq!(reloaded.version(), alter_end_version);
    let schema = reloaded.schema();
    assert_eq!(schema.fields().count(), 2 + num_columns);
    for name in &new_col_names {
        let field = schema.field(name).expect("added field should exist");
        assert_eq!(field.data_type(), &DataType::STRING);
        assert!(field.is_nullable());
    }

    // When CM is enabled: each new column must have a fresh id/physical name, and the
    // table's maxColumnId must have advanced past the original value. When CM is disabled:
    // the property must remain absent.
    if let Some(orig) = original_max_id {
        for name in &new_col_names {
            let field = schema.field(name).unwrap();
            let cm_id = field.column_mapping_id().expect("CM id should be assigned");
            assert!(
                cm_id > orig,
                "new column '{name}' id {cm_id} must exceed original max {orig}"
            );
            match field
                .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName)
                .expect("physical name should be assigned")
            {
                MetadataValue::String(s) => assert!(s.starts_with("col-")),
                other => panic!("expected String, got {other:?}"),
            }
        }
        assert!(max_column_id(&reloaded).expect("CM table must have maxColumnId") > orig);
    } else {
        assert!(reloaded
            .table_configuration()
            .metadata()
            .configuration()
            .get("delta.columnMapping.maxColumnId")
            .is_none());
    }

    // Scan back -- old rows should have NULL for every new column.
    let evolved_arrow_schema: delta_kernel::arrow::datatypes::SchemaRef =
        Arc::new(reloaded.schema().as_ref().try_into_arrow().unwrap());
    let scan = reloaded.scan_builder().build()?;
    let batches = test_utils::read_scan(&scan, engine.clone())?;
    assert!(!batches.is_empty());
    let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
    assert_eq!(total_rows, 2);
    assert_eq!(batches[0].num_columns(), 2 + num_columns);
    for name in &new_col_names {
        let col = batches[0].column_by_name(name).expect("new column");
        assert_eq!(col.null_count(), col.len());
    }

    // Write two more rows with all columns populated.
    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let mut new_arrays: Vec<Arc<dyn Array>> = vec![
        Arc::new(Int32Array::from(vec![3, 4])),
        Arc::new(StringArray::from(vec!["c", "d"])),
    ];
    for _ in 0..num_columns {
        new_arrays.push(Arc::new(StringArray::from(vec!["new_c", "new_d"])));
    }
    let batch2 = RecordBatch::try_new(evolved_arrow_schema, new_arrays).unwrap();
    let reloaded = Arc::new(reloaded);
    let _ = write_batch_to_table(&reloaded, engine.as_ref(), batch2, HashMap::new()).await?;

    // Scan back -- 4 rows total, each new column has 2 NULLs (old rows) and 2 values (new rows).
    let final_snap = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    assert_eq!(final_snap.version(), alter_end_version + 1);
    let scan = final_snap.scan_builder().build()?;
    let batches = test_utils::read_scan(&scan, engine.clone())?;
    let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
    assert_eq!(total_rows, 4);
    for name in &new_col_names {
        let null_count: usize = batches
            .iter()
            .map(|b| b.column_by_name(name).expect("new column").null_count())
            .sum();
        assert_eq!(null_count, 2, "column {name} should have 2 NULLs");
    }

    Ok(())
}

/// Adding columns of complex types (struct, array, map) -- with and without column mapping.
/// Verifies the data type round-trips and, under CM, that every reachable struct field
/// receives a distinct fresh ID and `maxColumnId` advances accordingly. `expected_id_count`
/// is the number of CM IDs the column should receive (1 for the parent + however many inner
/// struct fields the recursion reaches).
#[rstest]
#[case::struct_column(
    StructField::nullable(
        "address",
        schema! {
            nullable "city": STRING,
            nullable "zip": STRING,
        },
    ),
    3,
)]
#[case::array_of_primitive(
    StructField::nullable("tags", ArrayType::new(DataType::STRING, true)),
    1
)]
#[case::map_of_primitives(
    StructField::nullable("labels", MapType::new(DataType::STRING, DataType::INTEGER, true)),
    1
)]
#[case::array_of_struct(
    StructField::nullable(
        "items",
        ArrayType::new(
            schema! {
                nullable "a": STRING,
                nullable "b": INTEGER,
            },
            true,
        ),
    ),
    3,
)]
#[case::map_value_is_struct(
    StructField::nullable(
        "by_id",
        MapType::new(
            DataType::STRING,
            schema! {
                nullable "a": STRING,
                nullable "b": INTEGER,
            },
            true,
        ),
    ),
    3,
)]
#[case::map_key_is_struct(
    StructField::nullable(
        "lookup",
        MapType::new(
            schema! {
                nullable "a": STRING,
                nullable "b": INTEGER,
            },
            DataType::INTEGER,
            true,
        ),
    ),
    3,
)]
#[tokio::test]
async fn add_complex_type_column(
    #[case] field: StructField,
    #[case] expected_id_count: usize,
    #[values(None, Some("name"), Some("id"))] cm_mode: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let properties: Vec<(&str, &str)> = cm_mode
        .map(|m| vec![("delta.columnMapping.mode", m)])
        .unwrap_or_default();
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &properties)?;
    let original_max_id =
        cm_mode.map(|_| max_column_id(&snapshot).expect("CM table must have maxColumnId"));

    let field_name = field.name().to_string();
    let expected_type = field.data_type().clone();

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(table_path).build(engine.as_ref())?;
    let schema = reloaded.schema();
    let added = schema.field(&field_name).expect("added field should exist");

    if let Some(orig_max) = original_max_id {
        // Under CM, inner struct fields carry CM metadata that `expected_type` doesn't;
        // strict DataType equality won't hold. The ID-count check below implicitly verifies
        // that the type structure round-tripped correctly.
        let ids = added.collect_column_mapping_ids();
        let unique: HashSet<_> = ids.iter().copied().collect();
        assert_eq!(ids.len(), expected_id_count, "expected ID count mismatch");
        assert_eq!(unique.len(), ids.len(), "all assigned IDs must be distinct");
        assert!(
            ids.iter().all(|&id| id > orig_max),
            "all assigned IDs must exceed original max"
        );
        assert_eq!(
            max_column_id(&reloaded).expect("CM table must have maxColumnId"),
            ids.iter().copied().max().unwrap(),
            "table maxColumnId must equal the largest assigned ID",
        );
    } else {
        assert_eq!(added.data_type(), &expected_type);
    }
    Ok(())
}

#[rstest]
#[case::duplicate_column(&[], StructField::nullable("name", DataType::STRING), "already exists")]
#[case::duplicate_column_case_insensitive(
    &[],
    StructField::nullable("NAME", DataType::STRING),
    "already exists"
)]
#[case::timestamp_ntz_without_feature(
    &[],
    StructField::nullable("ts", DataType::TIMESTAMP_NTZ),
    "timestampNtz"
)]
#[case::non_nullable(&[], StructField::not_null("age", DataType::INTEGER), "non-nullable")]
#[tokio::test]
async fn add_column_failures(
    #[case] properties: &[(&str, &str)],
    #[case] field: StructField,
    #[case] error_contains: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), properties)?;

    let err = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer());
    assert!(err.is_err());
    assert!(err.unwrap_err().to_string().contains(error_contains));

    Ok(())
}

/// Back-to-back alters with a checkpoint in between, then a write against the evolved schema.
/// Exercises: create (v0) → alter add A (v1) → checkpoint at v1 → alter add B (v2) → write
/// row with values in both new columns (v3) → reload. The reload must rebuild the snapshot
/// from the checkpoint + alter commits + data commit and return the written values.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn back_to_back_alters_with_checkpoint() -> KernelResult<()> {
    // Checkpoint writing requires the multi-threaded engine (like `maintenance_ops.rs`).
    let (_temp_dir, table_path, engine) = test_table_setup_mt()?;

    // v0: create.
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &[])?;

    // v1: add column "a".
    let v1 = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(StructField::nullable("a", DataType::STRING))
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();
    let v1_snap = v1
        .post_commit_snapshot()
        .expect("post-commit snapshot at v1");

    // Checkpoint at v1.
    let (_, v1_ckpt) = v1_snap.clone().checkpoint(engine.as_ref(), None)?;

    // v2: add column "b" on top of the checkpointed snapshot.
    let v2 = v1_ckpt
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(StructField::nullable("b", DataType::INTEGER))
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();
    let v2_snap = v2
        .post_commit_snapshot()
        .expect("post-commit snapshot at v2");

    // v3: write one row populating both new columns.
    let evolved_arrow_schema: delta_kernel::arrow::datatypes::SchemaRef =
        Arc::new(v2_snap.schema().as_ref().try_into_arrow().unwrap());
    let batch = RecordBatch::try_new(
        evolved_arrow_schema,
        vec![
            Arc::new(Int32Array::from(vec![1])),
            Arc::new(StringArray::from(vec!["alice"])),
            Arc::new(StringArray::from(vec!["val_a"])),
            Arc::new(delta_kernel::arrow::array::Int32Array::from(vec![100])),
        ],
    )
    .unwrap();
    write_batch_to_table(v2_snap, engine.as_ref(), batch, HashMap::new()).await?;

    // Reload from scratch: kernel must rebuild from checkpoint + alter commits + data commit.
    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    assert_eq!(reloaded.version(), 3);
    let schema = reloaded.schema();
    assert!(
        schema.field("a").is_some(),
        "column added at v1 must survive checkpoint"
    );
    assert!(
        schema.field("b").is_some(),
        "column added at v2 must be present"
    );

    // Scan the row back and verify values in both newly-added columns.
    let scan = reloaded.scan_builder().build()?;
    let batches = test_utils::read_scan(&scan, engine.clone())?;
    let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
    assert_eq!(total_rows, 1);
    let a_col = batches[0]
        .column_by_name("a")
        .expect("column a")
        .as_any()
        .downcast_ref::<StringArray>()
        .expect("a is string");
    assert_eq!(a_col.value(0), "val_a");
    let b_col = batches[0]
        .column_by_name("b")
        .expect("column b")
        .as_any()
        .downcast_ref::<Int32Array>()
        .expect("b is int");
    assert_eq!(b_col.value(0), 100);

    Ok(())
}

// ============================================================================
// Add column at tests
// ============================================================================

#[rstest]
#[tokio::test]
async fn add_column_at_round_trip(
    #[values(None, Some("name"), Some("id"))] cm_mode: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let schema = schema_ref! {
        nullable "parent": {
            nullable "existing": INTEGER,
        },
    };
    let properties: Vec<(&str, &str)> = cm_mode
        .map(|mode| vec![("delta.columnMapping.mode", mode)])
        .unwrap_or_default();
    let snapshot =
        create_table_and_load_snapshot(&table_path, schema, engine.as_ref(), &properties)?;

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column_at(
            column_name!("parent"),
            StructField::nullable("added", DataType::STRING),
        )
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let reloaded_schema = reloaded.schema();
    let parent = reloaded_schema.field("parent").expect("parent must exist");
    let DataType::Struct(parent) = parent.data_type() else {
        panic!("parent must remain a struct")
    };
    assert!(parent.field("existing").is_some());

    let added = parent.field("added").expect("added field must exist");
    assert_eq!(added.data_type(), &DataType::STRING);
    assert!(added.is_nullable());
    assert_eq!(added.column_mapping_id().is_some(), cm_mode.is_some());
    assert_eq!(
        added
            .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName)
            .is_some(),
        cm_mode.is_some()
    );

    Ok(())
}

/// Nested `add_column_at` on a column-mapped table: commit + reload must persist the nested
/// field's CM id / physical name and advance `maxColumnId`.
#[rstest]
#[tokio::test]
async fn add_column_at_nested_struct_with_column_mapping(
    #[values("name", "id")] cm_mode: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let schema = Arc::new(StructType::try_new(vec![
        StructField::nullable("id", DataType::INTEGER),
        StructField::nullable(
            "address",
            StructType::try_new(vec![StructField::nullable("city", DataType::STRING)])?,
        ),
    ])?);
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        schema,
        engine.as_ref(),
        &[("delta.columnMapping.mode", cm_mode)],
    )?;
    let original_max = max_column_id(&snapshot).expect("CM table must have maxColumnId");

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column_at(
            column_name!("address"),
            StructField::nullable("zip", DataType::STRING),
        )
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let reloaded_schema = reloaded.schema();
    let zip = reloaded_schema.field_at_path(&["address".to_string(), "zip".to_string()]);
    let cm_id = zip
        .column_mapping_id()
        .expect("nested field must receive a column mapping id");
    assert!(
        cm_id > original_max,
        "nested field id {cm_id} must exceed original max {original_max}"
    );
    match zip
        .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName)
        .expect("physical name should be assigned")
    {
        MetadataValue::String(s) => assert!(s.starts_with("col-")),
        other => panic!("expected String, got {other:?}"),
    }
    assert_eq!(
        max_column_id(&reloaded).expect("CM table must have maxColumnId"),
        cm_id,
        "table maxColumnId must equal the nested field's assigned id",
    );
    Ok(())
}

#[rstest]
#[tokio::test]
async fn add_column_at_containers_round_trip(
    #[values(None, Some("name"), Some("id"))] cm_mode: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let schema = schema_ref! {
        nullable "items": [ nullable {
            nullable "existing": STRING,
        } ],
        nullable "by_key": {
            { nullable "existing": STRING } => nullable INTEGER
        },
        nullable "by_value": {
            INTEGER => nullable {
                nullable "existing": STRING,
            }
        },
    };
    let properties: Vec<(&str, &str)> = cm_mode
        .map(|mode| vec![("delta.columnMapping.mode", mode)])
        .unwrap_or_default();
    let snapshot =
        create_table_and_load_snapshot(&table_path, schema, engine.as_ref(), &properties)?;

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column_at(
            column_name!("items.element"),
            StructField::nullable("added", DataType::STRING),
        )
        .add_column_at(
            column_name!("by_key.key"),
            StructField::nullable("added", DataType::STRING),
        )
        .add_column_at(
            column_name!("by_value.value"),
            StructField::nullable("added", DataType::STRING),
        )
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let mut reloaded_type = DataType::from(reloaded.schema().as_ref().clone());
    for path in [
        column_name!("items.element"),
        column_name!("by_key.key"),
        column_name!("by_value.value"),
    ] {
        let target = reloaded_type
            .struct_at_path(path.path())
            .unwrap_or_else(|err| panic!("path '{path}' does not resolve to a struct: {err}"));
        assert!(target.field("existing").is_some());

        let added = target.field("added").expect("added field must exist");
        assert_eq!(added.data_type(), &DataType::STRING);
        assert!(added.is_nullable());
        assert_eq!(added.column_mapping_id().is_some(), cm_mode.is_some());
        assert_eq!(
            added
                .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName)
                .is_some(),
            cm_mode.is_some()
        );
    }

    Ok(())
}

#[tokio::test]
async fn add_column_at_struct_fields_named_like_container_segments() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let schema = schema_ref! {
        nullable "address": {
            nullable "element": {
                nullable "existing": STRING,
            },
            nullable "key": {
                nullable "existing": STRING,
            },
            nullable "value": {
                nullable "existing": STRING,
            },
        },
    };
    let snapshot = create_table_and_load_snapshot(&table_path, schema, engine.as_ref(), &[])?;

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column_at(
            column_name!("address.element"),
            StructField::nullable("added", DataType::STRING),
        )
        .add_column_at(
            column_name!("address.key"),
            StructField::nullable("added", DataType::STRING),
        )
        .add_column_at(
            column_name!("address.value"),
            StructField::nullable("added", DataType::STRING),
        )
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let mut reloaded_type = DataType::from(reloaded.schema().as_ref().clone());
    for path in [
        column_name!("address.element"),
        column_name!("address.key"),
        column_name!("address.value"),
    ] {
        let parent = reloaded_type
            .struct_at_path(path.path())
            .unwrap_or_else(|err| panic!("path '{path}' does not resolve to a struct: {err}"));
        assert!(parent.field("existing").is_some());
        assert_eq!(
            parent.field("added"),
            Some(&StructField::nullable("added", DataType::STRING))
        );
    }

    Ok(())
}

#[tokio::test]
async fn add_column_at_rejects_duplicate_field_in_same_builder() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let schema = schema_ref! {
        nullable "address": {
            nullable "existing": STRING,
        },
    };
    let snapshot = create_table_and_load_snapshot(&table_path, schema, engine.as_ref(), &[])?;

    let result = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column_at(
            column_name!("address"),
            StructField::nullable("dup", DataType::STRING),
        )
        .add_column_at(
            column_name!("address"),
            StructField::nullable("dup", DataType::STRING),
        )
        .build(engine.as_ref(), committer());
    assert_result_error_with_message(result, "already exists");

    Ok(())
}

#[tokio::test]
async fn add_column_at_rejects_non_struct_parent() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &[])?;

    let result = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column_at(
            column_name!("id"),
            StructField::nullable("added", DataType::STRING),
        )
        .build(engine.as_ref(), committer());
    assert_result_error_with_message(result, "path target is not a struct");

    Ok(())
}

/// Empty-schema tables are valid intermediate state, so they need to behave normally
/// once a column is added. This test runs the full lifecycle (create empty, ALTER ADD
/// COLUMN, write rows, scan them back, time-travel to v0) across all column-mapping
/// modes so the CM bookkeeping (maxColumnId, column-mapping id, physical name) is
/// exercised on a schema that started empty rather than being created with columns
/// up front.
#[rstest]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn empty_create_then_add_column(
    #[values(None, Some("name"), Some("id"))] cm_mode: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup_mt()?;
    let properties: Vec<(&str, &str)> = cm_mode
        .map(|m| vec![("delta.columnMapping.mode", m)])
        .unwrap_or_default();

    let empty_schema = schema_ref! {};
    let v0 =
        create_table_and_load_snapshot(&table_path, empty_schema, engine.as_ref(), &properties)?;
    assert_eq!(v0.version(), 0);
    assert_eq!(v0.schema().num_fields(), 0);
    assert_eq!(max_column_id(&v0), cm_mode.map(|_| 0));

    // Scans and blind appends against the empty-schema snapshot are blocked with
    // friendly errors that point users at ALTER TABLE ADD COLUMN.
    let scan_err = v0
        .clone()
        .scan_builder()
        .build()
        .expect_err("scan_builder().build() must reject empty-schema snapshots");
    assert!(
        scan_err.to_string().contains("empty schema")
            && scan_err.to_string().contains("ALTER TABLE ADD COLUMN"),
        "scan error must point at ALTER TABLE ADD COLUMN, got: {scan_err}"
    );
    let write_state_err = v0
        .clone()
        .update_table(UpdateTableOperation::Write, "EmptySchemaApp/0.1.0", true)
        .build(engine.as_ref(), committer())?
        .write_state()
        .expect_err("write_state() must reject empty-schema snapshots");
    assert!(
        write_state_err.to_string().contains("empty schema")
            && write_state_err.to_string().contains("update_table"),
        "write_state error must point at update_table, got: {write_state_err}"
    );

    v0.update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(StructField::nullable("id", DataType::INTEGER))
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let v1 = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    assert_eq!(v1.version(), 1);
    let schema = v1.schema();
    assert_eq!(schema.num_fields(), 1);
    let id_field = schema.field("id").expect("added field should exist");
    assert_eq!(id_field.data_type(), &DataType::INTEGER);
    assert!(id_field.is_nullable());
    assert_eq!(max_column_id(&v1), cm_mode.map(|_| 1));
    assert_eq!(id_field.column_mapping_id(), cm_mode.map(|_| 1i64));
    assert_eq!(
        id_field
            .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName)
            .is_some(),
        cm_mode.is_some(),
        "physical name presence must track column-mapping enablement",
    );

    // Write rows through the freshly-added column and confirm the read path round-trips
    // them. Under CM this also exercises the physical-name plumbing for a column whose
    // schema started empty.
    let arrow_schema = v1.schema().as_ref().try_into_arrow().unwrap();
    let batch = RecordBatch::try_new(
        Arc::new(arrow_schema),
        vec![Arc::new(Int32Array::from(vec![Some(1), None, Some(99)]))],
    )
    .unwrap();
    let v2 = write_batch_to_table(&v1, engine.as_ref(), batch, HashMap::new())
        .await
        .map_err(|e| {
            delta_kernel::KernelError::generic(format!("write_batch_to_table failed: {e}"))
        })?;
    assert_eq!(v2.version(), 2);

    let scan = v2.scan_builder().build()?;
    let batches = test_utils::read_scan(&scan, engine.clone())?;
    let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
    assert_eq!(
        total_rows, 3,
        "post-ALTER write should yield 3 rows on read"
    );

    // The empty-schema snapshot must still be loadable via time-travel after later
    // commits introduce a column and data files; pins the contract that the v0 history
    // survives schema evolution.
    let v0_after = Snapshot::builder_for(&table_path)
        .at_version(0)
        .build(engine.as_ref())?;
    assert_eq!(v0_after.version(), 0);
    assert_eq!(v0_after.schema().num_fields(), 0);

    Ok(())
}

// ============================================================================
// SET NULLABLE tests
// ============================================================================

/// Cross-product: 3 schema/column cases x 3 CM modes (off, name, id).
#[rstest]
#[case::already_nullable(simple_schema(), column_name!("name"))]
#[case::required_top_level(
    schema_ref! {
        not_null "id": INTEGER,
        nullable "name": STRING,
    },
    column_name!("id")
)]
#[case::required_nested(
    schema_ref! {
        nullable "id": INTEGER,
        nullable "address": {
            not_null "city": STRING,
            nullable "zip": STRING,
        },
    },
    column_name!("address.city")
)]
#[tokio::test]
async fn set_nullable_succeeds(
    #[case] schema: SchemaRef,
    #[case] column: ColumnName,
    #[values(None, Some("name"), Some("id"))] cm_mode: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let properties: Vec<(&str, &str)> = cm_mode
        .map(|m| vec![("delta.columnMapping.mode", m)])
        .unwrap_or_default();
    let snapshot =
        create_table_and_load_snapshot(&table_path, schema, engine.as_ref(), &properties)?;
    // Snapshot the field before the alter so we can prove set_nullable changes only the
    // nullable bit -- preserving name, data type, and ALL metadata (including column-mapping
    // id and physical name when CM is enabled).
    let before = snapshot.schema().field_at_path(column.path()).clone();

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .set_nullable(column.clone())
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(table_path).build(engine.as_ref())?;
    let reloaded_schema = reloaded.schema();
    let after = reloaded_schema.field_at_path(column.path());
    assert!(after.is_nullable());
    assert_eq!(after.name(), before.name());
    assert_eq!(after.data_type(), before.data_type());
    assert_eq!(
        after.metadata(),
        before.metadata(),
        "field metadata (incl. column mapping id/physical name) must be preserved"
    );
    Ok(())
}

/// End-to-end: create a table with a non-null layout column (partition or clustering),
/// write a row, flip the layout column to nullable, checkpoint, reload from scratch, scan.
/// Cross-product: layout kind (partitioned, clustered) x column-mapping mode (off, name, id).
#[rstest]
#[case::partition("date", DataLayout::partitioned(["date"]), "2026-01-01")]
#[case::clustered("region", DataLayout::clustered(["region"]), "us")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn set_nullable_on_layout_column_with_checkpoint(
    #[case] col_name: &str,
    #[case] layout: DataLayout,
    #[case] col_value: &str,
    #[values(None, Some("name"), Some("id"))] cm_mode: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup_mt()?;
    // Partition values live in the directory path; clustering values live in the row batch.
    let is_partitioned = matches!(layout, DataLayout::Partitioned { .. });

    // v0: create the table with the layout column as non-null.
    let schema = Arc::new(try_schema! {
        nullable "id": INTEGER,
        not_null (col_name): STRING,
    }?);
    let properties: Vec<(&str, &str)> = cm_mode
        .map(|m| vec![("delta.columnMapping.mode", m)])
        .unwrap_or_default();
    create_table(&table_path, schema.clone(), "Test/1.0", true)
        .with_data_layout(layout)
        .with_table_properties(properties)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();
    let v0 = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    assert!(!v0.schema().field(col_name).unwrap().is_nullable());

    // v1: write a single row.
    let v0_arc = Arc::new(v0);
    let v1 = if is_partitioned {
        // Partition cols are excluded from the row batch and passed via partition_values.
        let nonpartition_arrow_schema: delta_kernel::arrow::datatypes::SchemaRef =
            Arc::new(delta_kernel::arrow::datatypes::Schema::new(vec![
                delta_kernel::arrow::datatypes::Field::new(
                    "id",
                    delta_kernel::arrow::datatypes::DataType::Int32,
                    true,
                ),
            ]));
        let batch = RecordBatch::try_new(
            nonpartition_arrow_schema,
            vec![Arc::new(Int32Array::from(vec![1]))],
        )?;
        let mut partition_values = HashMap::new();
        partition_values.insert(col_name.to_string(), Scalar::String(col_value.to_string()));
        write_batch_to_table(&v0_arc, engine.as_ref(), batch, partition_values).await?
    } else {
        // Clustering cols are regular columns; partition_values is empty.
        let arrow_schema: delta_kernel::arrow::datatypes::SchemaRef =
            Arc::new(schema.as_ref().try_into_arrow().unwrap());
        let batch = RecordBatch::try_new(
            arrow_schema,
            vec![
                Arc::new(Int32Array::from(vec![1])),
                Arc::new(StringArray::from(vec![col_value])),
            ],
        )?;
        write_batch_to_table(&v0_arc, engine.as_ref(), batch, HashMap::new()).await?
    };

    // v2: ALTER TABLE -- set the layout column nullable.
    let v2 = v1
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .set_nullable(ColumnName::new([col_name]))
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();
    let v2_snap = v2
        .post_commit_snapshot()
        .expect("post-commit snapshot at v2");
    assert!(v2_snap.schema().field(col_name).unwrap().is_nullable());

    // Checkpoint at v2 so reload exercises the checkpoint path.
    v2_snap.clone().checkpoint(engine.as_ref(), None)?;

    // Reload from scratch and verify the schema and row survive.
    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    assert_eq!(reloaded.version(), 2);
    assert!(reloaded.schema().field(col_name).unwrap().is_nullable());

    let scan = reloaded.scan_builder().build()?;
    let batches = test_utils::read_scan(&scan, engine.clone())?;
    let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
    assert_eq!(total_rows, 1);
    let col = batches[0]
        .column_by_name(col_name)
        .expect("layout column")
        .as_any()
        .downcast_ref::<StringArray>()
        .expect("layout column is string");
    assert_eq!(col.value(0), col_value);
    Ok(())
}

#[tokio::test]
async fn set_nullable_nonexistent_column_fails() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &[])?;

    let err = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .set_nullable(column_name!("nonexistent"))
        .build(engine.as_ref(), committer());
    assert!(err.is_err());
    assert!(err.unwrap_err().to_string().contains("does not exist"));

    Ok(())
}

// ============================================================================
// CHAIN tests
// ============================================================================

/// Alternating chain: ADD COLUMN, SET NULLABLE, ADD COLUMN, SET NULLABLE. Verifies that
/// chaining mixed ops applies them in order and produces the expected final schema. Each
/// SET NULLABLE flips a still-NOT-NULL column from the original schema. Under CM, also
/// verifies existing fields' column mapping IDs are preserved by set_nullable while
/// add_column receives a new CM ID and bumps maxColumnId.
#[rstest]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn chain_add_column_and_set_nullable(
    #[values(None, Some("name"), Some("id"))] cm_mode: Option<&str>,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup_mt()?;
    let schema = schema_ref! {
        not_null "id": INTEGER,
        not_null "name": STRING,
    };
    let properties: Vec<(&str, &str)> = cm_mode
        .map(|m| vec![("delta.columnMapping.mode", m)])
        .unwrap_or_default();
    let snapshot =
        create_table_and_load_snapshot(&table_path, schema, engine.as_ref(), &properties)?;

    let original_id_cm_id = cm_mode.map(|_| {
        snapshot
            .schema()
            .field("id")
            .unwrap()
            .column_mapping_id()
            .expect("existing field should already have a column mapping ID")
    });
    let original_max_id =
        cm_mode.map(|_| max_column_id(&snapshot).expect("CM table must have maxColumnId"));

    // Two alter+checkpoint cycles: (add email + nullable id), (add age + nullable name).
    let v1 = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(StructField::nullable("email", DataType::STRING))
        .set_nullable(column_name!("id"))
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();
    let v1_snap = v1
        .post_commit_snapshot()
        .expect("post-commit snapshot at v1");
    let (_, v1_ckpt) = v1_snap.clone().checkpoint(engine.as_ref(), None)?;
    let v2 = v1_ckpt
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(StructField::nullable("age", DataType::INTEGER))
        .set_nullable(column_name!("name"))
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();
    let v2_snap = v2
        .post_commit_snapshot()
        .expect("post-commit snapshot at v2");
    v2_snap.clone().checkpoint(engine.as_ref(), None)?;

    let reloaded = Snapshot::builder_for(table_path).build(engine.as_ref())?;
    let schema = reloaded.schema();
    assert_eq!(schema.fields().count(), 4);
    for name in ["id", "name", "email", "age"] {
        let field = schema.field(name).expect("field should be present");
        assert!(field.is_nullable(), "field '{name}' should be nullable");
    }

    if let (Some(orig_id), Some(orig_max)) = (original_id_cm_id, original_max_id) {
        for added in ["email", "age"] {
            assert!(
                schema.field(added).unwrap().column_mapping_id().is_some(),
                "added field '{added}' should have a column mapping ID"
            );
        }
        let id_after = schema
            .field("id")
            .unwrap()
            .column_mapping_id()
            .expect("existing id column mapping");
        assert_eq!(id_after, orig_id, "existing id CM id must not change");
        assert!(
            max_column_id(&reloaded).expect("CM table must have maxColumnId") > orig_max,
            "chained add_column must bump maxColumnId"
        );
    }

    Ok(())
}

fn field_with_stray_key(name: &str, key: &ColumnMetadataKey, ty: DataType) -> StructField {
    let mut f = StructField::nullable(name, ty);
    f.metadata
        .insert(key.as_ref().to_string(), MetadataValue::Number(99));
    f
}

/// On a clean non-CM table, an ALTER that adds a column carrying stray CM metadata has that
/// metadata stripped (the commit introduces it into a previously-clean table), rather than
/// rejected -- matching delta-spark. Parametrized over each detected key and over placement
/// (top-level vs nested in a struct).
#[rstest]
#[case::top_level(false)]
#[case::nested_in_struct(true)]
#[tokio::test]
async fn add_column_with_stray_cm_metadata_on_non_cm_table_is_stripped(
    #[case] nested: bool,
    #[values(
        ColumnMetadataKey::ColumnMappingId,
        ColumnMetadataKey::ColumnMappingPhysicalName,
        ColumnMetadataKey::ColumnMappingNestedIds,
        ColumnMetadataKey::ParquetFieldId,
        ColumnMetadataKey::ParquetFieldNestedIds
    )]
    key: ColumnMetadataKey,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &[])?;

    let (field, stripped_path): (StructField, Vec<String>) = if nested {
        let outer = StructField::nullable(
            "outer",
            schema! {
                (field_with_stray_key("inner", &key, DataType::STRING)),
            },
        );
        (outer, vec!["outer".to_string(), "inner".to_string()])
    } else {
        (
            field_with_stray_key("tainted", &key, DataType::STRING),
            vec!["tainted".to_string()],
        )
    };

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    // Reload from disk so we assert on the persisted schemaString, not the in-memory config.
    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let reloaded_schema = reloaded.schema();
    let leaf = reloaded_schema.field_at_path(&stripped_path);
    assert!(
        leaf.get_config_value(&key).is_none(),
        "stray {} at {stripped_path:?} should be stripped",
        key.as_ref()
    );
    Ok(())
}

/// The ALTER strip is `None`-mode-only. Adding a column with pre-populated column-mapping metadata
/// preserves it when mapping is enabled (`id` / `name`, delta-spark's
/// `assignColumnIdAndPhysicalName` keeps existing ids) and strips it only in `None` mode.
#[rstest]
#[case::none(ColumnMappingMode::None, &[], false)]
#[case::id(ColumnMappingMode::Id, &[("delta.columnMapping.mode", "id")], true)]
#[case::name(ColumnMappingMode::Name, &[("delta.columnMapping.mode", "name")], true)]
#[tokio::test]
async fn add_column_strip_is_none_mode_only(
    #[case] expected_mode: ColumnMappingMode,
    #[case] properties: &[(&str, &str)],
    #[case] annotation_kept: bool,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), properties)?;
    assert_eq!(
        snapshot.table_configuration().column_mapping_mode(),
        expected_mode
    );

    // A well-formed id+physicalName pair so enabled modes have valid metadata to preserve.
    let field = fixtures::cm_field("added", 99, "phys-added", DataType::STRING);
    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let added = reloaded.schema().field("added").unwrap().clone();
    assert_eq!(
        added.column_mapping_id().is_some(),
        annotation_kept,
        "column mapping id under {expected_mode:?} mode"
    );
    Ok(())
}

#[rstest]
#[case::v2("delta.enableIcebergCompatV2", "icebergCompatV2")]
#[case::v3("delta.enableIcebergCompatV3", "icebergCompatV3")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn schema_changes_blocked_when_iceberg_compat_enabled(
    #[case] enablement_property: &str,
    #[case] feature_name: &str,
    #[values(
        UpdateTableOperation::Write,
        UpdateTableOperation::AlterTable,
        UpdateTableOperation::Custom("UNKNOWN".to_string())
    )]
    operation: UpdateTableOperation,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup_mt()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[(enablement_property, "true")],
    )?;

    let msg = snapshot
        .update_table(
            operation.clone(),
            "test-engine",
            operation != UpdateTableOperation::AlterTable,
        )
        .add_column(StructField::nullable("new_col", DataType::STRING))
        .build(engine.as_ref(), committer())
        .unwrap_err()
        .to_string();
    assert!(
        msg.contains(&format!(
            "Schema changes are not yet supported on tables with {feature_name} enabled"
        )),
        "unexpected error: {msg}",
    );

    Ok(())
}

#[tokio::test]
async fn add_column_with_orphan_default_metadata_succeeds() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot =
        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &[])?;
    let field = StructField::nullable("new_col", DataType::INTEGER).add_metadata([(
        ColumnMetadataKey::CurrentDefault.as_ref().to_string(),
        MetadataValue::String("42".to_string()),
    )]);

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let reloaded_schema = reloaded.schema();
    let default = reloaded_schema
        .field("new_col")
        .expect("new_col must exist after ALTER")
        .column_default()?
        .expect("CURRENT_DEFAULT metadata must survive ALTER");
    assert_eq!(default.raw_sql(), "42");

    let txn = reloaded
        .update_table(UpdateTableOperation::Write, "test-engine", true)
        .build(engine.as_ref(), committer())?;
    assert!(
        txn.top_level_column_defaults()?.is_empty(),
        "default metadata must remain inert without allowColumnDefaults",
    );

    Ok(())
}

#[rstest]
#[case(UpdateTableOperation::Write)]
#[case(UpdateTableOperation::AlterTable)]
#[case(UpdateTableOperation::Custom("UNKNOWN".to_string()))]
#[tokio::test]
async fn schema_changes_blocked_when_allow_column_defaults_enabled(
    #[case] operation: UpdateTableOperation,
) -> KernelResult<()> {
    let (store, engine, table_url) = engine_store_setup("alter_column_defaults", None);
    let table_url = create_test_table(
        store,
        table_url,
        simple_schema(),
        &[],
        true,
        vec![],
        vec!["allowColumnDefaults"],
    )
    .await?;
    let snapshot = Snapshot::builder_for(table_url).build(&engine)?;

    let msg = snapshot
        .update_table(
            operation.clone(),
            "test-engine",
            operation != UpdateTableOperation::AlterTable,
        )
        .add_column(StructField::nullable("new_col", DataType::STRING))
        .build(&engine, committer())
        .unwrap_err()
        .to_string();
    assert!(
        msg.contains(
            "Schema changes are not yet supported on tables with allowColumnDefaults enabled"
        ),
        "unexpected error: {msg}",
    );

    Ok(())
}

// ============================================================================
// ALTER TABLE ADD COLUMN preserves / fills pre-populated column mapping metadata
// (delta-spark parity per `DeltaColumnMapping.assignColumnIdAndPhysicalName`).
// See https://github.com/delta-io/delta-kernel-rs/issues/2377.
// ============================================================================

fn cm_id_for_field(field: &StructField) -> i64 {
    field
        .column_mapping_id()
        .expect("field must have a column mapping id")
}

fn physical_name_for_field(field: &StructField) -> &str {
    match field.get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName) {
        Some(MetadataValue::String(s)) => s.as_str(),
        other => panic!("expected physicalName string, got {other:?}"),
    }
}

/// ADD COLUMN with both `delta.columnMapping.id` and `delta.columnMapping.physicalName`
/// pre-populated: the connector-supplied metadata is preserved verbatim. `maxColumnId`
/// advances to the supplied id when it exceeds the existing max.
#[rstest]
#[tokio::test]
async fn add_column_preserves_complete_cm_metadata(
    #[values("name", "id")] cm_mode: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[("delta.columnMapping.mode", cm_mode)],
    )?;
    let original_max = max_column_id(&snapshot).expect("CM table must have maxColumnId");

    // Supplied id is well above the table's max so we can verify maxColumnId follows it.
    let supplied_id = original_max + 100;
    let field = fixtures::cm_field(
        "preserved",
        supplied_id,
        "user-supplied-physical",
        DataType::STRING,
    );

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let schema = reloaded.schema();
    let added = schema.field("preserved").unwrap();
    assert_eq!(cm_id_for_field(added), supplied_id);
    assert_eq!(physical_name_for_field(added), "user-supplied-physical");
    assert_eq!(
        max_column_id(&reloaded).expect("CM table must have maxColumnId"),
        supplied_id
    );
    Ok(())
}

/// ADD COLUMN with only `delta.columnMapping.physicalName` supplied: kernel allocates
/// `id = old maxColumnId + 1`, preserves the user-provided physical name, and bumps
/// `maxColumnId` to the new id.
#[rstest]
#[tokio::test]
async fn add_column_with_only_physical_name_allocates_id(
    #[values("name", "id")] cm_mode: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[("delta.columnMapping.mode", cm_mode)],
    )?;
    let original_max = max_column_id(&snapshot).expect("CM table must have maxColumnId");

    let field =
        fixtures::cm_field_physical_name_only("named_only", "phys-named-only", DataType::STRING);

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let schema = reloaded.schema();
    let added = schema.field("named_only").unwrap();
    assert_eq!(cm_id_for_field(added), original_max + 1);
    assert_eq!(physical_name_for_field(added), "phys-named-only");
    assert_eq!(
        max_column_id(&reloaded).expect("CM table must have maxColumnId"),
        original_max + 1
    );
    Ok(())
}

/// ADD COLUMN with only `delta.columnMapping.id` supplied: id is preserved, missing
/// `physicalName` is filled with `col-<uuid>`.
#[rstest]
#[tokio::test]
async fn add_column_with_only_id_fills_physical_name(
    #[values("name", "id")] cm_mode: &str,
) -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[("delta.columnMapping.mode", cm_mode)],
    )?;
    let original_max = max_column_id(&snapshot).expect("CM table must have maxColumnId");
    let supplied_id = original_max + 7;

    let field = fixtures::cm_field_id_only("id_only", supplied_id, DataType::STRING);

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let schema = reloaded.schema();
    let added = schema.field("id_only").unwrap();
    assert_eq!(cm_id_for_field(added), supplied_id);
    assert!(
        physical_name_for_field(added).starts_with("col-"),
        "physical name should be filled with col-<uuid>, got {}",
        physical_name_for_field(added)
    );
    assert_eq!(
        max_column_id(&reloaded).expect("CM table must have maxColumnId"),
        supplied_id
    );
    Ok(())
}

/// ADD COLUMN where the supplied `id` is *less than* the existing `maxColumnId` but does
/// not collide with any existing field's id: succeeds, with the supplied id preserved
/// verbatim and `maxColumnId` unchanged. Matches delta-spark; diverges from the Java Kernel
/// proposal in https://github.com/delta-io/delta/pull/4520, which would reject this.
#[tokio::test]
async fn add_column_with_id_below_max_column_id_succeeds() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;

    // Pre-populate the table with sparse ids (1, 100) using the create-table preserve path.
    let schema = schema_ref! {
        (fixtures::cm_field("a", 1, "phys-a", DataType::INTEGER)),
        (fixtures::cm_field("b", 100, "phys-b", DataType::STRING)),
    };
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        schema,
        engine.as_ref(),
        &[("delta.columnMapping.mode", "name")],
    )?;
    assert_eq!(
        max_column_id(&snapshot).expect("CM table must have maxColumnId"),
        100
    );

    // Now add a new column with id=50, which is well below maxColumnId=100 and not used.
    let field = fixtures::cm_field("inserted_below_max", 50, "phys-inserted", DataType::STRING);

    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())?
        .commit(engine.as_ref())?
        .unwrap_committed();

    let reloaded = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
    let schema = reloaded.schema();
    let added = schema.field("inserted_below_max").unwrap();
    assert_eq!(cm_id_for_field(added), 50);
    assert_eq!(physical_name_for_field(added), "phys-inserted");
    // maxColumnId stays at 100 because the supplied id (50) didn't exceed it.
    assert_eq!(
        max_column_id(&reloaded).expect("CM table must have maxColumnId"),
        100
    );
    Ok(())
}

/// ADD COLUMN where the supplied `id` collides with an existing field's id: fails. The
/// duplicate-id check happens when the alter builder constructs the new
/// `TableConfiguration` via `make_physical`.
#[tokio::test]
async fn add_column_with_id_colliding_existing_field_is_rejected() -> KernelResult<()> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        simple_schema(),
        engine.as_ref(),
        &[("delta.columnMapping.mode", "name")],
    )?;

    // Pick an id that already exists in the simple_schema (1, 2 typically).
    let existing_id = snapshot
        .schema()
        .field("id")
        .unwrap()
        .column_mapping_id()
        .expect("simple_schema 'id' must have a CM id under name mode");

    let field = fixtures::cm_field("colliding", existing_id, "phys-colliding", DataType::STRING);

    let err = snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(field)
        .build(engine.as_ref(), committer())
        .unwrap_err()
        .to_string();
    assert!(
        err.contains("Duplicate column mapping ID") && err.contains(&existing_id.to_string()),
        "expected duplicate-id error naming id {existing_id}, got: {err}"
    );
    Ok(())
}

/// A mapping-disabled table that already carries residual `delta.columnMapping.*` annotations is
/// left untouched by an ALTER: the pre-existing annotation on `value` survives, and whatever the
/// added column carries is persisted verbatim -- kernel strips only annotations a commit newly
/// introduces into a *clean* table, so an already-dirty table is never rewritten (matching
/// delta-spark). The clean-table strip is covered by
/// `add_column_with_stray_cm_metadata_on_non_cm_table_is_stripped`.
///
/// Cases: adding a clean column (stays clean) and adding a stray-annotated column (annotation
/// left in place, not stripped).
#[rstest]
#[case::clean_column(StructField::nullable("added", DataType::STRING), None)]
#[case::introduced_stray(
    field_with_stray_key("added", &ColumnMetadataKey::ColumnMappingId, DataType::STRING),
    Some(99)
)]
#[tokio::test]
async fn add_column_on_stale_table_leaves_schema_untouched(
    #[case] added_field: StructField,
    #[case] expected_added_cm_id: Option<i64>,
) -> KernelResult<()> {
    let (store, engine, table_url) = engine_store_setup("alter_stale_cm", None);

    // `value` carries a stale id; protocol omits columnMapping and no mode is set (resolves to
    // None) -- residual annotations already on the table.
    let stale_schema = schema! {
        nullable "id": INTEGER,
        (StructField::nullable("value", DataType::INTEGER)
            .add_metadata([("delta.columnMapping.id", MetadataValue::Number(2))])),
    };
    let escaped = serde_json::to_string(&serde_json::to_string(&stale_schema)?).unwrap();
    // v0 written directly to bypass create_table validation (which strips stale annotations).
    let v0 = format!(
        r#"{{"protocol":{{"minReaderVersion":1,"minWriterVersion":2}}}}
{{"metaData":{{"id":"alter-stale-cm","format":{{"provider":"parquet","options":{{}}}},"schemaString":{escaped},"partitionColumns":[],"configuration":{{}},"createdTime":1700000000000}}}}
"#
    );
    add_commit(table_url.as_str(), store.as_ref(), 0, v0)
        .await
        .unwrap();

    let snapshot = Snapshot::builder_for(table_url.clone()).build(&engine)?;
    snapshot
        .update_table(UpdateTableOperation::AlterTable, "test-engine", false)
        .add_column(added_field)
        .build(&engine, committer())?
        .commit(&engine)?
        .unwrap_committed();

    // Reload from disk so we assert on the persisted schemaString, not the in-memory config.
    let reloaded = Snapshot::builder_for(table_url).build(&engine)?;
    let schema = reloaded.schema();

    // The pre-existing annotation on the untouched `value` field survives verbatim.
    assert_eq!(schema.field("value").unwrap().column_mapping_id(), Some(2));
    // The added column is persisted verbatim -- clean stays clean, stray annotation is left in
    // place (not stripped, because the table was already dirty).
    assert_eq!(
        schema.field("added").unwrap().column_mapping_id(),
        expected_added_cm_id
    );
    Ok(())
}