use super::*;
use crate::embedded_rows::parse_timestamp_micros;
use crate::test_support::{
    TestAppendFile, TestDeleteFile, assert_snapshot_conflict as assert_snapshot_conflict_at,
    test_file_column_stats, test_inlined_file_deletion, test_schema_field as schema_field,
};
use crate::{
    AddColumn, AddColumnCommit, AppendFile, CatalogBackend, ColumnNullabilityCommit,
    CreateTableColumn, DeleteFileCommit, InlinedFileDeletionCommit, PartitionColumnCommit,
    PartitioningBatchCommit, PartitioningCommit, PartitioningTableCommit, TableChangeCommit,
    TruncateTableCommit,
};
use arrow::array::Int32Array;
use ducklake_catalog::{
    CatalogError, DeleteFileId, DuckLakeType, SUPPORTED_VERSION, SnapshotSelector, TableSchema,
};
use std::collections::HashMap;

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

    let backend = SqliteCatalog::new(temp.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(temp.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::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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 sqlite_create_drop_schema_commits_ducklake_metadata() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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: "sqlite 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: "sqlite create schema table".into(),
            })
            .await
            .unwrap();
        assert_eq!(created_table, SnapshotId(3));
    });

    let conn = Connection::open(temp.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: "sqlite 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: "sqlite 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: "sqlite cascade drop schema".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped, SnapshotId(5));
    });

    let conn = Connection::open(temp.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!(
        sqlite_snapshot_changes(temp.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 sqlite_create_drop_table_commits_ducklake_metadata() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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: "sqlite create table".into(),
            })
            .await
            .unwrap();
        assert_eq!(created, SnapshotId(2));
    });

    let conn = Connection::open(temp.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,
        (10, 2, None, 1, "created".into(), "created/".into(), true)
    );
    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 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));
    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: "sqlite drop table".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped, SnapshotId(3));
    });

    let conn = Connection::open(temp.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));
    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 = 10 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_10_2 WHERE end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(active_inlined_rows, 0);
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec![
            "created_table:\"main\".\"created\"".to_string(),
            "dropped_table:10".to_string(),
        ]
    );
}

#[test]
fn sqlite_drop_view_commits_ducklake_metadata() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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: "sqlite drop view".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped, SnapshotId(2));
    });

    let conn = Connection::open(temp.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));
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec!["dropped_view:9".to_string()]
    );
}

#[test]
fn sqlite_commit_partitioning_is_idempotent_like_duckdb() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    create_partitioning_catalog(temp.path());

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let first = backend
            .commit_partitioning(PartitioningCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                columns: vec![PartitionColumnCommit {
                    column_id: ColumnId(1),
                    transform: PartitionTransform::Identity,
                }],
                commit_message: "set partitioned by part_key".into(),
            })
            .await
            .unwrap();
        assert_eq!(first, SnapshotId(2));

        let repeated = backend
            .commit_partitioning(PartitioningCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                columns: vec![PartitionColumnCommit {
                    column_id: ColumnId(1),
                    transform: PartitionTransform::Identity,
                }],
                commit_message: "set partitioned by part_key".into(),
            })
            .await
            .unwrap();
        assert_eq!(repeated, SnapshotId(3));

        let changed = backend
            .commit_partitioning(PartitioningCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(3),
                columns: vec![
                    PartitionColumnCommit {
                        column_id: ColumnId(1),
                        transform: PartitionTransform::Identity,
                    },
                    PartitionColumnCommit {
                        column_id: ColumnId(2),
                        transform: PartitionTransform::Identity,
                    },
                ],
                commit_message: "set partitioned by two columns".into(),
            })
            .await
            .unwrap();
        assert_eq!(changed, SnapshotId(4));

        let reset = backend
            .commit_partitioning(PartitioningCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(4),
                columns: Vec::new(),
                commit_message: "reset partitioned by".into(),
            })
            .await
            .unwrap();
        assert_eq!(reset, SnapshotId(5));

        let repeated_reset = backend
            .commit_partitioning(PartitioningCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(5),
                columns: Vec::new(),
                commit_message: "reset partitioned by".into(),
            })
            .await
            .unwrap();
        assert_eq!(repeated_reset, SnapshotId(6));
    });

    let conn = Connection::open(temp.path()).unwrap();
    assert_eq!(
        partition_info_rows(&conn),
        vec![(2, 2, Some(4)), (4, 4, Some(5)), (5, 5, None)]
    );
    assert_eq!(
        partition_column_rows(&conn),
        vec![
            (2, 0, 1, "identity".to_string()),
            (4, 0, 1, "identity".to_string()),
            (4, 1, 2, "identity".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, (6, 6, 7, 0));
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec![
            "altered_table:1".to_string(),
            "altered_table:1".to_string(),
            "altered_table:1".to_string(),
            "altered_table:1".to_string(),
            "altered_table:1".to_string(),
        ]
    );
}

#[test]
fn sqlite_commit_partitioning_batch_matches_duckdb_multi_table_transaction() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    create_partitioning_multi_table_catalog(temp.path());

    let backend = SqliteCatalog::new(temp.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));
    });

    let conn = Connection::open(temp.path()).unwrap();
    assert_eq!(
        partition_info_rows_with_table(&conn),
        vec![(3, 3, None, 2), (4, 3, None, 1)]
    );
    assert_eq!(
        partition_column_rows(&conn),
        vec![
            (3, 0, 1, "identity".to_string()),
            (4, 0, 1, "identity".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, 5, 0));
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec!["altered_table:1,altered_table:2".to_string()]
    );
}

#[test]
fn sqlite_partitioning_batch_repeated_table_uses_final_state_and_duckdb_id_allocation() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    create_partitioning_multi_table_catalog(temp.path());

    let backend = SqliteCatalog::new(temp.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));
    });

    let conn = Connection::open(temp.path()).unwrap();
    assert_eq!(partition_info_rows_with_table(&conn), vec![(3, 3, None, 2)]);
    assert_eq!(
        partition_column_rows(&conn),
        vec![(3, 0, 1, "identity".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, 6, 0));
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec!["altered_table:1,altered_table:2".to_string()]
    );
}

#[test]
fn sqlite_create_view_commits_ducklake_metadata() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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: "sqlite create view".into(),
            })
            .await
            .unwrap();
        assert_eq!(created, SnapshotId(2));
    });

    let conn = Connection::open(temp.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));
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec!["created_view:\"main\".\"v_numbers\"".to_string()]
    );
}

#[test]
fn sqlite_create_table_and_view_reject_case_insensitive_object_conflicts() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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);

            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);
            INSERT INTO ducklake_view VALUES (3, '550e8400-e29b-41d4-a716-446655440003', 1, NULL, 1, 'v_numbers', 'duckdb', 'SELECT * FROM {DUCKLAKE_CATALOG}.numbers', NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let column = || {
            vec![CreateTableColumn {
                column_id: ColumnId(1),
                column_order: 1,
                column_name: "id".into(),
                column_type: "int32".into(),
                nulls_allowed: true,
                parent_column: None,
            }]
        };

        let table_name_conflict = backend
            .commit_create_table(CreateTableCommit {
                schema_name: "main".into(),
                table_name: "NUMBERS".into(),
                columns: column(),
                commit_message: "create conflicting table".into(),
            })
            .await
            .unwrap_err();
        assert!(table_name_conflict.to_string().contains("already exists"));

        let table_view_conflict = backend
            .commit_create_table(CreateTableCommit {
                schema_name: "main".into(),
                table_name: "V_NUMBERS".into(),
                columns: column(),
                commit_message: "create table over view".into(),
            })
            .await
            .unwrap_err();
        assert!(table_view_conflict.to_string().contains("already exists"));

        let view_table_conflict = backend
            .commit_create_view(CreateViewCommit {
                schema_name: "main".into(),
                view_name: "NUMBERS".into(),
                dialect: "duckdb".into(),
                sql: "SELECT * FROM {DUCKLAKE_CATALOG}.numbers".into(),
                column_aliases: None,
                commit_message: "create view over table".into(),
            })
            .await
            .unwrap_err();
        assert!(view_table_conflict.to_string().contains("already exists"));

        let view_name_conflict = backend
            .commit_create_view(CreateViewCommit {
                schema_name: "main".into(),
                view_name: "V_NUMBERS".into(),
                dialect: "duckdb".into(),
                sql: "SELECT * FROM {DUCKLAKE_CATALOG}.numbers".into(),
                column_aliases: None,
                commit_message: "create conflicting view".into(),
            })
            .await
            .unwrap_err();
        assert!(view_name_conflict.to_string().contains("already exists"));
    });

    assert_eq!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 1);
    assert_eq!(
        sqlite_table_count(temp.path(), "ducklake_snapshot_changes"),
        0
    );
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_table"), 1);
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_view"), 1);
}

#[test]
fn sqlite_schema_lookup_is_case_insensitive_for_catalog_commits() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let err = backend
            .commit_create_schema(CreateSchemaCommit {
                schema_name: "MAIN".into(),
                commit_message: "create conflicting schema".into(),
            })
            .await
            .unwrap_err();
        assert!(err.to_string().contains("already exists"));
        assert_eq!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 1);

        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: true,
                    parent_column: None,
                }],
                commit_message: "create table in main by case-insensitive schema".into(),
            })
            .await
            .unwrap();
        assert_eq!(created, SnapshotId(2));
    });

    let conn = Connection::open(temp.path()).unwrap();
    let schema_id: i64 = conn
        .query_row(
            "SELECT schema_id FROM ducklake_table WHERE table_name = 'created'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(schema_id, 1);
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_schema"), 1);
}

#[test]
fn sqlite_schema_lookup_prefers_exact_match_and_rejects_ambiguous_case() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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: "ambiguous schema lookup".into(),
            })
            .await
            .unwrap_err();
        assert!(
            ambiguous
                .to_string()
                .contains("ambiguous DuckLake schema identifier"),
            "{ambiguous}"
        );
        assert_eq!(sqlite_table_count(temp.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: "exact uppercase schema lookup".into(),
            })
            .await
            .unwrap();
        assert_eq!(created, SnapshotId(2));
    });

    let conn = Connection::open(temp.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!(
        sqlite_snapshot_changes(temp.path()),
        vec!["created_table:\"MAIN\".\"exact_upper\"".to_string()]
    );
}

#[test]
fn sqlite_table_and_view_lookup_is_case_insensitive_for_catalog_commits() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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: "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: "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: "drop view by case-insensitive source".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped_view, SnapshotId(4));
    });

    let conn = Connection::open(temp.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!(
        sqlite_snapshot_changes(temp.path()),
        vec![
            "created_table:\"main\".\"renamed\"".to_string(),
            "dropped_table:2".to_string(),
            "dropped_view:3".to_string(),
        ]
    );
}

#[test]
fn sqlite_table_and_view_lookup_prefers_exact_match_and_rejects_ambiguous_case() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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: "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: "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: "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: "drop exact uppercase view source".into(),
            })
            .await
            .unwrap();
        assert_eq!(dropped_view, SnapshotId(3));
    });

    let conn = Connection::open(temp.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!(
        sqlite_snapshot_changes(temp.path()),
        vec!["dropped_table:4".to_string(), "dropped_view:5".to_string()]
    );
}

#[test]
fn sqlite_column_nullability_commit_sets_and_drops_not_null() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    create_column_nullability_catalog(temp.path(), false);

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let set_snapshot = backend
            .commit_column_nullability(ColumnNullabilityCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(1),
                nulls_allowed: false,
                existing_values_validated: false,
                commit_message: "set not null".into(),
            })
            .await
            .unwrap();
        assert_eq!(set_snapshot, SnapshotId(2));

        let drop_snapshot = backend
            .commit_column_nullability(ColumnNullabilityCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(2),
                column_id: ColumnId(1),
                nulls_allowed: true,
                existing_values_validated: false,
                commit_message: "drop not null".into(),
            })
            .await
            .unwrap();
        assert_eq!(drop_snapshot, SnapshotId(3));
    });

    let conn = Connection::open(temp.path()).unwrap();
    let history = column_nullability_history(&conn);
    assert_eq!(
        history,
        vec![(1, Some(2), true), (2, Some(3), false), (3, None, true)]
    );
    let snapshots: Vec<(i64, i64)> = {
        let mut stmt = conn
            .prepare(
                "SELECT snapshot_id, schema_version FROM ducklake_snapshot ORDER BY snapshot_id",
            )
            .unwrap();
        collect_rows(&mut stmt, [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap()
    };
    assert_eq!(snapshots, vec![(1, 1), (2, 2), (3, 3)]);
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec!["altered_table:1".to_string(), "altered_table:1".to_string()]
    );
}

#[test]
fn sqlite_column_nullability_commit_rejects_existing_nulls() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    create_column_nullability_catalog(temp.path(), true);

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let err = backend
            .commit_column_nullability(ColumnNullabilityCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                column_id: ColumnId(1),
                nulls_allowed: false,
                existing_values_validated: false,
                commit_message: "set not null".into(),
            })
            .await
            .unwrap_err();
        assert!(err.to_string().contains("has NULL values"));
    });

    let conn = Connection::open(temp.path()).unwrap();
    assert_eq!(column_nullability_history(&conn), vec![(1, None, true)]);
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 1);
    assert_eq!(
        sqlite_table_count(temp.path(), "ducklake_snapshot_changes"),
        0
    );
}

#[test]
fn sqlite_column_nullability_commit_detects_snapshot_conflicts() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    create_column_nullability_catalog(temp.path(), false);

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let err = backend
            .commit_column_nullability(ColumnNullabilityCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(0),
                column_id: ColumnId(1),
                nulls_allowed: false,
                existing_values_validated: false,
                commit_message: "set not null".into(),
            })
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("alter-column-nullability conflict")
        );
    });
}

#[test]
fn sqlite_column_default_commit_sets_and_drops_default() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    create_column_alter_catalog(temp.path());

    let backend = SqliteCatalog::new(temp.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(temp.path()).unwrap();
    assert_eq!(
        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!(
        sqlite_snapshot_changes(temp.path()),
        vec!["altered_table:1".to_string(), "altered_table:1".to_string()]
    );
}

#[test]
fn sqlite_rename_column_commit_preserves_id_and_rejects_case_conflict() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    create_column_alter_catalog(temp.path());

    let backend = SqliteCatalog::new(temp.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(temp.path()).unwrap();
    assert_eq!(
        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();
        collect_rows(&mut stmt, [], |row| row.get(0)).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();
        collect_rows(&mut stmt, [], |row| {
            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
        })
        .unwrap()
    };
    assert_eq!(
        sort_rows,
        vec![
            (5, 1, Some(2), "name || id".to_string()),
            (10, 2, None, "label || id".to_string()),
        ]
    );
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec!["altered_table:1".to_string()]
    );
}

#[test]
fn sqlite_add_column_default_commits_column_and_table_stats() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.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: "sqlite add defaulted column".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(2));
    });

    let conn = Connection::open(temp.path()).unwrap();
    type ColumnRow = (
        i64,
        i64,
        Option<i64>,
        i64,
        String,
        String,
        String,
        String,
        bool,
        Option<i64>,
        String,
        String,
    );
    let column: ColumnRow = 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((
                        row.get(0)?,
                        row.get(1)?,
                        row.get(2)?,
                        row.get(3)?,
                        row.get(4)?,
                        row.get(5)?,
                        row.get(6)?,
                        row.get(7)?,
                        row.get(8)?,
                        row.get(9)?,
                        row.get(10)?,
                        row.get(11)?,
                    ))
                },
            )
            .unwrap();
    assert_eq!(
        column,
        (
            3,
            2,
            None,
            3,
            "note".into(),
            "varchar".into(),
            "hello".into(),
            "hello".into(),
            true,
            None,
            "literal".into(),
            "duckdb".into()
        )
    );
    let stats: (bool, Option<bool>, String, 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, (false, None, "hello".into(), "hello".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 changes: (String, String) = conn
            .query_row(
                "SELECT changes_made, commit_message FROM ducklake_snapshot_changes WHERE snapshot_id = 2",
                [],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
    assert_eq!(
        changes,
        (
            "altered_table:1".into(),
            "sqlite add defaulted column".into()
        )
    );
}

#[test]
fn cleanup_old_files_deletes_due_local_files_and_queue_rows() {
    let temp = tempfile::tempdir().unwrap();
    let data_dir = temp.path().join("lake");
    fs::create_dir(&data_dir).unwrap();
    let old_file = data_dir.join("old.parquet");
    let future_file = data_dir.join("future.parquet");
    let absolute_file = temp.path().join("absolute.parquet");
    fs::write(&old_file, b"old").unwrap();
    fs::write(&future_file, b"future").unwrap();
    fs::write(&absolute_file, b"absolute").unwrap();

    let catalog_path = temp.path().join("catalog.sqlite");
    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_dir.to_string_lossy().as_ref()],
    )
    .unwrap();
    conn.execute(
            "INSERT INTO ducklake_files_scheduled_for_deletion VALUES (3, ?1, false, '2026-01-01 00:00:00+00')",
            params![absolute_file.to_string_lossy().as_ref()],
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::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!(
            invalid
                .to_string()
                .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!(old_file.exists());
        assert!(future_file.exists());
        assert!(absolute_file.exists());
        assert_eq!(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();
        let mut ids: Vec<_> = cleaned.iter().map(|file| file.data_file_id.0).collect();
        ids.sort_unstable();
        assert_eq!(ids, vec![1, 3, 4]);
        assert!(!old_file.exists());
        assert!(future_file.exists());
        assert!(!absolute_file.exists());
        assert_eq!(scheduled_file_count(&catalog_path), 1);

        let cleaned = backend
            .cleanup_old_files(CleanupOldFilesOptions::cleanup_all())
            .await
            .unwrap();
        assert_eq!(cleaned[0].data_file_id, DataFileId(2));
        assert!(!future_file.exists());
        assert_eq!(scheduled_file_count(&catalog_path), 0);
    });
}

#[test]
fn cleanup_old_files_rejects_remote_paths_before_deleting_local_files() {
    let temp = tempfile::tempdir().unwrap();
    let local_file = temp.path().join("local.parquet");
    fs::write(&local_file, b"local").unwrap();

    let catalog_path = temp.path().join("catalog.sqlite");
    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_files_scheduled_for_deletion VALUES (1, ?1, false, '2026-01-01 00:00:00+00')",
            params![local_file.to_string_lossy().as_ref()],
        )
        .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 = SqliteCatalog::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!(
            err.to_string()
                .contains("Materialized object-store file access")
        );
        assert!(local_file.exists());
        assert_eq!(scheduled_file_count(&catalog_path), 2);
    });
}

#[test]
fn expire_snapshots_schedules_unreferenced_files_and_prunes_metadata() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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_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 VARCHAR, 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_inlined_data_tables(table_id BIGINT, table_name VARCHAR, schema_version BIGINT);
            CREATE TABLE ducklake_inlined_data_1_1(begin_snapshot BIGINT, end_snapshot BIGINT, row_id BIGINT, id INTEGER);

            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_metadata VALUES ('data_path', '/lake/', NULL, NULL);
            INSERT INTO ducklake_schema VALUES (1, 'schema-uuid', 1, NULL, 'main', 'main/', true);
            INSERT INTO ducklake_table VALUES (1, 'table-uuid', 1, 3, 1, 'numbers', 'numbers/', 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_inlined_data_1_1 VALUES (1, 3, 0, 7);
            INSERT INTO ducklake_inlined_data_tables VALUES (1, 'ducklake_inlined_data_1_1', 1);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::new(temp.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!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 3);
        assert_eq!(sqlite_table_count(temp.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!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 2);
        assert_eq!(
            sqlite_table_count(temp.path(), "ducklake_snapshot_changes"),
            2
        );
        assert_eq!(sqlite_table_count(temp.path(), "ducklake_data_file"), 0);
        assert_eq!(
            sqlite_table_count(temp.path(), "ducklake_file_column_stats"),
            0
        );
        assert_eq!(
            sqlite_table_count(temp.path(), "ducklake_inlined_data_tables"),
            1
        );
        assert!(sqlite_catalog_table_exists(
            temp.path(),
            "ducklake_inlined_data_1_1"
        ));
        assert_eq!(scheduled_file_count(temp.path()), 1);
        assert_eq!(
            scheduled_file_paths(temp.path()),
            vec!["main/numbers/old.parquet"]
        );

        let cutoff = parse_timestamp_micros("2026-01-03 00:00:00+00").unwrap();
        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!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 1);
        assert_eq!(sqlite_table_count(temp.path(), "ducklake_table"), 0);
        assert_eq!(sqlite_table_count(temp.path(), "ducklake_column"), 0);
        assert_eq!(sqlite_table_count(temp.path(), "ducklake_table_stats"), 0);
        assert_eq!(
            sqlite_table_count(temp.path(), "ducklake_schema_versions"),
            0
        );
        assert_eq!(
            sqlite_table_count(temp.path(), "ducklake_inlined_data_tables"),
            0
        );
        assert!(!sqlite_catalog_table_exists(
            temp.path(),
            "ducklake_inlined_data_1_1"
        ));
        assert_eq!(scheduled_file_count(temp.path()), 1);
    });
}

#[test]
fn expire_snapshots_rejects_ambiguous_criteria() {
    let invalid = ExpireSnapshotsOptions {
        dry_run: true,
        versions: vec![SnapshotId(1)],
        older_than_micros: Some(1),
    };
    let err = invalid.validate().unwrap_err();
    assert!(format!("{err}").contains("exactly one of versions or older_than"));

    let missing = ExpireSnapshotsOptions {
        dry_run: true,
        versions: Vec::new(),
        older_than_micros: None,
    };
    let err = missing.validate().unwrap_err();
    assert!(format!("{err}").contains("exactly one of versions or older_than"));
}

#[test]
fn public_read_helpers_reject_unsupported_catalog_version() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.path()).unwrap();
    conn.execute_batch(
            r#"
            CREATE TABLE ducklake_metadata(key VARCHAR, value VARCHAR, scope VARCHAR, scope_id BIGINT);
            INSERT INTO ducklake_metadata VALUES ('version', '2.0', NULL, NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        assert_unsupported_version(
            crate::plan_table_scan(&backend, "main", "numbers", SnapshotSelector::Latest).await,
        );
        assert_unsupported_version(
            crate::list_table_files(&backend, "main", "numbers", SnapshotSelector::Latest).await,
        );
        assert_unsupported_version(
            crate::list_table_info(&backend, SnapshotSelector::Latest).await,
        );
        assert_unsupported_version(
            crate::list_catalog_paths(&backend, SnapshotSelector::Latest).await,
        );
        assert_unsupported_version(
            crate::catalog_database_size_bytes(&backend, SnapshotSelector::Latest).await,
        );
        assert_unsupported_version(crate::list_snapshots(&backend).await);
        assert_unsupported_version(crate::current_snapshot_id(&backend).await);
        assert_unsupported_version(crate::catalog_settings(&backend).await);
        assert_unsupported_version(crate::list_options(&backend, SnapshotSelector::Latest).await);
        assert_unsupported_version(crate::list_tags(&backend, SnapshotSelector::Latest).await);
    });
}

#[test]
fn plan_table_scan_rejects_unsupported_files() {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();

    let format_temp = tempfile::NamedTempFile::new().unwrap();
    write_minimal_scan_catalog(format_temp.path(), None, None);
    let format_conn = Connection::open(format_temp.path()).unwrap();
    format_conn
        .execute("UPDATE ducklake_data_file SET file_format = 'csv'", [])
        .unwrap();
    drop(format_conn);
    let format_backend = SqliteCatalog::new(format_temp.path());
    runtime.block_on(async {
        let err =
            crate::plan_table_scan(&format_backend, "main", "numbers", SnapshotSelector::Latest)
                .await
                .unwrap_err();
        assert!(
            err.to_string()
                .contains("DuckLake data file 10 has unsupported file format csv")
        );
    });

    let delete_format_temp = tempfile::NamedTempFile::new().unwrap();
    write_minimal_scan_catalog(delete_format_temp.path(), None, None);
    let delete_format_conn = Connection::open(delete_format_temp.path()).unwrap();
    delete_format_conn
        .execute(
            "INSERT INTO ducklake_delete_file VALUES \
                 (11, 1, 1, NULL, 10, 'delete.orc', true, 'orc', 0, 10, NULL, NULL, NULL)",
            [],
        )
        .unwrap();
    drop(delete_format_conn);
    let delete_format_backend = SqliteCatalog::new(delete_format_temp.path());
    runtime.block_on(async {
        let err = crate::plan_table_scan(
            &delete_format_backend,
            "main",
            "numbers",
            SnapshotSelector::Latest,
        )
        .await
        .unwrap_err();
        assert!(
            err.to_string()
                .contains("DuckLake delete file 11 has unsupported format orc")
        );
    });

    let data_temp = tempfile::NamedTempFile::new().unwrap();
    write_minimal_scan_catalog(data_temp.path(), Some("data-key"), None);
    let data_backend = SqliteCatalog::new(data_temp.path());
    runtime.block_on(async {
        let err =
            crate::plan_table_scan(&data_backend, "main", "numbers", SnapshotSelector::Latest)
                .await
                .unwrap_err();
        assert!(
            err.to_string()
                .contains("Encrypted DuckLake data file 10 is not supported")
        );
    });

    let delete_temp = tempfile::NamedTempFile::new().unwrap();
    write_minimal_scan_catalog(delete_temp.path(), None, Some("delete-key"));
    let delete_backend = SqliteCatalog::new(delete_temp.path());
    runtime.block_on(async {
        let err =
            crate::plan_table_scan(&delete_backend, "main", "numbers", SnapshotSelector::Latest)
                .await
                .unwrap_err();
        assert!(
            err.to_string()
                .contains("Encrypted DuckLake delete file 11 is not supported")
        );
    });
}

#[test]
fn filters_versioned_sqlite_metadata_rows_by_snapshot() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        macro_rules! assert_snapshot {
            (
                    $snapshot:expr,
                    $schema_id:expr,
                    $table_id:expr,
                    $view_id:expr,
                    $tag_value:expr,
                    $column_tag_value:expr,
                    $column_id:expr,
                    $data_file_id:expr,
                    $delete_file_id:expr,
                    $partition_id:expr,
                    $macro_id:expr,
                    $sort_id:expr
                ) => {{
                let snapshot = SnapshotId($snapshot);
                let schemas = backend.schemas_at(snapshot).await.unwrap();
                assert_eq!(
                    schemas.iter().map(|row| row.schema_id).collect::<Vec<_>>(),
                    vec![SchemaId($schema_id)],
                    "schemas at snapshot {}",
                    snapshot.0
                );

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

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

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

                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($column_tag_value)],
                    "column tags at snapshot {}",
                    snapshot.0
                );

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

                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($data_file_id)],
                    "data files at snapshot {}",
                    snapshot.0
                );

                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($delete_file_id)],
                    "delete files at snapshot {}",
                    snapshot.0
                );

                let partitions = backend
                    .partition_info_at(TableId(20), snapshot)
                    .await
                    .unwrap();
                assert_eq!(
                    partitions
                        .iter()
                        .map(|row| row.partition_id)
                        .collect::<Vec<_>>(),
                    vec![PartitionId($partition_id)],
                    "partition info at snapshot {}",
                    snapshot.0
                );

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

                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($sort_id)],
                    "sort info at snapshot {}",
                    snapshot.0
                );
            }};
        }

        assert_snapshot!(2, 1, 10, 11, "old", "old", 100, 60, 90, 300, 600, 900);
        assert_snapshot!(
            3, 2, 20, 21, "current", "current", 200, 70, 100, 400, 700, 1000
        );
        assert_snapshot!(
            4, 3, 30, 31, "future", "future", 300, 80, 110, 500, 800, 1100
        );
    });
}

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

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

#[test]
fn reads_extended_sqlite_inlined_scalar_types() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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 VARCHAR,
                time_value VARCHAR,
                timestamp_value VARCHAR,
                bytes_value BLOB,
                uuid_value VARCHAR,
                ubig_value VARCHAR,
                huge_value VARCHAR,
                uhuge_value VARCHAR,
                wide_decimal VARCHAR
            );
            INSERT INTO ducklake_inlined_data_1_1 VALUES (
                1,
                NULL,
                1,
                '123.45',
                '2024-01-15',
                '12:30:00.123456',
                '2024-01-15 12:30:00.123456+00',
                X'00FF10',
                '550e8400-e29b-41d4-a716-446655440000',
                '18446744073709551615',
                '-17014118346046923173168730371588410572',
                '340282366920938463463374607431768211455',
                '123456789012345678901234567890123456789.12'
            );
            "#,
    )
    .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),
            schema_field(8, "huge_value", "int128", DuckLakeType::Int128),
            schema_field(9, "uhuge_value", "uint128", DuckLakeType::UInt128),
            schema_field(
                10,
                "wide_decimal",
                "decimal(41,2)",
                DuckLakeType::Decimal {
                    width: 41,
                    scale: 2,
                },
            ),
        ],
        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 = SqliteCatalog::new(temp.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);
        assert_eq!(
            batch.schema().field(7).data_type(),
            &DataType::Decimal256(39, 0)
        );
        assert_eq!(
            batch.schema().field(8).data_type(),
            &DataType::Decimal256(39, 0)
        );
        assert_eq!(
            batch.schema().field(9).data_type(),
            &DataType::Decimal256(41, 2)
        );

        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", "SQLite", 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,
                "SQLite",
                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);

        let huge = batch
            .column(7)
            .as_any()
            .downcast_ref::<Decimal256Array>()
            .unwrap();
        assert_eq!(
            huge.value(0),
            i256::from_i128(-17_014_118_346_046_923_173_168_730_371_588_410_572_i128)
        );

        let uhuge = batch
            .column(8)
            .as_any()
            .downcast_ref::<Decimal256Array>()
            .unwrap();
        assert_eq!(
            uhuge.value(0),
            "340282366920938463463374607431768211455"
                .parse::<i256>()
                .unwrap()
        );

        let wide_decimal = batch
            .column(9)
            .as_any()
            .downcast_ref::<Decimal256Array>()
            .unwrap();
        assert_eq!(
            wide_decimal.value(0),
            "12345678901234567890123456789012345678912"
                .parse::<i256>()
                .unwrap()
        );
    });
}

#[test]
fn sqlite_merge_adjacent_compaction_replaces_sources_and_schedules_cleanup() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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_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);

            INSERT INTO ducklake_metadata VALUES ('data_path', '/tmp/ducklake-data/', NULL, NULL);
            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', NULL, true);
            INSERT INTO ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440002', 1, NULL, 1, 'numbers', NULL, true);
            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 = SqliteCatalog::new(temp.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!(
        sqlite_data_file_ids(temp.path()),
        vec![(20, "compact-20.parquet".to_string())]
    );
    assert_eq!(
        sqlite_table_count(temp.path(), "ducklake_file_column_stats"),
        1
    );
    assert_eq!(
        sqlite_table_count(temp.path(), "ducklake_file_variant_stats"),
        0
    );
    assert_eq!(scheduled_file_count(temp.path()), 2);
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 2);
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec!["merge_adjacent:1".to_string()]
    );
    assert_eq!(sqlite_table_stats_file_size(temp.path(), TableId(1)), 95);
    assert_eq!(sqlite_latest_next_file_id(temp.path()), 21);
}

#[test]
fn sqlite_merge_adjacent_compaction_persists_partial_file_metadata() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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_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_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_files_scheduled_for_deletion(data_file_id BIGINT, path VARCHAR, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);

            INSERT INTO ducklake_metadata VALUES ('data_path', '/tmp/ducklake-data/', NULL, NULL);
            INSERT INTO ducklake_snapshot VALUES
                (1, '2026-06-26 00:00:00+00', 1, 10, 20),
                (2, '2026-06-26 00:01:00+00', 1, 10, 20);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', NULL, true);
            INSERT INTO ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440002', 1, NULL, 1, 'numbers', NULL, true);
            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, 2, NULL, 1, 'source-b.parquet', true, 'parquet', 1, 10, NULL, NULL, NULL, NULL, NULL, NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::new(temp.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(2),
                source_data_file_ids: vec![DataFileId(10), DataFileId(11)],
                compacted_files: vec![AppendFile {
                    begin_snapshot: Some(SnapshotId(1)),
                    partial_max: Some(SnapshotId(2)),
                    record_count: 2,
                    file_size_bytes: 20,
                    path: "compact-20.parquet".into(),
                    ..append_file(20, SnapshotId(2))
                }],
            })
            .await
            .unwrap();
        assert_eq!(new_snapshot, SnapshotId(3));
    });
    let conn = Connection::open(temp.path()).unwrap();
    let data_file: (i64, i64, Option<i64>) = conn
        .query_row(
            "SELECT data_file_id, begin_snapshot, partial_max FROM ducklake_data_file",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
        )
        .unwrap();
    assert_eq!(data_file, (20, 1, Some(2)));
    assert_eq!(scheduled_file_count(temp.path()), 2);
}

#[test]
fn sqlite_truncate_table_closes_active_files_without_allocating_file_ids() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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_1(row_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, value BIGINT);

            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 50);
            INSERT INTO ducklake_data_file VALUES
                (10, 1, 1, NULL, 0, 'file-a.parquet', true, 'parquet', 5, 40, NULL, NULL, NULL, NULL, NULL, NULL),
                (11, 1, 1, NULL, 1, 'file-b.parquet', true, 'parquet', 5, 60, NULL, NULL, NULL, NULL, NULL, NULL);
            INSERT INTO ducklake_delete_file VALUES
                (20, 1, 1, NULL, 10, 'delete-a.parquet', true, 'parquet', 1, 10, NULL, NULL, NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 11, 12, 100);
            INSERT INTO ducklake_inlined_data_1_1 VALUES
                (0, 1, NULL, 100),
                (1, 1, NULL, 101),
                (2, 1, 1, 102);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::new(temp.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: 11,
                inlined_data_deletions: vec![InlinedDataDeletionCommit {
                    table_name: "ducklake_inlined_data_1_1".into(),
                    row_ids: vec![0, 1],
                    rows_deleted: 2,
                }],
                changes_made: "deleted_from_table:1".into(),
                commit_message: "test truncate".into(),
            })
            .await
            .unwrap();
        assert_eq!(snapshot, SnapshotId(2));
    });

    let conn = Connection::open(temp.path()).unwrap();
    let active_data_files: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_data_file WHERE end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let active_delete_files: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_delete_file WHERE end_snapshot IS NULL",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let record_count: i64 = conn
        .query_row(
            "SELECT record_count FROM ducklake_table_stats WHERE table_id = 1",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let file_size_bytes: i64 = conn
        .query_row(
            "SELECT file_size_bytes FROM ducklake_table_stats WHERE table_id = 1",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let closed_inlined_rows: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_inlined_data_1_1 \
                 WHERE row_id IN (0, 1) AND end_snapshot = 2",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let visible_inlined_latest: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM ducklake_inlined_data_1_1 \
                 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_1 \
                 WHERE 1 >= begin_snapshot AND (1 < end_snapshot OR end_snapshot IS NULL)",
            [],
            |row| row.get(0),
        )
        .unwrap();

    assert_eq!(active_data_files, 0);
    assert_eq!(active_delete_files, 0);
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_data_file"), 2);
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_delete_file"), 1);
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 2);
    assert_eq!(sqlite_latest_next_file_id(temp.path()), 50);
    assert_eq!(
        sqlite_snapshot_changes(temp.path()),
        vec!["deleted_from_table:1"]
    );
    assert_eq!(record_count, 0);
    assert_eq!(file_size_bytes, 0);
    assert_eq!(closed_inlined_rows, 2);
    assert_eq!(visible_inlined_latest, 0);
    assert_eq!(visible_inlined_snapshot_one, 2);
}

#[test]
fn sqlite_write_commits_reject_stale_expected_snapshot() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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);

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

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        let append_err = backend
            .commit_append_files(vec![append_file(10, SnapshotId(1))])
            .await
            .unwrap_err();
        assert_snapshot_conflict_at(&append_err, "append", 1, 2);

        let delete_err = backend
            .commit_delete_files(vec![delete_file_commit(10, 1, SnapshotId(1))])
            .await
            .unwrap_err();
        assert_snapshot_conflict_at(&delete_err, "delete", 1, 2);

        let mixed_err = backend
            .commit_table_changes(TableChangeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                append_files: vec![append_file(10, SnapshotId(1))],
                delete_files: vec![delete_file_commit(11, 1, SnapshotId(1))],
                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: "test update".into(),
            })
            .await
            .unwrap_err();
        assert_snapshot_conflict_at(&mixed_err, "table-change", 1, 2);

        let inlined_err = backend
            .commit_table_changes(TableChangeCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                append_files: Vec::new(),
                delete_files: Vec::new(),
                inlined_file_deletions: vec![inlined_file_deletion_commit(1, SnapshotId(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: "test inlined delete".into(),
            })
            .await
            .unwrap_err();
        assert_snapshot_conflict_at(&inlined_err, "table-change", 1, 2);
    });
}

#[test]
fn sqlite_append_sets_row_id_start_from_table_stats() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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_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, 5, 5, 100);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        backend
            .commit_append_files(vec![AppendFile {
                record_count: 2,
                file_size_bytes: 25,
                ..append_file(10, SnapshotId(1))
            }])
            .await
            .unwrap();
    });

    let conn = Connection::open(temp.path()).unwrap();
    let appended_file: (i64, i64) = conn
        .query_row(
            "SELECT row_id_start, record_count FROM ducklake_data_file WHERE data_file_id = 10",
            [],
            |row| Ok((row.get(0)?, row.get(1)?)),
        )
        .unwrap();
    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 = 1",
                [],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();

    assert_eq!(appended_file, (5, 2));
    assert_eq!(table_stats, (7, 7, 125));
}

#[test]
fn sqlite_append_honors_begin_snapshot_override() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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);

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

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        backend
            .commit_append_files(vec![
                AppendFile {
                    begin_snapshot: Some(SnapshotId(1)),
                    ..append_file(10, SnapshotId(3))
                },
                append_file(11, SnapshotId(3)),
            ])
            .await
            .unwrap();
    });

    let conn = Connection::open(temp.path()).unwrap();
    let begin_snapshots: Vec<(i64, i64)> = conn
        .prepare(
            "SELECT data_file_id, begin_snapshot FROM ducklake_data_file ORDER BY data_file_id",
        )
        .unwrap()
        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
        .unwrap()
        .collect::<std::result::Result<_, _>>()
        .unwrap();

    assert_eq!(begin_snapshots, vec![(10, 1), (11, 4)]);
}

#[test]
fn sqlite_append_writes_file_variant_stats() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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_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_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_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, 'v', 'variant', NULL, NULL, true, NULL, 'literal', NULL);
            "#,
        )
        .unwrap();
    drop(conn);

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        backend
            .commit_append_files(vec![AppendFile {
                variant_stats: vec![FileVariantStats {
                    data_file_id: DataFileId(10),
                    table_id: TableId(1),
                    column_id: ColumnId(2),
                    variant_path: "\"a\"".into(),
                    shredded_type: "int32".into(),
                    column_size_bytes: Some(8),
                    value_count: Some(2),
                    null_count: Some(0),
                    min_value: Some("20".into()),
                    max_value: Some("30".into()),
                    contains_nan: Some(false),
                    extra_stats: None,
                }],
                ..append_file(10, SnapshotId(1))
            }])
            .await
            .unwrap();
    });

    let conn = Connection::open(temp.path()).unwrap();
    let stats: (i64, i64, String, String, Option<String>, Option<String>) = conn
        .query_row(
            "SELECT data_file_id, column_id, variant_path, shredded_type, min_value, max_value \
                 FROM ducklake_file_variant_stats",
            [],
            |row| {
                Ok((
                    row.get(0)?,
                    row.get(1)?,
                    row.get(2)?,
                    row.get(3)?,
                    row.get(4)?,
                    row.get(5)?,
                ))
            },
        )
        .unwrap();
    assert_eq!(
        stats,
        (
            10,
            2,
            "\"a\"".to_string(),
            "int32".to_string(),
            Some("20".to_string()),
            Some("30".to_string())
        )
    );

    let extra_stats: String = conn
        .query_row(
            "SELECT extra_stats \
                 FROM ducklake_table_column_stats WHERE table_id = 1 AND column_id = 2",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let extra_stats = serde_json::from_str::<Vec<serde_json::Value>>(&extra_stats).unwrap();
    assert_eq!(extra_stats.len(), 1);
    let variant = extra_stats[0].as_object().unwrap();
    assert_eq!(variant["field_name"], "\"a\"");
    assert_eq!(variant["shredded_type"], "int32");
    assert_eq!(variant["min"], "20");
    assert_eq!(variant["max"], "30");
    assert_eq!(variant["num_values"], 2);
    assert_eq!(variant["null_count"], 0);
}

#[test]
fn sqlite_append_updates_table_column_stats() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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_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_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 (1, 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, 1, false, false, '10', '20', NULL);
            "#,
        )
        .unwrap();
    drop(conn);

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

    let conn = Connection::open(temp.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 = 1",
            [],
            |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 sqlite_inlined_append_updates_table_column_stats() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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_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_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_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_column VALUES (1, 1, NULL, 1, 0, 'id', 'int32', NULL, NULL, true, NULL, 'literal', NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 1, 1, 100);
            INSERT INTO ducklake_table_column_stats VALUES (1, 1, false, false, '10', '20', NULL);
            "#,
    )
    .unwrap();
    drop(conn);

    let schema = TableSchema {
        columns: vec![schema_field(1, "id", "int32", DuckLakeType::Int32)],
        by_field_id: HashMap::from([(ducklake_catalog::ColumnId(1), vec![0])]),
    };
    let arrow_schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, true)]));
    let batch =
        RecordBatch::try_new(arrow_schema, vec![Arc::new(Int32Array::from(vec![5, 30]))]).unwrap();
    let column_stats = vec![test_file_column_stats(
        0,
        TableId(1),
        ducklake_catalog::ColumnId(1),
        None,
        2,
        0,
        "5",
        "30",
    )];

    let backend = SqliteCatalog::new(temp.path());
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(async {
        backend
            .commit_append_inlined_data(InlinedDataAppendCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(1),
                schema_version: 1,
                schema,
                batches: vec![batch],
                column_stats,
                rows_appended: 2,
                changes_made: "inserted_into_table:1".into(),
                commit_message: "sqlite inlined append stats".into(),
            })
            .await
            .unwrap();
    });

    let conn = Connection::open(temp.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 = 1",
            [],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .unwrap();
    assert_eq!(
        stats,
        (
            Some(false),
            Some(false),
            Some("5".to_string()),
            Some("30".to_string())
        )
    );
}

#[test]
fn sqlite_inlined_append_writes_reference_scalar_types() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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(41,2)",
                DuckLakeType::Decimal {
                    width: 41,
                    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::Decimal256(41, 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 = "12345678901234567890123456789012345678912"
        .parse::<i256>()
        .unwrap();
    let day_value = parse_inlined_date32("2024-01-15", "SQLite", 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, "SQLite", true)
            .unwrap();
    let timestamp_ms = parse_inlined_timestamp(
        "2024-01-15 12:30:00.123+00",
        TimeUnit::Millisecond,
        "SQLite",
        true,
    )
    .unwrap();
    let timestamp_value = parse_inlined_timestamp(
        "2024-01-15 12:30:00.123456+00",
        TimeUnit::Microsecond,
        "SQLite",
        true,
    )
    .unwrap();
    let timestamp_ns = parse_inlined_timestamp(
        "2024-01-15 12:30:00.123456789+00",
        TimeUnit::Nanosecond,
        "SQLite",
        true,
    )
    .unwrap();
    let timestamp_tz = parse_inlined_timestamp(
        "2024-01-15 12:30:00.123456+00",
        TimeUnit::Microsecond,
        "SQLite",
        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(
                Decimal256Array::from(vec![Some(wide_decimal_value), None])
                    .with_precision_and_scale(41, 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 = SqliteCatalog::new(temp.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: "sqlite 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::<Decimal256Array>()
            .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(temp.path()).unwrap();
    let mut stmt = conn
        .prepare("PRAGMA table_info(ducklake_inlined_data_1_1)")
        .unwrap();
    let column_types = stmt
        .query_map([], |row| {
            Ok((row.get::<_, String>(1)?, row.get::<_, String>(2)?))
        })
        .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"], "VARCHAR");
    assert_eq!(column_types["real64"], "VARCHAR");
    assert_eq!(column_types["payload"], "BLOB");
    assert_eq!(column_types["uuid_value"], "VARCHAR");
    assert_eq!(column_types["ubig"], "VARCHAR");
    assert_eq!(column_types["huge"], "VARCHAR");
    assert_eq!(column_types["uhuge"], "VARCHAR");
    assert_eq!(column_types["decimal_value"], "VARCHAR");
    assert_eq!(column_types["wide_decimal"], "VARCHAR");
    assert_eq!(column_types["day_value"], "VARCHAR");
    assert_eq!(column_types["time_value"], "VARCHAR");
    assert_eq!(column_types["time_ns_value"], "VARCHAR");
    assert_eq!(column_types["timestamp_value"], "VARCHAR");
    assert_eq!(column_types["timestamp_tz"], "VARCHAR");
    assert_eq!(column_types["timetz_value"], "VARCHAR");
    assert_eq!(column_types["interval_value"], "VARCHAR");
}

#[test]
fn sqlite_concurrent_append_commits_return_one_conflict() {
    let temp = tempfile::NamedTempFile::new().unwrap();
    let conn = Connection::open(temp.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);

            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 = SqliteCatalog::new(temp.path());
        let right = SqliteCatalog::new(temp.path());
        let (left, right) = tokio::join!(
            left.commit_append_files(vec![append_file(10, SnapshotId(1))]),
            right.commit_append_files(vec![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_at(&err, "append", 1, 2);
                    conflicts += 1;
                }
            }
        }
        assert_eq!(successes, 1);
        assert_eq!(conflicts, 1);
    });

    assert_eq!(sqlite_table_count(temp.path(), "ducklake_snapshot"), 2);
    assert_eq!(sqlite_table_count(temp.path(), "ducklake_data_file"), 1);
}

fn assert_unsupported_version<T>(result: Result<T>) {
    match result {
        Err(StorageError::Catalog(CatalogError::UnsupportedVersion { found, supported })) => {
            assert_eq!(found, "2.0");
            assert_eq!(supported, SUPPORTED_VERSION);
        }
        Err(other) => panic!("expected unsupported version error, got {other:?}"),
        Ok(_) => panic!("expected unsupported version error, got success"),
    }
}

fn write_minimal_scan_catalog(
    catalog_path: &Path,
    data_encryption_key: Option<&str>,
    delete_encryption_key: Option<&str>,
) {
    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_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', '', NULL, NULL),
                ('encrypted', 'false', NULL, NULL);
            INSERT INTO ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 20, 20);
            INSERT INTO ducklake_schema VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 'main', '', true);
            INSERT INTO ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440010', 1, NULL, 1, 'numbers', '', true);
            INSERT INTO ducklake_column VALUES (1, 1, NULL, 1, 0, 'id', 'int64', NULL, NULL, false, NULL, 'literal', NULL);
            "#,
        )
        .unwrap();
    conn.execute(
            "INSERT INTO ducklake_data_file VALUES \
             (10, 1, 1, NULL, 0, 'file.parquet', true, 'parquet', 1, 10, NULL, 0, NULL, ?1, NULL, NULL)",
            params![data_encryption_key],
        )
        .unwrap();
    if delete_encryption_key.is_some() {
        conn.execute(
            "INSERT INTO ducklake_delete_file VALUES \
                 (11, 1, 1, NULL, 10, 'delete.parquet', true, 'parquet', 0, 10, NULL, ?1, NULL)",
            params![delete_encryption_key],
        )
        .unwrap();
    }
}

fn append_file(data_file_id: i64, expected_snapshot: SnapshotId) -> AppendFile {
    TestAppendFile {
        table_id: TableId(1),
        data_file_id,
        expected_snapshot,
        path: format!("file-{data_file_id}.parquet"),
        record_count: 1,
        file_size_bytes: 10,
        column_stats: Vec::new(),
        partition_id: None,
        partition_values: Vec::new(),
    }
    .into_commit()
}

fn 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(1),
            None,
            value_count,
            null_count,
            min_value,
            max_value,
        )],
        ..append_file(data_file_id, expected_snapshot)
    }
}

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

fn inlined_file_deletion_commit(
    data_file_id: i64,
    expected_snapshot: SnapshotId,
) -> InlinedFileDeletionCommit {
    test_inlined_file_deletion(TableId(1), data_file_id, expected_snapshot, vec![0])
}

fn create_column_nullability_catalog(catalog_path: &Path, contains_null: bool) {
    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_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_column VALUES (1, 1, NULL, 1, 0, 'id', 'int32', NULL, NULL, true, NULL, 'literal', NULL);
            INSERT INTO ducklake_table_stats VALUES (1, 2, 2, 10);
            "#,
        )
        .unwrap();
    conn.execute(
        "INSERT INTO ducklake_table_column_stats VALUES (1, 1, ?1, false, '1', '2', NULL)",
        params![contains_null],
    )
    .unwrap();
}

fn create_column_alter_catalog(catalog_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();
}

fn create_partitioning_catalog(catalog_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_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 (1, '2026-06-26 00:00:00+00', 1, 2, 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');
            "#,
        )
        .unwrap();
}

fn create_partitioning_multi_table_catalog(catalog_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_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();
}

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

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

type ColumnDefaultHistoryRow = (
    i64,
    Option<i64>,
    String,
    Option<String>,
    Option<String>,
    Option<String>,
    Option<String>,
);

fn column_default_history(conn: &Connection, column_id: i64) -> Vec<ColumnDefaultHistoryRow> {
    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 = ?1 \
                 ORDER BY begin_snapshot",
        )
        .unwrap();
    stmt.query_map(params![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 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 = ?1 \
                 ORDER BY begin_snapshot",
        )
        .unwrap();
    stmt.query_map(params![column_id], |row| {
        Ok((row.get(0)?, row.get(1)?, row.get(2)?))
    })
    .unwrap()
    .collect::<std::result::Result<Vec<_>, _>>()
    .unwrap()
}

fn scheduled_file_count(catalog_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 scheduled_file_paths(catalog_path: &Path) -> Vec<String> {
    let conn = Connection::open(catalog_path).unwrap();
    let mut stmt = conn
        .prepare("SELECT path FROM ducklake_files_scheduled_for_deletion ORDER BY path")
        .unwrap();
    stmt.query_map([], |row| row.get(0))
        .unwrap()
        .collect::<rusqlite::Result<Vec<_>>>()
        .unwrap()
}

fn sqlite_table_count(catalog_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 sqlite_catalog_table_exists(catalog_path: &Path, table: &str) -> bool {
    let conn = Connection::open(catalog_path).unwrap();
    sqlite_table_exists(&conn, table).unwrap()
}

fn sqlite_data_file_ids(catalog_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 sqlite_snapshot_changes(catalog_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 sqlite_table_stats_file_size(catalog_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 sqlite_latest_next_file_id(catalog_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()
}
