delta_kernel 0.30.0

//! Integration tests for change-data-feed aware write paths.

use std::collections::HashMap;
use std::sync::Arc;

use delta_kernel::actions::deletion_vector::{DeletionVectorDescriptor, DeletionVectorStorageType};
use delta_kernel::arrow::array::{Int32Array, Int64Array, StringArray, StructArray};
use delta_kernel::arrow::record_batch::RecordBatch;
use delta_kernel::engine::arrow_conversion::TryIntoArrow as _;
use delta_kernel::engine::arrow_data::ArrowEngineData;
use delta_kernel::engine_data::FilteredEngineData;
use delta_kernel::schema::{DataType, SchemaRef, StructField};
use delta_kernel::transaction::{CommitResult, UpdateTableOperation};
use delta_kernel::{Snapshot, Version};
use rstest::rstest;
use tempfile::{tempdir, TempDir};
use test_utils::delta_kernel_default_engine::executor::tokio::TokioBackgroundExecutor;
use test_utils::delta_kernel_default_engine::DefaultEngine;
use test_utils::{
    assert_result_error_with_message, begin_transaction, begin_transaction_with,
    create_add_files_metadata, create_table, create_table_and_load_snapshot, engine_store_setup,
    into_record_batch, load_and_begin_transaction_with, read_actions_from_commit, test_table_setup,
};
use url::Url;

use crate::common::write_utils::{get_scan_files, get_simple_int_schema, resolve_struct_field};

// Helper function to create a table with CDF enabled
async fn create_cdf_table(
    table_name: &str,
    schema: SchemaRef,
) -> Result<(Url, Arc<DefaultEngine<TokioBackgroundExecutor>>, TempDir), Box<dyn std::error::Error>>
{
    let tmp_dir = tempdir()?;
    let tmp_test_dir_url = Url::from_directory_path(tmp_dir.path()).unwrap();

    let (store, engine, table_location) = engine_store_setup(table_name, Some(&tmp_test_dir_url));

    let table_url = create_table(
        store.clone(),
        table_location,
        schema.clone(),
        &[],
        true, // use protocol 3.7
        vec![],
        vec!["changeDataFeed"],
    )
    .await?;

    Ok((table_url, Arc::new(engine), tmp_dir))
}

// Helper function to write data to a table
async fn write_data_to_table(
    table_url: &Url,
    engine: &Arc<DefaultEngine<TokioBackgroundExecutor>>,
    schema: SchemaRef,
    values: Vec<i32>,
) -> Result<Version, Box<dyn std::error::Error>> {
    let mut txn = load_and_begin_transaction_with(
        table_url.clone(),
        engine.as_ref(),
        UpdateTableOperation::Write,
        "test",
        true,
        |builder| builder,
    )?;

    add_files_to_transaction(&mut txn, engine, schema, values).await?;

    let result = txn.commit(engine.as_ref())?;
    match result {
        CommitResult::Committed(committed) => Ok(committed.commit_version()),
        _ => panic!("Transaction should be committed"),
    }
}

// Helper function to add files to an existing transaction
async fn add_files_to_transaction(
    txn: &mut delta_kernel::transaction::Transaction,
    engine: &Arc<DefaultEngine<TokioBackgroundExecutor>>,
    schema: SchemaRef,
    values: Vec<i32>,
) -> Result<(), Box<dyn std::error::Error>> {
    let data = RecordBatch::try_new(
        Arc::new(schema.as_ref().try_into_arrow()?),
        vec![Arc::new(Int32Array::from(values))],
    )?;

    let write_context = txn.write_state()?.write_context_builder().build()?;
    let add_files_metadata = engine
        .write_parquet(&ArrowEngineData::new(data), &write_context)
        .await?;
    txn.add_files(add_files_metadata);
    Ok(())
}

#[tokio::test]
async fn test_cdf_write_all_adds_succeeds() -> Result<(), Box<dyn std::error::Error>> {
    // This test verifies that add-only transactions work with CDF enabled
    let _ = tracing_subscriber::fmt::try_init();

    let schema = get_simple_int_schema();

    let (table_url, engine, _tmp_dir) =
        create_cdf_table("test_cdf_all_adds", schema.clone()).await?;

    // Add files - this should succeed
    let version = write_data_to_table(&table_url, &engine, schema, vec![1, 2, 3]).await?;
    assert_eq!(version, 1);

    Ok(())
}

#[tokio::test]
async fn test_cdf_write_all_removes_succeeds() -> Result<(), Box<dyn std::error::Error>> {
    // This test verifies that remove-only transactions work with CDF enabled
    let _ = tracing_subscriber::fmt::try_init();

    let schema = get_simple_int_schema();

    let (table_url, engine, _tmp_dir) =
        create_cdf_table("test_cdf_all_removes", schema.clone()).await?;

    // First, add some data
    write_data_to_table(&table_url, &engine, schema, vec![1, 2, 3]).await?;

    // Now remove the files
    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
    let mut txn = begin_transaction_with(
        snapshot.clone(),
        engine.as_ref(),
        UpdateTableOperation::Write,
        "cdf remove test",
        true,
        |builder| builder,
    )?;

    let scan = snapshot.scan_builder().build()?;
    let scan_metadata = scan.scan_metadata(engine.as_ref())?.next().unwrap()?;
    let (data, selection_vector) = scan_metadata.scan_files.into_parts();
    txn.remove_files(FilteredEngineData::try_new(data, selection_vector)?);

    // This should succeed - remove-only transactions are allowed with CDF
    let result = txn.commit(engine.as_ref())?;
    match result {
        CommitResult::Committed(committed) => {
            assert_eq!(committed.commit_version(), 2);
        }
        _ => panic!("Transaction should be committed"),
    }

    Ok(())
}

#[tokio::test]
async fn test_cdf_write_mixed_no_data_change_succeeds() -> Result<(), Box<dyn std::error::Error>> {
    // This test verifies that mixed add+remove transactions work when dataChange=false.
    // It's allowed because the transaction does not contain any logical data changes.
    // This can happen when a table is being optimized/compacted.
    let _ = tracing_subscriber::fmt::try_init();

    let schema = get_simple_int_schema();

    let (table_url, engine, _tmp_dir) =
        create_cdf_table("test_cdf_mixed_no_data_change", schema.clone()).await?;

    // First, add some data
    write_data_to_table(&table_url, &engine, schema.clone(), vec![1, 2, 3]).await?;

    // Now create a transaction with both add AND remove files, but dataChange=false
    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
    let mut txn = begin_transaction_with(
        snapshot.clone(),
        engine.as_ref(),
        UpdateTableOperation::Write,
        "cdf mixed test",
        false,
        |builder| builder,
    )?; // dataChange=false is key here

    // Add new files
    add_files_to_transaction(&mut txn, &engine, schema, vec![4, 5, 6]).await?;

    // Also remove existing files
    let scan = snapshot.scan_builder().build()?;
    let scan_metadata = scan.scan_metadata(engine.as_ref())?.next().unwrap()?;
    let (data, selection_vector) = scan_metadata.scan_files.into_parts();
    txn.remove_files(FilteredEngineData::try_new(data, selection_vector)?);

    // This should succeed - mixed operations are allowed when dataChange=false
    let result = txn.commit(engine.as_ref())?;
    match result {
        CommitResult::Committed(committed) => {
            assert_eq!(committed.commit_version(), 2);
        }
        _ => panic!("Transaction should be committed"),
    }

    Ok(())
}

#[tokio::test]
async fn test_cdf_write_mixed_with_data_change_fails() -> Result<(), Box<dyn std::error::Error>> {
    // This test verifies that mixed add+remove transactions fail with helpful error when
    // dataChange=true
    let _ = tracing_subscriber::fmt::try_init();

    let schema = get_simple_int_schema();

    let (table_url, engine, _tmp_dir) =
        create_cdf_table("test_cdf_mixed_with_data_change", schema.clone()).await?;

    // First, add some data
    write_data_to_table(&table_url, &engine, schema.clone(), vec![1, 2, 3]).await?;

    // Now create a transaction with both add AND remove files with dataChange=true
    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
    let mut txn = begin_transaction_with(
        snapshot.clone(),
        engine.as_ref(),
        UpdateTableOperation::Write,
        "cdf mixed fail test",
        true,
        |builder| builder,
    )?; // dataChange=true - this should fail

    // Add new files
    add_files_to_transaction(&mut txn, &engine, schema, vec![4, 5, 6]).await?;

    // Also remove existing files
    let scan = snapshot.scan_builder().build()?;
    let scan_metadata = scan.scan_metadata(engine.as_ref())?.next().unwrap()?;
    let (data, selection_vector) = scan_metadata.scan_files.into_parts();
    txn.remove_files(FilteredEngineData::try_new(data, selection_vector)?);

    assert_result_error_with_message(
        txn.commit(engine.as_ref()),
        "Cannot add and remove data in the same transaction when Change Data Feed is enabled (delta.enableChangeDataFeed = true). \
         This would require writing CDC files for DML operations, which is not yet supported. \
         Consider using separate transactions: one to add files, another to remove files or update deletion vectors.",
    );

    Ok(())
}

#[rstest]
#[tokio::test]
async fn test_file_rewrite_validates_explicit_data_change(
    #[values(false, true)] data_change: bool,
    #[values(false, true)] cdf_enabled: bool,
    #[values(
        None,
        Some(UpdateTableOperation::Write),
        Some(UpdateTableOperation::AlterTable)
    )]
    schema_change_operation: Option<UpdateTableOperation>,
    #[values(false, true)] use_dv: bool,
) -> Result<(), Box<dyn std::error::Error>> {
    let (_temp_dir, table_path, engine) = test_table_setup()?;
    let snapshot = create_table_and_load_snapshot(
        &table_path,
        get_simple_int_schema(),
        engine.as_ref(),
        &[
            ("delta.enableDeletionVectors", "true"),
            (
                "delta.enableChangeDataFeed",
                if cdf_enabled { "true" } else { "false" },
            ),
        ],
    )?;
    let mut setup_txn = begin_transaction(snapshot, engine.as_ref())?;
    let existing_file = create_add_files_metadata(
        setup_txn.add_files_schema(),
        vec![(
            "existing.parquet",
            100,   /* size */
            1_000, /* modification_time */
            Some(3),
        )],
    )?;
    setup_txn.add_files(existing_file);
    let before = setup_txn
        .commit(engine.as_ref())?
        .unwrap_post_commit_snapshot();

    let schema_change = schema_change_operation.is_some();
    let operation = schema_change_operation.unwrap_or(UpdateTableOperation::Write);
    let is_alter = operation == UpdateTableOperation::AlterTable;
    let result = begin_transaction_with(
        before.clone(),
        engine.as_ref(),
        operation,
        "test-engine",
        data_change,
        |mut builder| {
            if schema_change {
                builder = builder.add_column(StructField::nullable("added", DataType::INTEGER));
            }
            builder
        },
    );
    if is_alter && data_change {
        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());
        return Ok(());
    }
    let mut txn = result?;
    let new_file = create_add_files_metadata(
        txn.add_files_schema(),
        vec![(
            "new.parquet",
            100,   /* size */
            2_000, /* modification_time */
            Some(if use_dv { 1 } else { 3 }),
        )],
    )?;
    txn.add_files(new_file);
    let old_files = get_scan_files(before.clone(), engine.as_ref())?;
    if use_dv {
        let descriptor = DeletionVectorDescriptor::try_new(
            DeletionVectorStorageType::PersistedAbsolute,
            "memory:///dv.bin",
            Some(0), /* offset */
            1,       /* size_in_bytes */
            1,       /* cardinality */
        )?;
        txn.update_deletion_vectors(
            HashMap::from([("existing.parquet".to_string(), descriptor)]),
            old_files.into_iter().map(Ok),
        )?;
    } else {
        for files in old_files {
            txn.remove_files(files);
        }
    }

    let commit_result = txn.commit(engine.as_ref());
    let expected_error = match (is_alter, cdf_enabled && data_change) {
        (true, _) => Some("ALTER TABLE cannot include data-file actions"),
        (false, true) => Some("Cannot add and remove data in the same transaction"),
        (false, false) => None,
    };
    if let Some(expected_error) = expected_error {
        assert_result_error_with_message(commit_result, expected_error);
        let fresh = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
        assert_eq!(fresh.version(), before.version());
        assert_eq!(fresh.schema(), before.schema());
    } else {
        let post = commit_result?.unwrap_post_commit_snapshot();
        let fresh = Snapshot::builder_for(&table_path).build(engine.as_ref())?;
        for snapshot in [post, fresh] {
            assert_eq!(snapshot.version(), 2);
            assert_eq!(snapshot.schema().contains("added"), schema_change);
            let mut active_files = 0;
            let mut existing_dv = None;
            for files in get_scan_files(snapshot, engine.as_ref())? {
                let batch = into_record_batch(files.apply_selection_vector()?);
                active_files += batch.num_rows();

                let batch = StructArray::from(batch);
                let paths: &StringArray = resolve_struct_field(&batch, &["path".into()]);
                if let Some(row) = paths
                    .iter()
                    .position(|path| path == Some("existing.parquet"))
                {
                    let dv_paths: &StringArray = resolve_struct_field(
                        &batch,
                        &["deletionVector".into(), "pathOrInlineDv".into()],
                    );
                    let cardinalities: &Int64Array = resolve_struct_field(
                        &batch,
                        &["deletionVector".into(), "cardinality".into()],
                    );
                    existing_dv = Some((dv_paths.value(row).to_string(), cardinalities.value(row)));
                }
            }
            assert_eq!(active_files, if use_dv { 2 } else { 1 });
            assert_eq!(
                existing_dv,
                use_dv.then(|| ("memory:///dv.bin".to_string(), 1)),
            );
        }
        for (kind, expected_count) in [("add", if use_dv { 2 } else { 1 }), ("remove", 1)] {
            let actions = read_actions_from_commit(before.table_root(), 2, kind)?;
            assert_eq!(actions.len(), expected_count);
            assert!(actions
                .iter()
                .all(|action| action["dataChange"] == data_change));
        }
    }

    Ok(())
}