use super::*;
use crate::embedded_rows::parse_timestamp_micros;
use crate::test_support::{
    TestAppendFile, TestDeleteFile, assert_snapshot_conflict, test_file_column_stats,
    test_file_partition_value, test_schema_field as schema_field,
    test_schema_field_with_children as schema_field_with_children,
};
use crate::{
    AddColumn, AppendFile, CatalogBackend, CreateTableColumn, CreateTableWithFilesCommit,
    FlushedInlinedFileDeletion, InlinedDataAppend, PartitionColumnCommit, PartitioningBatchCommit,
    PartitioningTableCommit, SortExpressionCommit, TableColumnStats,
};
use ducklake_catalog::{ColumnId, DeleteFileId, DeleteFormat, DuckLakeType};
use std::collections::HashMap;

#[test]
fn set_metadata_option_replaces_scoped_rows_case_insensitively() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            "CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);",
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        backend
            .set_metadata_option(
                "Data_Path".to_string(),
                "s3://one".to_string(),
                crate::MetadataOptionScope::global(),
            )
            .await
            .unwrap();
        backend
            .set_metadata_option(
                "DATA_PATH".to_string(),
                "s3://two".to_string(),
                crate::MetadataOptionScope::global(),
            )
            .await
            .unwrap();
        backend
            .set_metadata_option(
                "data_path".to_string(),
                "s3://table".to_string(),
                crate::MetadataOptionScope {
                    scope: Some("table".to_string()),
                    scope_id: Some(7),
                },
            )
            .await
            .unwrap();
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let count: i64 = conn
        .query_row("SELECT COUNT(*) FROM ducklake_metadata", [], |row| {
            row.get(0)
        })
        .unwrap();
    assert_eq!(count, 2);
    let (global_key, global_value): (String, String) = conn
        .query_row(
            "SELECT key, value FROM ducklake_metadata WHERE scope IS NULL",
            [],
            |row| Ok((row.get(0)?, row.get(1)?)),
        )
        .unwrap();
    assert_eq!(global_key, "data_path");
    assert_eq!(global_value, "s3://two");
    let scoped_value: String = conn
        .query_row(
            "SELECT value FROM ducklake_metadata WHERE scope = 'table' AND scope_id = 7",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(scoped_value, "s3://table");
}

#[test]
fn reads_extended_ducklake_metadata_tables() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_schema_versions(begin_snapshot BIGINT, schema_version BIGINT, table_id BIGINT);
            CREATE TABLE ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name VARCHAR, dialect VARCHAR, sql VARCHAR, column_aliases VARCHAR);
            CREATE TABLE ducklake_tag(object_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, key VARCHAR, value VARCHAR);
            CREATE TABLE ducklake_column_tag(table_id BIGINT, column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, key VARCHAR, value VARCHAR);
            CREATE TABLE ducklake_file_variant_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, variant_path VARCHAR, shredded_type VARCHAR, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_table_column_stats(table_id BIGINT, column_id BIGINT, contains_null BOOLEAN, contains_nan BOOLEAN, min_value VARCHAR, max_value VARCHAR, extra_stats VARCHAR);
            CREATE TABLE ducklake_partition_info(partition_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_partition_column(partition_id BIGINT, table_id BIGINT, partition_key_index BIGINT, column_id BIGINT, transform VARCHAR);
            CREATE TABLE ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value VARCHAR);
            CREATE TABLE ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
            CREATE TABLE ducklake_macro(schema_id BIGINT, macro_id BIGINT, macro_name VARCHAR, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_macro_impl(macro_id BIGINT, impl_id BIGINT, dialect VARCHAR, sql VARCHAR, type VARCHAR);
            CREATE TABLE ducklake_macro_parameters(macro_id BIGINT, impl_id BIGINT, column_id BIGINT, parameter_name VARCHAR, parameter_type VARCHAR, default_value VARCHAR, default_value_type VARCHAR);
            CREATE TABLE ducklake_sort_info(sort_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_sort_expression(sort_id BIGINT, table_id BIGINT, sort_key_index BIGINT, expression VARCHAR, dialect VARCHAR, sort_direction VARCHAR, null_order VARCHAR);

            INSERT INTO ducklake_snapshot_changes VALUES (1, 'inserted_into_table:1', 'alice', 'append', '{"job":"t"}');
            INSERT INTO ducklake_schema_versions VALUES (1, 7, 1);
            INSERT INTO ducklake_view VALUES (9, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 1, 'v_numbers', 'duckdb', 'SELECT * FROM numbers', 'id,n');
            INSERT INTO ducklake_tag VALUES (1, 1, NULL, 'owner', 'analytics');
            INSERT INTO ducklake_column_tag VALUES (1, 2, 1, NULL, 'semantic_type', 'dimension');
            INSERT INTO ducklake_file_variant_stats VALUES (10, 1, 2, '$.a', 'int64', 4, 5, 0, '1', '9', false, NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 5, 6, 1024);
            INSERT INTO ducklake_table_column_stats VALUES (1, 2, true, false, '1', '9', NULL);
            INSERT INTO ducklake_partition_info VALUES (3, 1, 1, NULL);
            INSERT INTO ducklake_partition_column VALUES (3, 1, 0, 2, 'bucket(8)');
            INSERT INTO ducklake_file_partition_value VALUES (10, 1, 0, '4');
            INSERT INTO ducklake_files_scheduled_for_deletion VALUES (10, 'old.parquet', true, '2026-01-01 00:00:00+00');
            INSERT INTO ducklake_macro VALUES (1, 4, 'plus_one', 1, NULL);
            INSERT INTO ducklake_macro_impl VALUES (4, 1, 'duckdb', 'x + 1', 'scalar');
            INSERT INTO ducklake_macro_parameters VALUES (4, 1, 0, 'x', 'int64', NULL, NULL);
            INSERT INTO ducklake_sort_info VALUES (5, 1, 1, NULL);
            INSERT INTO ducklake_sort_expression VALUES (5, 1, 0, 'n', 'duckdb', 'ASC', 'NULLS_LAST');
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let changes = backend.snapshot_changes().await.unwrap();
        assert_eq!(changes[0].author.as_deref(), Some("alice"));

        let versions = backend.schema_versions(TableId(1)).await.unwrap();
        assert_eq!(versions[0].schema_version, 7);

        let views = backend.views_at(SchemaId(1), SnapshotId(1)).await.unwrap();
        assert_eq!(views[0].view_name, "v_numbers");

        let tags = backend.tags_at(SnapshotId(1)).await.unwrap();
        assert_eq!(tags[0].key, "owner");

        let column_tags = backend
            .column_tags_at(TableId(1), SnapshotId(1))
            .await
            .unwrap();
        assert_eq!(column_tags[0].value.as_deref(), Some("dimension"));

        let variant_stats = backend.file_variant_stats(TableId(1)).await.unwrap();
        assert_eq!(variant_stats[0].variant_path, "$.a");

        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.next_row_id, 6);

        let column_stats = backend.table_column_stats(TableId(1)).await.unwrap();
        assert_eq!(column_stats[0].column_id, ColumnId(2));

        let partitions = backend
            .partition_info_at(TableId(1), SnapshotId(1))
            .await
            .unwrap();
        assert_eq!(partitions[0].partition_id, PartitionId(3));

        let partition_columns = backend.partition_columns(PartitionId(3)).await.unwrap();
        assert_eq!(
            partition_columns[0].transform,
            PartitionTransform::Bucket(8)
        );

        let partition_values = backend.file_partition_values(TableId(1)).await.unwrap();
        assert_eq!(partition_values[0].partition_value.as_deref(), Some("4"));

        let scheduled = backend.files_scheduled_for_deletion().await.unwrap();
        assert_eq!(scheduled[0].path, "old.parquet");
        assert!(scheduled[0].schedule_start > 0);

        let macros = backend.macros_at(SchemaId(1), SnapshotId(1)).await.unwrap();
        assert_eq!(macros[0].macro_id, MacroId(4));

        let impls = backend.macro_impls(MacroId(4)).await.unwrap();
        assert_eq!(impls[0].macro_type, "scalar");

        let parameters = backend
            .macro_parameters(MacroId(4), MacroImplId(1))
            .await
            .unwrap();
        assert_eq!(parameters[0].parameter_name, "x");

        let sort_info = backend
            .sort_info_at(TableId(1), SnapshotId(1))
            .await
            .unwrap();
        assert_eq!(sort_info[0].sort_id, SortId(5));

        let sort_expressions = backend.sort_expressions(SortId(5)).await.unwrap();
        assert_eq!(sort_expressions[0].expression, "n");
    });
}

#[test]
fn duckdb_file_commit_partitioning_batch_matches_duckdb_multi_table_transaction() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_partition_info(partition_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_partition_column(partition_id BIGINT, table_id BIGINT, partition_key_index BIGINT, column_id BIGINT, transform VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (2, '2026-06-26 00:00:00+00', 2, 3, 0);
            INSERT INTO ducklake_column VALUES (1, 1, NULL, 1, 1, 'part_key', 'int32', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (2, 1, NULL, 1, 2, 'values', 'varchar', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (1, 2, NULL, 2, 1, 'part_key', 'int32', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (2, 2, NULL, 2, 2, 'values', 'varchar', NULL, NULL, true, NULL, 'literal', 'duckdb');
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = backend
            .commit_partitioning_batch(PartitioningBatchCommit {
                expected_snapshot: SnapshotId(2),
                changes: vec![
                    PartitioningTableCommit {
                        table_id: TableId(1),
                        columns: vec![PartitionColumnCommit {
                            column_id: ColumnId(1),
                            transform: PartitionTransform::Identity,
                        }],
                    },
                    PartitioningTableCommit {
                        table_id: TableId(2),
                        columns: vec![PartitionColumnCommit {
                            column_id: ColumnId(1),
                            transform: PartitionTransform::Identity,
                        }],
                    },
                ],
                commit_message: "set partitioning for two tables".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(3));
    });

    assert_eq!(
        duckdb_partition_info_rows_with_table(&catalog_path),
        vec![(3, 3, None, 2), (4, 3, None, 1)]
    );
    assert_eq!(
        duckdb_partition_column_rows(&catalog_path),
        vec![
            (3, 0, 1, "identity".to_string()),
            (4, 0, 1, "identity".to_string()),
        ]
    );
    let conn = Connection::open(&catalog_path).unwrap();
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (3, 3, 5, 0));
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["altered_table:1,altered_table:2".to_string()]
    );
}

#[test]
fn duckdb_file_commit_sorted_by_updates_metadata_without_schema_bump() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_sort_info(sort_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_sort_expression(sort_id BIGINT, table_id BIGINT, sort_key_index BIGINT, expression VARCHAR, dialect VARCHAR, sort_direction VARCHAR, null_order VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (2, '2026-06-26 00:00:00+00', 7, 10, 20);
            INSERT INTO ducklake_column VALUES (1, 1, NULL, 1, 1, 'id', 'int32', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (2, 1, NULL, 1, 2, 'name', 'varchar', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_sort_info VALUES (5, 1, 1, NULL);
            INSERT INTO ducklake_sort_expression VALUES (5, 1, 0, 'id', 'duckdb', 'ASC', 'NULLS_LAST');
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = backend
            .commit_sorted_by(SortedByCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                expressions: vec![SortExpressionCommit {
                    expression: "name".into(),
                    dialect: "duckdb".into(),
                    sort_direction: "DESC".into(),
                    null_order: "NULLS_FIRST".into(),
                }],
                commit_message: "set sorted".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(3));

        let repeated = backend
            .commit_sorted_by(SortedByCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(3),
                expressions: vec![SortExpressionCommit {
                    expression: "name".into(),
                    dialect: "duckdb".into(),
                    sort_direction: "DESC".into(),
                    null_order: "NULLS_FIRST".into(),
                }],
                commit_message: "set sorted again".into(),
            })
            .await
            .unwrap();
        assert_eq!(repeated, SnapshotId(3));

        let reset = backend
            .commit_sorted_by(SortedByCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(3),
                expressions: Vec::new(),
                commit_message: "reset sorted".into(),
            })
            .await
            .unwrap();
        assert_eq!(reset, SnapshotId(4));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (4, 7, 11, 20));
    let sort_rows: Vec<(i64, i64, Option<i64>, String, String, String)> = {
        let mut stmt = conn
            .prepare(
                "SELECT si.sort_id, si.begin_snapshot, si.end_snapshot, se.expression, \
                            se.sort_direction, se.null_order \
                     FROM ducklake_sort_info si \
                     JOIN ducklake_sort_expression se USING (sort_id, table_id) \
                     WHERE si.table_id = 1 \
                     ORDER BY si.sort_id",
            )
            .unwrap();
        collect_rows(&mut stmt, [], |row| {
            Ok((
                row.get(0)?,
                row.get(1)?,
                row.get(2)?,
                row.get(3)?,
                row.get(4)?,
                row.get(5)?,
            ))
        })
        .unwrap()
    };
    assert_eq!(
        sort_rows,
        vec![
            (
                5,
                1,
                Some(3),
                "id".to_string(),
                "ASC".to_string(),
                "NULLS_LAST".to_string()
            ),
            (
                10,
                3,
                Some(4),
                "name".to_string(),
                "DESC".to_string(),
                "NULLS_FIRST".to_string()
            ),
        ]
    );
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["altered_table:1".to_string(), "altered_table:1".to_string()]
    );
}

#[test]
fn duckdb_file_partitioning_batch_repeated_table_uses_final_state_and_duckdb_id_allocation() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_partition_info(partition_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_partition_column(partition_id BIGINT, table_id BIGINT, partition_key_index BIGINT, column_id BIGINT, transform VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (2, '2026-06-26 00:00:00+00', 2, 3, 0);
            INSERT INTO ducklake_column VALUES (1, 1, NULL, 1, 1, 'part_key', 'int32', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (2, 1, NULL, 1, 2, 'values', 'varchar', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (1, 2, NULL, 2, 1, 'part_key', 'int32', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (2, 2, NULL, 2, 2, 'values', 'varchar', NULL, NULL, true, NULL, 'literal', 'duckdb');
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = backend
            .commit_partitioning_batch(PartitioningBatchCommit {
                expected_snapshot: SnapshotId(2),
                changes: vec![
                    PartitioningTableCommit {
                        table_id: TableId(1),
                        columns: vec![PartitionColumnCommit {
                            column_id: ColumnId(1),
                            transform: PartitionTransform::Identity,
                        }],
                    },
                    PartitioningTableCommit {
                        table_id: TableId(2),
                        columns: vec![PartitionColumnCommit {
                            column_id: ColumnId(1),
                            transform: PartitionTransform::Identity,
                        }],
                    },
                    PartitioningTableCommit {
                        table_id: TableId(1),
                        columns: Vec::new(),
                    },
                ],
                commit_message: "repeated partitioning in one transaction".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(3));
    });

    assert_eq!(
        duckdb_partition_info_rows_with_table(&catalog_path),
        vec![(3, 3, None, 2)]
    );
    assert_eq!(
        duckdb_partition_column_rows(&catalog_path),
        vec![(3, 0, 1, "identity".to_string())]
    );
    let conn = Connection::open(&catalog_path).unwrap();
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (3, 3, 6, 0));
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["altered_table:1,altered_table:2".to_string()]
    );
}

#[test]
fn reads_catalog_tables_from_non_default_metadata_schema() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
        r#"
            CREATE SCHEMA "quoted metadata schema";
            CREATE TABLE "quoted metadata schema".ducklake_metadata(
                key VARCHAR,
                value VARCHAR,
                scope VARCHAR,
                scope_id BIGINT
            );
            CREATE TABLE "quoted metadata schema".ducklake_snapshot(
                snapshot_id BIGINT PRIMARY KEY,
                snapshot_time TIMESTAMPTZ,
                schema_version BIGINT,
                next_catalog_id BIGINT,
                next_file_id BIGINT
            );
            INSERT INTO "quoted metadata schema".ducklake_metadata VALUES
                ('version', '1.0', NULL, NULL),
                ('data_path', 'data', NULL, NULL);
            INSERT INTO "quoted metadata schema".ducklake_snapshot VALUES
                (1, '2026-01-01 00:00:00+00', 1, 1, 1);
            "#,
    )
    .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let metadata = backend.metadata().await.unwrap();
        assert_eq!(metadata.version, "1.0");
        assert_eq!(metadata.data_path, "data");

        let snapshots = backend.snapshots().await.unwrap();
        assert_eq!(snapshots[0].snapshot_id, SnapshotId(1));
    });
}

#[test]
fn selects_explicit_metadata_schema_when_multiple_ducklake_catalogs_exist() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
        r#"
            CREATE SCHEMA metadata_s1;
            CREATE SCHEMA metadata_s2;
            CREATE TABLE metadata_s1.ducklake_metadata(
                key VARCHAR,
                value VARCHAR,
                scope VARCHAR,
                scope_id BIGINT
            );
            CREATE TABLE metadata_s1.ducklake_snapshot(
                snapshot_id BIGINT PRIMARY KEY,
                snapshot_time TIMESTAMPTZ,
                schema_version BIGINT,
                next_catalog_id BIGINT,
                next_file_id BIGINT
            );
            CREATE TABLE metadata_s2.ducklake_metadata(
                key VARCHAR,
                value VARCHAR,
                scope VARCHAR,
                scope_id BIGINT
            );
            CREATE TABLE metadata_s2.ducklake_snapshot(
                snapshot_id BIGINT PRIMARY KEY,
                snapshot_time TIMESTAMPTZ,
                schema_version BIGINT,
                next_catalog_id BIGINT,
                next_file_id BIGINT
            );
            INSERT INTO metadata_s1.ducklake_metadata VALUES
                ('version', '1.0', NULL, NULL),
                ('data_path', 'data_s1', NULL, NULL);
            INSERT INTO metadata_s1.ducklake_snapshot VALUES
                (1, '2026-01-01 00:00:00+00', 1, 1, 1);
            INSERT INTO metadata_s2.ducklake_metadata VALUES
                ('version', '1.0', NULL, NULL),
                ('data_path', 'data_s2', NULL, NULL);
            INSERT INTO metadata_s2.ducklake_snapshot VALUES
                (2, '2026-01-02 00:00:00+00', 1, 1, 1);
            "#,
    )
    .unwrap();
    drop(conn);

    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let ambiguous = DuckDbCatalog::new(&catalog_path).metadata().await;
        assert!(ambiguous.is_err());

        let s1 = DuckDbCatalog::with_metadata_schema(&catalog_path, "metadata_s1");
        assert_eq!(s1.metadata_schema(), Some("metadata_s1"));
        let s1_metadata = s1.metadata().await.unwrap();
        assert_eq!(s1_metadata.data_path, "data_s1");
        assert_eq!(s1.snapshots().await.unwrap()[0].snapshot_id, SnapshotId(1));

        let s2 = DuckDbCatalog::with_metadata_schema(&catalog_path, "metadata_s2");
        let s2_metadata = s2.metadata().await.unwrap();
        assert_eq!(s2_metadata.data_path, "data_s2");
        assert_eq!(s2.snapshots().await.unwrap()[0].snapshot_id, SnapshotId(2));
    });
}

#[test]
fn duckdb_file_cleanup_old_files_deletes_due_local_files_and_queue_rows() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let data_path = temp.path().join("data");
    std::fs::create_dir(&data_path).unwrap();
    std::fs::write(data_path.join("old.parquet"), b"old").unwrap();
    std::fs::write(data_path.join("future.parquet"), b"future").unwrap();
    let absolute_path = temp.path().join("absolute.parquet");
    std::fs::write(&absolute_path, b"absolute").unwrap();

    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);
            CREATE TABLE ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
            INSERT INTO ducklake_metadata VALUES ('version', '1.0', NULL, NULL);
            INSERT INTO ducklake_files_scheduled_for_deletion VALUES (1, 'old.parquet', true, '2026-01-01 00:00:00+00');
            INSERT INTO ducklake_files_scheduled_for_deletion VALUES (2, 'future.parquet', true, '2026-01-03 00:00:00+00');
            INSERT INTO ducklake_files_scheduled_for_deletion VALUES (4, 'already-gone.parquet', true, '2026-01-01 00:00:00+00');
            "#,
        )
        .unwrap();
    conn.execute(
        "INSERT INTO ducklake_metadata VALUES ('data_path', ?1, NULL, NULL)",
        params![data_path.display().to_string()],
    )
    .unwrap();
    conn.execute(
            "INSERT INTO ducklake_files_scheduled_for_deletion VALUES (3, ?1, false, '2026-01-01 00:00:00+00')",
            params![absolute_path.display().to_string()],
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let invalid = backend
            .cleanup_old_files(CleanupOldFilesOptions {
                dry_run: true,
                cleanup_all: false,
                older_than_micros: None,
            })
            .await
            .unwrap_err();
        assert!(format!("{invalid}").contains("requires exactly one of cleanup_all or older_than"));

        let dry_run = backend
            .cleanup_old_files(CleanupOldFilesOptions::dry_run_all())
            .await
            .unwrap();
        assert_eq!(dry_run.len(), 4);
        assert!(data_path.join("old.parquet").exists());
        assert!(absolute_path.exists());
        assert_eq!(duckdb_scheduled_file_count(&catalog_path), 4);

        let cutoff = parse_timestamp_micros("2026-01-02 00:00:00+00").unwrap();
        let cleaned = backend
            .cleanup_old_files(CleanupOldFilesOptions::older_than(cutoff))
            .await
            .unwrap();
        assert_eq!(
            cleaned
                .iter()
                .map(|file| file.data_file_id)
                .collect::<Vec<_>>(),
            vec![DataFileId(1), DataFileId(3), DataFileId(4)]
        );
        assert!(!data_path.join("old.parquet").exists());
        assert!(!absolute_path.exists());
        assert!(data_path.join("future.parquet").exists());
        assert_eq!(duckdb_scheduled_file_count(&catalog_path), 1);

        let cleaned = backend
            .cleanup_old_files(CleanupOldFilesOptions::cleanup_all())
            .await
            .unwrap();
        assert_eq!(cleaned.len(), 1);
        assert!(!data_path.join("future.parquet").exists());
        assert_eq!(duckdb_scheduled_file_count(&catalog_path), 0);
    });
}

#[test]
fn duckdb_file_cleanup_old_files_rejects_remote_paths_before_deleting_local_files() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let data_path = temp.path().join("data");
    std::fs::create_dir(&data_path).unwrap();
    let local_path = data_path.join("old.parquet");
    std::fs::write(&local_path, b"old").unwrap();

    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);
            CREATE TABLE ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
            INSERT INTO ducklake_metadata VALUES ('version', '1.0', NULL, NULL);
            "#,
        )
        .unwrap();
    conn.execute(
        "INSERT INTO ducklake_metadata VALUES ('data_path', ?1, NULL, NULL)",
        params![data_path.display().to_string()],
    )
    .unwrap();
    conn.execute(
            "INSERT INTO ducklake_files_scheduled_for_deletion VALUES (1, ?1, false, '2026-01-01 00:00:00+00')",
            params![local_path.display().to_string()],
        )
        .unwrap();
    conn.execute(
            "INSERT INTO ducklake_files_scheduled_for_deletion VALUES (2, 's3://bucket/old.parquet', false, '2026-01-01 00:00:00+00')",
            [],
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let err = backend
            .cleanup_old_files(CleanupOldFilesOptions::cleanup_all())
            .await
            .unwrap_err();
        assert!(format!("{err}").contains("Materialized object-store file access"));
        assert!(local_path.exists());
        assert_eq!(duckdb_scheduled_file_count(&catalog_path), 2);
    });
}

#[test]
fn duckdb_file_expire_snapshots_schedules_unreferenced_files_and_prunes_metadata() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_schema_versions(begin_snapshot BIGINT, schema_version BIGINT, table_id BIGINT);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
            CREATE TABLE ducklake_macro(schema_id BIGINT, macro_id BIGINT, macro_name VARCHAR, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_macro_impl(macro_id BIGINT, impl_id BIGINT, dialect VARCHAR, sql VARCHAR, type VARCHAR);
            CREATE TABLE ducklake_macro_parameters(macro_id BIGINT, impl_id BIGINT, column_id BIGINT, parameter_name VARCHAR, parameter_type VARCHAR, default_value VARCHAR, default_value_type VARCHAR);
            CREATE TABLE ducklake_column_mapping(mapping_id BIGINT, table_id BIGINT, type VARCHAR);
            CREATE TABLE ducklake_name_mapping(mapping_id BIGINT, column_id BIGINT, source_name VARCHAR, target_field_id BIGINT, parent_column BIGINT, is_partition BOOLEAN);

            INSERT INTO ducklake_metadata VALUES ('version', '1.0', NULL, NULL), ('data_path', 'file:///tmp/lake', NULL, NULL);
            INSERT INTO ducklake_snapshot VALUES
                (1, '2026-01-01 00:00:00+00', 1, 10, 10),
                (2, '2026-01-02 00:00:00+00', 1, 10, 11),
                (3, '2026-01-03 00:00:00+00', 1, 10, 11);
            INSERT INTO ducklake_snapshot_changes VALUES
                (1, 'created_table:1', NULL, NULL, NULL),
                (2, 'inserted_into_table:1', NULL, NULL, NULL),
                (3, 'deleted_from_table:1', NULL, NULL, NULL);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440000', 1, NULL, 'main', NULL, true);
            INSERT INTO ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, 3, 1, 'numbers', NULL, true);
            INSERT INTO ducklake_column VALUES (1, 1, NULL, 1, 0, 'id', 'int32', NULL, NULL, true, NULL, 'literal', NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 1, 1, 42);
            INSERT INTO ducklake_schema_versions VALUES (1, 1, 1);
            INSERT INTO ducklake_data_file VALUES (10, 1, 2, 3, 0, 'old.parquet', true, 'parquet', 1, 42, NULL, NULL, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_file_column_stats VALUES (10, 1, 1, 4, 1, 0, '1', '1', false, NULL);
            INSERT INTO ducklake_macro VALUES (1, 4, 'old_macro', 1, 3);
            INSERT INTO ducklake_macro_impl VALUES (4, 1, 'duckdb', 'x + 1', 'scalar');
            INSERT INTO ducklake_macro_parameters VALUES (4, 1, 0, 'x', 'int64', NULL, NULL);
            INSERT INTO ducklake_column_mapping VALUES (5, 1, 'name');
            INSERT INTO ducklake_name_mapping VALUES (5, 1, 'old_id', 1, NULL, false);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let dry_run = backend
            .expire_snapshots(ExpireSnapshotsOptions::dry_run_versions(vec![
                SnapshotId(2),
                SnapshotId(3),
            ]))
            .await
            .unwrap();
        assert_eq!(
            dry_run
                .iter()
                .map(|snapshot| snapshot.snapshot_id)
                .collect::<Vec<_>>(),
            vec![SnapshotId(2)]
        );
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_snapshot"), 3);
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_data_file"), 1);

        let expired = backend
            .expire_snapshots(ExpireSnapshotsOptions::versions(vec![SnapshotId(2)]))
            .await
            .unwrap();
        assert_eq!(expired[0].snapshot_id, SnapshotId(2));
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_snapshot"), 2);
        assert_eq!(
            duckdb_table_count(&catalog_path, "ducklake_snapshot_changes"),
            2
        );
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_data_file"), 0);
        assert_eq!(
            duckdb_table_count(&catalog_path, "ducklake_file_column_stats"),
            0
        );
        assert_eq!(duckdb_scheduled_file_count(&catalog_path), 1);

        let cutoff = backend
            .snapshots()
            .await
            .unwrap()
            .into_iter()
            .find(|snapshot| snapshot.snapshot_id == SnapshotId(3))
            .unwrap()
            .snapshot_time;
        let expired = backend
            .expire_snapshots(ExpireSnapshotsOptions::older_than(cutoff))
            .await
            .unwrap();
        assert_eq!(
            expired
                .iter()
                .map(|snapshot| snapshot.snapshot_id)
                .collect::<Vec<_>>(),
            vec![SnapshotId(1)]
        );
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_snapshot"), 1);
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_table"), 0);
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_column"), 0);
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_table_stats"), 0);
        assert_eq!(
            duckdb_table_count(&catalog_path, "ducklake_schema_versions"),
            0
        );
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_macro"), 0);
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_macro_impl"), 0);
        assert_eq!(
            duckdb_table_count(&catalog_path, "ducklake_macro_parameters"),
            0
        );
        assert_eq!(
            duckdb_table_count(&catalog_path, "ducklake_column_mapping"),
            0
        );
        assert_eq!(
            duckdb_table_count(&catalog_path, "ducklake_name_mapping"),
            0
        );
        assert_eq!(duckdb_scheduled_file_count(&catalog_path), 1);
    });
}

#[test]
fn duckdb_file_merge_adjacent_compaction_replaces_sources_and_schedules_cleanup() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_file_variant_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, variant_path VARCHAR, shredded_type VARCHAR, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);

            INSERT INTO ducklake_metadata VALUES ('version', '1.0', NULL, NULL), ('data_path', 'file:///tmp/lake', NULL, NULL);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440000', 1, NULL, 'main', NULL, true);
            INSERT INTO ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 1, 'numbers', NULL, true);
            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_data_file VALUES
                (10, 1, 1, NULL, 0, 'source-a.parquet', true, 'parquet', 1, 10, NULL, NULL, NULL, NULL, NULL, NULL),
                (11, 1, 1, NULL, 1, 'source-b.parquet', true, 'parquet', 2, 20, NULL, NULL, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_file_column_stats VALUES
                (10, 1, 1, 4, 1, 0, '1', '1', false, NULL),
                (11, 1, 1, 8, 2, 0, '2', '3', false, NULL);
            INSERT INTO ducklake_file_variant_stats VALUES
                (10, 1, 1, '$.a', 'int64', 4, 1, 0, '1', '1', false, NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 3, 3, 100);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let new_snapshot = backend
            .commit_merge_adjacent_compaction(MergeAdjacentCompactionCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                source_data_file_ids: vec![DataFileId(10), DataFileId(11)],
                compacted_files: vec![AppendFile {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(1),
                    data_file_id: DataFileId(20),
                    begin_snapshot: None,
                    path: "compact-20.parquet".into(),
                    path_is_relative: true,
                    record_count: 3,
                    file_size_bytes: 25,
                    footer_size: None,
                    row_id_start: None,
                    explicit_row_ids: false,
                    column_stats: vec![FileColumnStats {
                        data_file_id: DataFileId(20),
                        table_id: TableId(1),
                        column_id: ColumnId(1),
                        column_size_bytes: Some(12),
                        value_count: Some(3),
                        null_count: Some(0),
                        min_value: Some("1".into()),
                        max_value: Some("3".into()),
                        contains_nan: Some(false),
                        extra_stats: None,
                    }],
                    variant_stats: Vec::new(),
                    partition_id: None,
                    partition_values: Vec::new(),
                    encryption_key: None,
                    mapping_id: None,
                    partial_max: None,
                }],
            })
            .await
            .unwrap();
        assert_eq!(new_snapshot, SnapshotId(2));
    });

    assert_eq!(
        duckdb_data_file_ids(&catalog_path),
        vec![(20, "compact-20.parquet".to_string())]
    );
    assert_eq!(
        duckdb_table_count(&catalog_path, "ducklake_file_column_stats"),
        1
    );
    assert_eq!(
        duckdb_table_count(&catalog_path, "ducklake_file_variant_stats"),
        0
    );
    assert_eq!(duckdb_scheduled_file_count(&catalog_path), 2);
    assert_eq!(duckdb_table_count(&catalog_path, "ducklake_snapshot"), 2);
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["merge_adjacent:1".to_string()]
    );
    assert_eq!(duckdb_table_stats_file_size(&catalog_path, TableId(1)), 95);
    assert_eq!(duckdb_latest_next_file_id(&catalog_path), 21);
}

#[test]
fn duckdb_file_rewrite_data_files_closes_sources_without_cleanup() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);

            INSERT INTO ducklake_metadata VALUES ('version', '1.0', NULL, NULL), ('data_path', 'file:///tmp/lake', NULL, NULL);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440000', 1, NULL, 'main', NULL, true);
            INSERT INTO ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 1, 'numbers', NULL, true);
            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_data_file VALUES
                (10, 1, 1, NULL, 0, 'rewrite-source-a.parquet', true, 'parquet', 1, 10, NULL, NULL, NULL, NULL, NULL, NULL),
                (11, 1, 1, NULL, 1, 'rewrite-source-b.parquet', true, 'parquet', 2, 20, NULL, NULL, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_delete_file VALUES
                (12, 1, 1, NULL, 10, 'rewrite-delete-a.parquet', true, 'parquet', 1, 5, 1, NULL, NULL);
            INSERT INTO ducklake_file_column_stats VALUES
                (10, 1, 1, 4, 1, 0, '1', '1', false, NULL),
                (11, 1, 1, 8, 2, 0, '2', '3', false, NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 3, 3, 100);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let new_snapshot = backend
            .commit_rewrite_data_files(RewriteDataFilesCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                source_data_file_ids: vec![DataFileId(10), DataFileId(11)],
                rewritten_files: vec![AppendFile {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(1),
                    data_file_id: DataFileId(20),
                    begin_snapshot: None,
                    path: "rewrite-20.parquet".into(),
                    path_is_relative: true,
                    record_count: 3,
                    file_size_bytes: 25,
                    footer_size: None,
                    row_id_start: None,
                    explicit_row_ids: true,
                    column_stats: vec![FileColumnStats {
                        data_file_id: DataFileId(20),
                        table_id: TableId(1),
                        column_id: ColumnId(1),
                        column_size_bytes: Some(12),
                        value_count: Some(3),
                        null_count: Some(0),
                        min_value: Some("1".into()),
                        max_value: Some("3".into()),
                        contains_nan: Some(false),
                        extra_stats: None,
                    }],
                    variant_stats: Vec::new(),
                    partition_id: None,
                    partition_values: Vec::new(),
                    encryption_key: None,
                    mapping_id: None,
                    partial_max: None,
                }],
            })
            .await
            .unwrap();
        assert_eq!(new_snapshot, SnapshotId(2));

        let latest_files = backend
            .data_files_at(TableId(1), SnapshotId(2))
            .await
            .unwrap();
        assert_eq!(latest_files.len(), 1);
        assert_eq!(latest_files[0].data_file_id, DataFileId(20));
        let historical_files = backend
            .data_files_at(TableId(1), SnapshotId(1))
            .await
            .unwrap();
        assert_eq!(historical_files.len(), 2);
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let mut stmt = conn
        .prepare(
            "SELECT data_file_id, begin_snapshot, end_snapshot, path \
                 FROM ducklake_data_file ORDER BY data_file_id",
        )
        .unwrap();
    let data_file_rows = stmt
        .query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, Option<i64>>(2)?,
                row.get::<_, String>(3)?,
            ))
        })
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap();
    assert_eq!(
        data_file_rows,
        vec![
            (10, 1, Some(2), "rewrite-source-a.parquet".to_string()),
            (11, 1, Some(2), "rewrite-source-b.parquet".to_string()),
            (20, 2, None, "rewrite-20.parquet".to_string()),
        ]
    );
    let delete_end_snapshot: Option<i64> = conn
        .query_row(
            "SELECT end_snapshot FROM ducklake_delete_file WHERE delete_file_id = 12",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(delete_end_snapshot, Some(2));
    assert_eq!(duckdb_scheduled_file_count(&catalog_path), 0);
    assert_eq!(duckdb_table_count(&catalog_path, "ducklake_snapshot"), 2);
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["rewrite_data_files:1".to_string()]
    );
    assert_eq!(duckdb_table_stats_file_size(&catalog_path, TableId(1)), 95);
    assert_eq!(duckdb_latest_next_file_id(&catalog_path), 21);
}

#[test]
fn duckdb_file_materialized_overwrite_schedules_delete_cleanup_with_table_path() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

            INSERT INTO ducklake_metadata VALUES ('data_path', '/tmp/ducklake-data', NULL);
            INSERT INTO ducklake_schema VALUES (1, '00000000-0000-0000-0000-000000000001', 1, NULL, 'main', 'main', true);
            INSERT INTO ducklake_table VALUES (1, '00000000-0000-0000-0000-000000000002', 1, NULL, 1, 'numbers', 'numbers', true);
            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 11);
            INSERT INTO ducklake_data_file VALUES (1, 1, 1, NULL, 0, 'data.parquet', true, 'parquet', 4, 40, NULL, 0, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_delete_file VALUES (10, 1, 1, NULL, 1, 'old-delete.parquet', true, 'parquet', 1, 12, NULL, NULL, NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 3, 4, 40);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let replacement = DeleteFileCommit {
            table_id: TableId(1),
            expected_snapshot: SnapshotId(1),
            delete_file_id: DeleteFileId(11),
            begin_snapshot: Some(SnapshotId(1)),
            data_file_id: DataFileId(1),
            path: "replacement-delete.parquet".into(),
            path_is_relative: true,
            format: DeleteFormat::Parquet,
            delete_count: 1,
            rows_deleted: 0,
            file_size_bytes: 12,
            footer_size: None,
            partial_max: Some(SnapshotId(1)),
            overwritten_delete_file_id: Some(DeleteFileId(10)),
        };
        let snapshot = backend
            .commit_table_changes(TableChangeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                append_files: Vec::new(),
                delete_files: vec![replacement],
                inlined_file_deletions: Vec::new(),
                inlined_data_appends: Vec::new(),
                inlined_data_deletions: Vec::new(),
                flushed_inlined_data: Vec::new(),
                flushed_inlined_file_deletions: Vec::new(),
                changes_made: "flushed_inlined_data:1".into(),
                commit_message: "datafusion flush inlined deletes".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(2));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let cleanup: (i64, String, bool) = conn
        .query_row(
            "SELECT data_file_id, path, path_is_relative \
                 FROM ducklake_files_scheduled_for_deletion",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
        )
        .unwrap();
    assert_eq!(
        cleanup,
        (10, "main/numbers/old-delete.parquet".to_string(), true)
    );
}

#[test]
fn filters_versioned_duckdb_metadata_rows_by_snapshot() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name VARCHAR, dialect VARCHAR, sql VARCHAR, column_aliases VARCHAR);
            CREATE TABLE ducklake_tag(object_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, key VARCHAR, value VARCHAR);
            CREATE TABLE ducklake_column_tag(table_id BIGINT, column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, key VARCHAR, value VARCHAR);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_partition_info(partition_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_macro(schema_id BIGINT, macro_id BIGINT, macro_name VARCHAR, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_sort_info(sort_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);

            INSERT INTO ducklake_schema VALUES
                (1, '550e8400-e29b-41d4-a716-446655440001', 1, 3, 'old_schema', NULL, true),
                (2, '550e8400-e29b-41d4-a716-446655440002', 3, 4, 'current_schema', NULL, true),
                (3, '550e8400-e29b-41d4-a716-446655440003', 4, NULL, 'future_schema', NULL, true);
            INSERT INTO ducklake_table VALUES
                (10, '550e8400-e29b-41d4-a716-446655440010', 1, 3, 2, 'old_table', NULL, true),
                (20, '550e8400-e29b-41d4-a716-446655440020', 3, 4, 2, 'current_table', NULL, true),
                (30, '550e8400-e29b-41d4-a716-446655440030', 4, NULL, 2, 'future_table', NULL, true);
            INSERT INTO ducklake_view VALUES
                (11, '550e8400-e29b-41d4-a716-446655440011', 1, 3, 2, 'old_view', 'duckdb', 'SELECT 1', NULL),
                (21, '550e8400-e29b-41d4-a716-446655440021', 3, 4, 2, 'current_view', 'duckdb', 'SELECT 2', NULL),
                (31, '550e8400-e29b-41d4-a716-446655440031', 4, NULL, 2, 'future_view', 'duckdb', 'SELECT 3', NULL);
            INSERT INTO ducklake_tag VALUES
                (1, 1, 3, 'phase', 'old'),
                (1, 3, 4, 'phase', 'current'),
                (1, 4, NULL, 'phase', 'future');
            INSERT INTO ducklake_column_tag VALUES
                (20, 200, 1, 3, 'semantic', 'old'),
                (20, 200, 3, 4, 'semantic', 'current'),
                (20, 200, 4, NULL, 'semantic', 'future');
            INSERT INTO ducklake_column VALUES
                (100, 1, 3, 20, 0, 'old_col', 'int32', NULL, NULL, true, NULL, 'literal', NULL),
                (200, 3, 4, 20, 0, 'current_col', 'int32', NULL, NULL, true, NULL, 'literal', NULL),
                (300, 4, NULL, 20, 0, 'future_col', 'int32', NULL, NULL, true, NULL, 'literal', NULL);
            INSERT INTO ducklake_data_file VALUES
                (60, 20, 1, 3, 0, 'old.parquet', true, 'parquet', 1, 1, NULL, NULL, NULL, NULL, NULL, NULL),
                (70, 20, 3, 4, 0, 'current.parquet', true, 'parquet', 1, 1, NULL, NULL, NULL, NULL, NULL, NULL),
                (80, 20, 4, NULL, 0, 'future.parquet', true, 'parquet', 1, 1, NULL, NULL, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_delete_file VALUES
                (90, 20, 1, 3, 60, 'old-delete.parquet', true, 'parquet', 1, 1, NULL, NULL, NULL),
                (100, 20, 3, 4, 70, 'current-delete.parquet', true, 'parquet', 1, 1, NULL, NULL, NULL),
                (110, 20, 4, NULL, 80, 'future-delete.parquet', true, 'parquet', 1, 1, NULL, NULL, NULL);
            INSERT INTO ducklake_partition_info VALUES
                (300, 20, 1, 3),
                (400, 20, 3, 4),
                (500, 20, 4, NULL);
            INSERT INTO ducklake_macro VALUES
                (2, 600, 'old_macro', 1, 3),
                (2, 700, 'current_macro', 3, 4),
                (2, 800, 'future_macro', 4, NULL);
            INSERT INTO ducklake_sort_info VALUES
                (900, 20, 1, 3),
                (1000, 20, 3, 4),
                (1100, 20, 4, NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = SnapshotId(3);
        let schemas = backend.schemas_at(snapshot).await.unwrap();
        assert_eq!(
            schemas.iter().map(|row| row.schema_id).collect::<Vec<_>>(),
            vec![SchemaId(2)]
        );

        let tables = backend.tables_at(SchemaId(2), snapshot).await.unwrap();
        assert_eq!(
            tables.iter().map(|row| row.table_id).collect::<Vec<_>>(),
            vec![TableId(20)]
        );

        let views = backend.views_at(SchemaId(2), snapshot).await.unwrap();
        assert_eq!(
            views.iter().map(|row| row.view_id).collect::<Vec<_>>(),
            vec![ViewId(21)]
        );

        let tags = backend.tags_at(snapshot).await.unwrap();
        assert_eq!(
            tags.iter()
                .map(|row| row.value.as_deref())
                .collect::<Vec<_>>(),
            vec![Some("current")]
        );

        let column_tags = backend.column_tags_at(TableId(20), snapshot).await.unwrap();
        assert_eq!(
            column_tags
                .iter()
                .map(|row| row.value.as_deref())
                .collect::<Vec<_>>(),
            vec![Some("current")]
        );

        let columns = backend.columns_at(TableId(20), snapshot).await.unwrap();
        assert_eq!(
            columns.iter().map(|row| row.column_id).collect::<Vec<_>>(),
            vec![ColumnId(200)]
        );

        let data_files = backend.data_files_at(TableId(20), snapshot).await.unwrap();
        assert_eq!(
            data_files
                .iter()
                .map(|row| row.data_file_id)
                .collect::<Vec<_>>(),
            vec![DataFileId(70)]
        );

        let delete_files = backend
            .delete_files_at(TableId(20), snapshot)
            .await
            .unwrap();
        assert_eq!(
            delete_files
                .iter()
                .map(|row| row.delete_file_id)
                .collect::<Vec<_>>(),
            vec![DeleteFileId(100)]
        );

        let partitions = backend
            .partition_info_at(TableId(20), snapshot)
            .await
            .unwrap();
        assert_eq!(
            partitions
                .iter()
                .map(|row| row.partition_id)
                .collect::<Vec<_>>(),
            vec![PartitionId(400)]
        );

        let macros = backend.macros_at(SchemaId(2), snapshot).await.unwrap();
        assert_eq!(
            macros.iter().map(|row| row.macro_id).collect::<Vec<_>>(),
            vec![MacroId(700)]
        );

        let sort_info = backend.sort_info_at(TableId(20), snapshot).await.unwrap();
        assert_eq!(
            sort_info.iter().map(|row| row.sort_id).collect::<Vec<_>>(),
            vec![SortId(1000)]
        );
    });
}

#[test]
fn parses_duckdb_inlined_temporal_special_values() {
    assert_eq!(
        parse_inlined_date32("infinity", "DuckDB", true).unwrap(),
        i32::MAX
    );
    assert_eq!(
        parse_inlined_date32("+infinity", "DuckDB", true).unwrap(),
        i32::MAX
    );
    assert_eq!(
        parse_inlined_date32("-infinity", "DuckDB", true).unwrap(),
        -i32::MAX
    );

    for unit in [
        TimeUnit::Second,
        TimeUnit::Millisecond,
        TimeUnit::Microsecond,
        TimeUnit::Nanosecond,
    ] {
        assert_eq!(
            parse_inlined_timestamp("infinity", unit, "DuckDB", true).unwrap(),
            i64::MAX
        );
        assert_eq!(
            parse_inlined_timestamp("+infinity", unit, "DuckDB", true).unwrap(),
            i64::MAX
        );
        assert_eq!(
            parse_inlined_timestamp("-infinity", unit, "DuckDB", true).unwrap(),
            -i64::MAX
        );
    }
}

#[test]
fn reads_extended_duckdb_inlined_scalar_types() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
        r#"
            CREATE TABLE ducklake_inlined_data_1_1(
                begin_snapshot BIGINT,
                end_snapshot BIGINT,
                row_id BIGINT,
                price VARCHAR,
                day_value DATE,
                time_value TIME,
                timestamp_value TIMESTAMP,
                bytes_value VARCHAR,
                uuid_value VARCHAR,
                ubig_value UBIGINT
            );
            INSERT INTO ducklake_inlined_data_1_1 VALUES (
                1,
                NULL,
                1,
                '123.45',
                DATE '2024-01-15',
                TIME '12:30:00.123456',
                TIMESTAMP '2024-01-15 12:30:00.123456',
                '00ff10',
                '550e8400-e29b-41d4-a716-446655440000',
                CAST('18446744073709551615' AS UBIGINT)
            );
            "#,
    )
    .unwrap();
    drop(conn);

    let schema = TableSchema {
        columns: vec![
            schema_field(
                1,
                "price",
                "decimal(8,2)",
                DuckLakeType::Decimal { width: 8, scale: 2 },
            ),
            schema_field(2, "day_value", "date", DuckLakeType::Date),
            schema_field(3, "time_value", "time", DuckLakeType::Time),
            schema_field(
                4,
                "timestamp_value",
                "timestamptz",
                DuckLakeType::TimestampTz,
            ),
            schema_field(5, "bytes_value", "blob", DuckLakeType::Blob),
            schema_field(6, "uuid_value", "uuid", DuckLakeType::Uuid),
            schema_field(7, "ubig_value", "uint64", DuckLakeType::UInt64),
        ],
        by_field_id: HashMap::new(),
    };
    let table = InlinedDataTableRef {
        table_id: TableId(1),
        table_name: "ducklake_inlined_data_1_1".into(),
        schema_version: 1,
    };
    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();

    runtime.block_on(async {
        let batches = backend
            .inlined_data(&table, &schema, SnapshotId(1))
            .await
            .unwrap();
        assert_eq!(batches.len(), 1);
        let batch = &batches[0];
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(
            batch.schema().field(0).data_type(),
            &DataType::Decimal128(8, 2)
        );
        assert_eq!(batch.schema().field(1).data_type(), &DataType::Date32);
        assert_eq!(
            batch.schema().field(2).data_type(),
            &DataType::Time64(TimeUnit::Microsecond)
        );
        assert_eq!(
            batch.schema().field(3).data_type(),
            &DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()))
        );
        assert_eq!(batch.schema().field(4).data_type(), &DataType::Binary);
        assert_eq!(
            batch.schema().field(5).data_type(),
            &DataType::FixedSizeBinary(16)
        );
        assert_eq!(batch.schema().field(6).data_type(), &DataType::UInt64);

        let price = batch
            .column(0)
            .as_any()
            .downcast_ref::<Decimal128Array>()
            .unwrap();
        assert_eq!(price.value(0), 12_345);

        let day = batch
            .column(1)
            .as_any()
            .downcast_ref::<Date32Array>()
            .unwrap();
        assert_eq!(
            day.value(0),
            parse_inlined_date32("2024-01-15", "DuckDB", true).unwrap()
        );

        let time = batch
            .column(2)
            .as_any()
            .downcast_ref::<Time64MicrosecondArray>()
            .unwrap();
        assert_eq!(time.value(0), 45_000_123_456);

        let timestamp = batch
            .column(3)
            .as_any()
            .downcast_ref::<TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(
            timestamp.value(0),
            parse_inlined_timestamp(
                "2024-01-15 12:30:00.123456+00",
                TimeUnit::Microsecond,
                "DuckDB",
                true
            )
            .unwrap()
        );

        let bytes = batch
            .column(4)
            .as_any()
            .downcast_ref::<BinaryArray>()
            .unwrap();
        assert_eq!(bytes.value(0), &[0x00, 0xff, 0x10]);

        let uuid = batch
            .column(5)
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(
            uuid.value(0),
            &[
                0x55, 0x0e, 0x84, 0x00, 0xe2, 0x9b, 0x41, 0xd4, 0xa7, 0x16, 0x44, 0x66, 0x55, 0x44,
                0x00, 0x00
            ]
        );

        let ubig = batch
            .column(6)
            .as_any()
            .downcast_ref::<UInt64Array>()
            .unwrap();
        assert_eq!(ubig.value(0), u64::MAX);
    });
}

#[test]
fn reads_native_nested_duckdb_inlined_temporal_values() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
        r#"
            CREATE TABLE ducklake_inlined_data_1_1(
                begin_snapshot BIGINT,
                end_snapshot BIGINT,
                row_id BIGINT,
                payload STRUCT(
                    day_value DATE,
                    time_value TIME,
                    timestamp_value TIMESTAMP,
                    uuid_value UUID
                ),
                time_values TIME[],
                timestamp_values TIMESTAMP[]
            );
            INSERT INTO ducklake_inlined_data_1_1 VALUES (
                1,
                NULL,
                1,
                struct_pack(
                    day_value := DATE '2024-01-15',
                    time_value := TIME '12:30:00.123456',
                    timestamp_value := TIMESTAMP '2024-01-15 12:30:00.123456',
                    uuid_value := UUID '550e8400-e29b-41d4-a716-446655440000'
                ),
                [TIME '01:02:03.000004', TIME '05:06:07.000008'],
                [
                    TIMESTAMP '2024-01-15 12:30:00.123456',
                    TIMESTAMP '2024-01-16 01:02:03.000004'
                ]
            );
            "#,
    )
    .unwrap();
    drop(conn);

    let schema = TableSchema {
        columns: vec![
            schema_field_with_children(
                1,
                "payload",
                "struct",
                DuckLakeType::Struct,
                vec![
                    schema_field(2, "day_value", "date", DuckLakeType::Date),
                    schema_field(3, "time_value", "time", DuckLakeType::Time),
                    schema_field(4, "timestamp_value", "timestamp", DuckLakeType::Timestamp),
                    schema_field(5, "uuid_value", "uuid", DuckLakeType::Uuid),
                ],
            ),
            schema_field_with_children(
                6,
                "time_values",
                "time[]",
                DuckLakeType::List,
                vec![schema_field(7, "list", "time", DuckLakeType::Time)],
            ),
            schema_field_with_children(
                8,
                "timestamp_values",
                "timestamp[]",
                DuckLakeType::List,
                vec![schema_field(
                    9,
                    "list",
                    "timestamp",
                    DuckLakeType::Timestamp,
                )],
            ),
        ],
        by_field_id: HashMap::new(),
    };
    let table = InlinedDataTableRef {
        table_id: TableId(1),
        table_name: "ducklake_inlined_data_1_1".into(),
        schema_version: 1,
    };
    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();

    runtime.block_on(async {
        let batches = backend
            .inlined_data(&table, &schema, SnapshotId(1))
            .await
            .unwrap();
        assert_eq!(batches.len(), 1);
        let batch = &batches[0];
        assert_eq!(batch.num_rows(), 1);

        let payload = batch
            .column(0)
            .as_any()
            .downcast_ref::<StructArray>()
            .unwrap();
        let day = payload
            .column(0)
            .as_any()
            .downcast_ref::<Date32Array>()
            .unwrap();
        assert_eq!(
            day.value(0),
            parse_inlined_date32("2024-01-15", "DuckDB", true).unwrap()
        );

        let time = payload
            .column(1)
            .as_any()
            .downcast_ref::<Time64MicrosecondArray>()
            .unwrap();
        assert_eq!(time.value(0), 45_000_123_456);

        let timestamp = payload
            .column(2)
            .as_any()
            .downcast_ref::<TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(
            timestamp.value(0),
            parse_inlined_timestamp(
                "2024-01-15 12:30:00.123456",
                TimeUnit::Microsecond,
                "DuckDB",
                true
            )
            .unwrap()
        );

        let uuid = payload
            .column(3)
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(
            uuid.value(0),
            &[
                0x55, 0x0e, 0x84, 0x00, 0xe2, 0x9b, 0x41, 0xd4, 0xa7, 0x16, 0x44, 0x66, 0x55, 0x44,
                0x00, 0x00
            ]
        );

        let time_values = batch
            .column(1)
            .as_any()
            .downcast_ref::<ListArray>()
            .unwrap();
        let time_values_child = time_values.value(0);
        let time_values = time_values_child
            .as_any()
            .downcast_ref::<Time64MicrosecondArray>()
            .unwrap();
        assert_eq!(time_values.value(0), 3_723_000_004);
        assert_eq!(time_values.value(1), 18_367_000_008);

        let timestamp_values = batch
            .column(2)
            .as_any()
            .downcast_ref::<ListArray>()
            .unwrap();
        let timestamp_values_child = timestamp_values.value(0);
        let timestamp_values = timestamp_values_child
            .as_any()
            .downcast_ref::<TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(
            timestamp_values.value(0),
            parse_inlined_timestamp(
                "2024-01-15 12:30:00.123456",
                TimeUnit::Microsecond,
                "DuckDB",
                true
            )
            .unwrap()
        );
        assert_eq!(
            timestamp_values.value(1),
            parse_inlined_timestamp(
                "2024-01-16 01:02:03.000004",
                TimeUnit::Microsecond,
                "DuckDB",
                true
            )
            .unwrap()
        );
    });
}

#[test]
fn plans_duckdb_file_catalog_scan() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_metadata(key VARCHAR NOT NULL, value VARCHAR NOT NULL, scope VARCHAR, scope_id BIGINT);
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_inlined_data_tables(table_id BIGINT, table_name VARCHAR, schema_version BIGINT);

            INSERT INTO ducklake_metadata VALUES ('version', '1.0', NULL, NULL), ('data_path', 'file:///tmp/lake', NULL, NULL);
            INSERT INTO ducklake_snapshot VALUES (1, '2026-01-01 00:00:00+00', 1, 10, 10);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'schema-dir', true);
            INSERT INTO ducklake_table VALUES (2, '550e8400-e29b-41d4-a716-446655440002', 1, NULL, 1, 'numbers', 'table-dir', true);
            INSERT INTO ducklake_column VALUES (3, 1, NULL, 2, 0, 'n', 'int64', NULL, NULL, true, NULL, NULL, NULL);
            INSERT INTO ducklake_data_file VALUES (4, 2, 1, NULL, 0, 'part-0.parquet', true, 'parquet', 2, 128, NULL, 0, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_file_column_stats VALUES (4, 2, 3, 16, 2, 0, '1', '2', false, NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let plan = crate::plan_table_scan(
            &backend,
            "main",
            "numbers",
            ducklake_catalog::SnapshotSelector::Latest,
        )
        .await
        .unwrap();
        assert_eq!(plan.snapshot, SnapshotId(1));
        assert_eq!(plan.schema.columns[0].name, "n");
        assert_eq!(plan.files.len(), 1);
        assert_eq!(
            plan.files[0].resolved_path,
            "file:///tmp/lake/schema-dir/table-dir/part-0.parquet"
        );
        assert_eq!(plan.stats.len(), 1);
    });
}

#[test]
fn duckdb_file_create_drop_table_commits_ducklake_metadata() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_inlined_data_tables(table_id BIGINT, table_name VARCHAR, schema_version BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'main/', true);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let created = backend
            .commit_create_table(CreateTableCommit {
                schema_name: "main".into(),
                table_name: "created".into(),
                columns: vec![
                    CreateTableColumn {
                        column_id: ColumnId(1),
                        column_order: 1,
                        column_name: "id".into(),
                        column_type: "int32".into(),
                        nulls_allowed: false,
                        parent_column: None,
                    },
                    CreateTableColumn {
                        column_id: ColumnId(2),
                        column_order: 2,
                        column_name: "name".into(),
                        column_type: "varchar".into(),
                        nulls_allowed: true,
                        parent_column: None,
                    },
                ],
                commit_message: "duckdb-file create table".into(),
            })
            .await
            .unwrap();
        assert_eq!(created, SnapshotId(2));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let created_table: (i64, i64, Option<i64>, i64, String, String, bool) = conn
            .query_row(
                "SELECT table_id, begin_snapshot, end_snapshot, schema_id, table_name, path, path_is_relative \
                 FROM ducklake_table WHERE table_name = 'created'",
                [],
                |row| {
                    Ok((
                        row.get(0)?,
                        row.get(1)?,
                        row.get(2)?,
                        row.get(3)?,
                        row.get(4)?,
                        row.get(5)?,
                        row.get(6)?,
                    ))
                },
            )
            .unwrap();
    assert_eq!(created_table.0, 10);
    assert_eq!(created_table.1, 2);
    assert_eq!(created_table.2, None);
    assert_eq!(created_table.3, 1);
    assert_eq!(created_table.4, "created");
    assert_eq!(created_table.5, "created/");
    assert!(created_table.6);
    let columns = conn
        .prepare(
            "SELECT column_id, column_order, column_name, column_type, nulls_allowed \
                 FROM ducklake_column WHERE table_id = 10 ORDER BY column_id",
        )
        .unwrap()
        .query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, String>(2)?,
                row.get::<_, String>(3)?,
                row.get::<_, bool>(4)?,
            ))
        })
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap();
    assert_eq!(
        columns,
        vec![
            (1, 1, "id".to_string(), "int32".to_string(), false),
            (2, 2, "name".to_string(), "varchar".to_string(), true),
        ]
    );
    let snapshot: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(snapshot, (2, 2, 11, 20));
    let changes: String = conn
        .query_row(
            "SELECT changes_made FROM ducklake_snapshot_changes WHERE snapshot_id = 2",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(changes, "created_table:\"main\".\"created\"");
    conn.execute(
            "INSERT INTO ducklake_data_file VALUES (20, 10, 2, NULL, 20, 'created.parquet', true, 'parquet', 1, 10, NULL, 0, NULL, NULL, NULL, NULL)",
            [],
        )
        .unwrap();
    conn.execute(
            "INSERT INTO ducklake_delete_file VALUES (21, 10, 2, NULL, 20, 'created-delete.parquet', true, 'deletion-vector', 1, 10, 0, NULL, NULL)",
            [],
        )
        .unwrap();
    conn.execute(
            "CREATE TABLE ducklake_inlined_data_10_2(begin_snapshot BIGINT, end_snapshot BIGINT, row_id BIGINT, id INTEGER, name VARCHAR)",
            [],
        )
        .unwrap();
    conn.execute(
        "INSERT INTO ducklake_inlined_data_10_2 VALUES (2, NULL, 0, 7, 'seven')",
        [],
    )
    .unwrap();
    conn.execute(
        "INSERT INTO ducklake_inlined_data_tables VALUES (10, 'ducklake_inlined_data_10_2', 2)",
        [],
    )
    .unwrap();
    drop(conn);

    runtime.block_on(async {
        let dropped = backend
            .commit_drop_table(DropTableCommit {
                schema_name: "main".into(),
                table_name: "created".into(),
                commit_message: "duckdb-file drop table".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped, SnapshotId(3));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let table_end: Option<i64> = conn
        .query_row(
            "SELECT end_snapshot FROM ducklake_table WHERE table_id = 10",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(table_end, Some(3));
    let active_columns: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_column WHERE table_id = 10 AND end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let active_files: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_data_file WHERE table_id = 10 AND end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let active_deletes: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM ducklake_delete_file WHERE table_id = 10 AND end_snapshot IS NULL",
                [],
                |row| row.get(0),
            )
            .unwrap();
    let active_inlined_rows: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_inlined_data_10_2 WHERE end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(active_columns, 0);
    assert_eq!(active_files, 0);
    assert_eq!(active_deletes, 0);
    assert_eq!(active_inlined_rows, 0);
    let dropped_changes: String = conn
        .query_row(
            "SELECT changes_made FROM ducklake_snapshot_changes WHERE snapshot_id = 3",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(dropped_changes, "dropped_table:10");
}

#[test]
fn duckdb_file_create_drop_schema_commits_ducklake_metadata() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name VARCHAR, dialect VARCHAR, sql VARCHAR, column_aliases VARCHAR);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_inlined_data_tables(table_id BIGINT, table_name VARCHAR, schema_version BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'main/', true);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let created_schema = backend
            .commit_create_schema(CreateSchemaCommit {
                schema_name: "analytics".into(),
                commit_message: "duckdb-file create schema".into(),
            })
            .await
            .unwrap();
        assert_eq!(created_schema, SnapshotId(2));

        let created_table = backend
            .commit_create_table(CreateTableCommit {
                schema_name: "analytics".into(),
                table_name: "events".into(),
                columns: vec![
                    CreateTableColumn {
                        column_id: ColumnId(1),
                        column_order: 1,
                        column_name: "id".into(),
                        column_type: "int32".into(),
                        nulls_allowed: false,
                        parent_column: None,
                    },
                    CreateTableColumn {
                        column_id: ColumnId(2),
                        column_order: 2,
                        column_name: "name".into(),
                        column_type: "varchar".into(),
                        nulls_allowed: true,
                        parent_column: None,
                    },
                ],
                commit_message: "duckdb-file create schema table".into(),
            })
            .await
            .unwrap();
        assert_eq!(created_table, SnapshotId(3));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let created_schema: (i64, i64, Option<i64>, String, String, bool) = conn
        .query_row(
            "SELECT schema_id, begin_snapshot, end_snapshot, schema_name, path, path_is_relative \
                 FROM ducklake_schema WHERE schema_name = 'analytics'",
            [],
            |row| {
                Ok((
                    row.get(0)?,
                    row.get(1)?,
                    row.get(2)?,
                    row.get(3)?,
                    row.get(4)?,
                    row.get(5)?,
                ))
            },
        )
        .unwrap();
    assert_eq!(
        created_schema,
        (10, 2, None, "analytics".into(), "analytics/".into(), true)
    );
    conn.execute(
            "INSERT INTO ducklake_data_file VALUES (20, 11, 3, NULL, 20, 'analytics/events.parquet', true, 'parquet', 1, 10, NULL, 0, NULL, NULL, NULL, NULL)",
            [],
        )
        .unwrap();
    conn.execute(
            "INSERT INTO ducklake_delete_file VALUES (21, 11, 3, NULL, 20, 'analytics/events-delete.parquet', true, 'parquet', 1, 10, 0, NULL, NULL)",
            [],
        )
        .unwrap();
    conn.execute(
            "CREATE TABLE ducklake_inlined_data_11_3(begin_snapshot BIGINT, end_snapshot BIGINT, row_id BIGINT, id INTEGER, name VARCHAR)",
            [],
        )
        .unwrap();
    conn.execute(
        "INSERT INTO ducklake_inlined_data_11_3 VALUES (3, NULL, 0, 7, 'seven')",
        [],
    )
    .unwrap();
    conn.execute(
        "INSERT INTO ducklake_inlined_data_tables VALUES (11, 'ducklake_inlined_data_11_3', 3)",
        [],
    )
    .unwrap();
    drop(conn);

    runtime.block_on(async {
        let created_view = backend
            .commit_create_view(CreateViewCommit {
                schema_name: "analytics".into(),
                view_name: "event_names".into(),
                dialect: "duckdb".into(),
                sql: "SELECT name FROM {DUCKLAKE_CATALOG}.analytics.events".into(),
                column_aliases: Some(String::new()),
                commit_message: "duckdb-file create schema view".into(),
            })
            .await
            .unwrap();
        assert_eq!(created_view, SnapshotId(4));

        let restricted = backend
            .commit_drop_schema(DropSchemaCommit {
                schema_name: "analytics".into(),
                cascade: false,
                commit_message: "duckdb-file restricted drop schema".into(),
            })
            .await
            .unwrap_err();
        assert!(format!("{restricted}").contains("DuckLake schema analytics is not empty"));

        let dropped = backend
            .commit_drop_schema(DropSchemaCommit {
                schema_name: "analytics".into(),
                cascade: true,
                commit_message: "duckdb-file cascade drop schema".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped, SnapshotId(5));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let schema_end: Option<i64> = conn
        .query_row(
            "SELECT end_snapshot FROM ducklake_schema WHERE schema_id = 10",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let table_end: Option<i64> = conn
        .query_row(
            "SELECT end_snapshot FROM ducklake_table WHERE table_id = 11",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let view_end: Option<i64> = conn
        .query_row(
            "SELECT end_snapshot FROM ducklake_view WHERE view_id = 12",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(schema_end, Some(5));
    assert_eq!(table_end, Some(5));
    assert_eq!(view_end, Some(5));
    for table in [
        "ducklake_column",
        "ducklake_data_file",
        "ducklake_delete_file",
    ] {
        let active: i64 = conn
            .query_row(
                &format!(
                    "SELECT COUNT(*) FROM {table} WHERE table_id = 11 AND end_snapshot IS NULL"
                ),
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(active, 0, "{table}");
    }
    let active_inlined_rows: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_inlined_data_11_3 WHERE end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(active_inlined_rows, 0);
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec![
            "created_schema:\"analytics\"".to_string(),
            "created_table:\"analytics\".\"events\"".to_string(),
            "created_view:\"analytics\".\"event_names\"".to_string(),
            "dropped_schema:10,dropped_table:11,dropped_view:12".to_string(),
        ]
    );
}

#[test]
fn duckdb_file_create_view_commits_ducklake_metadata() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name VARCHAR, dialect VARCHAR, sql VARCHAR, column_aliases VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'main/', true);
            INSERT INTO ducklake_table VALUES (2, '550e8400-e29b-41d4-a716-446655440002', 1, NULL, 1, 'numbers', 'numbers/', true);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let created = backend
            .commit_create_view(CreateViewCommit {
                schema_name: "main".into(),
                view_name: "v_numbers".into(),
                dialect: "duckdb".into(),
                sql: "SELECT id, name FROM {DUCKLAKE_CATALOG}.main.numbers".into(),
                column_aliases: Some("\"id_alias\",\"name_alias\"".into()),
                commit_message: "duckdb-file create view".into(),
            })
            .await
            .unwrap();
        assert_eq!(created, SnapshotId(2));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let view: (
            i64,
            i64,
            Option<i64>,
            i64,
            String,
            String,
            String,
            String,
        ) = conn
            .query_row(
                "SELECT view_id, begin_snapshot, end_snapshot, schema_id, view_name, dialect, sql, column_aliases \
                 FROM ducklake_view WHERE view_name = 'v_numbers'",
                [],
                |row| {
                    Ok((
                        row.get(0)?,
                        row.get(1)?,
                        row.get(2)?,
                        row.get(3)?,
                        row.get(4)?,
                        row.get(5)?,
                        row.get(6)?,
                        row.get(7)?,
                    ))
                },
            )
            .unwrap();
    assert_eq!(view.0, 10);
    assert_eq!(view.1, 2);
    assert_eq!(view.2, None);
    assert_eq!(view.3, 1);
    assert_eq!(view.4, "v_numbers");
    assert_eq!(view.5, "duckdb");
    assert_eq!(
        view.6,
        "SELECT id, name FROM {DUCKLAKE_CATALOG}.main.numbers"
    );
    assert_eq!(view.7, "\"id_alias\",\"name_alias\"");
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (2, 2, 11, 20));
    let changes: String = conn
        .query_row(
            "SELECT changes_made FROM ducklake_snapshot_changes WHERE snapshot_id = 2",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(changes, "created_view:\"main\".\"v_numbers\"");
}

#[test]
fn duckdb_file_drop_view_commits_ducklake_metadata() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name VARCHAR, dialect VARCHAR, sql VARCHAR, column_aliases VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'main/', true);
            INSERT INTO ducklake_view VALUES (9, '550e8400-e29b-41d4-a716-446655440002', 1, NULL, 1, 'v_numbers', 'duckdb', 'SELECT * FROM numbers', 'id,n');
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let dropped = backend
            .commit_drop_view(DropViewCommit {
                schema_name: "main".into(),
                view_name: "v_numbers".into(),
                commit_message: "duckdb-file drop view".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped, SnapshotId(2));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let view_end: Option<i64> = conn
        .query_row(
            "SELECT end_snapshot FROM ducklake_view WHERE view_id = 9",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(view_end, Some(2));
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (2, 2, 10, 20));
    let changes: String = conn
        .query_row(
            "SELECT changes_made FROM ducklake_snapshot_changes WHERE snapshot_id = 2",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(changes, "dropped_view:9");
}

#[test]
fn duckdb_file_table_and_view_lookup_is_case_insensitive_for_catalog_commits() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name VARCHAR, dialect VARCHAR, sql VARCHAR, column_aliases VARCHAR);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_metadata VALUES ('data_path', '/tmp/ducklake-data', NULL, NULL);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'main/', true);
            INSERT INTO ducklake_table VALUES (2, '550e8400-e29b-41d4-a716-446655440002', 1, NULL, 1, 'numbers', 'numbers/', true);
            INSERT INTO ducklake_view VALUES (3, '550e8400-e29b-41d4-a716-446655440003', 1, NULL, 1, 'v_numbers', 'duckdb', 'SELECT * FROM {DUCKLAKE_CATALOG}.numbers', NULL);
            INSERT INTO ducklake_column VALUES (1, 1, NULL, 2, 1, 'id', 'int32', NULL, 'NULL', true, NULL, 'literal', 'duckdb');
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let renamed = backend
            .commit_rename_table(RenameTableCommit {
                schema_name: "main".into(),
                old_table_name: "NUMBERS".into(),
                new_schema_name: None,
                new_table_name: "renamed".into(),
                commit_message: "duckdb-file rename by case-insensitive source".into(),
            })
            .await
            .unwrap();
        assert_eq!(renamed, SnapshotId(2));

        let dropped = backend
            .commit_drop_table(DropTableCommit {
                schema_name: "main".into(),
                table_name: "RENAMED".into(),
                commit_message: "duckdb-file drop by case-insensitive source".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped, SnapshotId(3));

        let dropped_view = backend
            .commit_drop_view(DropViewCommit {
                schema_name: "main".into(),
                view_name: "V_NUMBERS".into(),
                commit_message: "duckdb-file drop view by case-insensitive source".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped_view, SnapshotId(4));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let active_tables: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_table WHERE end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let active_views: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_view WHERE end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!((active_tables, active_views), (0, 0));
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec![
            "created_table:\"main\".\"renamed\"".to_string(),
            "dropped_table:2".to_string(),
            "dropped_view:3".to_string(),
        ]
    );
}

#[test]
fn duckdb_file_table_and_view_lookup_prefers_exact_match_and_rejects_ambiguous_case() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name VARCHAR, dialect VARCHAR, sql VARCHAR, column_aliases VARCHAR);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_metadata VALUES ('data_path', '/tmp/ducklake-data', NULL, NULL);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'main/', true);
            INSERT INTO ducklake_table VALUES (2, '550e8400-e29b-41d4-a716-446655440002', 1, NULL, 1, 'numbers', 'numbers/', true);
            INSERT INTO ducklake_table VALUES (4, '550e8400-e29b-41d4-a716-446655440004', 1, NULL, 1, 'NUMBERS', 'upper-numbers/', true);
            INSERT INTO ducklake_view VALUES (3, '550e8400-e29b-41d4-a716-446655440003', 1, NULL, 1, 'v_numbers', 'duckdb', 'SELECT * FROM {DUCKLAKE_CATALOG}.numbers', NULL);
            INSERT INTO ducklake_view VALUES (5, '550e8400-e29b-41d4-a716-446655440005', 1, NULL, 1, 'V_NUMBERS', 'duckdb', 'SELECT * FROM {DUCKLAKE_CATALOG}."NUMBERS"', NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let ambiguous_table = backend
            .commit_rename_table(RenameTableCommit {
                schema_name: "main".into(),
                old_table_name: "NuMbErS".into(),
                new_schema_name: None,
                new_table_name: "renamed".into(),
                commit_message: "duckdb-file ambiguous rename source".into(),
            })
            .await
            .unwrap_err();
        assert!(
            ambiguous_table
                .to_string()
                .contains("ambiguous DuckLake table identifier"),
            "{ambiguous_table}"
        );

        let ambiguous_view = backend
            .commit_drop_view(DropViewCommit {
                schema_name: "main".into(),
                view_name: "V_NuMbErS".into(),
                commit_message: "duckdb-file ambiguous drop view source".into(),
            })
            .await
            .unwrap_err();
        assert!(
            ambiguous_view
                .to_string()
                .contains("ambiguous DuckLake view identifier"),
            "{ambiguous_view}"
        );
        assert_eq!(
            backend
                .snapshots()
                .await
                .unwrap()
                .last()
                .unwrap()
                .snapshot_id,
            SnapshotId(1)
        );

        let dropped_table = backend
            .commit_drop_table(DropTableCommit {
                schema_name: "main".into(),
                table_name: "NUMBERS".into(),
                commit_message: "duckdb-file drop exact uppercase source".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped_table, SnapshotId(2));

        let dropped_view = backend
            .commit_drop_view(DropViewCommit {
                schema_name: "main".into(),
                view_name: "V_NUMBERS".into(),
                commit_message: "duckdb-file drop exact uppercase view source".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped_view, SnapshotId(3));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let active_tables: Vec<(i64, String)> = {
        let mut stmt = conn
            .prepare(
                "SELECT table_id, table_name FROM ducklake_table \
                     WHERE end_snapshot IS NULL ORDER BY table_id",
            )
            .unwrap();
        collect_rows(&mut stmt, [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap()
    };
    let active_views: Vec<(i64, String)> = {
        let mut stmt = conn
            .prepare(
                "SELECT view_id, view_name FROM ducklake_view \
                     WHERE end_snapshot IS NULL ORDER BY view_id",
            )
            .unwrap();
        collect_rows(&mut stmt, [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap()
    };
    assert_eq!(active_tables, vec![(2, "numbers".to_string())]);
    assert_eq!(active_views, vec![(3, "v_numbers".to_string())]);
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["dropped_table:4".to_string(), "dropped_view:5".to_string()]
    );
}

#[test]
fn duckdb_file_schema_lookup_prefers_exact_match_and_rejects_ambiguous_case() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_schema VALUES
                (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'main/', true),
                (42, '550e8400-e29b-41d4-a716-446655440042', 1, NULL, 'MAIN', 'MAIN/', true);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let ambiguous = backend
            .commit_create_table(CreateTableCommit {
                schema_name: "mAiN".into(),
                table_name: "ambiguous".into(),
                columns: vec![CreateTableColumn {
                    column_id: ColumnId(1),
                    column_order: 1,
                    column_name: "id".into(),
                    column_type: "int32".into(),
                    nulls_allowed: true,
                    parent_column: None,
                }],
                commit_message: "duckdb-file ambiguous schema lookup".into(),
            })
            .await
            .unwrap_err();
        assert!(
            ambiguous
                .to_string()
                .contains("ambiguous DuckLake schema identifier"),
            "{ambiguous}"
        );
        assert_eq!(duckdb_table_count(&catalog_path, "ducklake_snapshot"), 1);

        let created = backend
            .commit_create_table(CreateTableCommit {
                schema_name: "MAIN".into(),
                table_name: "exact_upper".into(),
                columns: vec![CreateTableColumn {
                    column_id: ColumnId(1),
                    column_order: 1,
                    column_name: "id".into(),
                    column_type: "int32".into(),
                    nulls_allowed: true,
                    parent_column: None,
                }],
                commit_message: "duckdb-file exact uppercase schema lookup".into(),
            })
            .await
            .unwrap();
        assert_eq!(created, SnapshotId(2));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let schema_id: i64 = conn
        .query_row(
            "SELECT schema_id FROM ducklake_table WHERE table_name = 'exact_upper'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(schema_id, 42);
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["created_table:\"MAIN\".\"exact_upper\"".to_string()]
    );
}

#[test]
fn duckdb_file_create_table_with_files_commits_one_snapshot() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_table_column_stats(table_id BIGINT, column_id BIGINT, contains_null BOOLEAN, contains_nan BOOLEAN, min_value VARCHAR, max_value VARCHAR, extra_stats VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', 'main/', true);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let created = backend
            .commit_create_table_with_files(CreateTableWithFilesCommit {
                create: CreateTableCommit {
                    schema_name: "main".into(),
                    table_name: "ctas".into(),
                    columns: vec![
                        CreateTableColumn {
                            column_id: ColumnId(1),
                            column_order: 1,
                            column_name: "id".into(),
                            column_type: "int32".into(),
                            nulls_allowed: false,
                            parent_column: None,
                        },
                        CreateTableColumn {
                            column_id: ColumnId(2),
                            column_order: 2,
                            column_name: "name".into(),
                            column_type: "varchar".into(),
                            nulls_allowed: true,
                            parent_column: None,
                        },
                    ],
                    commit_message: "duckdb-file ctas".into(),
                },
                table_id: TableId(10),
                expected_snapshot: SnapshotId(1),
                files: vec![AppendFile {
                    table_id: TableId(10),
                    expected_snapshot: SnapshotId(1),
                    data_file_id: DataFileId(20),
                    begin_snapshot: None,
                    path: "ctas.parquet".into(),
                    path_is_relative: true,
                    record_count: 2,
                    file_size_bytes: 25,
                    footer_size: None,
                    row_id_start: None,
                    explicit_row_ids: false,
                    column_stats: vec![FileColumnStats {
                        data_file_id: DataFileId(20),
                        table_id: TableId(10),
                        column_id: ColumnId(1),
                        column_size_bytes: Some(8),
                        value_count: Some(2),
                        null_count: Some(0),
                        min_value: Some("1".into()),
                        max_value: Some("2".into()),
                        contains_nan: Some(false),
                        extra_stats: None,
                    }],
                    variant_stats: Vec::new(),
                    partition_id: None,
                    partition_values: Vec::new(),
                    encryption_key: None,
                    mapping_id: None,
                    partial_max: None,
                }],
            })
            .await
            .unwrap();
        assert_eq!(created, SnapshotId(2));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let snapshot: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(snapshot, (2, 2, 11, 21));
    let data_file: (i64, i64, i64, String, i64, Option<i64>) = conn
        .query_row(
            "SELECT data_file_id, table_id, begin_snapshot, path, record_count, row_id_start \
                 FROM ducklake_data_file WHERE table_id = 10",
            [],
            |row| {
                Ok((
                    row.get(0)?,
                    row.get(1)?,
                    row.get(2)?,
                    row.get(3)?,
                    row.get(4)?,
                    row.get(5)?,
                ))
            },
        )
        .unwrap();
    assert_eq!(data_file, (20, 10, 2, "ctas.parquet".into(), 2, Some(0)));
    let table_stats: (i64, i64, i64) = conn
            .query_row(
                "SELECT record_count, next_row_id, file_size_bytes FROM ducklake_table_stats WHERE table_id = 10",
                [],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
    assert_eq!(table_stats, (2, 2, 25));
    let table_column_stats: (Option<bool>, Option<String>, Option<String>) = conn
            .query_row(
                "SELECT contains_null, min_value, max_value FROM ducklake_table_column_stats WHERE table_id = 10 AND column_id = 1",
                [],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
    assert_eq!(
        table_column_stats,
        (Some(false), Some("1".into()), Some("2".into()))
    );
    let changes: String = conn
        .query_row(
            "SELECT changes_made FROM ducklake_snapshot_changes WHERE snapshot_id = 2",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(changes, "created_table:\"main\".\"ctas\"");
}

#[test]
fn duckdb_file_append_honors_begin_snapshot_override() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (3, '2026-06-26 00:00:00+00', 1, 10, 10);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let append_snapshot = backend
            .commit_append_files(vec![
                AppendFile {
                    begin_snapshot: Some(SnapshotId(1)),
                    ..duckdb_append_file(10, SnapshotId(3))
                },
                duckdb_append_file(11, SnapshotId(3)),
            ])
            .await
            .unwrap();
        assert_eq!(append_snapshot, SnapshotId(4));

        let data_files = backend
            .data_files_at(TableId(1), SnapshotId(4))
            .await
            .unwrap();
        let begin = |data_file_id: i64| {
            data_files
                .iter()
                .find(|file| file.data_file_id == DataFileId(data_file_id))
                .unwrap()
                .validity
                .begin
        };
        assert_eq!(begin(10), SnapshotId(1));
        assert_eq!(begin(11), SnapshotId(4));
    });
}

#[test]
fn duckdb_file_write_commits_update_catalog_metadata() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value VARCHAR);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_inlined_data_tables(table_id BIGINT, table_name VARCHAR, schema_version BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 7, 50, 10);
            INSERT INTO ducklake_data_file VALUES (1, 1, 1, NULL, 1, 'duckdb-base.parquet', true, 'parquet', 5, 100, NULL, 0, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 5, 5, 100);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let append_snapshot = backend
            .commit_append_files(vec![duckdb_append_file(10, SnapshotId(1))])
            .await
            .unwrap();
        assert_eq!(append_snapshot, SnapshotId(2));

        let snapshots = backend.snapshots().await.unwrap();
        assert_eq!(snapshots.last().unwrap().schema_version, 7);
        assert_eq!(snapshots.last().unwrap().next_catalog_id, 50);
        assert_eq!(snapshots.last().unwrap().next_file_id, 11);

        let data_files = backend
            .data_files_at(TableId(1), SnapshotId(2))
            .await
            .unwrap();
        assert!(
            data_files
                .iter()
                .any(|file| file.data_file_id == DataFileId(10)
                    && file.path == "duckdb-append-10.parquet"
                    && file.partition_id == Some(PartitionId(3)))
        );
        let stats = backend.file_stats(TableId(1)).await.unwrap();
        assert_eq!(stats.len(), 1);
        assert_eq!(stats[0].column_id, ColumnId(2));
        assert_eq!(stats[0].min_value.as_deref(), Some("10"));
        assert_eq!(stats[0].max_value.as_deref(), Some("20"));
        let partition_values = backend.file_partition_values(TableId(1)).await.unwrap();
        assert_eq!(partition_values.len(), 1);
        assert_eq!(partition_values[0].partition_value.as_deref(), Some("4"));
        let changes = backend.snapshot_changes().await.unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(
            changes[0].changes_made.as_deref(),
            Some("inserted_into_table:1")
        );
        assert_eq!(
            changes[0].commit_message.as_deref(),
            Some("datafusion append")
        );
        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 7);
        assert_eq!(table_stats.next_row_id, 7);
        assert_eq!(table_stats.file_size_bytes, 125);

        let stale_append = backend
            .commit_append_files(vec![duckdb_append_file(11, SnapshotId(1))])
            .await
            .unwrap_err();
        assert_snapshot_conflict(&stale_append, "append", 1, 2);

        let delete_snapshot = backend
            .commit_delete_files(vec![duckdb_delete_file(11, SnapshotId(2), 1, 1, None)])
            .await
            .unwrap();
        assert_eq!(delete_snapshot, SnapshotId(3));
        let active_delete_files = backend
            .delete_files_at(TableId(1), SnapshotId(3))
            .await
            .unwrap();
        assert_eq!(
            active_delete_files
                .iter()
                .map(|file| file.delete_file_id)
                .collect::<Vec<_>>(),
            vec![DeleteFileId(11)]
        );
        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 6);
        assert_eq!(table_stats.next_row_id, 7);
        assert_eq!(table_stats.file_size_bytes, 125);

        let mixed_snapshot = backend
            .commit_table_changes(TableChangeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(3),
                append_files: vec![duckdb_append_file(12, SnapshotId(3))],
                delete_files: vec![duckdb_delete_file(
                    13,
                    SnapshotId(3),
                    1,
                    1,
                    Some(DeleteFileId(11)),
                )],
                inlined_file_deletions: Vec::new(),
                inlined_data_appends: Vec::new(),
                inlined_data_deletions: Vec::new(),
                flushed_inlined_data: Vec::new(),
                flushed_inlined_file_deletions: Vec::new(),
                changes_made: "updated_table:1".into(),
                commit_message: "datafusion update".into(),
            })
            .await
            .unwrap();
        assert_eq!(mixed_snapshot, SnapshotId(4));
        let snapshots = backend.snapshots().await.unwrap();
        assert_eq!(snapshots.last().unwrap().next_file_id, 14);
        let active_delete_files = backend
            .delete_files_at(TableId(1), SnapshotId(4))
            .await
            .unwrap();
        assert_eq!(
            active_delete_files
                .iter()
                .map(|file| file.delete_file_id)
                .collect::<Vec<_>>(),
            vec![DeleteFileId(13)]
        );
        let conn = Connection::open(&catalog_path).unwrap();
        let overwritten_end_snapshot: Option<i64> = conn
            .query_row(
                "SELECT end_snapshot FROM ducklake_delete_file WHERE delete_file_id = 11",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(overwritten_end_snapshot, Some(4));
        drop(conn);
        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 7);
        assert_eq!(table_stats.next_row_id, 9);
        assert_eq!(table_stats.file_size_bytes, 150);

        let inlined_snapshot = backend
            .commit_table_changes(TableChangeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(4),
                append_files: Vec::new(),
                delete_files: Vec::new(),
                inlined_file_deletions: vec![InlinedFileDeletionCommit {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(4),
                    data_file_id: DataFileId(1),
                    row_ids: vec![4],
                    rows_deleted: 1,
                }],
                inlined_data_appends: Vec::new(),
                inlined_data_deletions: Vec::new(),
                flushed_inlined_data: Vec::new(),
                flushed_inlined_file_deletions: Vec::new(),
                changes_made: "deleted_from_table:1".into(),
                commit_message: "datafusion delete".into(),
            })
            .await
            .unwrap();
        assert_eq!(inlined_snapshot, SnapshotId(5));
        let snapshots = backend.snapshots().await.unwrap();
        assert_eq!(snapshots.last().unwrap().next_file_id, 14);
        let inlined = backend
            .inlined_file_deletions_at(TableId(1), SnapshotId(5))
            .await
            .unwrap();
        assert_eq!(inlined.len(), 1);
        assert_eq!(inlined[0].data_file_id, DataFileId(1));
        assert_eq!(inlined[0].row_id, 4);
        assert_eq!(inlined[0].begin_snapshot, SnapshotId(5));
        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 6);
        assert_eq!(table_stats.next_row_id, 9);
        assert_eq!(table_stats.file_size_bytes, 150);

        let stale_delete = backend
            .commit_delete_files(vec![duckdb_delete_file(14, SnapshotId(4), 1, 1, None)])
            .await
            .unwrap_err();
        assert_snapshot_conflict(&stale_delete, "delete", 4, 5);

        let inlined_schema = TableSchema {
            columns: vec![schema_field(2, "value", "int32", DuckLakeType::Int32)],
            by_field_id: HashMap::new(),
        };
        let arrow_schema = Arc::new(Schema::new(vec![Field::new(
            "value",
            DataType::Int32,
            true,
        )]));
        let inlined_batch = RecordBatch::try_new(
            Arc::clone(&arrow_schema),
            vec![Arc::new(Int32Array::from(vec![10, 20]))],
        )
        .unwrap();
        let inlined_append_snapshot = backend
            .commit_append_inlined_data(InlinedDataAppendCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(5),
                schema_version: 7,
                schema: inlined_schema.clone(),
                batches: vec![inlined_batch],
                column_stats: Vec::new(),
                rows_appended: 2,
                changes_made: "inserted_into_table:1".into(),
                commit_message: "datafusion inlined append".into(),
            })
            .await
            .unwrap();
        assert_eq!(inlined_append_snapshot, SnapshotId(6));
        let snapshots = backend.snapshots().await.unwrap();
        assert_eq!(snapshots.last().unwrap().next_file_id, 15);
        let inlined_table = InlinedDataTableRef {
            table_id: TableId(1),
            table_name: "ducklake_inlined_data_1_7".into(),
            schema_version: 7,
        };
        let visible = backend
            .inlined_data_with_metadata(&inlined_table, &inlined_schema, SnapshotId(6))
            .await
            .unwrap();
        assert_eq!(visible.len(), 1);
        assert_eq!(visible[0].row_ids.as_deref(), Some([9, 10].as_slice()));
        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 8);
        assert_eq!(table_stats.next_row_id, 11);
        assert_eq!(table_stats.file_size_bytes, 150);

        let postimage_batch = RecordBatch::try_new(
            arrow_schema,
            vec![Arc::new(Int32Array::from(vec![110, 120]))],
        )
        .unwrap();
        let inlined_update_snapshot = backend
            .commit_table_changes(TableChangeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(6),
                append_files: Vec::new(),
                delete_files: Vec::new(),
                inlined_file_deletions: Vec::new(),
                inlined_data_appends: vec![InlinedDataAppend {
                    schema_version: 7,
                    schema: inlined_schema.clone(),
                    batches: vec![postimage_batch],
                    column_stats: Vec::new(),
                    row_ids: Some(vec![9, 10]),
                    rows_appended: 2,
                }],
                inlined_data_deletions: vec![InlinedDataDeletionCommit {
                    table_name: "ducklake_inlined_data_1_7".into(),
                    row_ids: vec![9, 10],
                    rows_deleted: 2,
                }],
                flushed_inlined_data: Vec::new(),
                flushed_inlined_file_deletions: Vec::new(),
                changes_made: "updated_table:1".into(),
                commit_message: "datafusion inlined update".into(),
            })
            .await
            .unwrap();
        assert_eq!(inlined_update_snapshot, SnapshotId(7));
        let snapshots = backend.snapshots().await.unwrap();
        assert_eq!(snapshots.last().unwrap().next_file_id, 16);
        let visible = backend
            .inlined_data_with_metadata(&inlined_table, &inlined_schema, SnapshotId(7))
            .await
            .unwrap();
        assert_eq!(visible.len(), 1);
        assert_eq!(visible[0].row_ids.as_deref(), Some([9, 10].as_slice()));
        let values = visible[0].batch.column(0);
        let values = values.as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(values.values(), &[110, 120]);
        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 8);
        assert_eq!(table_stats.next_row_id, 11);
        assert_eq!(table_stats.file_size_bytes, 150);
    });
}

#[test]
fn duckdb_file_table_changes_flush_inlined_data_rows() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_inlined_data_tables(table_id BIGINT, table_name VARCHAR, schema_version BIGINT);
            CREATE TABLE ducklake_inlined_data_1_1(row_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, value INTEGER);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 10);
            INSERT INTO ducklake_snapshot VALUES (2, '2026-06-26 00:01:00+00', 1, 10, 10);
            INSERT INTO ducklake_table_stats VALUES (1, 2, 7, 0);
            INSERT INTO ducklake_inlined_data_tables VALUES (1, 'ducklake_inlined_data_1_1', 1);
            INSERT INTO ducklake_inlined_data_1_1 VALUES (5, 1, NULL, 10);
            INSERT INTO ducklake_inlined_data_1_1 VALUES (6, 2, NULL, 20);
            INSERT INTO ducklake_inlined_data_1_1 VALUES (7, 1, 2, 30);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let flushed_file = AppendFile {
            table_id: TableId(1),
            expected_snapshot: SnapshotId(2),
            data_file_id: DataFileId(10),
            begin_snapshot: Some(SnapshotId(1)),
            path: "duckdb-flushed-inlined.parquet".into(),
            path_is_relative: true,
            record_count: 2,
            file_size_bytes: 40,
            footer_size: None,
            row_id_start: Some(5),
            explicit_row_ids: true,
            column_stats: Vec::new(),
            variant_stats: Vec::new(),
            partition_id: None,
            partition_values: Vec::new(),
            encryption_key: None,
            mapping_id: None,
            partial_max: Some(SnapshotId(2)),
        };

        let snapshot = backend
            .commit_table_changes(TableChangeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                append_files: vec![flushed_file],
                delete_files: Vec::new(),
                inlined_file_deletions: Vec::new(),
                inlined_data_appends: Vec::new(),
                inlined_data_deletions: Vec::new(),
                flushed_inlined_data: vec![FlushedInlinedDataCommit {
                    table_name: "ducklake_inlined_data_1_1".into(),
                    row_ids: vec![5, 6],
                    rows_flushed: 2,
                }],
                flushed_inlined_file_deletions: Vec::new(),
                changes_made: "flushed_inlined_data:1".into(),
                commit_message: "datafusion flush inlined data".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(3));

        let snapshots = backend.snapshots().await.unwrap();
        assert_eq!(snapshots.last().unwrap().next_file_id, 11);
        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 2);
        assert_eq!(table_stats.next_row_id, 7);
        assert_eq!(table_stats.file_size_bytes, 40);
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let remaining_inlined_rows: Vec<(i64, i64, Option<i64>, i32)> = {
        let mut stmt = conn
            .prepare(
                "SELECT row_id, begin_snapshot, end_snapshot, value \
                     FROM ducklake_inlined_data_1_1 ORDER BY row_id",
            )
            .unwrap();
        stmt.query_map([], |row| {
            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
        })
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap()
    };
    assert_eq!(remaining_inlined_rows, vec![(7, 1, Some(2), 30)]);

    let flushed_file: (i64, i64, Option<i64>, Option<i64>, String, i64, Option<i64>) = conn
            .query_row(
                "SELECT data_file_id, begin_snapshot, end_snapshot, partial_max, path, record_count, row_id_start \
                 FROM ducklake_data_file WHERE data_file_id = 10",
                [],
                |row| {
                    Ok((
                        row.get(0)?,
                        row.get(1)?,
                        row.get(2)?,
                        row.get(3)?,
                        row.get(4)?,
                        row.get(5)?,
                        row.get(6)?,
                    ))
                },
            )
            .unwrap();
    assert_eq!(
        flushed_file,
        (
            10,
            1,
            None,
            Some(2),
            "duckdb-flushed-inlined.parquet".into(),
            2,
            Some(5)
        )
    );
}

#[test]
fn duckdb_file_table_changes_flush_inlined_file_deletions() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_inlined_delete_1(file_id BIGINT, row_id BIGINT, begin_snapshot BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 10);
            INSERT INTO ducklake_snapshot VALUES (2, '2026-06-26 00:01:00+00', 1, 10, 10);
            INSERT INTO ducklake_data_file VALUES (0, 1, 1, NULL, 0, 'duckdb-base-0.parquet', true, 'parquet', 5, 100, NULL, 0, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_data_file VALUES (1, 1, 1, NULL, 1, 'duckdb-base-1.parquet', true, 'parquet', 5, 100, NULL, 5, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 8, 10, 200);
            INSERT INTO ducklake_inlined_delete_1 VALUES (0, 2, 2);
            INSERT INTO ducklake_inlined_delete_1 VALUES (1, 4, 1);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let flushed_delete_file = DeleteFileCommit {
            table_id: TableId(1),
            expected_snapshot: SnapshotId(2),
            delete_file_id: DeleteFileId(10),
            begin_snapshot: Some(SnapshotId(1)),
            data_file_id: DataFileId(0),
            path: "duckdb-flushed-inlined-deletes.parquet".into(),
            path_is_relative: true,
            format: DeleteFormat::Parquet,
            delete_count: 2,
            rows_deleted: 0,
            file_size_bytes: 24,
            footer_size: Some(8),
            partial_max: Some(SnapshotId(2)),
            overwritten_delete_file_id: None,
        };

        let snapshot = backend
            .commit_table_changes(TableChangeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                append_files: Vec::new(),
                delete_files: vec![flushed_delete_file],
                inlined_file_deletions: Vec::new(),
                inlined_data_appends: Vec::new(),
                inlined_data_deletions: Vec::new(),
                flushed_inlined_data: Vec::new(),
                flushed_inlined_file_deletions: vec![FlushedInlinedFileDeletionCommit {
                    rows: vec![
                        FlushedInlinedFileDeletion {
                            data_file_id: DataFileId(0),
                            row_id: 2,
                            begin_snapshot: SnapshotId(2),
                        },
                        FlushedInlinedFileDeletion {
                            data_file_id: DataFileId(1),
                            row_id: 4,
                            begin_snapshot: SnapshotId(1),
                        },
                    ],
                }],
                changes_made: "flushed_inlined_file_deletions:1".into(),
                commit_message: "datafusion flush inlined file deletions".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(3));

        let snapshots = backend.snapshots().await.unwrap();
        assert_eq!(snapshots.last().unwrap().next_file_id, 11);
        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 8);
        assert_eq!(table_stats.next_row_id, 10);
        assert_eq!(table_stats.file_size_bytes, 200);
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let remaining_inlined_deletes: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_inlined_delete_1",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(remaining_inlined_deletes, 0);

    let flushed_delete_file: (i64, i64, Option<i64>, i64, i64, i64, Option<i64>, String) = conn
            .query_row(
                "SELECT delete_file_id, begin_snapshot, end_snapshot, data_file_id, delete_count, file_size_bytes, partial_max, format \
                 FROM ducklake_delete_file WHERE delete_file_id = 10",
                [],
                |row| {
                    Ok((
                        row.get(0)?,
                        row.get(1)?,
                        row.get(2)?,
                        row.get(3)?,
                        row.get(4)?,
                        row.get(5)?,
                        row.get(6)?,
                        row.get(7)?,
                    ))
                },
            )
            .unwrap();
    assert_eq!(
        flushed_delete_file,
        (10, 1, None, 0, 2, 24, Some(2), "parquet".into())
    );
}

#[test]
fn duckdb_file_inlined_append_writes_reference_scalar_types() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_inlined_data_tables(table_id BIGINT, table_name VARCHAR, schema_version BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 10);
            INSERT INTO ducklake_table_stats VALUES (1, 0, 0, 0);
            "#,
        )
        .unwrap();
    drop(conn);

    let schema = TableSchema {
        columns: vec![
            schema_field(1, "tiny", "int8", DuckLakeType::Int8),
            schema_field(2, "small", "int16", DuckLakeType::Int16),
            schema_field(3, "utiny", "uint8", DuckLakeType::UInt8),
            schema_field(4, "usmall", "uint16", DuckLakeType::UInt16),
            schema_field(5, "uint", "uint32", DuckLakeType::UInt32),
            schema_field(6, "real32", "float", DuckLakeType::Float32),
            schema_field(7, "real64", "double", DuckLakeType::Float64),
            schema_field(8, "payload", "blob", DuckLakeType::Blob),
            schema_field(9, "uuid_value", "uuid", DuckLakeType::Uuid),
            schema_field(10, "ubig", "uint64", DuckLakeType::UInt64),
            schema_field(11, "huge", "int128", DuckLakeType::Int128),
            schema_field(12, "uhuge", "uint128", DuckLakeType::UInt128),
            schema_field(
                13,
                "decimal_value",
                "decimal(8,2)",
                DuckLakeType::Decimal { width: 8, scale: 2 },
            ),
            schema_field(
                14,
                "wide_decimal",
                "decimal(38,2)",
                DuckLakeType::Decimal {
                    width: 38,
                    scale: 2,
                },
            ),
            schema_field(15, "day_value", "date", DuckLakeType::Date),
            schema_field(16, "time_value", "time", DuckLakeType::Time),
            schema_field(17, "time_ns_value", "time_ns", DuckLakeType::TimeNs),
            schema_field(18, "timestamp_s", "timestamp_s", DuckLakeType::TimestampS),
            schema_field(
                19,
                "timestamp_ms",
                "timestamp_ms",
                DuckLakeType::TimestampMs,
            ),
            schema_field(20, "timestamp_value", "timestamp", DuckLakeType::Timestamp),
            schema_field(
                21,
                "timestamp_ns",
                "timestamp_ns",
                DuckLakeType::TimestampNs,
            ),
            schema_field(22, "timestamp_tz", "timestamptz", DuckLakeType::TimestampTz),
            schema_field(23, "timetz_value", "timetz", DuckLakeType::TimeTz),
            schema_field(24, "interval_value", "interval", DuckLakeType::Interval),
        ],
        by_field_id: HashMap::new(),
    };
    let arrow_schema = Arc::new(Schema::new(vec![
        Field::new("tiny", DataType::Int8, true),
        Field::new("small", DataType::Int16, true),
        Field::new("utiny", DataType::UInt8, true),
        Field::new("usmall", DataType::UInt16, true),
        Field::new("uint", DataType::UInt32, true),
        Field::new("real32", DataType::Float32, true),
        Field::new("real64", DataType::Float64, true),
        Field::new("payload", DataType::Binary, true),
        Field::new("uuid_value", DataType::FixedSizeBinary(16), true),
        Field::new("ubig", DataType::UInt64, true),
        Field::new("huge", DataType::Decimal256(39, 0), true),
        Field::new("uhuge", DataType::Decimal256(39, 0), true),
        Field::new("decimal_value", DataType::Decimal128(8, 2), true),
        Field::new("wide_decimal", DataType::Decimal128(38, 2), true),
        Field::new("day_value", DataType::Date32, true),
        Field::new("time_value", DataType::Time64(TimeUnit::Microsecond), true),
        Field::new(
            "time_ns_value",
            DataType::Time64(TimeUnit::Nanosecond),
            true,
        ),
        Field::new(
            "timestamp_s",
            DataType::Timestamp(TimeUnit::Second, None),
            true,
        ),
        Field::new(
            "timestamp_ms",
            DataType::Timestamp(TimeUnit::Millisecond, None),
            true,
        ),
        Field::new(
            "timestamp_value",
            DataType::Timestamp(TimeUnit::Microsecond, None),
            true,
        ),
        Field::new(
            "timestamp_ns",
            DataType::Timestamp(TimeUnit::Nanosecond, None),
            true,
        ),
        Field::new(
            "timestamp_tz",
            DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
            true,
        ),
        Field::new("timetz_value", DataType::Utf8, true),
        Field::new(
            "interval_value",
            DataType::Interval(IntervalUnit::MonthDayNano),
            true,
        ),
    ]));
    let uuid_bytes = [
        0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee,
        0xff,
    ];
    let huge_value = -123_456_789_012_345_678_901_234_567_890_i128;
    let uhuge_value = "340282366920938463463374607431768211455"
        .parse::<i256>()
        .unwrap();
    let wide_decimal_value = 123_456_789_012_345_678_901_234_567_890_123_456_i128;
    let day_value = parse_inlined_date32("2024-01-15", "DuckDB", true).unwrap();
    let time_value = 45_296_123_456_i64;
    let time_ns_value = 45_296_123_456_789_i64;
    let timestamp_s =
        parse_inlined_timestamp("2024-01-15 12:30:00+00", TimeUnit::Second, "DuckDB", true)
            .unwrap();
    let timestamp_ms = parse_inlined_timestamp(
        "2024-01-15 12:30:00.123+00",
        TimeUnit::Millisecond,
        "DuckDB",
        true,
    )
    .unwrap();
    let timestamp_value = parse_inlined_timestamp(
        "2024-01-15 12:30:00.123456+00",
        TimeUnit::Microsecond,
        "DuckDB",
        true,
    )
    .unwrap();
    let timestamp_ns = parse_inlined_timestamp(
        "2024-01-15 12:30:00.123456789+00",
        TimeUnit::Nanosecond,
        "DuckDB",
        true,
    )
    .unwrap();
    let timestamp_tz = parse_inlined_timestamp(
        "2024-01-15 12:30:00.123456+00",
        TimeUnit::Microsecond,
        "DuckDB",
        true,
    )
    .unwrap();
    let interval_value = IntervalMonthDayNano::new(14, 3, 4_005_000);
    let payload_bytes = vec![0x00_u8, 0xff, 0x10];
    let payload_values = vec![Some(payload_bytes.as_slice()), None];
    let uuid_values = vec![Some(uuid_bytes.as_slice()), None];
    let batch = RecordBatch::try_new(
        arrow_schema,
        vec![
            Arc::new(Int8Array::from(vec![Some(-8), None])),
            Arc::new(Int16Array::from(vec![Some(-1_600), None])),
            Arc::new(UInt8Array::from(vec![Some(250), None])),
            Arc::new(UInt16Array::from(vec![Some(60_000), None])),
            Arc::new(UInt32Array::from(vec![Some(4_000_000_000), None])),
            Arc::new(Float32Array::from(vec![Some(1.5), None])),
            Arc::new(Float64Array::from(vec![Some(-2.25), None])),
            Arc::new(BinaryArray::from(payload_values)),
            Arc::new(
                FixedSizeBinaryArray::try_from_sparse_iter_with_size(uuid_values.into_iter(), 16)
                    .unwrap(),
            ),
            Arc::new(UInt64Array::from(vec![Some(u64::MAX), None])),
            Arc::new(
                Decimal256Array::from(vec![Some(i256::from_i128(huge_value)), None])
                    .with_precision_and_scale(39, 0)
                    .unwrap(),
            ),
            Arc::new(
                Decimal256Array::from(vec![Some(uhuge_value), None])
                    .with_precision_and_scale(39, 0)
                    .unwrap(),
            ),
            Arc::new(
                Decimal128Array::from(vec![Some(12_345), None])
                    .with_precision_and_scale(8, 2)
                    .unwrap(),
            ),
            Arc::new(
                Decimal128Array::from(vec![Some(wide_decimal_value), None])
                    .with_precision_and_scale(38, 2)
                    .unwrap(),
            ),
            Arc::new(Date32Array::from(vec![Some(day_value), None])),
            Arc::new(Time64MicrosecondArray::from(vec![Some(time_value), None])),
            Arc::new(Time64NanosecondArray::from(vec![Some(time_ns_value), None])),
            Arc::new(TimestampSecondArray::from(vec![Some(timestamp_s), None])),
            Arc::new(TimestampMillisecondArray::from(vec![
                Some(timestamp_ms),
                None,
            ])),
            Arc::new(TimestampMicrosecondArray::from(vec![
                Some(timestamp_value),
                None,
            ])),
            Arc::new(TimestampNanosecondArray::from(vec![
                Some(timestamp_ns),
                None,
            ])),
            Arc::new(
                TimestampMicrosecondArray::from(vec![Some(timestamp_tz), None])
                    .with_timezone("UTC"),
            ),
            Arc::new(StringArray::from(vec![Some("12:34:56+02"), None])),
            Arc::new(IntervalMonthDayNanoArray::from(vec![
                Some(interval_value),
                None,
            ])),
        ],
    )
    .unwrap();

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = backend
            .commit_append_inlined_data(InlinedDataAppendCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                schema_version: 1,
                schema: schema.clone(),
                batches: vec![batch],
                column_stats: Vec::new(),
                rows_appended: 2,
                changes_made: "inserted_into_table:1".into(),
                commit_message: "duckdb-file numeric inlined append".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(2));

        let table = InlinedDataTableRef {
            table_id: TableId(1),
            table_name: "ducklake_inlined_data_1_1".into(),
            schema_version: 1,
        };
        let visible = backend
            .inlined_data_with_metadata(&table, &schema, SnapshotId(2))
            .await
            .unwrap();
        assert_eq!(visible.len(), 1);
        assert_eq!(visible[0].row_ids.as_deref(), Some([0, 1].as_slice()));
        assert_eq!(
            visible[0].begin_snapshots.as_deref(),
            Some([SnapshotId(2), SnapshotId(2)].as_slice())
        );

        let batch = &visible[0].batch;
        let tiny = batch
            .column(0)
            .as_any()
            .downcast_ref::<Int8Array>()
            .unwrap();
        assert_eq!(tiny.value(0), -8);
        assert!(tiny.is_null(1));
        let small = batch
            .column(1)
            .as_any()
            .downcast_ref::<Int16Array>()
            .unwrap();
        assert_eq!(small.value(0), -1_600);
        assert!(small.is_null(1));
        let utiny = batch
            .column(2)
            .as_any()
            .downcast_ref::<UInt8Array>()
            .unwrap();
        assert_eq!(utiny.value(0), 250);
        assert!(utiny.is_null(1));
        let usmall = batch
            .column(3)
            .as_any()
            .downcast_ref::<UInt16Array>()
            .unwrap();
        assert_eq!(usmall.value(0), 60_000);
        assert!(usmall.is_null(1));
        let uint = batch
            .column(4)
            .as_any()
            .downcast_ref::<UInt32Array>()
            .unwrap();
        assert_eq!(uint.value(0), 4_000_000_000);
        assert!(uint.is_null(1));
        let real32 = batch
            .column(5)
            .as_any()
            .downcast_ref::<Float32Array>()
            .unwrap();
        assert_eq!(real32.value(0), 1.5);
        assert!(real32.is_null(1));
        let real64 = batch
            .column(6)
            .as_any()
            .downcast_ref::<Float64Array>()
            .unwrap();
        assert_eq!(real64.value(0), -2.25);
        assert!(real64.is_null(1));
        let payload = batch
            .column(7)
            .as_any()
            .downcast_ref::<BinaryArray>()
            .unwrap();
        assert_eq!(payload.value(0), &[0x00, 0xff, 0x10]);
        assert!(payload.is_null(1));
        let uuid_value = batch
            .column(8)
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(uuid_value.value(0), uuid_bytes.as_slice());
        assert!(uuid_value.is_null(1));
        let ubig = batch
            .column(9)
            .as_any()
            .downcast_ref::<UInt64Array>()
            .unwrap();
        assert_eq!(ubig.value(0), u64::MAX);
        assert!(ubig.is_null(1));
        let huge = batch
            .column(10)
            .as_any()
            .downcast_ref::<Decimal256Array>()
            .unwrap();
        assert_eq!(huge.value(0), i256::from_i128(huge_value));
        assert!(huge.is_null(1));
        let uhuge = batch
            .column(11)
            .as_any()
            .downcast_ref::<Decimal256Array>()
            .unwrap();
        assert_eq!(uhuge.value(0), uhuge_value);
        assert!(uhuge.is_null(1));
        let decimal_value = batch
            .column(12)
            .as_any()
            .downcast_ref::<Decimal128Array>()
            .unwrap();
        assert_eq!(decimal_value.value(0), 12_345);
        assert!(decimal_value.is_null(1));
        let wide_decimal = batch
            .column(13)
            .as_any()
            .downcast_ref::<Decimal128Array>()
            .unwrap();
        assert_eq!(wide_decimal.value(0), wide_decimal_value);
        assert!(wide_decimal.is_null(1));
        let day = batch
            .column(14)
            .as_any()
            .downcast_ref::<Date32Array>()
            .unwrap();
        assert_eq!(day.value(0), day_value);
        assert!(day.is_null(1));
        let time = batch
            .column(15)
            .as_any()
            .downcast_ref::<Time64MicrosecondArray>()
            .unwrap();
        assert_eq!(time.value(0), time_value);
        assert!(time.is_null(1));
        let time_ns = batch
            .column(16)
            .as_any()
            .downcast_ref::<Time64NanosecondArray>()
            .unwrap();
        assert_eq!(time_ns.value(0), time_ns_value);
        assert!(time_ns.is_null(1));
        let ts_s = batch
            .column(17)
            .as_any()
            .downcast_ref::<TimestampSecondArray>()
            .unwrap();
        assert_eq!(ts_s.value(0), timestamp_s);
        assert!(ts_s.is_null(1));
        let ts_ms = batch
            .column(18)
            .as_any()
            .downcast_ref::<TimestampMillisecondArray>()
            .unwrap();
        assert_eq!(ts_ms.value(0), timestamp_ms);
        assert!(ts_ms.is_null(1));
        let ts = batch
            .column(19)
            .as_any()
            .downcast_ref::<TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(ts.value(0), timestamp_value);
        assert!(ts.is_null(1));
        let ts_ns = batch
            .column(20)
            .as_any()
            .downcast_ref::<TimestampNanosecondArray>()
            .unwrap();
        assert_eq!(ts_ns.value(0), timestamp_ns);
        assert!(ts_ns.is_null(1));
        let ts_tz = batch
            .column(21)
            .as_any()
            .downcast_ref::<TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(ts_tz.value(0), timestamp_tz);
        assert!(ts_tz.is_null(1));
        let timetz = batch
            .column(22)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert_eq!(timetz.value(0), "12:34:56+02");
        assert!(timetz.is_null(1));
        let interval = batch
            .column(23)
            .as_any()
            .downcast_ref::<IntervalMonthDayNanoArray>()
            .unwrap();
        assert_eq!(interval.value(0), interval_value);
        assert!(interval.is_null(1));

        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 2);
        assert_eq!(table_stats.next_row_id, 2);
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let mut stmt = conn
        .prepare(
            "SELECT column_name, data_type \
                 FROM information_schema.columns \
                 WHERE table_name = 'ducklake_inlined_data_1_1' \
                 ORDER BY ordinal_position",
        )
        .unwrap();
    let column_types = stmt
        .query_map([], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        })
        .unwrap()
        .collect::<std::result::Result<HashMap<_, _>, _>>()
        .unwrap();
    assert_eq!(column_types["tiny"], "TINYINT");
    assert_eq!(column_types["small"], "SMALLINT");
    assert_eq!(column_types["utiny"], "UTINYINT");
    assert_eq!(column_types["usmall"], "USMALLINT");
    assert_eq!(column_types["uint"], "UINTEGER");
    assert_eq!(column_types["real32"], "FLOAT");
    assert_eq!(column_types["real64"], "DOUBLE");
    assert_eq!(column_types["payload"], "BLOB");
    assert_eq!(column_types["uuid_value"], "UUID");
    assert_eq!(column_types["ubig"], "UBIGINT");
    assert_eq!(column_types["huge"], "HUGEINT");
    assert_eq!(column_types["uhuge"], "UHUGEINT");
    assert_eq!(column_types["decimal_value"], "DECIMAL(8,2)");
    assert_eq!(column_types["wide_decimal"], "DECIMAL(38,2)");
    assert_eq!(column_types["day_value"], "DATE");
    assert_eq!(column_types["time_value"], "TIME");
    assert_eq!(column_types["time_ns_value"], "TIME_NS");
    assert_eq!(column_types["timestamp_value"], "TIMESTAMP");
    assert_eq!(column_types["timestamp_ns"], "TIMESTAMP_NS");
    assert_eq!(column_types["interval_value"], "INTERVAL");
}

#[test]
fn duckdb_file_truncate_table_closes_active_files_without_allocating_file_ids() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, format VARCHAR, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, partial_max BIGINT);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_inlined_data_1_7(row_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, value BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 7, 50, 20);
            INSERT INTO ducklake_data_file VALUES
                (10, 1, 1, NULL, 10, 'active-a.parquet', true, 'parquet', 5, 100, NULL, 0, NULL, NULL, NULL, NULL),
                (11, 1, 1, NULL, 11, 'active-b.parquet', true, 'parquet', 7, 140, NULL, 5, NULL, NULL, NULL, NULL),
                (12, 2, 1, NULL, 12, 'other-table.parquet', true, 'parquet', 3, 60, NULL, 0, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_delete_file VALUES
                (15, 1, 1, NULL, 10, 'active-delete.parquet', true, 'positional', 2, 30, 8, NULL, NULL),
                (16, 2, 1, NULL, 12, 'other-delete.parquet', true, 'positional', 1, 20, 8, NULL, NULL);
            INSERT INTO ducklake_table_stats VALUES
                (1, 14, 14, 240),
                (2, 3, 3, 60);
            INSERT INTO ducklake_inlined_data_1_7 VALUES
                (12, 1, NULL, 100),
                (13, 1, NULL, 101),
                (14, 1, 1, 102);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = backend
            .commit_truncate_table(TruncateTableCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                rows_deleted: 14,
                inlined_data_deletions: vec![InlinedDataDeletionCommit {
                    table_name: "ducklake_inlined_data_1_7".into(),
                    row_ids: vec![12, 13],
                    rows_deleted: 2,
                }],
                changes_made: "deleted_from_table:1".into(),
                commit_message: "datafusion truncate".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(2));

        let snapshots = backend.snapshots().await.unwrap();
        let latest = snapshots.last().unwrap();
        assert_eq!(latest.snapshot_id, SnapshotId(2));
        assert_eq!(latest.schema_version, 7);
        assert_eq!(latest.next_catalog_id, 50);
        assert_eq!(latest.next_file_id, 20);

        assert_eq!(
            backend
                .data_files_at(TableId(1), SnapshotId(2))
                .await
                .unwrap()
                .len(),
            0
        );
        assert_eq!(
            backend
                .delete_files_at(TableId(1), SnapshotId(2))
                .await
                .unwrap()
                .len(),
            0
        );
        assert_eq!(
            backend
                .data_files_at(TableId(1), SnapshotId(1))
                .await
                .unwrap()
                .len(),
            2
        );
        assert_eq!(
            backend
                .data_files_at(TableId(2), SnapshotId(2))
                .await
                .unwrap()
                .len(),
            1
        );

        let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
        assert_eq!(table_stats.record_count, 0);
        assert_eq!(table_stats.next_row_id, 14);
        assert_eq!(table_stats.file_size_bytes, 0);
        let other_stats = backend.table_stats(TableId(2)).await.unwrap().unwrap();
        assert_eq!(other_stats.record_count, 3);
        assert_eq!(other_stats.file_size_bytes, 60);
        let conn = Connection::open(&catalog_path).unwrap();
        let closed_inlined_rows: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM ducklake_inlined_data_1_7 \
                     WHERE row_id IN (12, 13) AND end_snapshot = 2",
                [],
                |row| row.get(0),
            )
            .unwrap();
        let visible_inlined_latest: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM ducklake_inlined_data_1_7 \
                     WHERE 2 >= begin_snapshot AND (2 < end_snapshot OR end_snapshot IS NULL)",
                [],
                |row| row.get(0),
            )
            .unwrap();
        let visible_inlined_snapshot_one: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM ducklake_inlined_data_1_7 \
                     WHERE 1 >= begin_snapshot AND (1 < end_snapshot OR end_snapshot IS NULL)",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(closed_inlined_rows, 2);
        assert_eq!(visible_inlined_latest, 0);
        assert_eq!(visible_inlined_snapshot_one, 2);
        drop(conn);

        let changes = backend.snapshot_changes().await.unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(
            changes[0].changes_made.as_deref(),
            Some("deleted_from_table:1")
        );
        assert_eq!(
            changes[0].commit_message.as_deref(),
            Some("datafusion truncate")
        );

        let stale = backend
            .commit_truncate_table(TruncateTableCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                rows_deleted: 0,
                inlined_data_deletions: Vec::new(),
                changes_made: "deleted_from_table:1".into(),
                commit_message: "datafusion truncate".into(),
            })
            .await
            .unwrap_err();
        assert_snapshot_conflict(&stale, "truncate", 1, 2);
    });
}

#[test]
fn duckdb_file_append_updates_table_column_stats() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_dialect VARCHAR, default_value_type VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
            CREATE TABLE ducklake_table_column_stats(table_id BIGINT, column_id BIGINT, contains_null BOOLEAN, contains_nan BOOLEAN, min_value VARCHAR, max_value VARCHAR, extra_stats VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 10);
            INSERT INTO ducklake_column VALUES (2, 1, NULL, 1, 0, 'id', 'int32', NULL, NULL, true, NULL, 'literal', NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 3, 3, 100);
            INSERT INTO ducklake_table_column_stats VALUES (1, 2, false, false, '10', '20', NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        backend
            .commit_append_files(vec![
                duckdb_append_file_with_stats(10, SnapshotId(1), 2, 0, "5", "12"),
                duckdb_append_file_with_stats(11, SnapshotId(1), 2, 1, "18", "30"),
            ])
            .await
            .unwrap();
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let stats: (Option<bool>, Option<bool>, Option<String>, Option<String>) = conn
        .query_row(
            "SELECT contains_null, contains_nan, min_value, max_value \
                 FROM ducklake_table_column_stats WHERE table_id = 1 AND column_id = 2",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(
        stats,
        (
            Some(true),
            Some(false),
            Some("5".to_string()),
            Some("30".to_string())
        )
    );
}

#[test]
fn duckdb_file_add_column_default_commits_column_and_table_stats() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name VARCHAR, path VARCHAR, path_is_relative BOOLEAN);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_table_column_stats(table_id BIGINT, column_id BIGINT, contains_null BOOLEAN, contains_nan BOOLEAN, min_value VARCHAR, max_value VARCHAR, extra_stats VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 3, 10);
            INSERT INTO ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 1, 'numbers', 'numbers/', true);
            INSERT INTO ducklake_column VALUES (1, 1, NULL, 1, 1, 'id', 'int32', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (2, 1, NULL, 1, 2, 'name', 'varchar', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_table_column_stats VALUES (1, 1, false, false, '1', '3', NULL);
            INSERT INTO ducklake_table_column_stats VALUES (1, 2, false, NULL, 'a', 'c', NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = backend
            .commit_add_columns(AddColumnCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                columns: vec![AddColumn {
                    column_id: ColumnId(3),
                    column_order: 3,
                    column_name: "note".into(),
                    column_type: "varchar".into(),
                    initial_default: Some("hello".into()),
                    default_value: Some("hello".into()),
                    nulls_allowed: true,
                    parent_column: None,
                    default_value_type: Some("literal".into()),
                    default_value_dialect: Some("duckdb".into()),
                }],
                table_column_stats: vec![TableColumnStats {
                    table_id: TableId(1),
                    column_id: ColumnId(3),
                    contains_null: Some(false),
                    contains_nan: None,
                    min_value: Some("hello".into()),
                    max_value: Some("hello".into()),
                    extra_stats: None,
                }],
                commit_message: "duckdb add defaulted column".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(2));
    });

    #[derive(Debug, PartialEq, Eq)]
    struct ColumnRow {
        column_id: i64,
        begin_snapshot: i64,
        end_snapshot: Option<i64>,
        column_order: i64,
        column_name: String,
        column_type: String,
        initial_default: String,
        default_value: String,
        nulls_allowed: bool,
        parent_column: Option<i64>,
        default_value_type: String,
        default_value_dialect: String,
    }

    let conn = Connection::open(&catalog_path).unwrap();
    let column = conn
            .query_row(
                "SELECT column_id, begin_snapshot, end_snapshot, column_order, column_name, column_type, \
                        initial_default, default_value, nulls_allowed, parent_column, default_value_type, \
                        default_value_dialect \
                 FROM ducklake_column WHERE column_id = 3",
                [],
                |row| {
                    Ok(ColumnRow {
                        column_id: row.get(0)?,
                        begin_snapshot: row.get(1)?,
                        end_snapshot: row.get(2)?,
                        column_order: row.get(3)?,
                        column_name: row.get(4)?,
                        column_type: row.get(5)?,
                        initial_default: row.get(6)?,
                        default_value: row.get(7)?,
                        nulls_allowed: row.get(8)?,
                        parent_column: row.get(9)?,
                        default_value_type: row.get(10)?,
                        default_value_dialect: row.get(11)?,
                    })
                },
            )
            .unwrap();
    assert_eq!(
        column,
        ColumnRow {
            column_id: 3,
            begin_snapshot: 2,
            end_snapshot: None,
            column_order: 3,
            column_name: "note".into(),
            column_type: "varchar".into(),
            initial_default: "hello".into(),
            default_value: "hello".into(),
            nulls_allowed: true,
            parent_column: None,
            default_value_type: "literal".into(),
            default_value_dialect: "duckdb".into(),
        }
    );

    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (2, 2, 4, 10));

    let stats: (Option<bool>, Option<bool>, Option<String>, Option<String>) = conn
        .query_row(
            "SELECT contains_null, contains_nan, min_value, max_value \
                 FROM ducklake_table_column_stats WHERE table_id = 1 AND column_id = 3",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(
        stats,
        (
            Some(false),
            None,
            Some("hello".to_string()),
            Some("hello".to_string())
        )
    );

    let changes = duckdb_snapshot_changes(&catalog_path);
    assert_eq!(changes, vec!["altered_table:1".to_string()]);
}

#[test]
fn duckdb_file_column_default_commit_sets_and_drops_default() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    create_duckdb_column_alter_catalog(&catalog_path);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let set_snapshot = backend
            .commit_column_default(ColumnDefaultCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(2),
                default_value: Some("lower('FUTURE')".into()),
                default_value_type: Some("expression".into()),
                default_value_dialect: Some("duckdb".into()),
                commit_message: "set default".into(),
            })
            .await
            .unwrap();
        assert_eq!(set_snapshot, SnapshotId(2));

        let drop_snapshot = backend
            .commit_column_default(ColumnDefaultCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                column_id: ColumnId(2),
                default_value: Some("NULL".into()),
                default_value_type: Some("literal".into()),
                default_value_dialect: Some("duckdb".into()),
                commit_message: "drop default".into(),
            })
            .await
            .unwrap();
        assert_eq!(drop_snapshot, SnapshotId(3));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    assert_eq!(
        duckdb_column_default_history(&conn, 2),
        vec![
            (
                1,
                Some(2),
                "name".to_string(),
                Some("'legacy'".to_string()),
                Some("'legacy'".to_string()),
                Some("literal".to_string()),
                Some("duckdb".to_string())
            ),
            (
                2,
                Some(3),
                "name".to_string(),
                Some("'legacy'".to_string()),
                Some("lower('FUTURE')".to_string()),
                Some("expression".to_string()),
                Some("duckdb".to_string())
            ),
            (
                3,
                None,
                "name".to_string(),
                Some("'legacy'".to_string()),
                Some("NULL".to_string()),
                Some("literal".to_string()),
                Some("duckdb".to_string())
            ),
        ]
    );
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (3, 3, 10, 20));
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["altered_table:1".to_string(), "altered_table:1".to_string()]
    );
}

#[test]
fn duckdb_file_column_type_commit_promotes_noops_and_rejects_narrowing() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    create_duckdb_column_alter_catalog(&catalog_path);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let promoted = backend
            .commit_column_type(ColumnTypeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(1),
                new_column_type: "int64".into(),
                commit_message: "promote column".into(),
            })
            .await
            .unwrap();
        assert_eq!(promoted, SnapshotId(2));

        let noop = backend
            .commit_column_type(ColumnTypeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                column_id: ColumnId(1),
                new_column_type: "int64".into(),
                commit_message: "same type".into(),
            })
            .await
            .unwrap();
        assert_eq!(noop, SnapshotId(2));

        let err = backend
            .commit_column_type(ColumnTypeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                column_id: ColumnId(1),
                new_column_type: "int16".into(),
                commit_message: "narrow column".into(),
            })
            .await
            .unwrap_err();
        assert!(err.to_string().contains("only widening"), "{err}");
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let mut stmt = conn
        .prepare(
            "SELECT begin_snapshot, end_snapshot, column_type \
                 FROM ducklake_column \
                 WHERE table_id = 1 AND column_id = 1 \
                 ORDER BY begin_snapshot",
        )
        .unwrap();
    let history = stmt
        .query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, Option<i64>>(1)?,
                row.get::<_, String>(2)?,
            ))
        })
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap();
    assert_eq!(
        history,
        vec![
            (1, Some(2), "int32".to_string()),
            (2, None, "int64".to_string())
        ]
    );
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (2, 2, 10, 20));
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["altered_table:1".to_string()]
    );
}

#[test]
fn duckdb_file_column_type_tree_commit_replaces_struct_children() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    create_duckdb_column_alter_catalog(&catalog_path);
    {
        let conn = Connection::open(&catalog_path).unwrap();
        conn.execute_batch(
                r#"
                INSERT INTO ducklake_column VALUES (3, 1, NULL, 1, 3, 'profile', 'struct', NULL, NULL, true, NULL, NULL, NULL);
                INSERT INTO ducklake_column VALUES (4, 1, NULL, 1, 0, 'age', 'int32', NULL, NULL, true, 3, NULL, NULL);
                INSERT INTO ducklake_column VALUES (5, 1, NULL, 1, 1, 'city', 'varchar', NULL, NULL, true, 3, NULL, NULL);
                "#,
            )
            .unwrap();
    }

    let replacement = crate::ColumnTypeTree {
        column_name: "profile".into(),
        column_type: "struct".into(),
        nulls_allowed: true,
        children: vec![
            crate::ColumnTypeTree {
                column_name: "age".into(),
                column_type: "int64".into(),
                nulls_allowed: true,
                children: vec![],
            },
            crate::ColumnTypeTree {
                column_name: "city".into(),
                column_type: "varchar".into(),
                nulls_allowed: true,
                children: vec![],
            },
            crate::ColumnTypeTree {
                column_name: "score".into(),
                column_type: "int32".into(),
                nulls_allowed: true,
                children: vec![],
            },
        ],
    };

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = backend
            .commit_column_type_tree(ColumnTypeTreeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(3),
                replacement: replacement.clone(),
                commit_message: "replace struct".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(2));

        let noop = backend
            .commit_column_type_tree(ColumnTypeTreeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                column_id: ColumnId(3),
                replacement,
                commit_message: "same struct".into(),
            })
            .await
            .unwrap();
        assert_eq!(noop, SnapshotId(2));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    let mut stmt = conn
            .prepare(
                "SELECT column_id, begin_snapshot, end_snapshot, column_order, column_name, \
                        column_type, parent_column, default_value, default_value_type, default_value_dialect \
                 FROM ducklake_column \
                 WHERE table_id = 1 AND (column_id IN (3, 4, 5, 10)) \
                 ORDER BY column_id, begin_snapshot",
            )
            .unwrap();
    let rows = stmt
        .query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, Option<i64>>(2)?,
                row.get::<_, i64>(3)?,
                row.get::<_, String>(4)?,
                row.get::<_, String>(5)?,
                row.get::<_, Option<i64>>(6)?,
                row.get::<_, Option<String>>(7)?,
                row.get::<_, Option<String>>(8)?,
                row.get::<_, Option<String>>(9)?,
            ))
        })
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap();
    assert_eq!(
        rows,
        vec![
            (
                3,
                1,
                Some(2),
                3,
                "profile".to_string(),
                "struct".to_string(),
                None,
                None,
                None,
                None
            ),
            (
                3,
                2,
                None,
                3,
                "profile".to_string(),
                "struct".to_string(),
                None,
                None,
                None,
                None
            ),
            (
                4,
                1,
                Some(2),
                0,
                "age".to_string(),
                "int32".to_string(),
                Some(3),
                None,
                None,
                None
            ),
            (
                4,
                2,
                None,
                0,
                "age".to_string(),
                "int64".to_string(),
                Some(3),
                None,
                None,
                None
            ),
            (
                5,
                1,
                Some(2),
                1,
                "city".to_string(),
                "varchar".to_string(),
                Some(3),
                None,
                None,
                None
            ),
            (
                5,
                2,
                None,
                1,
                "city".to_string(),
                "varchar".to_string(),
                Some(3),
                None,
                None,
                None
            ),
            (
                10,
                2,
                None,
                2,
                "score".to_string(),
                "int32".to_string(),
                Some(3),
                Some("NULL".to_string()),
                Some("literal".to_string()),
                Some("duckdb".to_string())
            ),
        ]
    );
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (2, 2, 11, 20));
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["altered_table:1".to_string()]
    );
}

#[test]
fn duckdb_file_rename_column_commit_preserves_id_and_rejects_case_conflict() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    create_duckdb_column_alter_catalog(&catalog_path);

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let rename_snapshot = backend
            .commit_rename_column(RenameColumnCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(2),
                new_column_name: "label".into(),
                commit_message: "rename column".into(),
            })
            .await
            .unwrap();
        assert_eq!(rename_snapshot, SnapshotId(2));

        let err = backend
            .commit_rename_column(RenameColumnCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                column_id: ColumnId(2),
                new_column_name: "ID".into(),
                commit_message: "rename conflict".into(),
            })
            .await
            .unwrap_err();
        assert!(err.to_string().contains("already exists"));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    assert_eq!(
        duckdb_column_name_history(&conn, 2),
        vec![
            (1, Some(2), "name".to_string()),
            (2, None, "label".to_string())
        ]
    );
    let active_names: Vec<String> = {
        let mut stmt = conn
            .prepare(
                "SELECT column_name FROM ducklake_column \
                     WHERE table_id = 1 AND end_snapshot IS NULL \
                     ORDER BY column_order",
            )
            .unwrap();
        stmt.query_map([], |row| row.get(0))
            .unwrap()
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap()
    };
    assert_eq!(active_names, vec!["id".to_string(), "label".to_string()]);
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (2, 2, 11, 20));
    let sort_rows: Vec<(i64, i64, Option<i64>, String)> = {
        let mut stmt = conn
            .prepare(
                "SELECT si.sort_id, si.begin_snapshot, si.end_snapshot, se.expression \
                     FROM ducklake_sort_info si \
                     JOIN ducklake_sort_expression se USING (sort_id, table_id) \
                     WHERE si.table_id = 1 \
                     ORDER BY si.sort_id",
            )
            .unwrap();
        stmt.query_map([], |row| {
            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
        })
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap()
    };
    assert_eq!(
        sort_rows,
        vec![
            (5, 1, Some(2), "name || id".to_string()),
            (10, 2, None, "label || id".to_string()),
        ]
    );
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["altered_table:1".to_string()]
    );
}

#[test]
fn duckdb_file_drop_column_commit_closes_column_and_rejects_last_column() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    create_duckdb_column_alter_catalog(&catalog_path);
    {
        let conn = Connection::open(&catalog_path).unwrap();
        conn.execute(
            "UPDATE ducklake_sort_expression SET expression = 'id' WHERE sort_id = 5",
            [],
        )
        .unwrap();
    }

    let backend = DuckDbCatalog::new(&catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let snapshot = backend
            .commit_drop_column(DropColumnCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(2),
                commit_message: "drop column".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(2));

        let err = backend
            .commit_drop_column(DropColumnCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                column_id: ColumnId(1),
                commit_message: "drop last column".into(),
            })
            .await
            .unwrap_err();
        assert!(err.to_string().contains("only has one column"));
    });

    let conn = Connection::open(&catalog_path).unwrap();
    assert_eq!(
        duckdb_column_name_history(&conn, 2),
        vec![(1, Some(2), "name".to_string())]
    );
    let active_names: Vec<String> = {
        let mut stmt = conn
            .prepare(
                "SELECT column_name FROM ducklake_column \
                     WHERE table_id = 1 AND end_snapshot IS NULL \
                     ORDER BY column_order",
            )
            .unwrap();
        stmt.query_map([], |row| row.get(0))
            .unwrap()
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap()
    };
    assert_eq!(active_names, vec!["id".to_string()]);
    let latest: (i64, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                 FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(latest, (2, 2, 10, 20));
    assert_eq!(
        duckdb_snapshot_changes(&catalog_path),
        vec!["altered_table:1".to_string()]
    );
}

#[test]
fn duckdb_file_drop_column_commit_rejects_sort_and_partition_columns() {
    let temp = tempfile::tempdir().unwrap();
    let sorted_catalog_path = temp.path().join("sorted.duckdb");
    create_duckdb_column_alter_catalog(&sorted_catalog_path);

    let backend = DuckDbCatalog::new(&sorted_catalog_path);
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let err = backend
            .commit_drop_column(DropColumnCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(2),
                commit_message: "drop sorted column".into(),
            })
            .await
            .unwrap_err();
        assert!(err.to_string().contains("sorted by this column"));
    });

    let partition_catalog_path = temp.path().join("partitioned.duckdb");
    create_duckdb_column_alter_catalog(&partition_catalog_path);
    {
        let conn = Connection::open(&partition_catalog_path).unwrap();
        conn.execute_batch(
                "
                CREATE TABLE ducklake_partition_info(partition_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
                CREATE TABLE ducklake_partition_column(partition_id BIGINT, table_id BIGINT, partition_key_index BIGINT, column_id BIGINT, transform VARCHAR);
                INSERT INTO ducklake_partition_info VALUES (7, 1, 1, NULL);
                INSERT INTO ducklake_partition_column VALUES (7, 1, 0, 2, 'identity');
                ",
            )
            .unwrap();
        conn.execute(
            "UPDATE ducklake_sort_expression SET expression = 'id' WHERE sort_id = 5",
            [],
        )
        .unwrap();
    }

    let backend = DuckDbCatalog::new(&partition_catalog_path);
    runtime.block_on(async {
        let err = backend
            .commit_drop_column(DropColumnCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(2),
                commit_message: "drop partition column".into(),
            })
            .await
            .unwrap_err();
        assert!(err.to_string().contains("partitioned by this column"));
    });
}

#[test]
fn duckdb_file_concurrent_append_commits_return_one_conflict() {
    let temp = tempfile::tempdir().unwrap();
    let catalog_path = temp.path().join("catalog.duckdb");
    let conn = Connection::open(&catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, mapping_id BIGINT, partial_max BIGINT);
            CREATE TABLE ducklake_file_column_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value VARCHAR, max_value VARCHAR, contains_nan BOOLEAN, extra_stats VARCHAR);
            CREATE TABLE ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value VARCHAR);
            CREATE TABLE ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 10);
            INSERT INTO ducklake_table_stats VALUES (1, 0, 0, 0);
            "#,
        )
        .unwrap();
    drop(conn);

    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
        let left = DuckDbCatalog::new(&catalog_path);
        let right = DuckDbCatalog::new(&catalog_path);
        let (left, right) = tokio::join!(
            left.commit_append_files(vec![duckdb_append_file(10, SnapshotId(1))]),
            right.commit_append_files(vec![duckdb_append_file(10, SnapshotId(1))])
        );

        let mut successes = 0;
        let mut conflicts = 0;
        for result in [left, right] {
            match result {
                Ok(snapshot) => {
                    assert_eq!(snapshot, SnapshotId(2));
                    successes += 1;
                }
                Err(err) => {
                    assert_snapshot_conflict(&err, "append", 1, 2);
                    conflicts += 1;
                }
            }
        }
        assert_eq!(successes, 1);
        assert_eq!(conflicts, 1);
    });

    let snapshots = duckdb_table_count(&catalog_path, "ducklake_snapshot");
    let data_files = duckdb_table_count(&catalog_path, "ducklake_data_file");
    assert_eq!(snapshots, 2);
    assert_eq!(data_files, 1);
}

fn duckdb_append_file(data_file_id: i64, expected_snapshot: SnapshotId) -> AppendFile {
    TestAppendFile {
        table_id: TableId(1),
        data_file_id,
        expected_snapshot,
        path: format!("duckdb-append-{data_file_id}.parquet"),
        record_count: 2,
        file_size_bytes: 25,
        column_stats: vec![test_file_column_stats(
            data_file_id,
            TableId(1),
            ColumnId(2),
            Some(8),
            2,
            0,
            "10",
            "20",
        )],
        partition_id: Some(PartitionId(3)),
        partition_values: vec![test_file_partition_value(data_file_id, TableId(1), 0, "4")],
    }
    .into_commit()
}

fn duckdb_append_file_with_stats(
    data_file_id: i64,
    expected_snapshot: SnapshotId,
    value_count: i64,
    null_count: i64,
    min_value: &str,
    max_value: &str,
) -> AppendFile {
    AppendFile {
        column_stats: vec![test_file_column_stats(
            data_file_id,
            TableId(1),
            ColumnId(2),
            None,
            value_count,
            null_count,
            min_value,
            max_value,
        )],
        partition_id: None,
        partition_values: Vec::new(),
        ..duckdb_append_file(data_file_id, expected_snapshot)
    }
}

fn duckdb_delete_file(
    delete_file_id: i64,
    expected_snapshot: SnapshotId,
    data_file_id: i64,
    rows_deleted: i64,
    overwritten_delete_file_id: Option<DeleteFileId>,
) -> DeleteFileCommit {
    TestDeleteFile {
        table_id: TableId(1),
        delete_file_id,
        expected_snapshot,
        data_file_id,
        path: format!("duckdb-delete-{delete_file_id}.parquet"),
        rows_deleted,
        file_size_bytes: 12,
        footer_size: None,
        overwritten_delete_file_id,
    }
    .into_commit()
}

fn duckdb_scheduled_file_count(catalog_path: &std::path::Path) -> i64 {
    let conn = Connection::open(catalog_path).unwrap();
    conn.query_row(
        "SELECT COUNT(*) FROM ducklake_files_scheduled_for_deletion",
        [],
        |row| row.get(0),
    )
    .unwrap()
}

fn duckdb_table_count(catalog_path: &std::path::Path, table: &str) -> i64 {
    let conn = Connection::open(catalog_path).unwrap();
    conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| {
        row.get(0)
    })
    .unwrap()
}

fn duckdb_data_file_ids(catalog_path: &std::path::Path) -> Vec<(i64, String)> {
    let conn = Connection::open(catalog_path).unwrap();
    let mut stmt = conn
        .prepare("SELECT data_file_id, path FROM ducklake_data_file ORDER BY data_file_id")
        .unwrap();
    stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap()
}

fn duckdb_snapshot_changes(catalog_path: &std::path::Path) -> Vec<String> {
    let conn = Connection::open(catalog_path).unwrap();
    let mut stmt = conn
        .prepare("SELECT changes_made FROM ducklake_snapshot_changes ORDER BY snapshot_id")
        .unwrap();
    stmt.query_map([], |row| row.get(0))
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap()
}

fn create_duckdb_column_alter_catalog(catalog_path: &std::path::Path) {
    let conn = Connection::open(catalog_path).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
            CREATE TABLE ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made VARCHAR, author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR);
            CREATE TABLE ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name VARCHAR, column_type VARCHAR, initial_default VARCHAR, default_value VARCHAR, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type VARCHAR, default_value_dialect VARCHAR);
            CREATE TABLE ducklake_sort_info(sort_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
            CREATE TABLE ducklake_sort_expression(sort_id BIGINT, table_id BIGINT, sort_key_index BIGINT, expression VARCHAR, dialect VARCHAR, sort_direction VARCHAR, null_order VARCHAR);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
            INSERT INTO ducklake_column VALUES (1, 1, NULL, 1, 1, 'id', 'int32', NULL, NULL, true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_column VALUES (2, 1, NULL, 1, 2, 'name', 'varchar', '''legacy''', '''legacy''', true, NULL, 'literal', 'duckdb');
            INSERT INTO ducklake_sort_info VALUES (5, 1, 1, NULL);
            INSERT INTO ducklake_sort_expression VALUES (5, 1, 0, 'name || id', 'duckdb', 'ASC', 'NULLS_LAST');
            "#,
        )
        .unwrap();
}

// The tuple mirrors the selected `ducklake_column` row shape; a named type would
// not read more clearly for a one-off test helper.
#[allow(clippy::type_complexity)]
fn duckdb_column_default_history(
    conn: &Connection,
    column_id: i64,
) -> Vec<(
    i64,
    Option<i64>,
    String,
    Option<String>,
    Option<String>,
    Option<String>,
    Option<String>,
)> {
    let mut stmt = conn
        .prepare(
            "SELECT begin_snapshot, end_snapshot, column_name, initial_default, default_value, \
                        default_value_type, default_value_dialect \
                 FROM ducklake_column \
                 WHERE table_id = 1 AND column_id = ? \
                 ORDER BY begin_snapshot",
        )
        .unwrap();
    stmt.query_map([column_id], |row| {
        Ok((
            row.get(0)?,
            row.get(1)?,
            row.get(2)?,
            row.get(3)?,
            row.get(4)?,
            row.get(5)?,
            row.get(6)?,
        ))
    })
    .unwrap()
    .collect::<std::result::Result<Vec<_>, _>>()
    .unwrap()
}

fn duckdb_column_name_history(
    conn: &Connection,
    column_id: i64,
) -> Vec<(i64, Option<i64>, String)> {
    let mut stmt = conn
        .prepare(
            "SELECT begin_snapshot, end_snapshot, column_name \
                 FROM ducklake_column \
                 WHERE table_id = 1 AND column_id = ? \
                 ORDER BY begin_snapshot",
        )
        .unwrap();
    stmt.query_map([column_id], |row| {
        Ok((row.get(0)?, row.get(1)?, row.get(2)?))
    })
    .unwrap()
    .collect::<std::result::Result<Vec<_>, _>>()
    .unwrap()
}

fn duckdb_partition_info_rows_with_table(
    catalog_path: &std::path::Path,
) -> Vec<(i64, i64, Option<i64>, i64)> {
    let conn = Connection::open(catalog_path).unwrap();
    let mut stmt = conn
        .prepare(
            "SELECT partition_id, begin_snapshot, end_snapshot, table_id \
                 FROM ducklake_partition_info ORDER BY partition_id",
        )
        .unwrap();
    stmt.query_map([], |row| {
        Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
    })
    .unwrap()
    .collect::<std::result::Result<Vec<_>, _>>()
    .unwrap()
}

fn duckdb_partition_column_rows(catalog_path: &std::path::Path) -> Vec<(i64, i64, i64, String)> {
    let conn = Connection::open(catalog_path).unwrap();
    let mut stmt = conn
        .prepare(
            "SELECT partition_id, partition_key_index, column_id, transform \
                 FROM ducklake_partition_column ORDER BY partition_id, partition_key_index",
        )
        .unwrap();
    stmt.query_map([], |row| {
        Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
    })
    .unwrap()
    .collect::<std::result::Result<Vec<_>, _>>()
    .unwrap()
}

fn duckdb_table_stats_file_size(catalog_path: &std::path::Path, table_id: TableId) -> i64 {
    let conn = Connection::open(catalog_path).unwrap();
    conn.query_row(
        "SELECT file_size_bytes FROM ducklake_table_stats WHERE table_id = ?1",
        params![table_id.0],
        |row| row.get(0),
    )
    .unwrap()
}

fn duckdb_latest_next_file_id(catalog_path: &std::path::Path) -> i64 {
    let conn = Connection::open(catalog_path).unwrap();
    conn.query_row(
        "SELECT next_file_id FROM ducklake_snapshot ORDER BY snapshot_id DESC LIMIT 1",
        [],
        |row| row.get(0),
    )
    .unwrap()
}
