use super::*;
use crate::test_support::{
    TestAppendFile, TestDeleteFile, assert_snapshot_conflict, test_file_column_stats,
    test_file_partition_value, test_schema_field,
};
use crate::{
    AddColumn, AppendFile, CatalogBackend, CreateTableColumn, CreateTableWithFilesCommit,
    PartitionColumnCommit, PartitioningBatchCommit, PartitioningTableCommit, SortExpressionCommit,
};
use ducklake_catalog::{DeleteFileId, DeleteFormat, DuckLakeType, PartitionTransform, ViewId};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

#[test]
fn qualifies_metadata_table_names_with_quoted_schema() {
    let catalog = PostgresCatalog::with_metadata_schema("host=localhost", "ducklake meta");
    assert_eq!(
        catalog.table("ducklake_snapshot"),
        "\"ducklake meta\".\"ducklake_snapshot\""
    );
}

#[test]
fn escapes_quoted_identifiers() {
    assert_eq!(quote_ident("a\"b"), "\"a\"\"b\"");
}

#[test]
fn inlined_projection_casts_scalars_to_text_and_bytea_to_hex() {
    let text_field = scalar_field("name", DuckLakeType::Varchar);
    assert_eq!(
        crate::inlined_values::inlined_read_projection_expr(
            &text_field,
            crate::inlined_values::InlinedSqlDialect::Postgres,
            quote_ident,
        )
        .unwrap(),
        "encode(\"name\", 'hex') AS \"name\""
    );

    let blob_field = scalar_field("payload", DuckLakeType::Blob);
    assert_eq!(
        crate::inlined_values::inlined_read_projection_expr(
            &blob_field,
            crate::inlined_values::InlinedSqlDialect::Postgres,
            quote_ident,
        )
        .unwrap(),
        "encode(\"payload\", 'hex') AS \"payload\""
    );

    let uhuge_field = scalar_field("uhuge", DuckLakeType::UInt128);
    assert_eq!(
        inlined_arrow_field_from_schema(&uhuge_field)
            .unwrap()
            .data_type(),
        &DataType::Decimal256(39, 0)
    );
}

#[test]
fn postgres_inlined_append_scalar_helpers_follow_reference_mapping() {
    let text = scalar_field("label", DuckLakeType::Varchar);
    assert_eq!(
        inlined_column_type(&text, InlinedSqlDialect::Postgres).unwrap(),
        "BYTEA"
    );
    let values = StringArray::from(vec![Some("hello"), None]);
    assert!(postgres_inlined_value(&text, &values, 0).is_ok());
    assert!(postgres_inlined_value(&text, &values, 1).is_ok());

    let json = scalar_field("payload", DuckLakeType::Json);
    assert_eq!(
        inlined_column_type(&json, InlinedSqlDialect::Postgres).unwrap(),
        "BYTEA"
    );
    let values = StringArray::from(vec![Some("{\"key\":\"value\"}"), None]);
    assert!(postgres_inlined_value(&json, &values, 0).is_ok());
    assert!(postgres_inlined_value(&json, &values, 1).is_ok());

    let blob = scalar_field("bytes", DuckLakeType::Blob);
    assert_eq!(
        inlined_column_type(&blob, InlinedSqlDialect::Postgres).unwrap(),
        "BYTEA"
    );
    let blob_bytes = vec![0x00_u8, 0xff, 0x10];
    let values = BinaryArray::from(vec![Some(blob_bytes.as_slice()), None]);
    assert!(postgres_inlined_value(&blob, &values, 0).is_ok());
    assert!(postgres_inlined_value(&blob, &values, 1).is_ok());

    let uuid = scalar_field("uuid_value", DuckLakeType::Uuid);
    assert_eq!(
        inlined_column_type(&uuid, InlinedSqlDialect::Postgres).unwrap(),
        "UUID"
    );
    let uuid_bytes = [
        0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee,
        0xff,
    ];
    let uuid_values = vec![Some(uuid_bytes.as_slice()), None];
    let values =
        FixedSizeBinaryArray::try_from_sparse_iter_with_size(uuid_values.into_iter(), 16).unwrap();
    assert!(postgres_inlined_value(&uuid, &values, 0).is_ok());
    assert!(postgres_inlined_value(&uuid, &values, 1).is_ok());

    let decimal = scalar_field(
        "decimal_value",
        DuckLakeType::Decimal { width: 8, scale: 2 },
    );
    let time = scalar_field("time_value", DuckLakeType::Time);
    let timetz = scalar_field("timetz_value", DuckLakeType::TimeTz);
    let interval = scalar_field("interval_value", DuckLakeType::Interval);
    let placeholder_schema = TableSchema {
        columns: vec![
            text.clone(),
            decimal.clone(),
            time.clone(),
            timetz.clone(),
            interval.clone(),
            uuid.clone(),
        ],
        by_field_id: HashMap::new(),
    };
    assert_eq!(
        postgres_inlined_insert_placeholders(&placeholder_schema).unwrap(),
        "$1, $2, $3, $4, CAST($5 AS DECIMAL(8,2)), CAST($6 AS TIME), CAST($7 AS TIMETZ), CAST($8 AS INTERVAL), CAST($9 AS UUID)"
    );

    let tiny = scalar_field("tiny", DuckLakeType::Int8);
    assert_eq!(
        inlined_column_type(&tiny, InlinedSqlDialect::Postgres).unwrap(),
        "SMALLINT"
    );
    let values = Int8Array::from(vec![Some(-8), None]);
    assert!(postgres_inlined_value(&tiny, &values, 0).is_ok());
    assert!(postgres_inlined_value(&tiny, &values, 1).is_ok());

    let small = scalar_field("small", DuckLakeType::Int16);
    assert_eq!(
        inlined_column_type(&small, InlinedSqlDialect::Postgres).unwrap(),
        "SMALLINT"
    );
    let values = Int16Array::from(vec![Some(-1_600), None]);
    assert!(postgres_inlined_value(&small, &values, 0).is_ok());
    assert!(postgres_inlined_value(&small, &values, 1).is_ok());

    let utiny = scalar_field("utiny", DuckLakeType::UInt8);
    assert_eq!(
        inlined_column_type(&utiny, InlinedSqlDialect::Postgres).unwrap(),
        "INTEGER"
    );
    let values = UInt8Array::from(vec![Some(250), None]);
    assert!(postgres_inlined_value(&utiny, &values, 0).is_ok());
    assert!(postgres_inlined_value(&utiny, &values, 1).is_ok());

    let usmall = scalar_field("usmall", DuckLakeType::UInt16);
    assert_eq!(
        inlined_column_type(&usmall, InlinedSqlDialect::Postgres).unwrap(),
        "INTEGER"
    );
    let values = UInt16Array::from(vec![Some(60_000), None]);
    assert!(postgres_inlined_value(&usmall, &values, 0).is_ok());
    assert!(postgres_inlined_value(&usmall, &values, 1).is_ok());

    let uint = scalar_field("uint", DuckLakeType::UInt32);
    assert_eq!(
        inlined_column_type(&uint, InlinedSqlDialect::Postgres).unwrap(),
        "BIGINT"
    );
    let values = UInt32Array::from(vec![Some(4_000_000_000), None]);
    assert!(postgres_inlined_value(&uint, &values, 0).is_ok());
    assert!(postgres_inlined_value(&uint, &values, 1).is_ok());

    let ubig = scalar_field("ubig", DuckLakeType::UInt64);
    assert_eq!(
        inlined_column_type(&ubig, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = UInt64Array::from(vec![Some(u64::MAX), None]);
    assert!(postgres_inlined_value(&ubig, &values, 0).is_ok());
    assert!(postgres_inlined_value(&ubig, &values, 1).is_ok());

    let huge = scalar_field("huge", DuckLakeType::Int128);
    assert_eq!(
        inlined_column_type(&huge, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = Decimal256Array::from(vec![
        Some(i256::from_i128(
            -123_456_789_012_345_678_901_234_567_890_i128,
        )),
        None,
    ])
    .with_precision_and_scale(39, 0)
    .unwrap();
    assert!(postgres_inlined_value(&huge, &values, 0).is_ok());
    assert!(postgres_inlined_value(&huge, &values, 1).is_ok());

    let uhuge = scalar_field("uhuge", DuckLakeType::UInt128);
    assert_eq!(
        inlined_column_type(&uhuge, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let uhuge_value = "340282366920938463463374607431768211455"
        .parse::<i256>()
        .unwrap();
    let values = Decimal256Array::from(vec![Some(uhuge_value), None])
        .with_precision_and_scale(39, 0)
        .unwrap();
    assert!(postgres_inlined_value(&uhuge, &values, 0).is_ok());
    assert!(postgres_inlined_value(&uhuge, &values, 1).is_ok());

    let real32 = scalar_field("real32", DuckLakeType::Float32);
    assert_eq!(
        inlined_column_type(&real32, InlinedSqlDialect::Postgres).unwrap(),
        "REAL"
    );
    let values = Float32Array::from(vec![Some(1.5), None]);
    assert!(postgres_inlined_value(&real32, &values, 0).is_ok());
    assert!(postgres_inlined_value(&real32, &values, 1).is_ok());

    let real64 = scalar_field("real64", DuckLakeType::Float64);
    assert_eq!(
        inlined_column_type(&real64, InlinedSqlDialect::Postgres).unwrap(),
        "DOUBLE PRECISION"
    );
    let values = Float64Array::from(vec![Some(-2.25), None]);
    assert!(postgres_inlined_value(&real64, &values, 0).is_ok());
    assert!(postgres_inlined_value(&real64, &values, 1).is_ok());

    assert_eq!(
        inlined_column_type(&decimal, InlinedSqlDialect::Postgres).unwrap(),
        "DECIMAL(8,2)"
    );
    let values = Decimal128Array::from(vec![Some(12_345), None])
        .with_precision_and_scale(8, 2)
        .unwrap();
    assert!(postgres_inlined_value(&decimal, &values, 0).is_ok());
    assert!(postgres_inlined_value(&decimal, &values, 1).is_ok());

    let wide_decimal = scalar_field(
        "wide_decimal",
        DuckLakeType::Decimal {
            width: 41,
            scale: 2,
        },
    );
    assert_eq!(
        inlined_column_type(&wide_decimal, InlinedSqlDialect::Postgres).unwrap(),
        "DECIMAL(41,2)"
    );
    let wide_decimal_value = "12345678901234567890123456789012345678912"
        .parse::<i256>()
        .unwrap();
    let values = Decimal256Array::from(vec![Some(wide_decimal_value), None])
        .with_precision_and_scale(41, 2)
        .unwrap();
    assert!(postgres_inlined_value(&wide_decimal, &values, 0).is_ok());
    assert!(postgres_inlined_value(&wide_decimal, &values, 1).is_ok());

    let day = scalar_field("day_value", DuckLakeType::Date);
    assert_eq!(
        inlined_column_type(&day, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = Date32Array::from(vec![Some(1), None]);
    assert!(postgres_inlined_value(&day, &values, 0).is_ok());
    assert!(postgres_inlined_value(&day, &values, 1).is_ok());

    assert_eq!(
        inlined_column_type(&time, InlinedSqlDialect::Postgres).unwrap(),
        "TIME"
    );
    let values = Time64MicrosecondArray::from(vec![Some(45_296_123_456), None]);
    assert!(postgres_inlined_value(&time, &values, 0).is_ok());
    assert!(postgres_inlined_value(&time, &values, 1).is_ok());

    let time_ns = scalar_field("time_ns_value", DuckLakeType::TimeNs);
    assert_eq!(
        inlined_column_type(&time_ns, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = Time64NanosecondArray::from(vec![Some(45_296_123_456_789), None]);
    assert!(postgres_inlined_value(&time_ns, &values, 0).is_ok());
    assert!(postgres_inlined_value(&time_ns, &values, 1).is_ok());

    let timestamp_s = scalar_field("timestamp_s", DuckLakeType::TimestampS);
    assert_eq!(
        inlined_column_type(&timestamp_s, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = TimestampSecondArray::from(vec![Some(1), None]);
    assert!(postgres_inlined_value(&timestamp_s, &values, 0).is_ok());
    assert!(postgres_inlined_value(&timestamp_s, &values, 1).is_ok());

    let timestamp_ms = scalar_field("timestamp_ms", DuckLakeType::TimestampMs);
    assert_eq!(
        inlined_column_type(&timestamp_ms, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = TimestampMillisecondArray::from(vec![Some(1_234), None]);
    assert!(postgres_inlined_value(&timestamp_ms, &values, 0).is_ok());
    assert!(postgres_inlined_value(&timestamp_ms, &values, 1).is_ok());

    let timestamp = scalar_field("timestamp_value", DuckLakeType::Timestamp);
    assert_eq!(
        inlined_column_type(&timestamp, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = TimestampMicrosecondArray::from(vec![Some(1_234_567), None]);
    assert!(postgres_inlined_value(&timestamp, &values, 0).is_ok());
    assert!(postgres_inlined_value(&timestamp, &values, 1).is_ok());

    let timestamp_ns = scalar_field("timestamp_ns", DuckLakeType::TimestampNs);
    assert_eq!(
        inlined_column_type(&timestamp_ns, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = TimestampNanosecondArray::from(vec![Some(1_234_567_890), None]);
    assert!(postgres_inlined_value(&timestamp_ns, &values, 0).is_ok());
    assert!(postgres_inlined_value(&timestamp_ns, &values, 1).is_ok());

    let timestamp_tz = scalar_field("timestamp_tz", DuckLakeType::TimestampTz);
    assert_eq!(
        inlined_column_type(&timestamp_tz, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    let values = TimestampMicrosecondArray::from(vec![Some(1_234_567), None]).with_timezone("UTC");
    assert!(postgres_inlined_value(&timestamp_tz, &values, 0).is_ok());
    assert!(postgres_inlined_value(&timestamp_tz, &values, 1).is_ok());

    assert_eq!(
        inlined_column_type(&timetz, InlinedSqlDialect::Postgres).unwrap(),
        "TIMETZ"
    );
    let values = StringArray::from(vec![Some("12:34:56+02"), None]);
    assert!(postgres_inlined_value(&timetz, &values, 0).is_ok());
    assert!(postgres_inlined_value(&timetz, &values, 1).is_ok());

    assert_eq!(
        inlined_column_type(&interval, InlinedSqlDialect::Postgres).unwrap(),
        "INTERVAL"
    );
    let values = IntervalMonthDayNanoArray::from(vec![
        Some(IntervalMonthDayNano::new(14, 3, 4_005_000)),
        None,
    ]);
    assert!(postgres_inlined_value(&interval, &values, 0).is_ok());
    assert!(postgres_inlined_value(&interval, &values, 1).is_ok());
}

#[test]
fn parses_postgres_inlined_scalar_text_values() {
    assert!(parse_inlined_bool_text("t", "PostgreSQL").unwrap());
    assert!(!parse_inlined_bool_text("false", "PostgreSQL").unwrap());
    assert_eq!(parse_inlined_i64_text("-7", "PostgreSQL").unwrap(), -7);
    assert_eq!(parse_inlined_u64_text("7", "PostgreSQL").unwrap(), 7);
    assert_eq!(
        parse_inlined_u64_text("18446744073709551615", "PostgreSQL").unwrap(),
        u64::MAX
    );
    assert_eq!(
        parse_inlined_decimal128("12.30", 2, "PostgreSQL").unwrap(),
        1230
    );
    assert_eq!(
        parse_inlined_decimal256("340282366920938463463374607431768211455", 0, "PostgreSQL")
            .unwrap(),
        "340282366920938463463374607431768211455"
            .parse::<i256>()
            .unwrap()
    );
    assert_eq!(
        parse_inlined_decimal256(
            "123456789012345678901234567890123456789.12",
            2,
            "PostgreSQL"
        )
        .unwrap(),
        "12345678901234567890123456789012345678912"
            .parse::<i256>()
            .unwrap()
    );
    assert_eq!(
        parse_inlined_hex_bytes("0aff", "PostgreSQL").unwrap(),
        vec![0x0a, 0xff]
    );
    assert_eq!(
        parse_inlined_utf8_hex_bytes("68656c6c6f", "PostgreSQL").unwrap(),
        "hello"
    );
    assert_eq!(
        parse_inlined_uuid("00112233-4455-6677-8899-aabbccddeeff", "PostgreSQL").unwrap(),
        [
            0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd,
            0xee, 0xff
        ]
    );
}

#[test]
fn rejects_out_of_range_postgres_inlined_unsigned_values() {
    let mut tiny = InlinedColumn::new(
        &scalar_field("tiny", DuckLakeType::UInt8),
        InlinedSqlDialect::Postgres,
    )
    .unwrap();
    let decoder = TextInlinedValueDecoder::new("PostgreSQL", false, false);
    push_decoded_inlined_column(&mut tiny, Some("255".into()), &decoder).unwrap();
    let tiny = tiny
        .into_array()
        .unwrap()
        .as_any()
        .downcast_ref::<UInt8Array>()
        .unwrap()
        .value(0);
    assert_eq!(tiny, u8::MAX);

    let mut tiny = InlinedColumn::new(
        &scalar_field("tiny", DuckLakeType::UInt8),
        InlinedSqlDialect::Postgres,
    )
    .unwrap();
    let err = push_decoded_inlined_column(&mut tiny, Some("256".into()), &decoder).unwrap_err();
    assert!(format!("{err}").contains("out of range for UInt8"));

    let mut unsigned = InlinedColumn::new(
        &scalar_field("ubig", DuckLakeType::UInt64),
        InlinedSqlDialect::Postgres,
    )
    .unwrap();
    push_decoded_inlined_column(&mut unsigned, Some(u64::MAX.to_string()), &decoder).unwrap();
    let unsigned = unsigned
        .into_array()
        .unwrap()
        .as_any()
        .downcast_ref::<UInt64Array>()
        .unwrap()
        .value(0);
    assert_eq!(unsigned, u64::MAX);
}

#[test]
fn live_postgres_create_drop_table_commits_ducklake_metadata() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_inlined_data_tables(table_id BIGINT, table_name TEXT, schema_version BIGINT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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: "postgres create table".into(),
                })
                .await
                .unwrap();
            assert_eq!(created, SnapshotId(2));

            let created_row = client
                .query_one(
                    &format!(
                        "SELECT table_id, begin_snapshot, end_snapshot, schema_id, table_name, path, path_is_relative \
                         FROM {} WHERE table_name = 'created'",
                        postgres_qualified_table(&schema, "ducklake_table")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(created_row.get::<_, i64>(0), 10);
            assert_eq!(created_row.get::<_, i64>(1), 2);
            assert_eq!(created_row.get::<_, Option<i64>>(2), None);
            assert_eq!(created_row.get::<_, i64>(3), 1);
            assert_eq!(created_row.get::<_, String>(4), "created");
            assert_eq!(created_row.get::<_, String>(5), "created/");
            assert!(created_row.get::<_, bool>(6));
            let columns = client
                .query(
                    &format!(
                        "SELECT column_id, column_order, column_name, column_type, nulls_allowed \
                         FROM {} WHERE table_id = 10 ORDER BY column_id",
                        postgres_qualified_table(&schema, "ducklake_column")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .iter()
                .map(|row| {
                    (
                        row.get::<_, i64>(0),
                        row.get::<_, i64>(1),
                        row.get::<_, String>(2),
                        row.get::<_, String>(3),
                        row.get::<_, bool>(4),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                columns,
                vec![
                    (1, 1, "id".to_string(), "int32".to_string(), false),
                    (2, 2, "name".to_string(), "varchar".to_string(), true),
                ]
            );
            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3)
                ),
                (2, 2, 11, 20)
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["created_table:\"main\".\"created\"".to_string()]
            );

            client
                .batch_execute(&format!(
                    r#"
                    INSERT INTO {qualified}.ducklake_data_file VALUES (20, 10, 2, NULL, 20, 'created.parquet', true, 'parquet', 1, 10, NULL, 0, NULL, NULL, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_delete_file VALUES (21, 10, 2, NULL, 20, 'created-delete.parquet', true, 'deletion-vector', 1, 10, 0, NULL, NULL);
                    CREATE TABLE {qualified}.ducklake_inlined_data_10_2(begin_snapshot BIGINT, end_snapshot BIGINT, row_id BIGINT, id INTEGER, name BYTEA);
                    INSERT INTO {qualified}.ducklake_inlined_data_10_2 VALUES (2, NULL, 0, 7, decode('736576656e', 'hex'));
                    INSERT INTO {qualified}.ducklake_inlined_data_tables VALUES (10, 'ducklake_inlined_data_10_2', 2);
                    "#
                ))
                .await
                .unwrap();

            let dropped = backend
                .commit_drop_table(DropTableCommit {
                    schema_name: "main".into(),
                    table_name: "created".into(),
                    commit_message: "postgres drop table".into(),
                })
                .await
                .unwrap();
            assert_eq!(dropped, SnapshotId(3));
            let table_end: Option<i64> = client
                .query_one(
                    &format!(
                        "SELECT end_snapshot FROM {} WHERE table_id = 10",
                        postgres_qualified_table(&schema, "ducklake_table")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(table_end, Some(3));
            for table in [
                "ducklake_column",
                "ducklake_data_file",
                "ducklake_delete_file",
            ] {
                let active: i64 = client
                    .query_one(
                        &format!(
                            "SELECT COUNT(*) FROM {} WHERE table_id = 10 AND end_snapshot IS NULL",
                            postgres_qualified_table(&schema, table)
                        ),
                        &[],
                    )
                    .await
                    .unwrap()
                    .get(0);
                assert_eq!(active, 0, "{table}");
            }
            let active_inlined_rows: i64 = client
                .query_one(
                    &format!(
                        "SELECT COUNT(*) FROM {} WHERE end_snapshot IS NULL",
                        postgres_qualified_table(&schema, "ducklake_inlined_data_10_2")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(active_inlined_rows, 0);
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec![
                    "created_table:\"main\".\"created\"".to_string(),
                    "dropped_table:10".to_string()
                ]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_create_drop_schema_commits_ducklake_metadata() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name TEXT, dialect TEXT, sql TEXT, column_aliases TEXT);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_inlined_data_tables(table_id BIGINT, table_name TEXT, schema_version BIGINT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let created_schema = backend
                .commit_create_schema(CreateSchemaCommit {
                    schema_name: "analytics".into(),
                    commit_message: "postgres create schema".into(),
                })
                .await
                .unwrap();
            assert_eq!(created_schema, SnapshotId(2));

            let schema_row = client
                .query_one(
                    &format!(
                        "SELECT schema_id, begin_snapshot, end_snapshot, schema_name, path, path_is_relative \
                         FROM {} WHERE schema_name = 'analytics'",
                        postgres_qualified_table(&schema, "ducklake_schema")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(schema_row.get::<_, i64>(0), 10);
            assert_eq!(schema_row.get::<_, i64>(1), 2);
            assert_eq!(schema_row.get::<_, Option<i64>>(2), None);
            assert_eq!(schema_row.get::<_, String>(3), "analytics");
            assert_eq!(schema_row.get::<_, String>(4), "analytics/");
            assert!(schema_row.get::<_, bool>(5));

            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: "postgres create schema table".into(),
                })
                .await
                .unwrap();
            assert_eq!(created_table, SnapshotId(3));

            client
                .batch_execute(&format!(
                    r#"
                    INSERT INTO {qualified}.ducklake_data_file VALUES (20, 11, 3, NULL, 20, 'analytics/events.parquet', true, 'parquet', 1, 10, NULL, 0, NULL, NULL, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_delete_file VALUES (21, 11, 3, NULL, 20, 'analytics/events-delete.parquet', true, 'parquet', 1, 10, 0, NULL, NULL);
                    CREATE TABLE {qualified}.ducklake_inlined_data_11_3(begin_snapshot BIGINT, end_snapshot BIGINT, row_id BIGINT, id INTEGER, name BYTEA);
                    INSERT INTO {qualified}.ducklake_inlined_data_11_3 VALUES (3, NULL, 0, 7, decode('736576656e', 'hex'));
                    INSERT INTO {qualified}.ducklake_inlined_data_tables VALUES (11, 'ducklake_inlined_data_11_3', 3);
                    "#
                ))
                .await
                .unwrap();

            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: "postgres 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: "postgres 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: "postgres cascade drop schema".into(),
                })
                .await
                .unwrap();
            assert_eq!(dropped, SnapshotId(5));

            let schema_end: Option<i64> = client
                .query_one(
                    &format!(
                        "SELECT end_snapshot FROM {} WHERE schema_id = 10",
                        postgres_qualified_table(&schema, "ducklake_schema")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(schema_end, Some(5));
            let table_end: Option<i64> = client
                .query_one(
                    &format!(
                        "SELECT end_snapshot FROM {} WHERE table_id = 11",
                        postgres_qualified_table(&schema, "ducklake_table")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(table_end, Some(5));
            let view_end: Option<i64> = client
                .query_one(
                    &format!(
                        "SELECT end_snapshot FROM {} WHERE view_id = 12",
                        postgres_qualified_table(&schema, "ducklake_view")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(view_end, Some(5));

            for table in [
                "ducklake_column",
                "ducklake_data_file",
                "ducklake_delete_file",
            ] {
                let active: i64 = client
                    .query_one(
                        &format!(
                            "SELECT COUNT(*) FROM {} WHERE table_id = 11 AND end_snapshot IS NULL",
                            postgres_qualified_table(&schema, table)
                        ),
                        &[],
                    )
                    .await
                    .unwrap()
                    .get(0);
                assert_eq!(active, 0, "{table}");
            }
            let active_inlined_rows: i64 = client
                .query_one(
                    &format!(
                        "SELECT COUNT(*) FROM {} WHERE end_snapshot IS NULL",
                        postgres_qualified_table(&schema, "ducklake_inlined_data_11_3")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(active_inlined_rows, 0);
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                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(),
                ]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_create_view_commits_ducklake_metadata() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name TEXT, dialect TEXT, sql TEXT, column_aliases TEXT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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: "postgres create view".into(),
                })
                .await
                .unwrap();
            assert_eq!(created, SnapshotId(2));

            let view = client
                .query_one(
                    &format!(
                        "SELECT view_id, begin_snapshot, end_snapshot, schema_id, view_name, dialect, sql, column_aliases \
                         FROM {} WHERE view_name = 'v_numbers'",
                        postgres_qualified_table(&schema, "ducklake_view")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(view.get::<_, i64>(0), 10);
            assert_eq!(view.get::<_, i64>(1), 2);
            assert_eq!(view.get::<_, Option<i64>>(2), None);
            assert_eq!(view.get::<_, i64>(3), 1);
            assert_eq!(view.get::<_, String>(4), "v_numbers");
            assert_eq!(view.get::<_, String>(5), "duckdb");
            assert_eq!(
                view.get::<_, String>(6),
                "SELECT id, name FROM {DUCKLAKE_CATALOG}.main.numbers"
            );
            assert_eq!(view.get::<_, String>(7), "\"id_alias\",\"name_alias\"");
            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (2, 2, 11, 20)
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["created_view:\"main\".\"v_numbers\"".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_drop_view_commits_ducklake_metadata() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name TEXT, dialect TEXT, sql TEXT, column_aliases TEXT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let dropped = backend
                .commit_drop_view(DropViewCommit {
                    schema_name: "main".into(),
                    view_name: "v_numbers".into(),
                    commit_message: "postgres drop view".into(),
                })
                .await
                .unwrap();
            assert_eq!(dropped, SnapshotId(2));

            let view_end: Option<i64> = client
                .query_one(
                    &format!(
                        "SELECT end_snapshot FROM {} WHERE view_id = 9",
                        postgres_qualified_table(&schema, "ducklake_view")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(view_end, Some(2));
            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (2, 2, 10, 20)
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["dropped_view:9".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_table_and_view_lookup_is_case_insensitive_for_catalog_commits() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_metadata(key TEXT, value TEXT, scope TEXT, scope_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name TEXT, dialect TEXT, sql TEXT, column_aliases TEXT);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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: "postgres 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: "postgres 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: "postgres drop view by case-insensitive source".into(),
                })
                .await
                .unwrap();
            assert_eq!(dropped_view, SnapshotId(4));

            let active_tables: i64 = client
                .query_one(
                    &format!(
                        "SELECT COUNT(*) FROM {} WHERE end_snapshot IS NULL",
                        postgres_qualified_table(&schema, "ducklake_table")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            let active_views: i64 = client
                .query_one(
                    &format!(
                        "SELECT COUNT(*) FROM {} WHERE end_snapshot IS NULL",
                        postgres_qualified_table(&schema, "ducklake_view")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!((active_tables, active_views), (0, 0));
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec![
                    "created_table:\"main\".\"renamed\"".to_string(),
                    "dropped_table:2".to_string(),
                    "dropped_view:3".to_string(),
                ]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_table_and_view_lookup_prefers_exact_match_and_rejects_ambiguous_case() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_metadata(key TEXT, value TEXT, scope TEXT, scope_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name TEXT, dialect TEXT, sql TEXT, column_aliases TEXT);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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: "postgres 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: "postgres ambiguous drop view source".into(),
                })
                .await
                .unwrap_err();
            assert!(
                ambiguous_view
                    .to_string()
                    .contains("ambiguous DuckLake view identifier"),
                "{ambiguous_view}"
            );
            let latest_snapshot: i64 = client
                .query_one(
                    &format!(
                        "SELECT MAX(snapshot_id) FROM {}",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(latest_snapshot, 1);

            let dropped_table = backend
                .commit_drop_table(DropTableCommit {
                    schema_name: "main".into(),
                    table_name: "NUMBERS".into(),
                    commit_message: "postgres 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: "postgres drop exact uppercase view source".into(),
                })
                .await
                .unwrap();
            assert_eq!(dropped_view, SnapshotId(3));

            let active_tables = client
                .query(
                    &format!(
                        "SELECT table_id, table_name FROM {} \
                         WHERE end_snapshot IS NULL ORDER BY table_id",
                        postgres_qualified_table(&schema, "ducklake_table")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| (row.get::<_, i64>(0), row.get::<_, String>(1)))
                .collect::<Vec<_>>();
            let active_views = client
                .query(
                    &format!(
                        "SELECT view_id, view_name FROM {} \
                         WHERE end_snapshot IS NULL ORDER BY view_id",
                        postgres_qualified_table(&schema, "ducklake_view")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| (row.get::<_, i64>(0), row.get::<_, String>(1)))
                .collect::<Vec<_>>();
            assert_eq!(active_tables, vec![(2, "numbers".to_string())]);
            assert_eq!(active_views, vec![(3, "v_numbers".to_string())]);
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["dropped_table:4".to_string(), "dropped_view:5".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_schema_lookup_prefers_exact_match_and_rejects_ambiguous_case() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
                    INSERT INTO {qualified}.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);
                    "#
                ))
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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: "postgres ambiguous schema lookup".into(),
                })
                .await
                .unwrap_err();
            assert!(
                ambiguous
                    .to_string()
                    .contains("ambiguous DuckLake schema identifier"),
                "{ambiguous}"
            );
            let snapshot_count: i64 = client
                .query_one(
                    &format!(
                        "SELECT COUNT(*) FROM {}",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(snapshot_count, 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: "postgres exact uppercase schema lookup".into(),
                })
                .await
                .unwrap();
            assert_eq!(created, SnapshotId(2));
            let schema_id: i64 = client
                .query_one(
                    &format!(
                        "SELECT schema_id FROM {} WHERE table_name = 'exact_upper'",
                        postgres_qualified_table(&schema, "ducklake_table")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(schema_id, 42);
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["created_table:\"MAIN\".\"exact_upper\"".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_partitioning_batch_matches_duckdb_multi_table_transaction() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_partition_info(partition_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_partition_column(partition_id BIGINT, table_id BIGINT, partition_key_index BIGINT, column_id BIGINT, transform TEXT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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: "postgres set partitioning for two tables".into(),
                })
                .await
                .unwrap();
            assert_eq!(snapshot, SnapshotId(3));

            let partition_rows = client
                .query(
                    &format!(
                        "SELECT partition_id, begin_snapshot, end_snapshot, table_id \
                         FROM {} ORDER BY partition_id",
                        postgres_qualified_table(&schema, "ducklake_partition_info")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| {
                    (
                        row.get::<_, i64>(0),
                        row.get::<_, i64>(1),
                        row.get::<_, Option<i64>>(2),
                        row.get::<_, i64>(3),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                partition_rows,
                vec![(3, 3, None, 2), (4, 3, None, 1)]
            );

            let partition_columns = client
                .query(
                    &format!(
                        "SELECT partition_id, partition_key_index, column_id, transform \
                         FROM {} ORDER BY partition_id, partition_key_index",
                        postgres_qualified_table(&schema, "ducklake_partition_column")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| {
                    (
                        row.get::<_, i64>(0),
                        row.get::<_, i64>(1),
                        row.get::<_, i64>(2),
                        row.get::<_, String>(3),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                partition_columns,
                vec![
                    (3, 0, 1, "identity".to_string()),
                    (4, 0, 1, "identity".to_string()),
                ]
            );

            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (3, 3, 5, 0)
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["altered_table:1,altered_table:2".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_create_table_with_files_commits_one_snapshot() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.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 TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
                    CREATE TABLE {qualified}.ducklake_table_column_stats(table_id BIGINT, column_id BIGINT, contains_null BOOLEAN, contains_nan BOOLEAN, min_value TEXT, max_value TEXT, extra_stats TEXT);

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

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

            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (2, 2, 11, 21)
            );
            let data_file = client
                .query_one(
                    &format!(
                        "SELECT data_file_id, table_id, begin_snapshot, path, record_count, row_id_start \
                         FROM {} WHERE table_id = 10",
                        postgres_qualified_table(&schema, "ducklake_data_file")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(data_file.get::<_, i64>(0), 20);
            assert_eq!(data_file.get::<_, i64>(1), 10);
            assert_eq!(data_file.get::<_, i64>(2), 2);
            assert_eq!(data_file.get::<_, String>(3), "ctas.parquet");
            assert_eq!(data_file.get::<_, i64>(4), 2);
            assert_eq!(data_file.get::<_, Option<i64>>(5), Some(0));
            let table_stats = client
                .query_one(
                    &format!(
                        "SELECT record_count, next_row_id, file_size_bytes FROM {} WHERE table_id = 10",
                        postgres_qualified_table(&schema, "ducklake_table_stats")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    table_stats.get::<_, i64>(0),
                    table_stats.get::<_, i64>(1),
                    table_stats.get::<_, i64>(2),
                ),
                (2, 2, 25)
            );
            let table_column_stats = client
                .query_one(
                    &format!(
                        "SELECT contains_null, min_value, max_value FROM {} WHERE table_id = 10 AND column_id = 1",
                        postgres_qualified_table(&schema, "ducklake_table_column_stats")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(table_column_stats.get::<_, Option<bool>>(0), Some(false));
            assert_eq!(
                table_column_stats.get::<_, Option<String>>(1),
                Some("1".into())
            );
            assert_eq!(
                table_column_stats.get::<_, Option<String>>(2),
                Some("2".into())
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["created_table:\"main\".\"ctas\"".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_add_column_default_commits_column_and_table_stats() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_table_column_stats(table_id BIGINT, column_id BIGINT, contains_null BOOLEAN, contains_nan BOOLEAN, min_value TEXT, max_value TEXT, extra_stats TEXT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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: "postgres add defaulted column".into(),
                })
                .await
                .unwrap();
            assert_eq!(snapshot, SnapshotId(2));

            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (2, 2, 4, 10)
            );

            let column = client
                .query_one(
                    &format!(
                        "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 {} WHERE column_id = 3",
                        postgres_qualified_table(&schema, "ducklake_column")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(column.get::<_, i64>(0), 3);
            assert_eq!(column.get::<_, i64>(1), 2);
            assert_eq!(column.get::<_, Option<i64>>(2), None);
            assert_eq!(column.get::<_, i64>(3), 3);
            assert_eq!(column.get::<_, String>(4), "note");
            assert_eq!(column.get::<_, String>(5), "varchar");
            assert_eq!(column.get::<_, String>(6), "hello");
            assert_eq!(column.get::<_, String>(7), "hello");
            assert!(column.get::<_, bool>(8));
            assert_eq!(column.get::<_, Option<i64>>(9), None);
            assert_eq!(column.get::<_, String>(10), "literal");
            assert_eq!(column.get::<_, String>(11), "duckdb");

            let table_column_stats = client
                .query_one(
                    &format!(
                        "SELECT contains_null, contains_nan, min_value, max_value \
                         FROM {} WHERE table_id = 1 AND column_id = 3",
                        postgres_qualified_table(&schema, "ducklake_table_column_stats")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(table_column_stats.get::<_, Option<bool>>(0), Some(false));
            assert_eq!(table_column_stats.get::<_, Option<bool>>(1), None);
            assert_eq!(
                table_column_stats.get::<_, Option<String>>(2),
                Some("hello".into())
            );
            assert_eq!(
                table_column_stats.get::<_, Option<String>>(3),
                Some("hello".into())
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["altered_table:1".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_sorted_by_commit_updates_metadata_when_enabled() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_sort_info(sort_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_sort_expression(sort_id BIGINT, table_id BIGINT, sort_key_index BIGINT, expression TEXT, dialect TEXT, sort_direction TEXT, null_order TEXT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            assert_eq!(
                backend
                    .commit_sorted_by(SortedByCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(2),
                        expressions: vec![SortExpressionCommit {
                            expression: "name".into(),
                            dialect: "duckdb".into(),
                            sort_direction: "DESC".into(),
                            null_order: "NULLS_FIRST".into(),
                        }],
                        commit_message: "postgres set sorted".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(3)
            );
            assert_eq!(
                backend
                    .commit_sorted_by(SortedByCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(3),
                        expressions: vec![SortExpressionCommit {
                            expression: "name".into(),
                            dialect: "duckdb".into(),
                            sort_direction: "DESC".into(),
                            null_order: "NULLS_FIRST".into(),
                        }],
                        commit_message: "postgres set sorted again".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(3)
            );
            assert_eq!(
                backend
                    .commit_sorted_by(SortedByCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(3),
                        expressions: Vec::new(),
                        commit_message: "postgres reset sorted".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(4)
            );

            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (4, 7, 11, 20)
            );
            let sort_rows = client
                .query(
                    &format!(
                        "SELECT si.sort_id, si.begin_snapshot, si.end_snapshot, se.expression, \
                                se.sort_direction, se.null_order \
                         FROM {} si \
                         JOIN {} se USING (sort_id, table_id) \
                         WHERE si.table_id = 1 \
                         ORDER BY si.sort_id",
                        postgres_qualified_table(&schema, "ducklake_sort_info"),
                        postgres_qualified_table(&schema, "ducklake_sort_expression")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| {
                    (
                        row.get::<_, i64>(0),
                        row.get::<_, i64>(1),
                        row.get::<_, Option<i64>>(2),
                        row.get::<_, String>(3),
                        row.get::<_, String>(4),
                        row.get::<_, String>(5),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                sort_rows,
                vec![
                    (
                        5,
                        1,
                        Some(3),
                        "id".to_string(),
                        "ASC".to_string(),
                        "NULLS_LAST".to_string()
                    ),
                    (
                        10,
                        3,
                        Some(4),
                        "name".to_string(),
                        "DESC".to_string(),
                        "NULLS_FIRST".to_string()
                    ),
                ]
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["altered_table:1".to_string(), "altered_table:1".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_column_default_and_rename_column_commits_metadata() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_sort_info(sort_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_sort_expression(sort_id BIGINT, table_id BIGINT, sort_key_index BIGINT, expression TEXT, dialect TEXT, sort_direction TEXT, null_order TEXT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
                    INSERT INTO {qualified}.ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, NULL, 1, 'numbers', 'numbers/', true);
                    INSERT INTO {qualified}.ducklake_column VALUES (1, 1, NULL, 1, 1, 'id', 'int32', NULL, NULL, true, NULL, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_column VALUES (2, 1, NULL, 1, 2, 'name', 'varchar', NULL, NULL, true, NULL, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_sort_info VALUES (5, 1, 1, NULL);
                    INSERT INTO {qualified}.ducklake_sort_expression VALUES (5, 1, 0, 'name || id', 'duckdb', 'ASC', 'NULLS_LAST');
                    "#
                ))
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            assert_eq!(
                backend
                    .commit_column_default(ColumnDefaultCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(1),
                        column_id: ColumnId(2),
                        default_value: Some("guest".into()),
                        default_value_type: Some("literal".into()),
                        default_value_dialect: Some("duckdb".into()),
                        commit_message: "postgres set default".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(2)
            );
            assert_eq!(
                backend
                    .commit_column_default(ColumnDefaultCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(2),
                        column_id: ColumnId(2),
                        default_value: None,
                        default_value_type: None,
                        default_value_dialect: None,
                        commit_message: "postgres drop default".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(3)
            );
            assert_eq!(
                backend
                    .commit_rename_column(RenameColumnCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(3),
                        column_id: ColumnId(2),
                        new_column_name: "label".into(),
                        commit_message: "postgres rename column".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(4)
            );

            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (4, 4, 11, 20)
            );

            let column_rows = client
                .query(
                    &format!(
                        "SELECT column_name, begin_snapshot, end_snapshot, default_value, \
                                default_value_type, default_value_dialect \
                         FROM {} WHERE column_id = 2 ORDER BY begin_snapshot",
                        postgres_qualified_table(&schema, "ducklake_column")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| {
                    (
                        row.get::<_, String>(0),
                        row.get::<_, i64>(1),
                        row.get::<_, Option<i64>>(2),
                        row.get::<_, Option<String>>(3),
                        row.get::<_, Option<String>>(4),
                        row.get::<_, Option<String>>(5),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                column_rows,
                vec![
                    ("name".into(), 1, Some(2), None, None, None),
                    (
                        "name".into(),
                        2,
                        Some(3),
                        Some("guest".into()),
                        Some("literal".into()),
                        Some("duckdb".into())
                    ),
                    ("name".into(), 3, Some(4), None, None, None),
                    ("label".into(), 4, None, None, None, None),
                ]
            );

            let sort_rows = client
                .query(
                    &format!(
                        "SELECT si.sort_id, si.begin_snapshot, si.end_snapshot, se.expression \
                         FROM {} si \
                         JOIN {} se USING (sort_id, table_id) \
                         ORDER BY si.sort_id",
                        postgres_qualified_table(&schema, "ducklake_sort_info"),
                        postgres_qualified_table(&schema, "ducklake_sort_expression")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| {
                    (
                        row.get::<_, i64>(0),
                        row.get::<_, i64>(1),
                        row.get::<_, Option<i64>>(2),
                        row.get::<_, String>(3),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                sort_rows,
                vec![
                    (5, 1, Some(4), "name || id".into()),
                    (10, 4, None, "label || id".into()),
                ]
            );

            let sorted_drop_err = backend
                .commit_drop_column(DropColumnCommit {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(4),
                    column_id: ColumnId(2),
                    commit_message: "postgres drop sorted column".into(),
                })
                .await
                .unwrap_err();
            assert!(
                sorted_drop_err
                    .to_string()
                    .contains("sorted by this column"),
                "{sorted_drop_err}"
            );
            client
                .execute(
                    &format!(
                        "UPDATE {} SET expression = 'id' WHERE sort_id = 10 AND table_id = 1",
                        postgres_qualified_table(&schema, "ducklake_sort_expression")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                backend
                    .commit_drop_column(DropColumnCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(4),
                        column_id: ColumnId(2),
                        commit_message: "postgres drop column".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(5)
            );
            assert_eq!(
                backend
                    .commit_column_type(ColumnTypeCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(5),
                        column_id: ColumnId(1),
                        new_column_type: "int64".into(),
                        commit_message: "postgres promote column".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(6)
            );
            assert_eq!(
                backend
                    .commit_column_type(ColumnTypeCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(6),
                        column_id: ColumnId(1),
                        new_column_type: "int64".into(),
                        commit_message: "postgres same type".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(6)
            );
            let narrow_err = backend
                .commit_column_type(ColumnTypeCommit {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(6),
                    column_id: ColumnId(1),
                    new_column_type: "int16".into(),
                    commit_message: "postgres narrow column".into(),
                })
                .await
                .unwrap_err();
            assert!(narrow_err.to_string().contains("only widening"), "{narrow_err}");

            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (6, 6, 11, 20)
            );
            let active_columns = client
                .query(
                    &format!(
                        "SELECT column_name FROM {} \
                         WHERE table_id = 1 AND end_snapshot IS NULL ORDER BY column_order",
                        postgres_qualified_table(&schema, "ducklake_column")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| row.get::<_, String>(0))
                .collect::<Vec<_>>();
            assert_eq!(active_columns, vec!["id".to_string()]);
            let id_type_history = client
                .query(
                    &format!(
                        "SELECT begin_snapshot, end_snapshot, column_type \
                         FROM {} \
                         WHERE table_id = 1 AND column_id = 1 ORDER BY begin_snapshot",
                        postgres_qualified_table(&schema, "ducklake_column")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| {
                    (
                        row.get::<_, i64>(0),
                        row.get::<_, Option<i64>>(1),
                        row.get::<_, String>(2),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                id_type_history,
                vec![
                    (1, Some(6), "int32".to_string()),
                    (6, None, "int64".to_string())
                ]
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                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(),
                ]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_column_type_tree_commit_replaces_struct_children() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
                    INSERT INTO {qualified}.ducklake_column VALUES (3, 1, NULL, 1, 3, 'profile', 'struct', NULL, NULL, true, NULL, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_column VALUES (4, 1, NULL, 1, 0, 'age', 'int32', NULL, NULL, true, 3, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_column VALUES (5, 1, NULL, 1, 1, 'city', 'varchar', NULL, NULL, true, 3, NULL, NULL);
                    "#
                ))
                .await
                .unwrap();

            let replacement = crate::ColumnTypeTree {
                column_name: "profile".into(),
                column_type: "struct".into(),
                nulls_allowed: true,
                children: vec![
                    crate::ColumnTypeTree {
                        column_name: "age".into(),
                        column_type: "int64".into(),
                        nulls_allowed: true,
                        children: vec![],
                    },
                    crate::ColumnTypeTree {
                        column_name: "city".into(),
                        column_type: "varchar".into(),
                        nulls_allowed: true,
                        children: vec![],
                    },
                    crate::ColumnTypeTree {
                        column_name: "score".into(),
                        column_type: "int32".into(),
                        nulls_allowed: true,
                        children: vec![],
                    },
                ],
            };
            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            assert_eq!(
                backend
                    .commit_column_type_tree(ColumnTypeTreeCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(1),
                        column_id: ColumnId(3),
                        replacement: replacement.clone(),
                        commit_message: "postgres replace struct".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(2)
            );
            assert_eq!(
                backend
                    .commit_column_type_tree(ColumnTypeTreeCommit {
                        table_id: TableId(1),
                        expected_snapshot: SnapshotId(2),
                        column_id: ColumnId(3),
                        replacement,
                        commit_message: "postgres same struct".into(),
                    })
                    .await
                    .unwrap(),
                SnapshotId(2)
            );

            let rows = client
                .query(
                    &format!(
                        "SELECT column_id, begin_snapshot, end_snapshot, column_order, column_name, \
                                column_type, parent_column, default_value, default_value_type, default_value_dialect \
                         FROM {} \
                         WHERE table_id = 1 AND column_id IN (3, 4, 5, 10) \
                         ORDER BY column_id, begin_snapshot",
                        postgres_qualified_table(&schema, "ducklake_column")
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| {
                    (
                        row.get::<_, i64>(0),
                        row.get::<_, i64>(1),
                        row.get::<_, Option<i64>>(2),
                        row.get::<_, i64>(3),
                        row.get::<_, String>(4),
                        row.get::<_, String>(5),
                        row.get::<_, Option<i64>>(6),
                        row.get::<_, Option<String>>(7),
                        row.get::<_, Option<String>>(8),
                        row.get::<_, Option<String>>(9),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                rows,
                vec![
                    (3, 1, Some(2), 3, "profile".into(), "struct".into(), None, None, None, None),
                    (3, 2, None, 3, "profile".into(), "struct".into(), None, None, None, None),
                    (4, 1, Some(2), 0, "age".into(), "int32".into(), Some(3), None, None, None),
                    (4, 2, None, 0, "age".into(), "int64".into(), Some(3), None, None, None),
                    (5, 1, Some(2), 1, "city".into(), "varchar".into(), Some(3), None, None, None),
                    (5, 2, None, 1, "city".into(), "varchar".into(), Some(3), None, None, None),
                    (
                        10,
                        2,
                        None,
                        2,
                        "score".into(),
                        "int32".into(),
                        Some(3),
                        Some("NULL".into()),
                        Some("literal".into()),
                        Some("duckdb".into())
                    ),
                ]
            );
            let latest = client
                .query_one(
                    &format!(
                        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
                         FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                        postgres_qualified_table(&schema, "ducklake_snapshot")
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(
                (
                    latest.get::<_, i64>(0),
                    latest.get::<_, i64>(1),
                    latest.get::<_, i64>(2),
                    latest.get::<_, i64>(3),
                ),
                (2, 2, 11, 20)
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["altered_table:1".to_string()]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_reads_extended_metadata_tables() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_schema_versions(begin_snapshot BIGINT, schema_version BIGINT, table_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name TEXT, dialect TEXT, sql TEXT, column_aliases TEXT);
                    CREATE TABLE {qualified}.ducklake_tag(object_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, key TEXT, value TEXT);
                    CREATE TABLE {qualified}.ducklake_column_tag(table_id BIGINT, column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, key TEXT, value TEXT);
                    CREATE TABLE {qualified}.ducklake_file_variant_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, variant_path TEXT, shredded_type TEXT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
                    CREATE TABLE {qualified}.ducklake_table_column_stats(table_id BIGINT, column_id BIGINT, contains_null BOOLEAN, contains_nan BOOLEAN, min_value TEXT, max_value TEXT, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_partition_info(partition_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_partition_column(partition_id BIGINT, table_id BIGINT, partition_key_index BIGINT, column_id BIGINT, transform TEXT);
                    CREATE TABLE {qualified}.ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value TEXT);
                    CREATE TABLE {qualified}.ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
                    CREATE TABLE {qualified}.ducklake_column_mapping(mapping_id BIGINT, table_id BIGINT, type TEXT);
                    CREATE TABLE {qualified}.ducklake_name_mapping(mapping_id BIGINT, column_id BIGINT, source_name TEXT, target_field_id BIGINT, parent_column BIGINT, is_partition BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_inlined_data_tables(table_id BIGINT, table_name TEXT, schema_version BIGINT);
                    CREATE TABLE {qualified}.ducklake_macro(schema_id BIGINT, macro_id BIGINT, macro_name TEXT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_macro_impl(macro_id BIGINT, impl_id BIGINT, dialect TEXT, sql TEXT, type TEXT);
                    CREATE TABLE {qualified}.ducklake_macro_parameters(macro_id BIGINT, impl_id BIGINT, column_id BIGINT, parameter_name TEXT, parameter_type TEXT, default_value TEXT, default_value_type TEXT);
                    CREATE TABLE {qualified}.ducklake_sort_info(sort_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_sort_expression(sort_id BIGINT, table_id BIGINT, sort_key_index BIGINT, expression TEXT, dialect TEXT, sort_direction TEXT, null_order TEXT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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 (mapping, entries) = backend.column_mapping(MappingId(11)).await.unwrap();
            assert_eq!(mapping.mapping_type, "name");
            assert_eq!(entries[0].source_name, "old_name");
            let inlined = backend.inlined_tables(TableId(1), 7).await.unwrap();
            assert_eq!(inlined[0].table_name, "ducklake_inlined_data_1_7");
            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");

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_filters_versioned_metadata_rows_by_snapshot() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_view(view_id BIGINT, view_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, view_name TEXT, dialect TEXT, sql TEXT, column_aliases TEXT);
                    CREATE TABLE {qualified}.ducklake_tag(object_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, key TEXT, value TEXT);
                    CREATE TABLE {qualified}.ducklake_column_tag(table_id BIGINT, column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, key TEXT, value TEXT);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_partition_info(partition_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_macro(schema_id BIGINT, macro_id BIGINT, macro_name TEXT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_sort_info(sort_id BIGINT, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT);

                    INSERT INTO {qualified}.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 {qualified}.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 {qualified}.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 {qualified}.ducklake_tag VALUES
                        (1, 1, 3, 'phase', 'old'),
                        (1, 3, 4, 'phase', 'current'),
                        (1, 4, NULL, 'phase', 'future');
                    INSERT INTO {qualified}.ducklake_column_tag VALUES
                        (20, 200, 1, 3, 'semantic', 'old'),
                        (20, 200, 3, 4, 'semantic', 'current'),
                        (20, 200, 4, NULL, 'semantic', 'future');
                    INSERT INTO {qualified}.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 {qualified}.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 {qualified}.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 {qualified}.ducklake_partition_info VALUES
                        (300, 20, 1, 3),
                        (400, 20, 3, 4),
                        (500, 20, 4, NULL);
                    INSERT INTO {qualified}.ducklake_macro VALUES
                        (2, 600, 'old_macro', 1, 3),
                        (2, 700, 'current_macro', 3, 4),
                        (2, 800, 'future_macro', 4, NULL);
                    INSERT INTO {qualified}.ducklake_sort_info VALUES
                        (900, 20, 1, 3),
                        (1000, 20, 3, 4),
                        (1100, 20, 4, NULL);
                    "#
                ))
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let snapshot = SnapshotId(3);
            let schemas = backend.schemas_at(snapshot).await.unwrap();
            assert_eq!(
                schemas.iter().map(|row| row.schema_id).collect::<Vec<_>>(),
                vec![SchemaId(2)]
            );
            let tables = backend.tables_at(SchemaId(2), snapshot).await.unwrap();
            assert_eq!(
                tables.iter().map(|row| row.table_id).collect::<Vec<_>>(),
                vec![TableId(20)]
            );
            let views = backend.views_at(SchemaId(2), snapshot).await.unwrap();
            assert_eq!(
                views.iter().map(|row| row.view_id).collect::<Vec<_>>(),
                vec![ViewId(21)]
            );
            let tags = backend.tags_at(snapshot).await.unwrap();
            assert_eq!(
                tags.iter()
                    .map(|row| row.value.as_deref())
                    .collect::<Vec<_>>(),
                vec![Some("current")]
            );
            let column_tags = backend.column_tags_at(TableId(20), snapshot).await.unwrap();
            assert_eq!(
                column_tags
                    .iter()
                    .map(|row| row.value.as_deref())
                    .collect::<Vec<_>>(),
                vec![Some("current")]
            );
            let columns = backend.columns_at(TableId(20), snapshot).await.unwrap();
            assert_eq!(
                columns.iter().map(|row| row.column_id).collect::<Vec<_>>(),
                vec![ColumnId(200)]
            );
            let data_files = backend.data_files_at(TableId(20), snapshot).await.unwrap();
            assert_eq!(
                data_files
                    .iter()
                    .map(|row| row.data_file_id)
                    .collect::<Vec<_>>(),
                vec![DataFileId(70)]
            );
            let delete_files = backend
                .delete_files_at(TableId(20), snapshot)
                .await
                .unwrap();
            assert_eq!(
                delete_files
                    .iter()
                    .map(|row| row.delete_file_id)
                    .collect::<Vec<_>>(),
                vec![DeleteFileId(100)]
            );
            let partitions = backend
                .partition_info_at(TableId(20), snapshot)
                .await
                .unwrap();
            assert_eq!(
                partitions
                    .iter()
                    .map(|row| row.partition_id)
                    .collect::<Vec<_>>(),
                vec![PartitionId(400)]
            );
            let macros = backend.macros_at(SchemaId(2), snapshot).await.unwrap();
            assert_eq!(
                macros.iter().map(|row| row.macro_id).collect::<Vec<_>>(),
                vec![MacroId(700)]
            );
            let sort_info = backend.sort_info_at(TableId(20), snapshot).await.unwrap();
            assert_eq!(
                sort_info.iter().map(|row| row.sort_id).collect::<Vec<_>>(),
                vec![SortId(1000)]
            );

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_cleanup_old_files_deletes_due_local_files_and_queue_rows() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let temp = tempfile::tempdir().unwrap();
    let data_path = temp.path().join("data");
    std::fs::create_dir(&data_path).unwrap();
    std::fs::write(data_path.join("old.parquet"), b"old").unwrap();
    std::fs::write(data_path.join("future.parquet"), b"future").unwrap();
    let absolute_path = temp.path().join("absolute.parquet");
    std::fs::write(&absolute_path, b"absolute").unwrap();

    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_metadata(key TEXT, value TEXT, scope TEXT, scope_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
                    INSERT INTO {qualified}.ducklake_metadata VALUES ('version', '1.0', NULL, NULL);
                    INSERT INTO {qualified}.ducklake_files_scheduled_for_deletion VALUES (1, 'old.parquet', true, '2026-01-01 00:00:00+00');
                    INSERT INTO {qualified}.ducklake_files_scheduled_for_deletion VALUES (2, 'future.parquet', true, '2026-01-03 00:00:00+00');
                    INSERT INTO {qualified}.ducklake_files_scheduled_for_deletion VALUES (4, 'already-gone.parquet', true, '2026-01-01 00:00:00+00');
                    "#
                ))
                .await
                .unwrap();
            client
                .execute(
                    &format!("INSERT INTO {qualified}.ducklake_metadata VALUES ('data_path', $1, NULL, NULL)"),
                    &[&data_path.display().to_string()],
                )
                .await
                .unwrap();
            client
                .execute(
                    &format!(
                        "INSERT INTO {qualified}.ducklake_files_scheduled_for_deletion VALUES (3, $1, false, '2026-01-01 00:00:00+00')"
                    ),
                    &[&absolute_path.display().to_string()],
                )
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let invalid = backend
                .cleanup_old_files(CleanupOldFilesOptions {
                    dry_run: true,
                    cleanup_all: false,
                    older_than_micros: None,
                })
                .await
                .unwrap_err();
            assert!(
                format!("{invalid}").contains("requires exactly one of cleanup_all or older_than")
            );

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

            let scheduled = backend.files_scheduled_for_deletion().await.unwrap();
            let cutoff = scheduled
                .iter()
                .find(|file| file.data_file_id == DataFileId(2))
                .unwrap()
                .schedule_start;
            let cleaned = backend
                .cleanup_old_files(CleanupOldFilesOptions::older_than(cutoff))
                .await
                .unwrap();
            assert_eq!(
                cleaned
                    .iter()
                    .map(|file| file.data_file_id)
                    .collect::<Vec<_>>(),
                vec![DataFileId(1), DataFileId(3), DataFileId(4)]
            );
            assert!(!data_path.join("old.parquet").exists());
            assert!(!absolute_path.exists());
            assert!(data_path.join("future.parquet").exists());
            assert_eq!(postgres_scheduled_file_count(&client, &schema).await, 1);

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

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_cleanup_old_files_rejects_remote_paths_before_deleting_local_files() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let temp = tempfile::tempdir().unwrap();
    let data_path = temp.path().join("data");
    std::fs::create_dir(&data_path).unwrap();
    let local_path = data_path.join("old.parquet");
    std::fs::write(&local_path, b"old").unwrap();

    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_metadata(key TEXT, value TEXT, scope TEXT, scope_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
                    INSERT INTO {qualified}.ducklake_metadata VALUES ('version', '1.0', NULL, NULL);
                    INSERT INTO {qualified}.ducklake_files_scheduled_for_deletion VALUES (2, 's3://bucket/old.parquet', false, '2026-01-01 00:00:00+00');
                    "#
                ))
                .await
                .unwrap();
            client
                .execute(
                    &format!("INSERT INTO {qualified}.ducklake_metadata VALUES ('data_path', $1, NULL, NULL)"),
                    &[&data_path.display().to_string()],
                )
                .await
                .unwrap();
            client
                .execute(
                    &format!(
                        "INSERT INTO {qualified}.ducklake_files_scheduled_for_deletion VALUES (1, $1, false, '2026-01-01 00:00:00+00')"
                    ),
                    &[&local_path.display().to_string()],
                )
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let err = backend
                .cleanup_old_files(CleanupOldFilesOptions::cleanup_all())
                .await
                .unwrap_err();
            assert!(format!("{err}").contains("Materialized object-store file access"));
            assert!(local_path.exists());
            assert_eq!(postgres_scheduled_file_count(&client, &schema).await, 2);

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_expire_snapshots_schedules_unreferenced_files_and_prunes_metadata() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);
                    CREATE TABLE {qualified}.ducklake_schema_versions(begin_snapshot BIGINT, schema_version BIGINT, table_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.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 TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
                    CREATE TABLE {qualified}.ducklake_macro(schema_id BIGINT, macro_id BIGINT, macro_name TEXT, begin_snapshot BIGINT, end_snapshot BIGINT);
                    CREATE TABLE {qualified}.ducklake_macro_impl(macro_id BIGINT, impl_id BIGINT, dialect TEXT, sql TEXT, type TEXT);
                    CREATE TABLE {qualified}.ducklake_macro_parameters(macro_id BIGINT, impl_id BIGINT, column_id BIGINT, parameter_name TEXT, parameter_type TEXT, default_value TEXT, default_value_type TEXT);
                    CREATE TABLE {qualified}.ducklake_column_mapping(mapping_id BIGINT, table_id BIGINT, type TEXT);
                    CREATE TABLE {qualified}.ducklake_name_mapping(mapping_id BIGINT, column_id BIGINT, source_name TEXT, target_field_id BIGINT, parent_column BIGINT, is_partition BOOLEAN);

                    INSERT INTO {qualified}.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 {qualified}.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 {qualified}.ducklake_table VALUES (1, '550e8400-e29b-41d4-a716-446655440001', 1, 3, 1, 'numbers', NULL, true);
                    INSERT INTO {qualified}.ducklake_column VALUES (1, 1, NULL, 1, 0, 'id', 'int32', NULL, NULL, true, NULL, 'literal', NULL);
                    INSERT INTO {qualified}.ducklake_table_stats VALUES (1, 1, 1, 42);
                    INSERT INTO {qualified}.ducklake_schema_versions VALUES (1, 1, 1);
                    INSERT INTO {qualified}.ducklake_data_file VALUES (10, 1, 2, 3, 0, 'old.parquet', true, 'parquet', 1, 42, NULL, NULL, NULL, NULL, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_file_column_stats VALUES (10, 1, 1, 4, 1, 0, '1', '1', false, NULL);
                    INSERT INTO {qualified}.ducklake_macro VALUES (1, 4, 'old_macro', 1, 3);
                    INSERT INTO {qualified}.ducklake_macro_impl VALUES (4, 1, 'duckdb', 'x + 1', 'scalar');
                    INSERT INTO {qualified}.ducklake_macro_parameters VALUES (4, 1, 0, 'x', 'int64', NULL, NULL);
                    INSERT INTO {qualified}.ducklake_column_mapping VALUES (5, 1, 'name');
                    INSERT INTO {qualified}.ducklake_name_mapping VALUES (5, 1, 'old_id', 1, NULL, false);
                    "#
                ))
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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!(postgres_table_count(&client, &schema, "ducklake_snapshot").await, 3);
            assert_eq!(
                postgres_table_count(&client, &schema, "ducklake_data_file").await,
                1
            );

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

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

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_merge_adjacent_compaction_replaces_sources_and_schedules_cleanup() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.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 TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_file_variant_stats(data_file_id BIGINT, table_id BIGINT, column_id BIGINT, variant_path TEXT, shredded_type TEXT, column_size_bytes BIGINT, value_count BIGINT, null_count BIGINT, min_value TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
                    INSERT INTO {qualified}.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 {qualified}.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 {qualified}.ducklake_file_variant_stats VALUES
                        (10, 1, 1, '$.a', 'int64', 4, 1, 0, '1', '1', false, NULL);
                    INSERT INTO {qualified}.ducklake_table_stats VALUES (1, 3, 3, 100);
                    "#
                ))
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            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!(
                postgres_data_file_ids(&client, &schema).await,
                vec![(20, "compact-20.parquet".to_string())]
            );
            assert_eq!(
                postgres_table_count(&client, &schema, "ducklake_file_column_stats").await,
                1
            );
            assert_eq!(
                postgres_table_count(&client, &schema, "ducklake_file_variant_stats").await,
                0
            );
            assert_eq!(postgres_scheduled_file_count(&client, &schema).await, 2);
            assert_eq!(
                postgres_table_count(&client, &schema, "ducklake_snapshot").await,
                2
            );
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["merge_adjacent:1".to_string()]
            );
            assert_eq!(
                postgres_table_stats_file_size(&client, &schema, TableId(1)).await,
                95
            );
            assert_eq!(postgres_latest_next_file_id(&client, &schema).await, 21);

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_rewrite_data_files_closes_sources_without_cleanup() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.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 TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 1, 10, 20);
                    INSERT INTO {qualified}.ducklake_data_file VALUES
                        (10, 1, 1, NULL, 0, 'rewrite-source-a.parquet', true, 'parquet', 1, 10, NULL, NULL, NULL, NULL, NULL, NULL),
                        (11, 1, 1, NULL, 1, 'rewrite-source-b.parquet', true, 'parquet', 2, 20, NULL, NULL, NULL, NULL, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_delete_file VALUES
                        (12, 1, 1, NULL, 10, 'rewrite-delete-a.parquet', true, 'parquet', 1, 5, 1, NULL, NULL);
                    INSERT INTO {qualified}.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 {qualified}.ducklake_table_stats VALUES (1, 3, 3, 100);
                    "#
                ))
                .await
                .unwrap();

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

            let latest_files = backend.data_files_at(TableId(1), SnapshotId(2)).await.unwrap();
            assert_eq!(latest_files.len(), 1);
            assert_eq!(latest_files[0].data_file_id, DataFileId(20));
            let historical_files = backend.data_files_at(TableId(1), SnapshotId(1)).await.unwrap();
            assert_eq!(historical_files.len(), 2);
            let rows = client
                .query(
                    &format!(
                        "SELECT data_file_id, begin_snapshot, end_snapshot, path \
                         FROM {qualified}.ducklake_data_file ORDER BY data_file_id"
                    ),
                    &[],
                )
                .await
                .unwrap()
                .into_iter()
                .map(|row| {
                    (
                        row.get::<_, i64>(0),
                        row.get::<_, i64>(1),
                        row.get::<_, Option<i64>>(2),
                        row.get::<_, String>(3),
                    )
                })
                .collect::<Vec<_>>();
            assert_eq!(
                rows,
                vec![
                    (10, 1, Some(2), "rewrite-source-a.parquet".to_string()),
                    (11, 1, Some(2), "rewrite-source-b.parquet".to_string()),
                    (20, 2, None, "rewrite-20.parquet".to_string()),
                ]
            );
            let delete_end_snapshot: Option<i64> = client
                .query_one(
                    &format!(
                        "SELECT end_snapshot FROM {qualified}.ducklake_delete_file WHERE delete_file_id = 12"
                    ),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(delete_end_snapshot, Some(2));
            assert_eq!(postgres_scheduled_file_count(&client, &schema).await, 0);
            assert_eq!(
                postgres_snapshot_changes(&client, &schema).await,
                vec!["rewrite_data_files:1".to_string()]
            );
            assert_eq!(
                postgres_table_stats_file_size(&client, &schema, TableId(1)).await,
                95
            );
            assert_eq!(postgres_latest_next_file_id(&client, &schema).await, 21);

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_materialized_overwrite_schedules_delete_cleanup_with_table_path() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_metadata(key TEXT, value TEXT, scope TEXT, scope_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_schema(schema_id BIGINT PRIMARY KEY, schema_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_table(table_id BIGINT, table_uuid UUID, begin_snapshot BIGINT, end_snapshot BIGINT, schema_id BIGINT, table_name TEXT, path TEXT, path_is_relative BOOLEAN);
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_files_scheduled_for_deletion(data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, schedule_start TIMESTAMPTZ);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

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

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

            let cleanup = client
                .query_one(
                    &format!(
                        "SELECT data_file_id, path, path_is_relative \
                         FROM {qualified}.ducklake_files_scheduled_for_deletion"
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(cleanup.get::<_, i64>(0), 10);
            assert_eq!(
                cleanup.get::<_, String>(1),
                "main/numbers/old-delete.parquet"
            );
            assert!(cleanup.get::<_, bool>(2));

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_commit_append_files_updates_catalog_metadata() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_column(column_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, table_id BIGINT, column_order BIGINT, column_name TEXT, column_type TEXT, initial_default TEXT, default_value TEXT, nulls_allowed BOOLEAN, parent_column BIGINT, default_value_type TEXT, default_value_dialect TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.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 TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_table_column_stats(table_id BIGINT, column_id BIGINT, contains_null BOOLEAN, contains_nan BOOLEAN, min_value TEXT, max_value TEXT, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value TEXT);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 7, 50, 10);
                    INSERT INTO {qualified}.ducklake_column VALUES (2, 1, NULL, 1, 0, 'value', 'int64', NULL, NULL, false, NULL, 'literal', NULL);
                    INSERT INTO {qualified}.ducklake_table_column_stats VALUES (1, 2, false, false, '1', '9', NULL);
                    INSERT INTO {qualified}.ducklake_table_stats VALUES (1, 5, 5, 100);
                    "#
                ))
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let new_snapshot = backend
                .commit_append_files(vec![postgres_append_file(10, SnapshotId(1))])
                .await
                .unwrap();
            assert_eq!(new_snapshot, SnapshotId(2));

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

            let data_files = backend.data_files_at(TableId(1), SnapshotId(2)).await.unwrap();
            assert_eq!(data_files.len(), 1);
            assert_eq!(data_files[0].data_file_id, DataFileId(10));
            assert_eq!(data_files[0].path, "postgres-append.parquet");
            assert_eq!(data_files[0].partition_id, Some(PartitionId(3)));

            let stats = backend.file_stats(TableId(1)).await.unwrap();
            assert_eq!(stats.len(), 1);
            assert_eq!(stats[0].column_id, ColumnId(2));
            assert_eq!(stats[0].min_value.as_deref(), Some("10"));
            assert_eq!(stats[0].max_value.as_deref(), Some("20"));

            let table_column_stats = backend.table_column_stats(TableId(1)).await.unwrap();
            assert_eq!(table_column_stats.len(), 1);
            assert_eq!(table_column_stats[0].column_id, ColumnId(2));
            assert_eq!(table_column_stats[0].contains_null, Some(false));
            assert_eq!(table_column_stats[0].contains_nan, Some(false));
            assert_eq!(table_column_stats[0].min_value.as_deref(), Some("1"));
            assert_eq!(table_column_stats[0].max_value.as_deref(), Some("20"));

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

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

            let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
            assert_eq!(table_stats.record_count, 7);
            assert_eq!(table_stats.next_row_id, 7);
            assert_eq!(table_stats.file_size_bytes, 125);

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

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_append_honors_begin_snapshot_override() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.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 TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value TEXT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (3, '2026-06-26 00:00:00+00', 1, 10, 10);
                    "#
                ))
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let new_snapshot = backend
                .commit_append_files(vec![
                    AppendFile {
                        begin_snapshot: Some(SnapshotId(1)),
                        ..postgres_append_file(10, SnapshotId(3))
                    },
                    postgres_append_file(11, SnapshotId(3)),
                ])
                .await
                .unwrap();
            assert_eq!(new_snapshot, SnapshotId(4));

            let rows = client
                .query(
                    &format!(
                        "SELECT data_file_id, begin_snapshot FROM {qualified}.ducklake_data_file \
                         ORDER BY data_file_id"
                    ),
                    &[],
                )
                .await
                .unwrap();
            let begin_snapshots = rows
                .iter()
                .map(|row| (row.get::<_, i64>(0), row.get::<_, i64>(1)))
                .collect::<Vec<_>>();
            assert_eq!(begin_snapshots, vec![(10, 1), (11, 4)]);

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_truncate_table_closes_active_files_without_allocating_file_ids() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 7, 50, 20);
                    INSERT INTO {qualified}.ducklake_data_file VALUES
                        (10, 1, 1, NULL, 10, 'postgres-active-a.parquet', true, 'parquet', 5, 100, NULL, 0, NULL, NULL, NULL, NULL),
                        (11, 1, 1, NULL, 11, 'postgres-active-b.parquet', true, 'parquet', 7, 140, NULL, 5, NULL, NULL, NULL, NULL),
                        (12, 2, 1, NULL, 12, 'postgres-other.parquet', true, 'parquet', 3, 60, NULL, 0, NULL, NULL, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_delete_file VALUES
                        (15, 1, 1, NULL, 10, 'postgres-active-delete.parquet', true, 'positional', 2, 30, 8, NULL, NULL),
                        (16, 2, 1, NULL, 12, 'postgres-other-delete.parquet', true, 'positional', 1, 20, 8, NULL, NULL);
                    INSERT INTO {qualified}.ducklake_table_stats VALUES
                        (1, 12, 12, 240),
                        (2, 3, 3, 60);
                    "#
                ))
                .await
                .unwrap();

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let snapshot = backend
                .commit_truncate_table(TruncateTableCommit {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(1),
                    rows_deleted: 12,
                    inlined_data_deletions: Vec::new(),
                    changes_made: "deleted_from_table:1".into(),
                    commit_message: "datafusion truncate".into(),
                })
                .await
                .unwrap();
            assert_eq!(snapshot, SnapshotId(2));

            let snapshots = backend.snapshots().await.unwrap();
            let latest = snapshots.last().unwrap();
            assert_eq!(latest.next_file_id, 20);
            assert_eq!(
                backend
                    .data_files_at(TableId(1), SnapshotId(2))
                    .await
                    .unwrap()
                    .len(),
                0
            );
            assert_eq!(
                backend
                    .delete_files_at(TableId(1), SnapshotId(2))
                    .await
                    .unwrap()
                    .len(),
                0
            );
            assert_eq!(
                backend
                    .data_files_at(TableId(1), SnapshotId(1))
                    .await
                    .unwrap()
                    .len(),
                2
            );
            assert_eq!(
                backend
                    .data_files_at(TableId(2), SnapshotId(2))
                    .await
                    .unwrap()
                    .len(),
                1
            );

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

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

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_concurrent_append_commits_return_one_conflict() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.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 TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value TEXT);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

                    INSERT INTO {qualified}.ducklake_snapshot VALUES (1, '2026-06-26 00:00:00+00', 7, 50, 10);
                    INSERT INTO {qualified}.ducklake_table_stats VALUES (1, 5, 5, 100);
                    "#
                ))
                .await
                .unwrap();

            let left = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let right = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let (left, right) = tokio::join!(
                left.commit_append_files(vec![postgres_append_file(10, SnapshotId(1))]),
                right.commit_append_files(vec![postgres_append_file(10, SnapshotId(1))])
            );

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

            drop_test_schema(&client, &schema).await;
        });
}

#[test]
fn live_postgres_commit_delete_and_table_changes_update_catalog_metadata() {
    let Some(dsn) = postgres_test_dsn() else {
        eprintln!("skipping live Postgres storage test: DUCKLAKE_POSTGRES_TEST_DSN is not set");
        return;
    };
    let schema = unique_test_schema();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
            let client = connect_test_postgres(&dsn).await;
            create_test_schema(&client, &schema).await;
            let qualified = quote_ident(&schema);
            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_snapshot(snapshot_id BIGINT PRIMARY KEY, snapshot_time TIMESTAMPTZ, schema_version BIGINT, next_catalog_id BIGINT, next_file_id BIGINT);
                    CREATE TABLE {qualified}.ducklake_snapshot_changes(snapshot_id BIGINT PRIMARY KEY, changes_made TEXT, author TEXT, commit_message TEXT, commit_extra_info TEXT);
                    CREATE TABLE {qualified}.ducklake_data_file(data_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, file_order BIGINT, path TEXT, path_is_relative BOOLEAN, file_format TEXT, record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, row_id_start BIGINT, partition_id BIGINT, encryption_key TEXT, mapping_id BIGINT, partial_max BIGINT);
                    CREATE TABLE {qualified}.ducklake_delete_file(delete_file_id BIGINT PRIMARY KEY, table_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, data_file_id BIGINT, path TEXT, path_is_relative BOOLEAN, format TEXT, delete_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, encryption_key TEXT, partial_max BIGINT);
                    CREATE TABLE {qualified}.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 TEXT, max_value TEXT, contains_nan BOOLEAN, extra_stats TEXT);
                    CREATE TABLE {qualified}.ducklake_file_partition_value(data_file_id BIGINT, table_id BIGINT, partition_key_index BIGINT, partition_value TEXT);
                    CREATE TABLE {qualified}.ducklake_table_stats(table_id BIGINT, record_count BIGINT, next_row_id BIGINT, file_size_bytes BIGINT);

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

            let backend = PostgresCatalog::with_metadata_schema(&dsn, &schema);
            let delete_snapshot = backend
                .commit_delete_files(vec![postgres_delete_file(
                    10,
                    SnapshotId(1),
                    1,
                    1,
                    None,
                )])
                .await
                .unwrap();
            assert_eq!(delete_snapshot, SnapshotId(2));

            let active_delete_files = backend
                .delete_files_at(TableId(1), SnapshotId(2))
                .await
                .unwrap();
            assert_eq!(active_delete_files.len(), 1);
            assert_eq!(active_delete_files[0].delete_file_id, DeleteFileId(10));
            assert_eq!(active_delete_files[0].data_file_id, DataFileId(1));
            assert_eq!(active_delete_files[0].delete_count, 1);

            let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
            assert_eq!(table_stats.record_count, 4);
            assert_eq!(table_stats.next_row_id, 5);
            assert_eq!(table_stats.file_size_bytes, 100);

            let mixed_snapshot = backend
                .commit_table_changes(TableChangeCommit {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(2),
                    append_files: vec![postgres_append_file(11, SnapshotId(2))],
                    delete_files: vec![postgres_delete_file(
                        12,
                        SnapshotId(2),
                        1,
                        1,
                        Some(DeleteFileId(10)),
                    )],
                    inlined_file_deletions: Vec::new(),
                    inlined_data_appends: Vec::new(),
                    inlined_data_deletions: Vec::new(),
                    flushed_inlined_data: Vec::new(),
                    flushed_inlined_file_deletions: Vec::new(),
                    changes_made: "updated_table:1".into(),
                    commit_message: "datafusion update".into(),
                })
                .await
                .unwrap();
            assert_eq!(mixed_snapshot, SnapshotId(3));

            let snapshots = backend.snapshots().await.unwrap();
            assert_eq!(snapshots.last().unwrap().snapshot_id, SnapshotId(3));
            assert_eq!(snapshots.last().unwrap().next_file_id, 13);

            let data_files = backend.data_files_at(TableId(1), SnapshotId(3)).await.unwrap();
            assert!(
                data_files
                    .iter()
                    .any(|file| file.data_file_id == DataFileId(11))
            );
            let active_delete_files = backend
                .delete_files_at(TableId(1), SnapshotId(3))
                .await
                .unwrap();
            assert_eq!(
                active_delete_files
                    .iter()
                    .map(|file| file.delete_file_id)
                    .collect::<Vec<_>>(),
                vec![DeleteFileId(12)]
            );
            let overwritten = client
                .query_one(
                    &format!(
                        "SELECT end_snapshot FROM {qualified}.ducklake_delete_file WHERE delete_file_id = 10"
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(overwritten.get::<_, Option<i64>>(0), Some(3));

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

            let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
            assert_eq!(table_stats.record_count, 5);
            assert_eq!(table_stats.next_row_id, 7);
            assert_eq!(table_stats.file_size_bytes, 125);

            let err = backend
                .commit_delete_files(vec![postgres_delete_file(
                    13,
                    SnapshotId(2),
                    1,
                    1,
                    None,
                )])
                .await
                .unwrap_err();
            assert_snapshot_conflict(&err, "delete", 2, 3);

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

            client
                .batch_execute(&format!(
                    r#"
                    CREATE TABLE {qualified}.ducklake_inlined_data_1_7(row_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT, value INTEGER);
                    INSERT INTO {qualified}.ducklake_inlined_data_1_7 VALUES (7, 2, NULL, 70);
                    INSERT INTO {qualified}.ducklake_inlined_data_1_7 VALUES (8, 4, NULL, 80);
                    INSERT INTO {qualified}.ducklake_inlined_data_1_7 VALUES (9, 2, 4, 90);
                    UPDATE {qualified}.ducklake_table_stats
                    SET record_count = 6, next_row_id = 9
                    WHERE table_id = 1;
                    "#
                ))
                .await
                .unwrap();
            let flushed_file = AppendFile {
                begin_snapshot: Some(SnapshotId(2)),
                row_id_start: Some(7),
                explicit_row_ids: true,
                record_count: 2,
                file_size_bytes: 30,
                partition_id: None,
                partition_values: Vec::new(),
                partial_max: Some(SnapshotId(4)),
                ..postgres_append_file(13, SnapshotId(4))
            };
            let flush_snapshot = backend
                .commit_table_changes(TableChangeCommit {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(4),
                    append_files: vec![flushed_file],
                    delete_files: Vec::new(),
                    inlined_file_deletions: Vec::new(),
                    inlined_data_appends: Vec::new(),
                    inlined_data_deletions: Vec::new(),
                    flushed_inlined_data: vec![FlushedInlinedDataCommit {
                        table_name: "ducklake_inlined_data_1_7".into(),
                        row_ids: vec![7, 8],
                        rows_flushed: 2,
                    }],
                    flushed_inlined_file_deletions: Vec::new(),
                    changes_made: "inline_flush:1".into(),
                    commit_message: "datafusion flush inlined data".into(),
                })
                .await
                .unwrap();
            assert_eq!(flush_snapshot, SnapshotId(5));
            let remaining_inlined: Vec<i64> = client
                .query(
                    &format!(
                        "SELECT row_id FROM {qualified}.ducklake_inlined_data_1_7 ORDER BY row_id"
                    ),
                    &[],
                )
                .await
                .unwrap()
                .iter()
                .map(|row| row.get(0))
                .collect();
            assert_eq!(remaining_inlined, vec![9]);
            let flushed_file_metadata = client
                .query_one(
                    &format!(
                        "SELECT begin_snapshot, partial_max, row_id_start, record_count \
                         FROM {qualified}.ducklake_data_file WHERE data_file_id = 13"
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(flushed_file_metadata.get::<_, i64>(0), 2);
            assert_eq!(flushed_file_metadata.get::<_, Option<i64>>(1), Some(4));
            assert_eq!(flushed_file_metadata.get::<_, Option<i64>>(2), Some(7));
            assert_eq!(flushed_file_metadata.get::<_, i64>(3), 2);
            let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
            assert_eq!(table_stats.record_count, 6);
            assert_eq!(table_stats.next_row_id, 9);
            assert_eq!(table_stats.file_size_bytes, 155);

            let flushed_delete = DeleteFileCommit {
                begin_snapshot: Some(SnapshotId(4)),
                rows_deleted: 0,
                delete_count: 1,
                partial_max: Some(SnapshotId(4)),
                ..postgres_delete_file(14, SnapshotId(5), 1, 0, None)
            };
            let flush_delete_snapshot = backend
                .commit_table_changes(TableChangeCommit {
                    table_id: TableId(1),
                    expected_snapshot: SnapshotId(5),
                    append_files: Vec::new(),
                    delete_files: vec![flushed_delete],
                    inlined_file_deletions: Vec::new(),
                    inlined_data_appends: Vec::new(),
                    inlined_data_deletions: Vec::new(),
                    flushed_inlined_data: Vec::new(),
                    flushed_inlined_file_deletions: vec![FlushedInlinedFileDeletionCommit {
                        rows: vec![crate::FlushedInlinedFileDeletion {
                            data_file_id: DataFileId(1),
                            row_id: 4,
                            begin_snapshot: SnapshotId(4),
                        }],
                    }],
                    changes_made: "inline_flush:1".into(),
                    commit_message: "datafusion flush inlined deletes".into(),
                })
                .await
                .unwrap();
            assert_eq!(flush_delete_snapshot, SnapshotId(6));
            let remaining_inlined_deletes: i64 = client
                .query_one(
                    &format!("SELECT COUNT(*) FROM {qualified}.ducklake_inlined_delete_1"),
                    &[],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(remaining_inlined_deletes, 0);
            let flushed_delete_file = client
                .query_one(
                    &format!(
                        "SELECT begin_snapshot, partial_max, delete_count \
                         FROM {qualified}.ducklake_delete_file WHERE delete_file_id = 14"
                    ),
                    &[],
                )
                .await
                .unwrap();
            assert_eq!(flushed_delete_file.get::<_, i64>(0), 4);
            assert_eq!(flushed_delete_file.get::<_, Option<i64>>(1), Some(4));
            assert_eq!(flushed_delete_file.get::<_, i64>(2), 1);
            let table_stats = backend.table_stats(TableId(1)).await.unwrap().unwrap();
            assert_eq!(table_stats.record_count, 6);
            assert_eq!(table_stats.next_row_id, 9);
            assert_eq!(table_stats.file_size_bytes, 155);

            drop_test_schema(&client, &schema).await;
        });
}

fn postgres_test_dsn() -> Option<String> {
    std::env::var("DUCKLAKE_POSTGRES_TEST_DSN").ok()
}

fn unique_test_schema() -> String {
    static NEXT_SCHEMA_SUFFIX: AtomicU64 = AtomicU64::new(0);

    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    let suffix = NEXT_SCHEMA_SUFFIX.fetch_add(1, Ordering::Relaxed);
    format!(
        "ducklake_storage_test_{}_{}_{}",
        std::process::id(),
        nanos,
        suffix
    )
}

async fn connect_test_postgres(dsn: &str) -> Client {
    let (client, connection) = tokio_postgres::connect(dsn, NoTls).await.unwrap();
    tokio::spawn(async move {
        let _ = connection.await;
    });
    client
}

async fn create_test_schema(client: &Client, schema: &str) {
    client
        .batch_execute(&format!(
            "DROP SCHEMA IF EXISTS {} CASCADE; CREATE SCHEMA {};",
            quote_ident(schema),
            quote_ident(schema)
        ))
        .await
        .unwrap();
}

async fn drop_test_schema(client: &Client, schema: &str) {
    client
        .batch_execute(&format!(
            "DROP SCHEMA IF EXISTS {} CASCADE;",
            quote_ident(schema)
        ))
        .await
        .unwrap();
}

async fn postgres_scheduled_file_count(client: &Client, schema: &str) -> i64 {
    client
        .query_one(
            &format!(
                "SELECT COUNT(*) FROM {}.ducklake_files_scheduled_for_deletion",
                quote_ident(schema)
            ),
            &[],
        )
        .await
        .unwrap()
        .get(0)
}

async fn postgres_table_count(client: &Client, schema: &str, table: &str) -> i64 {
    client
        .query_one(
            &format!(
                "SELECT COUNT(*) FROM {}",
                postgres_qualified_table(schema, table)
            ),
            &[],
        )
        .await
        .unwrap()
        .get(0)
}

async fn postgres_data_file_ids(client: &Client, schema: &str) -> Vec<(i64, String)> {
    client
        .query(
            &format!(
                "SELECT data_file_id, path FROM {} ORDER BY data_file_id",
                postgres_qualified_table(schema, "ducklake_data_file")
            ),
            &[],
        )
        .await
        .unwrap()
        .iter()
        .map(|row| (row.get(0), row.get(1)))
        .collect()
}

async fn postgres_snapshot_changes(client: &Client, schema: &str) -> Vec<String> {
    client
        .query(
            &format!(
                "SELECT changes_made FROM {} ORDER BY snapshot_id",
                postgres_qualified_table(schema, "ducklake_snapshot_changes")
            ),
            &[],
        )
        .await
        .unwrap()
        .iter()
        .map(|row| row.get(0))
        .collect()
}

async fn postgres_table_stats_file_size(client: &Client, schema: &str, table_id: TableId) -> i64 {
    client
        .query_one(
            &format!(
                "SELECT file_size_bytes FROM {} WHERE table_id = $1",
                postgres_qualified_table(schema, "ducklake_table_stats")
            ),
            &[&table_id.0],
        )
        .await
        .unwrap()
        .get(0)
}

async fn postgres_latest_next_file_id(client: &Client, schema: &str) -> i64 {
    client
        .query_one(
            &format!(
                "SELECT next_file_id FROM {} ORDER BY snapshot_id DESC LIMIT 1",
                postgres_qualified_table(schema, "ducklake_snapshot")
            ),
            &[],
        )
        .await
        .unwrap()
        .get(0)
}

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

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

fn scalar_field(name: &str, ducklake_type: DuckLakeType) -> SchemaField {
    let mut field = test_schema_field(1, name, &ducklake_type.to_string(), ducklake_type);
    field.column_order = 0;
    field
}
