use super::*;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use arrow::array::{
    ArrayRef, BinaryArray, FixedSizeBinaryArray, Int32Array, StringArray, StructArray,
};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use arrow_schema::{DataType, Field};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use ducklake_catalog::ResolvedFile;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use ducklake_catalog::{
    ColumnId, DeleteFormat, DuckLakeType, PartitionTransform, ResolvedMappingTarget, SchemaField,
};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use ducklake_catalog::{
    DataFileId, DeleteFile, InlinedFileDeletion, SnapshotId, TableId, Validity,
};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use ducklake_catalog::{DeleteFileId, PartitionId, SchemaEntry, SchemaId, TableEntry, ViewId};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use std::collections::HashSet;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_schema(schema_id: i64, path: Option<&str>, path_is_relative: bool) -> SchemaEntry {
    SchemaEntry {
        schema_id: SchemaId(schema_id),
        schema_uuid: format!("schema-{schema_id}"),
        validity: Validity {
            begin: SnapshotId(1),
            end: None,
        },
        schema_name: format!("schema_{schema_id}"),
        path: path.map(ToString::to_string),
        path_is_relative,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_table(path: Option<&str>, path_is_relative: bool) -> TableEntry {
    TableEntry {
        table_id: TableId(1),
        table_uuid: "table-1".into(),
        validity: Validity {
            begin: SnapshotId(1),
            end: None,
        },
        schema_id: SchemaId(1),
        table_name: "numbers".into(),
        path: path.map(ToString::to_string),
        path_is_relative,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_append_file(
    data_file_id: i64,
    table_id: TableId,
    expected_snapshot: SnapshotId,
    record_count: i64,
) -> AppendFile {
    AppendFile {
        table_id,
        expected_snapshot,
        data_file_id: DataFileId(data_file_id),
        begin_snapshot: None,
        path: format!("data-{data_file_id}.parquet"),
        path_is_relative: true,
        record_count,
        file_size_bytes: record_count * 10,
        footer_size: None,
        row_id_start: None,
        explicit_row_ids: false,
        column_stats: Vec::new(),
        variant_stats: Vec::new(),
        partition_id: None,
        partition_values: Vec::new(),
        encryption_key: None,
        mapping_id: None,
        partial_max: None,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_delete_file(
    delete_file_id: i64,
    table_id: TableId,
    expected_snapshot: SnapshotId,
) -> DeleteFileCommit {
    DeleteFileCommit {
        table_id,
        expected_snapshot,
        delete_file_id: DeleteFileId(delete_file_id),
        begin_snapshot: None,
        data_file_id: DataFileId(99),
        path: format!("delete-{delete_file_id}.parquet"),
        path_is_relative: true,
        format: DeleteFormat::Parquet,
        delete_count: 1,
        rows_deleted: 1,
        file_size_bytes: 10,
        footer_size: Some(0),
        partial_max: None,
        overwritten_delete_file_id: None,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn empty_table_change() -> TableChangeCommit {
    TableChangeCommit {
        table_id: TableId(1),
        expected_snapshot: SnapshotId(7),
        append_files: Vec::new(),
        delete_files: Vec::new(),
        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: "changed_table:1".into(),
        commit_message: "test change".into(),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_truncate(rows_deleted: i64) -> TruncateTableCommit {
    TruncateTableCommit {
        table_id: TableId(1),
        expected_snapshot: SnapshotId(7),
        rows_deleted,
        inlined_data_deletions: Vec::new(),
        changes_made: "deleted_from_table:1".into(),
        commit_message: "test truncate".into(),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn one_row_batch() -> RecordBatch {
    let schema = std::sync::Arc::new(arrow_schema::Schema::new(vec![Field::new(
        "id",
        DataType::Int64,
        false,
    )]));
    RecordBatch::try_new(
        schema,
        vec![
            std::sync::Arc::new(arrow::array::Int64Array::from(vec![1_i64]))
                as arrow::array::ArrayRef,
        ],
    )
    .unwrap()
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn three_row_int32_batch(name: &str) -> RecordBatch {
    let schema = std::sync::Arc::new(arrow_schema::Schema::new(vec![Field::new(
        name,
        DataType::Int32,
        true,
    )]));
    RecordBatch::try_new(
        schema,
        vec![std::sync::Arc::new(Int32Array::from(vec![1_i32, 2, 3])) as ArrayRef],
    )
    .unwrap()
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_merge_adjacent_compaction() -> MergeAdjacentCompactionCommit {
    MergeAdjacentCompactionCommit {
        table_id: TableId(1),
        expected_snapshot: SnapshotId(7),
        source_data_file_ids: vec![DataFileId(11), DataFileId(10)],
        compacted_files: vec![
            test_append_file(90, TableId(1), SnapshotId(7), 5),
            test_append_file(91, TableId(1), SnapshotId(7), 6),
        ],
    }
}

#[cfg(feature = "sqlite")]
fn test_rewrite_data_files() -> RewriteDataFilesCommit {
    RewriteDataFilesCommit {
        table_id: TableId(1),
        expected_snapshot: SnapshotId(7),
        source_data_file_ids: vec![DataFileId(10), DataFileId(11)],
        rewritten_files: vec![test_append_file(90, TableId(1), SnapshotId(7), 11)],
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_scheduled_file(
    data_file_id: i64,
    path: &str,
    path_is_relative: bool,
    schedule_start: i64,
) -> FileScheduledForDeletion {
    FileScheduledForDeletion {
        data_file_id: DataFileId(data_file_id),
        path: path.into(),
        path_is_relative,
        schedule_start,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_cleanup_metadata_file(id: i64) -> CleanupMetadataFile {
    CleanupMetadataFile {
        id,
        path: format!("file-{id}.parquet"),
        path_is_relative: true,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_snapshot(snapshot_id: i64, snapshot_time: i64) -> Snapshot {
    Snapshot {
        snapshot_id: SnapshotId(snapshot_id),
        snapshot_time,
        schema_version: 1,
        next_catalog_id: 10,
        next_file_id: 20,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_catalog_snapshot_ids() -> CatalogSnapshotIds {
    CatalogSnapshotIds {
        snapshot_id: 7,
        schema_version: 3,
        next_catalog_id: 40,
        next_file_id: 90,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_column_entry() -> ColumnEntry {
    ColumnEntry {
        column_id: ColumnId(3),
        validity: Validity {
            begin: SnapshotId(1),
            end: None,
        },
        table_id: TableId(1),
        column_order: 1,
        column_name: "value".into(),
        column_type: "int32".into(),
        initial_default: None,
        default_value: Some("0".into()),
        nulls_allowed: true,
        parent_column: None,
        default_value_type: Some("literal".into()),
        default_value_dialect: Some("duckdb".into()),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_sort_expression(expression: &str) -> SortExpressionCommit {
    SortExpressionCommit {
        expression: expression.into(),
        dialect: "duckdb".into(),
        sort_direction: "ASC".into(),
        null_order: "NULLS_LAST".into(),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_partition_column(column_id: i64) -> PartitionColumnCommit {
    PartitionColumnCommit {
        column_id: ColumnId(column_id),
        transform: PartitionTransform::Identity,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_create_table_column(column_id: i64, name: &str) -> CreateTableColumn {
    CreateTableColumn {
        column_id: ColumnId(column_id),
        column_order: column_id,
        column_name: name.into(),
        column_type: "int32".into(),
        nulls_allowed: true,
        parent_column: None,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn append_files_commit_plan_rejects_empty_and_mismatched_inputs() {
    assert!(matches!(
        plan_append_files_commit(&[]),
        Err(StorageError::Backend(message))
            if message == "DuckLake append commit received no files"
    ));

    let mixed_tables = vec![
        test_append_file(10, TableId(1), SnapshotId(7), 3),
        test_append_file(11, TableId(2), SnapshotId(7), 4),
    ];
    assert!(matches!(
        plan_append_files_commit(&mixed_tables),
        Err(StorageError::Backend(message))
            if message == "DuckLake append files span multiple tables"
    ));

    let mixed_snapshots = vec![
        test_append_file(10, TableId(1), SnapshotId(7), 3),
        test_append_file(11, TableId(1), SnapshotId(8), 4),
    ];
    assert!(matches!(
        plan_append_files_commit(&mixed_snapshots),
        Err(StorageError::Backend(message))
            if message == "DuckLake append files have inconsistent expected snapshots"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn append_files_commit_plan_allocates_row_ids_and_totals() {
    let files = vec![
        test_append_file(10, TableId(1), SnapshotId(7), 3),
        test_append_file(11, TableId(1), SnapshotId(7), 4),
    ];
    let plan = plan_append_files_commit(&files).unwrap();
    assert_eq!(plan.table_id, TableId(1));
    assert_eq!(plan.expected_snapshot, SnapshotId(7));

    let snapshot_plan = plan.plan_snapshot(7, 10, Some(100), &files).unwrap();
    assert_eq!(snapshot_plan.new_snapshot, SnapshotId(8));
    assert_eq!(snapshot_plan.new_next_file_id, 12);
    assert_eq!(snapshot_plan.total_record_count, 7);
    assert_eq!(snapshot_plan.total_allocated_row_id_count, 7);
    assert_eq!(snapshot_plan.total_file_size_bytes, 70);
    assert_eq!(snapshot_plan.files.len(), 2);
    assert_eq!(snapshot_plan.files[0].row_id_start, Some(100));
    assert_eq!(snapshot_plan.files[1].row_id_start, Some(103));
    assert_eq!(snapshot_plan.files[0].begin_snapshot, 8);
    assert_eq!(snapshot_plan.files[1].begin_snapshot, 8);
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn append_files_commit_plan_resolves_begin_snapshot_overrides() {
    let mut rewritten = test_append_file(10, TableId(1), SnapshotId(7), 3);
    rewritten.begin_snapshot = Some(SnapshotId(3));
    let normal = test_append_file(11, TableId(1), SnapshotId(7), 4);
    let files = vec![rewritten, normal];
    let plan = plan_append_files_commit(&files).unwrap();

    let snapshot_plan = plan.plan_snapshot(7, 10, Some(100), &files).unwrap();
    assert_eq!(snapshot_plan.files[0].begin_snapshot, 3);
    assert_eq!(snapshot_plan.files[1].begin_snapshot, 8);
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn append_files_commit_plan_rejects_future_begin_snapshot() {
    let mut future = test_append_file(10, TableId(1), SnapshotId(7), 3);
    future.begin_snapshot = Some(SnapshotId(9));
    let files = vec![future];
    let plan = plan_append_files_commit(&files).unwrap();

    assert!(matches!(
        plan.plan_snapshot(7, 10, Some(100), &files),
        Err(StorageError::Backend(message))
            if message == "DuckLake append begin_snapshot 9 is greater than the new snapshot 8"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn append_files_commit_plan_preserves_explicit_row_ids_without_allocating() {
    let mut explicit = test_append_file(10, TableId(1), SnapshotId(7), 3);
    explicit.explicit_row_ids = true;
    explicit.row_id_start = Some(900);
    let normal = test_append_file(11, TableId(1), SnapshotId(7), 4);
    let files = vec![explicit, normal];
    let plan = plan_append_files_commit(&files).unwrap();

    let snapshot_plan = plan.plan_snapshot(7, 10, Some(100), &files).unwrap();
    assert_eq!(snapshot_plan.total_record_count, 7);
    assert_eq!(snapshot_plan.total_allocated_row_id_count, 4);
    assert_eq!(snapshot_plan.files[0].row_id_start, Some(900));
    assert_eq!(snapshot_plan.files[1].row_id_start, Some(100));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn append_files_commit_plan_rejects_conflicts_and_file_id_gaps() {
    let files = vec![
        test_append_file(10, TableId(1), SnapshotId(7), 3),
        test_append_file(12, TableId(1), SnapshotId(7), 4),
    ];
    let plan = plan_append_files_commit(&files).unwrap();

    assert!(matches!(
        plan.plan_snapshot(8, 10, Some(100), &files),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 7,
            found_snapshot: 8,
        }) if operation == "append"
    ));

    assert!(matches!(
        plan.plan_snapshot(7, 10, Some(100), &files),
        Err(StorageError::Backend(message))
            if message == "DuckLake next_file_id sequence expected append file id 11, found 12"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn table_change_commit_plan_rejects_empty_and_mismatched_inputs() {
    assert!(matches!(
        plan_table_change_commit(&empty_table_change()),
        Err(StorageError::Backend(message))
            if message == "DuckLake table-change commit received no changes"
    ));

    let mut mixed_append = empty_table_change();
    mixed_append.append_files = vec![test_append_file(10, TableId(2), SnapshotId(7), 3)];
    assert!(matches!(
        plan_table_change_commit(&mixed_append),
        Err(StorageError::Backend(message))
            if message == "DuckLake table-change append files span multiple tables"
    ));

    let mut mixed_delete = empty_table_change();
    mixed_delete.delete_files = vec![test_delete_file(10, TableId(1), SnapshotId(8))];
    assert!(matches!(
        plan_table_change_commit(&mixed_delete),
        Err(StorageError::Backend(message))
            if message == "DuckLake table-change delete files have inconsistent expected snapshots"
    ));

    let mut empty_inlined_delete = empty_table_change();
    empty_inlined_delete.inlined_file_deletions = vec![InlinedFileDeletionCommit {
        table_id: TableId(1),
        expected_snapshot: SnapshotId(7),
        data_file_id: DataFileId(10),
        row_ids: Vec::new(),
        rows_deleted: 0,
    }];
    assert!(matches!(
        plan_table_change_commit(&empty_inlined_delete),
        Err(StorageError::Backend(message))
            if message == "DuckLake table-change inlined file deletion has no rows"
    ));

    let mut mismatched_inlined_delete = empty_table_change();
    mismatched_inlined_delete.inlined_file_deletions = vec![InlinedFileDeletionCommit {
        table_id: TableId(1),
        expected_snapshot: SnapshotId(7),
        data_file_id: DataFileId(10),
        row_ids: vec![1, 2],
        rows_deleted: 1,
    }];
    assert!(matches!(
        plan_table_change_commit(&mismatched_inlined_delete),
        Err(StorageError::Backend(message))
            if message
                == "DuckLake table-change inlined file deletion row count does not match row ids"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn table_change_commit_plan_sequences_files_and_allocates_append_rows() {
    let mut changes = empty_table_change();
    changes.append_files = vec![
        test_append_file(10, TableId(1), SnapshotId(7), 3),
        test_append_file(12, TableId(1), SnapshotId(7), 4),
    ];
    changes.delete_files = vec![test_delete_file(11, TableId(1), SnapshotId(7))];
    let plan = plan_table_change_commit(&changes).unwrap();
    assert_eq!(plan.table_id, TableId(1));
    assert_eq!(plan.expected_snapshot, SnapshotId(7));

    let snapshot_plan = plan.plan_snapshot(7, 10, Some(100), &changes).unwrap();
    assert_eq!(snapshot_plan.new_snapshot, SnapshotId(8));
    assert_eq!(snapshot_plan.new_next_file_id, 13);
    assert_eq!(snapshot_plan.next_row_id_after_appends, Some(107));
    assert_eq!(snapshot_plan.total_appended_record_count, 7);
    assert_eq!(snapshot_plan.total_allocated_row_id_count, 7);
    assert_eq!(snapshot_plan.total_appended_file_size_bytes, 70);
    assert_eq!(snapshot_plan.total_declared_deleted_rows, 1);
    assert_eq!(snapshot_plan.append_files.len(), 2);
    assert_eq!(snapshot_plan.append_files[0].row_id_start, Some(100));
    assert_eq!(snapshot_plan.append_files[1].row_id_start, Some(103));
    assert!(snapshot_plan.inlined_data_appends.is_empty());
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn table_change_commit_plan_counts_inlined_file_deletions() {
    let mut changes = empty_table_change();
    changes.delete_files = vec![test_delete_file(10, TableId(1), SnapshotId(7))];
    changes.inlined_file_deletions = vec![InlinedFileDeletionCommit {
        table_id: TableId(1),
        expected_snapshot: SnapshotId(7),
        data_file_id: DataFileId(20),
        row_ids: vec![3, 4],
        rows_deleted: 2,
    }];

    let plan = plan_table_change_commit(&changes).unwrap();
    let snapshot_plan = plan.plan_snapshot(7, 10, Some(100), &changes).unwrap();
    assert_eq!(snapshot_plan.total_declared_deleted_rows, 3);
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn table_change_commit_plan_reserves_file_id_for_inlined_only_appends() {
    let mut changes = empty_table_change();
    changes.inlined_data_appends = vec![InlinedDataAppend {
        schema_version: 1,
        schema: TableSchema {
            columns: Vec::new(),
            by_field_id: HashMap::new(),
        },
        batches: vec![one_row_batch()],
        column_stats: Vec::new(),
        row_ids: None,
        rows_appended: 1,
    }];
    let plan = plan_table_change_commit(&changes).unwrap();

    let snapshot_plan = plan.plan_snapshot(7, 10, Some(100), &changes).unwrap();
    assert_eq!(snapshot_plan.new_next_file_id, 11);
    assert_eq!(snapshot_plan.next_row_id_after_appends, Some(101));
    assert_eq!(snapshot_plan.total_appended_record_count, 1);
    assert_eq!(snapshot_plan.total_allocated_row_id_count, 1);
    assert!(snapshot_plan.append_files.is_empty());
    assert_eq!(snapshot_plan.inlined_data_appends.len(), 1);
    assert_eq!(snapshot_plan.inlined_data_appends[0].row_id_start, 100);
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn table_change_commit_plan_preserves_inlined_row_ids_without_allocating() {
    let mut changes = empty_table_change();
    changes.inlined_data_appends = vec![InlinedDataAppend {
        schema_version: 1,
        schema: TableSchema {
            columns: Vec::new(),
            by_field_id: HashMap::new(),
        },
        batches: vec![one_row_batch()],
        column_stats: Vec::new(),
        row_ids: Some(vec![900]),
        rows_appended: 1,
    }];
    let plan = plan_table_change_commit(&changes).unwrap();

    let snapshot_plan = plan.plan_snapshot(7, 10, Some(100), &changes).unwrap();
    assert_eq!(snapshot_plan.next_row_id_after_appends, Some(100));
    assert_eq!(snapshot_plan.total_appended_record_count, 1);
    assert_eq!(snapshot_plan.total_allocated_row_id_count, 0);
    assert_eq!(snapshot_plan.inlined_data_appends[0].row_id_start, 100);
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn table_change_commit_plan_validates_inlined_append_rows() {
    let mut changes = empty_table_change();
    changes.inlined_data_appends = vec![InlinedDataAppend {
        schema_version: 1,
        schema: TableSchema {
            columns: Vec::new(),
            by_field_id: HashMap::new(),
        },
        batches: Vec::new(),
        column_stats: Vec::new(),
        row_ids: Some(vec![1]),
        rows_appended: 1,
    }];

    assert!(matches!(
        plan_table_change_commit(&changes),
        Err(StorageError::Backend(message))
            if message == "DuckLake table-change inlined append row count does not match batches"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn table_change_commit_plan_rejects_snapshot_conflicts_and_file_id_gaps() {
    let mut changes = empty_table_change();
    changes.append_files = vec![test_append_file(11, TableId(1), SnapshotId(7), 3)];
    let plan = plan_table_change_commit(&changes).unwrap();

    assert!(matches!(
        plan.plan_snapshot(8, 10, Some(100), &changes),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 7,
            found_snapshot: 8,
        }) if operation == "table-change"
    ));
    assert!(matches!(
        plan.plan_snapshot(7, 10, Some(100), &changes),
        Err(StorageError::Backend(message))
            if message == "DuckLake next_file_id sequence expected file id 10, found 11"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn delete_files_commit_plan_rejects_empty_and_mismatched_inputs() {
    assert!(matches!(
        plan_delete_files_commit(&[]),
        Err(StorageError::Backend(message))
            if message == "DuckLake delete commit received no files"
    ));

    let mixed_tables = vec![
        test_delete_file(10, TableId(1), SnapshotId(7)),
        test_delete_file(11, TableId(2), SnapshotId(7)),
    ];
    assert!(matches!(
        plan_delete_files_commit(&mixed_tables),
        Err(StorageError::Backend(message))
            if message == "DuckLake delete files span multiple tables"
    ));

    let mixed_snapshots = vec![
        test_delete_file(10, TableId(1), SnapshotId(7)),
        test_delete_file(11, TableId(1), SnapshotId(8)),
    ];
    assert!(matches!(
        plan_delete_files_commit(&mixed_snapshots),
        Err(StorageError::Backend(message))
            if message == "DuckLake delete files have inconsistent expected snapshots"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn delete_files_commit_plan_sequences_files_and_totals_deleted_rows() {
    let mut first = test_delete_file(10, TableId(1), SnapshotId(7));
    first.rows_deleted = 3;
    let mut second = test_delete_file(11, TableId(1), SnapshotId(7));
    second.rows_deleted = 4;
    let files = vec![first, second];
    let plan = plan_delete_files_commit(&files).unwrap();
    assert_eq!(plan.table_id, TableId(1));
    assert_eq!(plan.expected_snapshot, SnapshotId(7));

    let snapshot_plan = plan.plan_snapshot(7, 10, &files).unwrap();
    assert_eq!(snapshot_plan.new_snapshot, SnapshotId(8));
    assert_eq!(snapshot_plan.new_next_file_id, 12);
    assert_eq!(snapshot_plan.total_delete_count, 7);
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn delete_files_commit_plan_rejects_snapshot_conflicts_and_file_id_gaps() {
    let files = vec![test_delete_file(11, TableId(1), SnapshotId(7))];
    let plan = plan_delete_files_commit(&files).unwrap();

    assert!(matches!(
        plan.plan_snapshot(8, 10, &files),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 7,
            found_snapshot: 8,
        }) if operation == "delete"
    ));
    assert!(matches!(
        plan.plan_snapshot(7, 10, &files),
        Err(StorageError::Backend(message))
            if message == "DuckLake next_file_id sequence expected delete file id 10, found 11"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn truncate_table_commit_plan_rejects_negative_row_counts() {
    assert!(matches!(
        plan_truncate_table_commit(&test_truncate(-1)),
        Err(StorageError::Backend(message))
            if message == "DuckLake truncate row count is negative"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn truncate_table_commit_plan_preserves_file_ids_and_validates_counts() {
    let plan = plan_truncate_table_commit(&test_truncate(5)).unwrap();
    assert_eq!(plan.table_id, TableId(1));
    assert_eq!(plan.expected_snapshot, SnapshotId(7));
    assert_eq!(plan.rows_deleted, 5);

    let snapshot_plan = plan.plan_snapshot(7, 42).unwrap();
    assert_eq!(snapshot_plan.new_snapshot, SnapshotId(8));
    assert_eq!(snapshot_plan.next_file_id, 42);
    plan.validate_inlined_rows_deleted(5).unwrap();
    assert!(matches!(
        plan.validate_inlined_rows_deleted(6),
        Err(StorageError::Backend(message))
            if message == "DuckLake truncate inlined row count exceeds total row count"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn truncate_table_commit_plan_rejects_snapshot_conflicts() {
    let plan = plan_truncate_table_commit(&test_truncate(5)).unwrap();
    assert!(matches!(
        plan.plan_snapshot(8, 42),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 7,
            found_snapshot: 8,
        }) if operation == "truncate"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn maintenance_merge_adjacent_sources_validate_input_set() {
    let mut empty = test_merge_adjacent_compaction();
    empty.source_data_file_ids.clear();
    assert!(matches!(
        plan_merge_adjacent_compaction_sources(&empty),
        Err(StorageError::Backend(message))
            if message == "DuckLake merge-adjacent compaction received no source files"
    ));

    let mut duplicate = test_merge_adjacent_compaction();
    duplicate.source_data_file_ids = vec![DataFileId(10), DataFileId(10)];
    assert!(matches!(
        plan_merge_adjacent_compaction_sources(&duplicate),
        Err(StorageError::Backend(message))
            if message == "DuckLake merge-adjacent compaction source files contain duplicates"
    ));

    let mut overlapping = test_merge_adjacent_compaction();
    overlapping.compacted_files[0].data_file_id = DataFileId(10);
    assert!(matches!(
        plan_merge_adjacent_compaction_sources(&overlapping),
        Err(StorageError::Backend(message))
            if message == "DuckLake merge-adjacent compacted file id overlaps a source file"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn maintenance_merge_adjacent_plan_allocates_snapshot_and_file_size_delta() {
    let compaction = test_merge_adjacent_compaction();
    let source_ids = plan_merge_adjacent_compaction_sources(&compaction).unwrap();
    assert_eq!(source_ids, vec![10, 11]);

    let plan = plan_merge_adjacent_compaction_commit(
        &compaction,
        test_catalog_snapshot_ids(),
        &source_ids,
        2,
        [20, 30],
    )
    .unwrap();
    assert_eq!(plan.source_ids, vec![10, 11]);
    assert_eq!(plan.new_snapshot, 8);
    assert_eq!(plan.schema_version, 3);
    assert_eq!(plan.next_catalog_id, 40);
    assert_eq!(plan.new_next_file_id, 92);
    assert_eq!(plan.source_file_size_bytes, 50);
    assert_eq!(plan.output_file_size_bytes, 110);
    assert_eq!(plan.changes_made, "merge_adjacent:1");
    assert_eq!(plan.commit_message, "datafusion merge adjacent");
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn maintenance_merge_adjacent_plan_rejects_conflicts_and_missing_sources() {
    let compaction = test_merge_adjacent_compaction();
    let source_ids = plan_merge_adjacent_compaction_sources(&compaction).unwrap();
    let mut moved = test_catalog_snapshot_ids();
    moved.snapshot_id = 8;
    assert!(matches!(
        plan_merge_adjacent_compaction_commit(&compaction, moved, &source_ids, 2, [20, 30]),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 7,
            found_snapshot: 8,
        }) if operation == "merge-adjacent"
    ));
    assert!(matches!(
        plan_merge_adjacent_compaction_commit(
            &compaction,
            test_catalog_snapshot_ids(),
            &source_ids,
            1,
            [20]
        ),
        Err(StorageError::Backend(message))
            if message == "DuckLake merge-adjacent source files must all be active data files for the table"
    ));
}

#[cfg(feature = "sqlite")]
#[test]
fn maintenance_rewrite_plan_allocates_snapshot_and_file_size_delta() {
    let rewrite = test_rewrite_data_files();
    let source_ids = plan_rewrite_data_files_sources(&rewrite).unwrap();
    let plan = plan_rewrite_data_files_commit(
        &rewrite,
        test_catalog_snapshot_ids(),
        &source_ids,
        2,
        [45, 55],
    )
    .unwrap();
    assert_eq!(plan.source_ids, vec![10, 11]);
    assert_eq!(plan.new_snapshot, 8);
    assert_eq!(plan.new_next_file_id, 91);
    assert_eq!(plan.source_file_size_bytes, 100);
    assert_eq!(plan.output_file_size_bytes, 110);
    assert_eq!(plan.changes_made, "rewrite_data_files:1");
    assert_eq!(plan.commit_message, "datafusion rewrite data files");
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn cleanup_old_files_plan_filters_and_resolves_paths() {
    let rows = vec![
        test_scheduled_file(1, "main/a.parquet", true, 10),
        test_scheduled_file(2, "main/future.parquet", true, 200),
        test_scheduled_file(3, "/abs/c.parquet", false, 20),
    ];
    let due =
        plan_cleanup_old_files(CleanupOldFilesOptions::older_than(100), "/lake/", &rows).unwrap();
    assert_eq!(
        due,
        vec![
            CleanupFile {
                data_file_id: DataFileId(1),
                path: "/lake/main/a.parquet".into(),
            },
            CleanupFile {
                data_file_id: DataFileId(3),
                path: "/abs/c.parquet".into(),
            },
        ]
    );

    let all =
        plan_cleanup_old_files(CleanupOldFilesOptions::dry_run_all(), "/lake", &rows).unwrap();
    assert_eq!(all.len(), 3);
    assert_eq!(all[1].path, "/lake/main/future.parquet");
    assert_eq!(
        join_cleanup_path_components("/lake/", "/main/a.parquet"),
        "/lake/main/a.parquet"
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn cleanup_old_files_plan_rejects_ambiguous_options() {
    let rows = vec![test_scheduled_file(1, "main/a.parquet", true, 10)];
    assert!(matches!(
        plan_cleanup_old_files(
            CleanupOldFilesOptions {
                dry_run: false,
                cleanup_all: false,
                older_than_micros: None,
            },
            "/lake",
            &rows,
        ),
        Err(StorageError::Backend(message))
            if message == "DuckLake cleanup_old_files requires exactly one of cleanup_all or older_than"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn expirable_snapshots_plan_filters_versions_and_keeps_latest() {
    let snapshots = vec![
        test_snapshot(1, 10),
        test_snapshot(2, 20),
        test_snapshot(3, 30),
    ];
    let versions = plan_expirable_snapshots(
        &ExpireSnapshotsOptions::versions(vec![SnapshotId(1), SnapshotId(3)]),
        3,
        snapshots.clone(),
    )
    .unwrap();
    assert_eq!(
        versions
            .iter()
            .map(|snapshot| snapshot.snapshot_id)
            .collect::<Vec<_>>(),
        vec![SnapshotId(1)]
    );

    let older_than =
        plan_expirable_snapshots(&ExpireSnapshotsOptions::older_than(25), 3, snapshots).unwrap();
    assert_eq!(
        older_than
            .iter()
            .map(|snapshot| snapshot.snapshot_id)
            .collect::<Vec<_>>(),
        vec![SnapshotId(1), SnapshotId(2)]
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn expirable_snapshots_plan_rejects_ambiguous_options() {
    assert!(matches!(
        plan_expirable_snapshots(
            &ExpireSnapshotsOptions {
                dry_run: false,
                versions: Vec::new(),
                older_than_micros: None,
            },
            1,
            vec![test_snapshot(1, 10)],
        ),
        Err(StorageError::Backend(message))
            if message == "DuckLake expire_snapshots requires exactly one of versions or older_than"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn snapshot_expiration_prune_plan_derives_ordered_ids() {
    let snapshots = vec![test_snapshot(1, 10), test_snapshot(2, 20)];
    let data_files = vec![
        test_cleanup_metadata_file(10),
        test_cleanup_metadata_file(11),
    ];
    let delete_files = vec![test_cleanup_metadata_file(20)];
    let plan = plan_snapshot_expiration_prune(&snapshots, vec![7, 8], &data_files, &delete_files);
    assert_eq!(plan.snapshot_ids, vec![1, 2]);
    assert_eq!(plan.deleted_table_ids, vec![7, 8]);
    assert_eq!(plan.deleted_data_file_ids, vec![10, 11]);
    assert_eq!(plan.deleted_delete_file_ids, vec![20]);
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn cleanup_queue_removal_plan_sorts_and_deduplicates_ids() {
    assert_eq!(
        plan_remove_files_scheduled_for_deletion(vec![
            DataFileId(3),
            DataFileId(1),
            DataFileId(3),
            DataFileId(2),
        ]),
        vec![1, 2, 3]
    );
    assert!(plan_remove_files_scheduled_for_deletion(Vec::new()).is_empty());
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn metadata_mutation_plan_bumps_schema_and_preserves_file_ids() {
    let metadata = test_catalog_snapshot_ids()
        .schema_metadata_mutation(
            "alter-column-type",
            SnapshotId(7),
            TableId(1),
            44,
            "metadata change".into(),
        )
        .unwrap();

    assert_eq!(metadata.table_id, TableId(1));
    assert_eq!(metadata.new_snapshot, 8);
    assert_eq!(metadata.new_schema_version, 4);
    assert_eq!(metadata.next_catalog_id, 44);
    assert_eq!(metadata.next_file_id, 90);
    assert_eq!(metadata.changes_made, "altered_table:1");
    assert_eq!(metadata.commit_message, "metadata change");
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn table_metadata_mutation_plan_preserves_schema_version() {
    let metadata = test_catalog_snapshot_ids()
        .table_metadata_mutation(
            "alter-sorted-by",
            SnapshotId(7),
            TableId(1),
            41,
            "set sorted".into(),
        )
        .unwrap();

    assert_eq!(metadata.table_id, TableId(1));
    assert_eq!(metadata.new_snapshot, 8);
    assert_eq!(metadata.new_schema_version, 3);
    assert_eq!(metadata.next_catalog_id, 41);
    assert_eq!(metadata.next_file_id, 90);
    assert_eq!(metadata.changes_made, "altered_table:1");
    assert_eq!(metadata.commit_message, "set sorted");
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn metadata_mutation_plan_rejects_conflicts_and_overflow() {
    assert!(matches!(
        test_catalog_snapshot_ids().schema_metadata_mutation(
            "drop-column",
            SnapshotId(6),
            TableId(1),
            40,
            "drop".into(),
        ),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 6,
            found_snapshot: 7,
        }) if operation == "drop-column"
    ));

    let overflow_snapshot = CatalogSnapshotIds {
        snapshot_id: i64::MAX,
        schema_version: 3,
        next_catalog_id: 40,
        next_file_id: 90,
    };
    assert!(matches!(
        overflow_snapshot.schema_metadata_mutation(
            "alter-column-type",
            SnapshotId(i64::MAX),
            TableId(1),
            40,
            "overflow".into(),
        ),
        Err(StorageError::Backend(message))
            if message == "DuckLake metadata snapshot id overflow"
    ));

    let overflow_schema = CatalogSnapshotIds {
        snapshot_id: 7,
        schema_version: i64::MAX,
        next_catalog_id: 40,
        next_file_id: 90,
    };
    assert!(matches!(
        overflow_schema.schema_metadata_mutation(
            "alter-column-type",
            SnapshotId(7),
            TableId(1),
            40,
            "overflow".into(),
        ),
        Err(StorageError::Backend(message))
            if message == "DuckLake metadata schema version overflow"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn column_default_metadata_plan_replaces_default_fields() {
    let plan = plan_column_default_mutation(
        ColumnDefaultCommit {
            table_id: TableId(1),
            expected_snapshot: SnapshotId(7),
            column_id: ColumnId(3),
            default_value: Some("42".into()),
            default_value_type: Some("literal".into()),
            default_value_dialect: Some("duckdb".into()),
            commit_message: "set default".into(),
        },
        test_catalog_snapshot_ids(),
        Some(test_column_entry()),
    )
    .unwrap();

    assert_eq!(plan.metadata.table_id, TableId(1));
    assert_eq!(plan.metadata.new_snapshot, 8);
    assert_eq!(plan.metadata.new_schema_version, 4);
    assert_eq!(plan.metadata.next_catalog_id, 40);
    assert_eq!(plan.metadata.next_file_id, 90);
    assert_eq!(plan.metadata.changes_made, "altered_table:1");
    assert_eq!(plan.metadata.commit_message, "set default");
    assert_eq!(plan.column.column_id, ColumnId(3));
    assert_eq!(plan.column.column_name, "value");
    assert_eq!(plan.column.column_type, "int32");
    assert_eq!(plan.column.validity.begin, SnapshotId(8));
    assert_eq!(plan.column.validity.end, None);
    assert_eq!(plan.column.default_value.as_deref(), Some("42"));
    assert_eq!(plan.column.default_value_type.as_deref(), Some("literal"));
    assert_eq!(plan.column.default_value_dialect.as_deref(), Some("duckdb"));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn column_nullability_metadata_plan_reports_validation_and_noops() {
    let plan = plan_column_nullability_mutation(
        ColumnNullabilityCommit {
            table_id: TableId(1),
            expected_snapshot: SnapshotId(7),
            column_id: ColumnId(3),
            nulls_allowed: false,
            existing_values_validated: false,
            commit_message: "set not null".into(),
        },
        test_catalog_snapshot_ids(),
        Some(test_column_entry()),
    )
    .unwrap();
    assert!(plan.requires_existing_value_validation);
    assert!(!plan.replacement.column.nulls_allowed);
    assert_eq!(plan.replacement.column.validity.begin, SnapshotId(8));
    assert_eq!(plan.replacement.metadata.commit_message, "set not null");

    let validated = plan_column_nullability_mutation(
        ColumnNullabilityCommit {
            table_id: TableId(1),
            expected_snapshot: SnapshotId(7),
            column_id: ColumnId(3),
            nulls_allowed: false,
            existing_values_validated: true,
            commit_message: "set not null".into(),
        },
        test_catalog_snapshot_ids(),
        Some(test_column_entry()),
    )
    .unwrap();
    assert!(!validated.requires_existing_value_validation);

    assert!(matches!(
        plan_column_nullability_mutation(
            ColumnNullabilityCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(7),
                column_id: ColumnId(3),
                nulls_allowed: true,
                existing_values_validated: true,
                commit_message: "drop null".into(),
            },
            test_catalog_snapshot_ids(),
            Some(test_column_entry()),
        ),
        Err(StorageError::Backend(message))
            if message == "Cannot DROP NULL on column value - it has no NOT NULL constraint defined"
    ));

    let mut not_null_column = test_column_entry();
    not_null_column.nulls_allowed = false;
    assert!(matches!(
        plan_column_nullability_mutation(
            ColumnNullabilityCommit {
                table_id: TableId(1),
                expected_snapshot: SnapshotId(7),
                column_id: ColumnId(3),
                nulls_allowed: false,
                existing_values_validated: true,
                commit_message: "set not null".into(),
            },
            test_catalog_snapshot_ids(),
            Some(not_null_column),
        ),
        Err(StorageError::Backend(message))
            if message == "Cannot SET NOT NULL on column value - it already has a NOT NULL constraint"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn rename_column_metadata_plan_rewrites_name_and_conflicts() {
    let start = plan_rename_column_mutation_start(
        RenameColumnCommit {
            table_id: TableId(1),
            expected_snapshot: SnapshotId(7),
            column_id: ColumnId(3),
            new_column_name: "renamed".into(),
            commit_message: "rename".into(),
        },
        test_catalog_snapshot_ids(),
        Some(test_column_entry()),
    )
    .unwrap();

    assert_eq!(start.metadata.new_snapshot, 8);
    assert_eq!(start.column.column_name, "value");

    let plan = start.finish(None, 45).unwrap();
    assert_eq!(plan.metadata.next_catalog_id, 45);
    assert_eq!(plan.metadata.new_schema_version, 4);
    assert_eq!(plan.metadata.commit_message, "rename");
    assert_eq!(plan.column.column_id, ColumnId(3));
    assert_eq!(plan.column.column_name, "renamed");
    assert_eq!(plan.column.validity.begin, SnapshotId(8));
    assert_eq!(plan.column.validity.end, None);

    let conflict = plan_rename_column_mutation_start(
        RenameColumnCommit {
            table_id: TableId(1),
            expected_snapshot: SnapshotId(7),
            column_id: ColumnId(3),
            new_column_name: "renamed".into(),
            commit_message: "rename".into(),
        },
        test_catalog_snapshot_ids(),
        Some(test_column_entry()),
    )
    .unwrap();
    assert!(matches!(
        conflict.finish(Some("existing".into()), 45),
        Err(StorageError::Backend(message))
            if message == "DuckLake column existing already exists in table 1"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn sorted_by_metadata_plan_preserves_schema_version_and_allocates_sort_id() {
    let active = vec![ActiveSortSpec {
        sort_id: SortId(39),
        expressions: vec![test_sort_expression("value")],
    }];
    let requested = vec![test_sort_expression("renamed")];
    let plan = plan_sorted_by_commit(
        TableId(1),
        SnapshotId(7),
        test_catalog_snapshot_ids(),
        &active,
        &requested,
        "set sorted".into(),
    )
    .unwrap();
    let SortedByCommitPlan::Change(plan) = plan else {
        panic!("expected sorted-by change");
    };

    assert_eq!(plan.metadata.table_id, TableId(1));
    assert_eq!(plan.metadata.new_snapshot, 8);
    assert_eq!(plan.metadata.new_schema_version, 3);
    assert_eq!(plan.metadata.next_catalog_id, 41);
    assert_eq!(plan.metadata.next_file_id, 90);
    assert_eq!(plan.metadata.changes_made, "altered_table:1");
    assert_eq!(plan.metadata.commit_message, "set sorted");
    assert_eq!(plan.close_sort_id, Some(SortId(39)));
    assert_eq!(plan.insert_sort_id, Some(SortId(40)));

    let reset = plan_sorted_by_commit(
        TableId(1),
        SnapshotId(7),
        test_catalog_snapshot_ids(),
        &active,
        &[],
        "reset sorted".into(),
    )
    .unwrap();
    let SortedByCommitPlan::Change(reset) = reset else {
        panic!("expected sorted-by reset");
    };
    assert_eq!(reset.metadata.new_schema_version, 3);
    assert_eq!(reset.metadata.next_catalog_id, 40);
    assert_eq!(reset.close_sort_id, Some(SortId(39)));
    assert_eq!(reset.insert_sort_id, None);

    let no_op = plan_sorted_by_commit(
        TableId(1),
        SnapshotId(7),
        test_catalog_snapshot_ids(),
        &active,
        &active[0].expressions,
        "set sorted".into(),
    )
    .unwrap();
    assert_eq!(no_op, SortedByCommitPlan::NoOp(SnapshotId(7)));

    assert!(matches!(
        plan_sorted_by_commit(
            TableId(1),
            SnapshotId(6),
            test_catalog_snapshot_ids(),
            &active,
            &requested,
            "set sorted".into(),
        ),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 6,
            found_snapshot: 7,
        }) if operation == "alter-sorted-by"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn partitioning_batch_plan_squashes_final_changes_and_reserves_catalog_ids() {
    let partitioning = PartitioningBatchCommit {
        expected_snapshot: SnapshotId(7),
        changes: vec![
            PartitioningTableCommit {
                table_id: TableId(1),
                columns: vec![test_partition_column(1)],
            },
            PartitioningTableCommit {
                table_id: TableId(2),
                columns: vec![test_partition_column(2)],
            },
            PartitioningTableCommit {
                table_id: TableId(1),
                columns: Vec::new(),
            },
        ],
        commit_message: "partition batch".into(),
    };
    let final_changes = final_partitioning_changes(&partitioning.changes);
    assert_eq!(final_changes[0].table_id, TableId(1));
    assert_eq!(final_changes[1].table_id, TableId(2));

    let plan = plan_partitioning_batch_commit(
        &partitioning,
        test_catalog_snapshot_ids(),
        vec![
            PartitioningTableFacts {
                change: final_changes[0].clone(),
                active: vec![ActivePartitionSpec {
                    partition_id: PartitionId(37),
                    columns: vec![test_partition_column(1)],
                }],
            },
            PartitioningTableFacts {
                change: final_changes[1].clone(),
                active: Vec::new(),
            },
        ],
    )
    .unwrap();

    assert_eq!(plan.new_snapshot, 8);
    assert_eq!(plan.new_schema_version, 4);
    assert_eq!(plan.next_catalog_id, 43);
    assert_eq!(plan.next_file_id, 90);
    assert_eq!(plan.changes_made, "altered_table:1,altered_table:2");
    assert_eq!(plan.commit_message, "partition batch");
    assert_eq!(plan.table_changes.len(), 2);
    assert_eq!(plan.table_changes[0].table_id, TableId(1));
    assert_eq!(
        plan.table_changes[0].close_partition_id,
        Some(PartitionId(37))
    );
    assert_eq!(
        plan.table_changes[0].insert_partition_id,
        Some(PartitionId(40))
    );
    assert!(plan.table_changes[0].columns.is_empty());
    assert_eq!(plan.table_changes[1].table_id, TableId(2));
    assert_eq!(plan.table_changes[1].close_partition_id, None);
    assert_eq!(
        plan.table_changes[1].insert_partition_id,
        Some(PartitionId(41))
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn partitioning_batch_plan_preserves_noop_snapshots_and_rejects_bad_facts() {
    let partitioning = PartitioningBatchCommit {
        expected_snapshot: SnapshotId(7),
        changes: vec![PartitioningTableCommit {
            table_id: TableId(1),
            columns: vec![test_partition_column(1)],
        }],
        commit_message: "partition noop".into(),
    };
    let plan = plan_partitioning_batch_commit(
        &partitioning,
        test_catalog_snapshot_ids(),
        vec![PartitioningTableFacts {
            change: partitioning.changes[0].clone(),
            active: vec![ActivePartitionSpec {
                partition_id: PartitionId(39),
                columns: vec![test_partition_column(1)],
            }],
        }],
    )
    .unwrap();

    assert_eq!(plan.new_snapshot, 8);
    assert_eq!(plan.new_schema_version, 4);
    assert_eq!(plan.next_catalog_id, 41);
    assert!(plan.table_changes.is_empty());

    assert!(matches!(
        plan_partitioning_batch_commit(
            &PartitioningBatchCommit {
                expected_snapshot: SnapshotId(6),
                changes: partitioning.changes.clone(),
                commit_message: "partition conflict".into(),
            },
            test_catalog_snapshot_ids(),
            Vec::new(),
        ),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 6,
            found_snapshot: 7,
        }) if operation == "alter-partitioning"
    ));

    assert!(matches!(
        plan_partitioning_batch_commit(&partitioning, test_catalog_snapshot_ids(), Vec::new()),
        Err(StorageError::Backend(message))
            if message == "DuckLake partitioning planner expected 1 final table changes, found 0 fact rows"
    ));

    assert!(matches!(
        plan_partitioning_batch_commit(
            &partitioning,
            test_catalog_snapshot_ids(),
            vec![PartitioningTableFacts {
                change: partitioning.changes[0].clone(),
                active: vec![
                    ActivePartitionSpec {
                        partition_id: PartitionId(39),
                        columns: vec![test_partition_column(1)],
                    },
                    ActivePartitionSpec {
                        partition_id: PartitionId(40),
                        columns: vec![test_partition_column(1)],
                    },
                ],
            }],
        ),
        Err(StorageError::Backend(message))
            if message == "DuckLake table 1 has 2 active partition specs at snapshot 7"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn catalog_object_plans_allocate_ids_and_snapshot_changes() {
    let create_schema = plan_create_schema_catalog_mutation(
        &CreateSchemaCommit {
            schema_name: "analytics".into(),
            commit_message: "create schema".into(),
        },
        test_catalog_snapshot_ids(),
    )
    .unwrap();
    assert_eq!(create_schema.new_snapshot, 8);
    assert_eq!(create_schema.new_schema_version, 4);
    assert_eq!(create_schema.allocated_catalog_id, Some(40));
    assert_eq!(create_schema.next_catalog_id, 41);
    assert_eq!(create_schema.next_file_id, 90);
    assert_eq!(create_schema.changes_made, "created_schema:\"analytics\"");
    assert_eq!(create_schema.commit_message, "create schema");

    let create_table = plan_create_table_catalog_mutation(
        &CreateTableCommit {
            schema_name: "analytics".into(),
            table_name: "numbers".into(),
            columns: vec![test_create_table_column(1, "value")],
            commit_message: "create table".into(),
        },
        test_catalog_snapshot_ids(),
    )
    .unwrap();
    assert_eq!(create_table.allocated_catalog_id, Some(40));
    assert_eq!(create_table.next_catalog_id, 41);
    assert_eq!(
        create_table.changes_made,
        "created_table:\"analytics\".\"numbers\""
    );

    let create_view = plan_create_view_catalog_mutation(
        &CreateViewCommit {
            schema_name: "analytics".into(),
            view_name: "v\"n".into(),
            dialect: "duckdb".into(),
            sql: "select 1".into(),
            column_aliases: None,
            commit_message: "create view".into(),
        },
        test_catalog_snapshot_ids(),
    )
    .unwrap();
    assert_eq!(create_view.allocated_catalog_id, Some(40));
    assert_eq!(
        create_view.changes_made,
        "created_view:\"analytics\".\"v\"\"n\""
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn create_table_with_inlined_data_plan_allocates_snapshot_without_file_id() {
    let create = CreateTableWithInlinedDataCommit {
        create: CreateTableCommit {
            schema_name: "analytics".into(),
            table_name: "numbers".into(),
            columns: vec![test_create_table_column(1, "value")],
            commit_message: "ctas inline".into(),
        },
        table_id: TableId(40),
        expected_snapshot: SnapshotId(7),
        batches: vec![three_row_int32_batch("value")],
        column_stats: Vec::new(),
        rows_inserted: 3,
    };

    let plan =
        plan_create_table_with_inlined_data_commit(&create, test_catalog_snapshot_ids(), true)
            .unwrap();

    assert_eq!(plan.catalog.new_snapshot, 8);
    assert_eq!(plan.catalog.new_schema_version, 4);
    assert_eq!(plan.catalog.allocated_catalog_id, Some(40));
    assert_eq!(plan.catalog.next_catalog_id, 41);
    assert_eq!(plan.catalog.next_file_id, 90);
    assert_eq!(
        plan.catalog.changes_made,
        "created_table:\"analytics\".\"numbers\""
    );
    assert_eq!(plan.catalog.commit_message, "ctas inline");
    assert_eq!(plan.total_record_count, 3);
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn create_table_with_inlined_data_plan_rejects_conflicts_and_missing_registry() {
    let create = CreateTableWithInlinedDataCommit {
        create: CreateTableCommit {
            schema_name: "analytics".into(),
            table_name: "numbers".into(),
            columns: vec![test_create_table_column(1, "value")],
            commit_message: "ctas inline".into(),
        },
        table_id: TableId(40),
        expected_snapshot: SnapshotId(7),
        batches: vec![three_row_int32_batch("value")],
        column_stats: Vec::new(),
        rows_inserted: 3,
    };

    let mut moved = test_catalog_snapshot_ids();
    moved.snapshot_id = 8;
    assert!(matches!(
        plan_create_table_with_inlined_data_commit(&create, moved, true),
        Err(StorageError::Conflict {
            operation,
            expected_snapshot: 7,
            found_snapshot: 8,
        }) if operation == "create-table-as"
    ));

    let mut wrong_table = create.clone();
    wrong_table.table_id = TableId(41);
    assert!(matches!(
        plan_create_table_with_inlined_data_commit(&wrong_table, test_catalog_snapshot_ids(), true),
        Err(StorageError::Backend(message))
            if message == "DuckLake CREATE TABLE AS expected table id 40, found 41"
    ));

    assert!(matches!(
        plan_create_table_with_inlined_data_commit(&create, test_catalog_snapshot_ids(), false),
        Err(StorageError::Backend(message))
            if message == "DuckLake CREATE TABLE AS inlined commit requires ducklake_inlined_data_tables"
    ));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn catalog_object_plans_drop_and_rename_without_allocating_ids() {
    let drop_table = plan_drop_table_catalog_mutation(
        &DropTableCommit {
            schema_name: "main".into(),
            table_name: "numbers".into(),
            commit_message: "drop table".into(),
        },
        test_catalog_snapshot_ids(),
        TableId(12),
    )
    .unwrap();
    assert_eq!(drop_table.new_snapshot, 8);
    assert_eq!(drop_table.new_schema_version, 4);
    assert_eq!(drop_table.allocated_catalog_id, None);
    assert_eq!(drop_table.next_catalog_id, 40);
    assert_eq!(drop_table.changes_made, "dropped_table:12");

    let rename = plan_rename_table_catalog_mutation(
        &RenameTableCommit {
            schema_name: "main".into(),
            old_table_name: "numbers".into(),
            new_schema_name: Some("analytics".into()),
            new_table_name: "renamed".into(),
            commit_message: "rename table".into(),
        },
        test_catalog_snapshot_ids(),
        "analytics",
    )
    .unwrap();
    assert_eq!(rename.allocated_catalog_id, None);
    assert_eq!(
        rename.changes_made,
        "created_table:\"analytics\".\"renamed\""
    );

    let drop_view = plan_drop_view_catalog_mutation(
        &DropViewCommit {
            schema_name: "main".into(),
            view_name: "v".into(),
            commit_message: "drop view".into(),
        },
        test_catalog_snapshot_ids(),
        ViewId(14),
    )
    .unwrap();
    assert_eq!(drop_view.changes_made, "dropped_view:14");

    let drop_schema = plan_drop_schema_catalog_mutation(
        &DropSchemaCommit {
            schema_name: "analytics".into(),
            cascade: true,
            commit_message: "drop schema".into(),
        },
        test_catalog_snapshot_ids(),
        SchemaId(9),
        &[12, 13],
        &[14],
    )
    .unwrap();
    assert_eq!(
        drop_schema.changes_made,
        "dropped_schema:9,dropped_table:12,dropped_table:13,dropped_view:14"
    );

    assert!(matches!(
        plan_drop_schema_catalog_mutation(
            &DropSchemaCommit {
                schema_name: "analytics".into(),
                cascade: false,
                commit_message: "drop schema".into(),
            },
            test_catalog_snapshot_ids(),
            SchemaId(9),
            &[12],
            &[],
        ),
        Err(StorageError::Backend(message))
            if message == "DuckLake schema analytics is not empty"
    ));
}

#[test]
fn resolves_relative_local_path() {
    assert_eq!(
        resolve_object_store_path("lake/table/file.parquet").unwrap(),
        ResolvedObjectStorePath {
            object_store_url: LOCAL_OBJECT_STORE_URL.to_string(),
            object_path: "lake/table/file.parquet".to_string(),
            is_local: true,
        }
    );
}

#[test]
fn resolves_absolute_file_url_to_local_path() {
    assert_eq!(
        resolve_object_store_path("file:///tmp/lake/table/file.parquet").unwrap(),
        ResolvedObjectStorePath {
            object_store_url: LOCAL_OBJECT_STORE_URL.to_string(),
            object_path: "/tmp/lake/table/file.parquet".to_string(),
            is_local: true,
        }
    );
    assert_eq!(
        local_filesystem_path("file:///tmp/lake/table/file.parquet").unwrap(),
        "/tmp/lake/table/file.parquet"
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn renamed_table_path_preserves_same_schema_storage_path() {
    let schema = test_schema(1, Some("main/"), true);
    let table = test_table(Some("numbers/"), true);

    assert_eq!(
        renamed_table_path_for_target_schema("/lake/", &schema, &schema, &table),
        (Some("numbers/".to_string()), true)
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn renamed_table_path_materializes_absolute_path_when_moved_outside_target_schema_path() {
    let source_schema = test_schema(1, None, true);
    let target_schema = test_schema(10, Some("analytics/"), true);
    let table = test_table(None, true);

    assert_eq!(
        renamed_table_path_for_target_schema("/lake/", &source_schema, &target_schema, &table),
        (Some("/lake/".to_string()), false)
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn renamed_table_path_stays_relative_when_old_path_is_under_target_schema_path() {
    let source_schema = test_schema(1, Some("analytics/"), true);
    let target_schema = test_schema(10, Some("analytics/"), true);
    let table = test_table(Some("numbers/"), true);

    assert_eq!(
        renamed_table_path_for_target_schema("/lake/", &source_schema, &target_schema, &table),
        (Some("numbers/".to_string()), true)
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn rewrite_sort_expression_updates_only_unqualified_column_refs() {
    assert_eq!(
        rewrite_sort_expression_for_renamed_column("sort_key_1", "sort_key_1", "renamed").unwrap(),
        "renamed"
    );
    assert_eq!(
        rewrite_sort_expression_for_renamed_column(
            "sort_key_1 || sort_key_2",
            "sort_key_1",
            "renamed",
        )
        .unwrap(),
        "renamed || sort_key_2"
    );
    assert_eq!(
        rewrite_sort_expression_for_renamed_column(
            "tbl.sort_key_1 || sort_key_1",
            "sort_key_1",
            "renamed",
        )
        .unwrap(),
        "tbl.sort_key_1 || renamed"
    );
    assert_eq!(
        rewrite_sort_expression_for_renamed_column("sort_key_1", "sort_key_1", "new name").unwrap(),
        "\"new name\""
    );
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn sort_expression_reference_check_matches_unqualified_column_refs() {
    assert!(
        sort_expression_references_unqualified_column("sort_key_1 || sort_key_2", "sort_key_1")
            .unwrap()
    );
    assert!(
        sort_expression_references_unqualified_column("tbl.sort_key_1 || sort_key_2", "sort_key_2")
            .unwrap()
    );
    assert!(
        !sort_expression_references_unqualified_column("tbl.sort_key_1", "sort_key_1").unwrap()
    );
    assert!(!sort_expression_references_unqualified_column("sort_key_10", "sort_key_1").unwrap());
}

#[test]
fn extracts_hive_partition_values_from_path_segments() {
    assert_eq!(
        hive_partition_value_from_path(
            "s3://bucket/lake/part_key=alpha%20beta/part_key2=10/data.parquet",
            "part_key",
        ),
        Some(Some("alpha beta".to_string()))
    );
    assert_eq!(
        hive_partition_value_from_path(
            "/tmp/lake/part_key=__HIVE_DEFAULT_PARTITION__/data.parquet",
            "part_key",
        ),
        Some(None)
    );
    assert_eq!(
        hive_partition_value_from_path("/tmp/lake/other=1/data.parquet", "part_key"),
        None
    );
}

#[test]
fn resolves_catalog_identifier_exact_match_first() {
    let value = resolve_catalog_identifier(
        vec!["Mixed".to_string(), "mixed".to_string()],
        "mixed",
        "table",
        |name| name.as_str(),
    )
    .unwrap();

    assert_eq!(value, "mixed");
}

#[test]
fn resolves_catalog_identifier_case_insensitive_fallback() {
    let value =
        resolve_catalog_identifier(vec!["my_table".to_string()], "MY_TABLE", "table", |name| {
            name.as_str()
        })
        .unwrap();

    assert_eq!(value, "my_table");
}

#[test]
fn rejects_ambiguous_case_insensitive_catalog_identifier() {
    let err = resolve_catalog_identifier(
        vec!["Mixed".to_string(), "MIXED".to_string()],
        "mixed",
        "table",
        |name| name.as_str(),
    )
    .unwrap_err();

    assert!(matches!(
        err,
        StorageError::Catalog(CatalogError::InvalidCatalog(message))
            if message.contains("ambiguous table identifier")
    ));
}

#[test]
fn resolves_s3_url_to_store_and_object_path() {
    assert_eq!(
        resolve_object_store_path("s3://bucket/lake/table/file.parquet").unwrap(),
        ResolvedObjectStorePath {
            object_store_url: "s3://bucket".to_string(),
            object_path: "lake/table/file.parquet".to_string(),
            is_local: false,
        }
    );
}

#[test]
fn resolves_https_url_to_store_and_object_path() {
    assert_eq!(
        resolve_object_store_path("https://example.com/lake/table/file.parquet").unwrap(),
        ResolvedObjectStorePath {
            object_store_url: "https://example.com".to_string(),
            object_path: "lake/table/file.parquet".to_string(),
            is_local: false,
        }
    );
}

#[test]
fn materialized_local_path_rejects_remote_object_store() {
    let err = local_filesystem_path("s3://bucket/lake/table/file.parquet").unwrap_err();
    assert!(
        err.to_string()
            .contains("Materialized object-store file access")
    );
}

#[test]
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn parses_ducklake_inlined_interval_literals() {
    let value =
        parse_interval_month_day_nano("14 months 3 days 4005 microseconds", "test").unwrap();
    assert_eq!(value.months, 14);
    assert_eq!(value.days, 3);
    assert_eq!(value.nanoseconds, 4_005_000);

    let value = parse_interval_month_day_nano("1 mon 2 days 00:00:03.004005", "test").unwrap();
    assert_eq!(value.months, 1);
    assert_eq!(value.days, 2);
    assert_eq!(value.nanoseconds, 3_004_005_000);
}

#[test]
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn nested_inlined_binary_and_uuid_literals_decode() {
    let field = test_schema_field(
        1,
        "payload",
        "struct",
        DuckLakeType::Struct,
        vec![
            test_schema_field(2, "bytes_value", "blob", DuckLakeType::Blob, vec![]),
            test_schema_field(3, "uuid_value", "uuid", DuckLakeType::Uuid, vec![]),
        ],
    );

    let array = inlined_nested_values_to_array(
        &field,
        vec![Some(
            "{bytes_value: X'00FF10', uuid_value: '550e8400-e29b-41d4-a716-446655440000'}".into(),
        )],
    )
    .unwrap();
    let struct_array = array.as_any().downcast_ref::<StructArray>().unwrap();
    let bytes = struct_array
        .column(0)
        .as_any()
        .downcast_ref::<BinaryArray>()
        .unwrap();
    assert_eq!(bytes.value(0), &[0x00, 0xff, 0x10]);
    let uuid = struct_array
        .column(1)
        .as_any()
        .downcast_ref::<FixedSizeBinaryArray>()
        .unwrap();
    assert_eq!(
        uuid.value(0),
        &[
            0x55, 0x0e, 0x84, 0x00, 0xe2, 0x9b, 0x41, 0xd4, 0xa7, 0x16, 0x44, 0x66, 0x55, 0x44,
            0x00, 0x00
        ]
    );
}

#[test]
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn nested_inlined_column_types_follow_backend_storage_model() {
    let field = test_schema_field(
        1,
        "payload",
        "struct",
        DuckLakeType::Struct,
        vec![
            test_schema_field(2, "a", "integer", DuckLakeType::Int32, vec![]),
            test_schema_field(3, "b", "varchar", DuckLakeType::Varchar, vec![]),
        ],
    );

    assert_eq!(
        inlined_column_type(&field, InlinedSqlDialect::Sqlite).unwrap(),
        "VARCHAR"
    );
    assert_eq!(
        inlined_column_type(&field, InlinedSqlDialect::Postgres).unwrap(),
        "VARCHAR"
    );
    assert_eq!(
        inlined_column_type(&field, InlinedSqlDialect::DuckDbFile).unwrap(),
        "STRUCT(\"a\" INTEGER, \"b\" VARCHAR)"
    );
    assert_eq!(
        inlined_column_type(&field, InlinedSqlDialect::Quack).unwrap(),
        "STRUCT(\"a\" INTEGER, \"b\" VARCHAR)"
    );
}

#[test]
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn renders_arrow_struct_as_duckdb_inlined_literal() {
    let field = test_schema_field(
        1,
        "payload",
        "struct",
        DuckLakeType::Struct,
        vec![
            test_schema_field(2, "a", "integer", DuckLakeType::Int32, vec![]),
            test_schema_field(3, "b", "varchar", DuckLakeType::Varchar, vec![]),
        ],
    );
    let array = StructArray::from(vec![
        (
            std::sync::Arc::new(Field::new("a", DataType::Int32, true)),
            std::sync::Arc::new(Int32Array::from(vec![Some(42)])) as ArrayRef,
        ),
        (
            std::sync::Arc::new(Field::new("b", DataType::Utf8, true)),
            std::sync::Arc::new(StringArray::from(vec![Some("forty-two")])) as ArrayRef,
        ),
    ]);

    let literal = inlined_duckdb_literal_value(
        &field,
        &array,
        0,
        "test",
        InlinedIntervalFormat::PreserveNanoseconds,
    )
    .unwrap();
    assert_eq!(literal, "{'a': 42, 'b': 'forty-two'}");

    let decoded = inlined_nested_values_to_array(&field, vec![Some(literal)]).unwrap();
    let decoded = decoded.as_any().downcast_ref::<StructArray>().unwrap();
    let values = decoded
        .column(0)
        .as_any()
        .downcast_ref::<Int32Array>()
        .unwrap();
    let labels = decoded
        .column(1)
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    assert_eq!(values.value(0), 42);
    assert_eq!(labels.value(0), "forty-two");
}

#[test]
fn rejects_file_url_with_authority() {
    let err = resolve_object_store_path("file://localhost/tmp/file.parquet").unwrap_err();
    assert!(err.to_string().contains("not an absolute local path"));
}

#[test]
fn rejects_object_store_url_with_query() {
    let err = resolve_object_store_path("s3://bucket/file.parquet?version=1").unwrap_err();
    assert!(err.to_string().contains("query or fragment"));
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn test_schema_field(
    id: i64,
    name: &str,
    type_name: &str,
    ducklake_type: DuckLakeType,
    children: Vec<SchemaField>,
) -> SchemaField {
    SchemaField {
        field_id: ColumnId(id),
        name: name.into(),
        type_name: type_name.into(),
        ducklake_type,
        nulls_allowed: true,
        initial_default: None,
        column_order: id,
        children,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn mapped_physical_schema_handles_list_and_map_children() {
    let arrow_schema = arrow_schema::Schema::new(vec![
        Field::new(
            "items",
            DataType::List(std::sync::Arc::new(Field::new(
                "element",
                DataType::Int32,
                true,
            ))),
            true,
        ),
        Field::new(
            "attrs",
            DataType::Map(
                std::sync::Arc::new(Field::new(
                    "entries",
                    DataType::Struct(
                        vec![
                            Field::new("k", DataType::Utf8, false),
                            Field::new("v", DataType::Int64, true),
                        ]
                        .into(),
                    ),
                    false,
                )),
                false,
            ),
            true,
        ),
    ]);

    let mut by_parent_and_name = std::collections::HashMap::new();
    by_parent_and_name.insert(
        (None, "items".to_string()),
        ResolvedMappingTarget {
            mapping_column_id: ColumnId(40),
            target_field_id: ColumnId(10),
        },
    );
    by_parent_and_name.insert(
        (Some(ColumnId(40)), "list".to_string()),
        ResolvedMappingTarget {
            mapping_column_id: ColumnId(41),
            target_field_id: ColumnId(11),
        },
    );
    by_parent_and_name.insert(
        (None, "attrs".to_string()),
        ResolvedMappingTarget {
            mapping_column_id: ColumnId(50),
            target_field_id: ColumnId(20),
        },
    );
    by_parent_and_name.insert(
        (Some(ColumnId(50)), "k".to_string()),
        ResolvedMappingTarget {
            mapping_column_id: ColumnId(51),
            target_field_id: ColumnId(21),
        },
    );
    by_parent_and_name.insert(
        (Some(ColumnId(50)), "v".to_string()),
        ResolvedMappingTarget {
            mapping_column_id: ColumnId(52),
            target_field_id: ColumnId(22),
        },
    );

    let file = ResolvedFile {
        data: DataFile {
            data_file_id: DataFileId(1),
            table_id: TableId(1),
            validity: Validity {
                begin: SnapshotId(1),
                end: None,
            },
            file_order: None,
            path: "file.parquet".into(),
            path_is_relative: false,
            file_format: "parquet".into(),
            record_count: 0,
            file_size_bytes: 0,
            footer_size: None,
            row_id_start: None,
            partition_id: None,
            encryption_key: None,
            mapping_id: Some(MappingId(1)),
            partial_max: None,
        },
        delete: None::<DeleteFile>,
        inlined_deletions: Vec::<InlinedFileDeletion>::new(),
        mapping: Some(ducklake_catalog::ResolvedMapping {
            mapping_type: "name".into(),
            by_parent_and_name,
        }),
        snapshot_row_filter: None,
        resolved_path: "file.parquet".into(),
        resolved_delete_path: None,
    };

    let physical = physical_schema_from_arrow_schema(&arrow_schema, &file).unwrap();

    assert_eq!(
        physical.by_field_id[&ColumnId(10)].physical_path,
        vec!["items"]
    );
    assert_eq!(
        physical.by_field_id[&ColumnId(11)].physical_path,
        vec!["items", "element"]
    );
    assert_eq!(physical.by_field_id[&ColumnId(11)].physical_type, "int32");
    assert_eq!(physical.by_field_id[&ColumnId(20)].physical_type, "map");
    assert_eq!(
        physical.by_field_id[&ColumnId(21)].physical_path,
        vec!["attrs", "k"]
    );
    assert_eq!(physical.by_field_id[&ColumnId(21)].physical_type, "varchar");
    assert_eq!(
        physical.by_field_id[&ColumnId(22)].physical_path,
        vec!["attrs", "v"]
    );
    assert_eq!(physical.by_field_id[&ColumnId(22)].physical_type, "int64");
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[test]
fn parquet_enum_annotation_overrides_arrow_binary_physical_type() {
    let mut metadata = std::collections::HashMap::new();
    metadata.insert(PARQUET_FIELD_ID_META_KEY.to_string(), "7".to_string());
    let arrow_schema = arrow_schema::Schema::new(vec![
        Field::new("event_type", DataType::Binary, true).with_metadata(metadata),
    ]);
    let file = ResolvedFile {
        data: DataFile {
            data_file_id: DataFileId(1),
            table_id: TableId(1),
            validity: Validity {
                begin: SnapshotId(1),
                end: None,
            },
            file_order: None,
            path: "file.parquet".into(),
            path_is_relative: false,
            file_format: "parquet".into(),
            record_count: 0,
            file_size_bytes: 0,
            footer_size: None,
            row_id_start: None,
            partition_id: None,
            encryption_key: None,
            mapping_id: None,
            partial_max: None,
        },
        delete: None::<DeleteFile>,
        inlined_deletions: Vec::<InlinedFileDeletion>::new(),
        mapping: None,
        snapshot_row_filter: None,
        resolved_path: "file.parquet".into(),
        resolved_delete_path: None,
    };

    let physical_without_override = physical_schema_from_arrow_schema(&arrow_schema, &file)
        .unwrap()
        .by_field_id
        .remove(&ColumnId(7))
        .unwrap();
    assert_eq!(physical_without_override.physical_type, "blob");

    let enum_paths = HashSet::from([vec!["event_type".to_string()]]);
    let physical_with_override =
        physical_schema_from_arrow_schema_with_overrides(&arrow_schema, &file, &enum_paths)
            .unwrap()
            .by_field_id
            .remove(&ColumnId(7))
            .unwrap();
    assert_eq!(physical_with_override.physical_type, "varchar");
}
