#![allow(dead_code)]
// Catalog query descriptors are shared across backend feature profiles. Narrow
// builds intentionally compile renderers/helpers whose consumers live behind
// other backend features.

use std::borrow::Cow;
use std::collections::{HashMap, HashSet};

use ducklake_catalog::{
    ColumnEntry, ColumnId, DuckLakeType, FileColumnStats, FilePartitionValue, FileVariantStats,
    PartitionId, PartitionTransform, SchemaEntry, SchemaId, SnapshotId, SortExpression, SortId,
    TableColumnStats, TableEntry, TableId, ViewEntry, ViewId,
};

use crate::ducklake_sql_string_literal as sql_string;
use crate::embedded_reads::quote_identifier;
use crate::embedded_rows::EmbeddedRow;
use crate::sql_utils::join_i64s;
use crate::{
    AddColumn, AppendFile, CreateTableColumn, DeleteFileCommit, PartitionColumnCommit,
    PlannedColumnTypeTreeRow, Result, SortExpressionCommit, StorageError,
};

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum CatalogParam {
    I64(i64),
    String(String),
    OptI64(Option<i64>),
    OptString(Option<String>),
    OptBool(Option<bool>),
    Bool(bool),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum UuidParamStyle {
    Text,
    CastAsUuid,
    /// Casts the bound parameter through `TEXT` before `UUID`
    /// (`CAST(CAST(? AS TEXT) AS UUID)`). Postgres infers a bare
    /// `CAST($n AS UUID)` parameter as the native `uuid` type, which
    /// tokio-postgres cannot serialize from a Rust `String`; the extra text cast
    /// makes it infer `text` instead so the string binds and is then cast.
    CastTextAsUuid,
}

pub(crate) fn metadata_rows() -> CatalogQuery {
    CatalogQuery::new(
        "SELECT key, value, scope, scope_id FROM {ducklake_metadata}",
        Vec::new(),
    )
}

pub(crate) fn global_metadata_value(key: &str) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT value FROM {ducklake_metadata} WHERE key = ?1 AND scope IS NULL",
        vec![key.into()],
    )
}

pub(crate) fn latest_snapshot_ids() -> CatalogQuery {
    CatalogQuery::new(
        "SELECT snapshot_id, schema_version, next_catalog_id, next_file_id \
         FROM {ducklake_snapshot} ORDER BY snapshot_id DESC LIMIT 1",
        Vec::new(),
    )
}

pub(crate) fn snapshot_next_file_id(snapshot_id: i64) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT next_file_id FROM {ducklake_snapshot} WHERE snapshot_id = ?1",
        vec![snapshot_id.into()],
    )
}

pub(crate) fn max_snapshot_id() -> CatalogQuery {
    CatalogQuery::new(
        "SELECT MAX(snapshot_id) FROM {ducklake_snapshot}",
        Vec::new(),
    )
}

pub(crate) fn delete_snapshots_after(snapshot_id: i64) -> CatalogQuery {
    CatalogQuery::new(
        "DELETE FROM {ducklake_snapshot} WHERE snapshot_id > ?1",
        vec![snapshot_id.into()],
    )
}

pub(crate) fn insert_snapshot_with_timestamp_text(
    snapshot_id: i64,
    snapshot_time: String,
    schema_version: i64,
    next_catalog_id: i64,
    next_file_id: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_snapshot} VALUES (?1, ?2, ?3, ?4, ?5)",
        vec![
            snapshot_id.into(),
            snapshot_time.into(),
            schema_version.into(),
            next_catalog_id.into(),
            next_file_id.into(),
        ],
    )
}

pub(crate) fn insert_snapshot_current_timestamp(
    snapshot_id: i64,
    schema_version: i64,
    next_catalog_id: i64,
    next_file_id: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_snapshot} VALUES (?1, CURRENT_TIMESTAMP, ?2, ?3, ?4)",
        vec![
            snapshot_id.into(),
            schema_version.into(),
            next_catalog_id.into(),
            next_file_id.into(),
        ],
    )
}

pub(crate) fn insert_snapshot_change(
    snapshot_id: i64,
    changes_made: &str,
    commit_message: &str,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_snapshot_changes} VALUES (?1, ?2, NULL, ?3, NULL)",
        vec![
            snapshot_id.into(),
            changes_made.into(),
            commit_message.into(),
        ],
    )
}

pub(crate) fn next_row_id_from_table_stats(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT next_row_id FROM {ducklake_table_stats} WHERE table_id = ?1",
        vec![table_id.0.into()],
    )
}

pub(crate) fn coalesced_next_row_id_from_table_stats(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT COALESCE(next_row_id, 0) FROM {ducklake_table_stats} WHERE table_id = ?1",
        vec![table_id.0.into()],
    )
}

pub(crate) fn next_row_id_from_data_files(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT MAX(row_id_start + record_count) \
         FROM {ducklake_data_file} \
         WHERE table_id = ?1 AND row_id_start IS NOT NULL",
        vec![table_id.0.into()],
    )
}

pub(crate) fn close_schema(schema_id: SchemaId, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_schema} \
         SET end_snapshot = ?2 \
         WHERE schema_id = ?1 AND end_snapshot IS NULL",
        vec![schema_id.0.into(), new_snapshot.into()],
    )
}

pub(crate) fn close_tables_for_schema(
    schema_id: SchemaId,
    snapshot: SnapshotId,
    new_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table} \
         SET end_snapshot = ?2 \
         WHERE schema_id = ?1 \
           AND begin_snapshot <= ?3 \
           AND end_snapshot IS NULL",
        vec![schema_id.0.into(), new_snapshot.into(), snapshot.0.into()],
    )
}

pub(crate) fn close_views_for_schema(
    schema_id: SchemaId,
    snapshot: SnapshotId,
    new_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_view} \
         SET end_snapshot = ?2 \
         WHERE schema_id = ?1 \
           AND begin_snapshot <= ?3 \
           AND end_snapshot IS NULL",
        vec![schema_id.0.into(), new_snapshot.into(), snapshot.0.into()],
    )
}

pub(crate) fn close_table(table_id: TableId, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table} \
         SET end_snapshot = ?2 \
         WHERE table_id = ?1 AND end_snapshot IS NULL",
        vec![table_id.0.into(), new_snapshot.into()],
    )
    .with_affected_count(
        "SELECT COUNT(*) FROM {ducklake_table} WHERE table_id = ?1 AND end_snapshot IS NULL",
    )
}

pub(crate) fn close_view(view_id: ViewId, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_view} \
         SET end_snapshot = ?2 \
         WHERE view_id = ?1 AND end_snapshot IS NULL",
        vec![view_id.0.into(), new_snapshot.into()],
    )
}

pub(crate) fn close_table_columns(table_id: TableId, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_column} \
         SET end_snapshot = ?2 \
         WHERE table_id = ?1 AND end_snapshot IS NULL",
        vec![table_id.0.into(), new_snapshot.into()],
    )
}

pub(crate) fn close_column(
    table_id: TableId,
    column_id: ColumnId,
    new_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_column} \
         SET end_snapshot = ?3 \
         WHERE table_id = ?1 AND column_id = ?2 AND end_snapshot IS NULL",
        vec![table_id.0.into(), column_id.0.into(), new_snapshot.into()],
    )
    .with_affected_count(
        "SELECT COUNT(*) FROM {ducklake_column} \
         WHERE table_id = ?1 AND column_id = ?2 AND end_snapshot IS NULL",
    )
}

pub(crate) fn close_column_subtree(
    table_id: TableId,
    column_id: ColumnId,
    latest_snapshot: SnapshotId,
    new_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "WITH RECURSIVE dropped(column_id) AS ( \
           SELECT column_id \
           FROM {ducklake_column} \
           WHERE table_id = ?1 \
             AND column_id = ?2 \
             AND begin_snapshot <= ?3 \
             AND (end_snapshot IS NULL OR ?3 < end_snapshot) \
           UNION ALL \
           SELECT c.column_id \
           FROM {ducklake_column} c \
           JOIN dropped d ON c.parent_column = d.column_id \
           WHERE c.table_id = ?1 \
             AND c.begin_snapshot <= ?3 \
             AND (c.end_snapshot IS NULL OR ?3 < c.end_snapshot) \
         ) \
         UPDATE {ducklake_column} \
         SET end_snapshot = ?4 \
         WHERE table_id = ?1 \
           AND end_snapshot IS NULL \
           AND column_id IN (SELECT column_id FROM dropped)",
        vec![
            table_id.0.into(),
            column_id.0.into(),
            latest_snapshot.0.into(),
            new_snapshot.into(),
        ],
    )
    .with_affected_count(
        "WITH RECURSIVE dropped(column_id) AS ( \
           SELECT column_id \
           FROM {ducklake_column} \
           WHERE table_id = ?1 \
             AND column_id = ?2 \
             AND begin_snapshot <= ?3 \
             AND (end_snapshot IS NULL OR ?3 < end_snapshot) \
           UNION ALL \
           SELECT c.column_id \
           FROM {ducklake_column} c \
           JOIN dropped d ON c.parent_column = d.column_id \
           WHERE c.table_id = ?1 \
             AND c.begin_snapshot <= ?3 \
             AND (c.end_snapshot IS NULL OR ?3 < c.end_snapshot) \
         ) \
         SELECT COUNT(*) \
         FROM {ducklake_column} \
         WHERE table_id = ?1 \
           AND end_snapshot IS NULL \
           AND column_id IN (SELECT column_id FROM dropped)",
    )
}

pub(crate) fn insert_column(column: &ColumnEntry, new_snapshot: i64) -> CatalogQuery {
    insert_column_values(ColumnInsertValues {
        column_id: column.column_id,
        table_id: column.table_id,
        new_snapshot,
        column_order: column.column_order,
        column_name: column.column_name.clone(),
        column_type: column.column_type.clone(),
        initial_default: column.initial_default.clone(),
        default_value: column.default_value.clone(),
        nulls_allowed: column.nulls_allowed,
        parent_column: column.parent_column,
        default_value_type: column.default_value_type.clone(),
        default_value_dialect: column.default_value_dialect.clone(),
    })
}

pub(crate) fn insert_type_tree_column(
    table_id: TableId,
    column: &PlannedColumnTypeTreeRow,
    new_snapshot: i64,
) -> CatalogQuery {
    insert_column_values(ColumnInsertValues {
        column_id: column.column_id,
        table_id,
        new_snapshot,
        column_order: column.column_order,
        column_name: column.column_name.clone(),
        column_type: column.column_type.clone(),
        initial_default: column.initial_default.clone(),
        default_value: column.default_value.clone(),
        nulls_allowed: column.nulls_allowed,
        parent_column: column.parent_column,
        default_value_type: column.default_value_type.clone(),
        default_value_dialect: column.default_value_dialect.clone(),
    })
}

pub(crate) fn insert_add_column(
    table_id: TableId,
    column: &AddColumn,
    new_snapshot: i64,
) -> CatalogQuery {
    insert_column_values(ColumnInsertValues {
        column_id: column.column_id,
        table_id,
        new_snapshot,
        column_order: column.column_order,
        column_name: column.column_name.clone(),
        column_type: column.column_type.clone(),
        initial_default: column.initial_default.clone(),
        default_value: column.default_value.clone(),
        nulls_allowed: column.nulls_allowed,
        parent_column: column.parent_column,
        default_value_type: column.default_value_type.clone(),
        default_value_dialect: column.default_value_dialect.clone(),
    })
}

pub(crate) fn insert_create_table_column(
    table_id: i64,
    column: &CreateTableColumn,
    new_snapshot: i64,
) -> CatalogQuery {
    insert_column_values(ColumnInsertValues {
        column_id: column.column_id,
        table_id: TableId(table_id),
        new_snapshot,
        column_order: column.column_order,
        column_name: column.column_name.clone(),
        column_type: column.column_type.clone(),
        initial_default: None,
        default_value: Some("NULL".into()),
        nulls_allowed: column.nulls_allowed,
        parent_column: column.parent_column,
        default_value_type: Some("literal".into()),
        default_value_dialect: Some("duckdb".into()),
    })
}

pub(crate) fn insert_data_file(
    file: &AppendFile,
    row_id_start: Option<i64>,
    begin_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_data_file} VALUES \
         (?1, ?2, ?3, NULL, ?1, ?4, ?5, 'parquet', ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
        vec![
            file.data_file_id.0.into(),
            file.table_id.0.into(),
            begin_snapshot.into(),
            file.path.clone().into(),
            file.path_is_relative.into(),
            file.record_count.into(),
            file.file_size_bytes.into(),
            CatalogParam::OptI64(file.footer_size),
            CatalogParam::OptI64(row_id_start),
            CatalogParam::OptI64(file.partition_id.map(|partition_id| partition_id.0)),
            CatalogParam::OptString(file.encryption_key.clone()),
            CatalogParam::OptI64(file.mapping_id.map(|mapping_id| mapping_id.0)),
            CatalogParam::OptI64(file.partial_max.map(|snapshot| snapshot.0)),
        ],
    )
}

pub(crate) fn insert_file_partition_value(value: &FilePartitionValue) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_file_partition_value} VALUES (?1, ?2, ?3, ?4)",
        vec![
            value.data_file_id.0.into(),
            value.table_id.0.into(),
            value.partition_key_index.into(),
            CatalogParam::OptString(value.partition_value.clone()),
        ],
    )
}

pub(crate) fn insert_file_column_stats(stats: &FileColumnStats) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_file_column_stats} VALUES \
         (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
        vec![
            stats.data_file_id.0.into(),
            stats.table_id.0.into(),
            stats.column_id.0.into(),
            CatalogParam::OptI64(stats.column_size_bytes),
            CatalogParam::OptI64(stats.value_count),
            CatalogParam::OptI64(stats.null_count),
            CatalogParam::OptString(stats.min_value.clone()),
            CatalogParam::OptString(stats.max_value.clone()),
            CatalogParam::OptBool(stats.contains_nan),
            CatalogParam::OptString(stats.extra_stats.clone()),
        ],
    )
}

pub(crate) fn insert_file_variant_stats(stats: &FileVariantStats) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_file_variant_stats} VALUES \
         (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
        vec![
            stats.data_file_id.0.into(),
            stats.table_id.0.into(),
            stats.column_id.0.into(),
            stats.variant_path.clone().into(),
            stats.shredded_type.clone().into(),
            CatalogParam::OptI64(stats.column_size_bytes),
            CatalogParam::OptI64(stats.value_count),
            CatalogParam::OptI64(stats.null_count),
            CatalogParam::OptString(stats.min_value.clone()),
            CatalogParam::OptString(stats.max_value.clone()),
            CatalogParam::OptBool(stats.contains_nan),
            CatalogParam::OptString(stats.extra_stats.clone()),
        ],
    )
}

pub(crate) fn insert_delete_file(
    file: &DeleteFileCommit,
    begin_snapshot: i64,
    format: &str,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_delete_file} VALUES \
         (?1, ?2, ?3, NULL, ?4, ?5, ?6, ?7, ?8, ?9, ?10, NULL, ?11)",
        vec![
            file.delete_file_id.0.into(),
            file.table_id.0.into(),
            begin_snapshot.into(),
            file.data_file_id.0.into(),
            file.path.clone().into(),
            file.path_is_relative.into(),
            format.into(),
            file.delete_count.into(),
            file.file_size_bytes.into(),
            CatalogParam::OptI64(file.footer_size),
            CatalogParam::OptI64(file.partial_max.map(|snapshot| snapshot.0)),
        ],
    )
}

pub(crate) fn close_overwritten_delete_file(
    file: &DeleteFileCommit,
    new_snapshot: i64,
) -> Option<CatalogQuery> {
    let overwritten = file.overwritten_delete_file_id?;
    Some(
        CatalogQuery::new(
            "UPDATE {ducklake_delete_file} \
             SET end_snapshot = ?1 \
             WHERE delete_file_id = ?2 AND table_id = ?3 AND end_snapshot IS NULL",
            vec![
                new_snapshot.into(),
                overwritten.0.into(),
                file.table_id.0.into(),
            ],
        )
        .with_affected_count(
            "SELECT COUNT(*) FROM {ducklake_delete_file} \
             WHERE delete_file_id = ?2 AND table_id = ?3 AND end_snapshot IS NULL",
        ),
    )
}

pub(crate) fn delete_overwritten_delete_file(file: &DeleteFileCommit) -> Option<CatalogQuery> {
    let overwritten = file.overwritten_delete_file_id?;
    Some(
        CatalogQuery::new(
            "DELETE FROM {ducklake_delete_file} WHERE delete_file_id = ?1 AND table_id = ?2",
            vec![overwritten.0.into(), file.table_id.0.into()],
        )
        .with_affected_count(
            "SELECT COUNT(*) FROM {ducklake_delete_file} \
             WHERE delete_file_id = ?1 AND table_id = ?2",
        ),
    )
}

pub(crate) fn overwritten_delete_file_cleanup_row(
    overwritten: ducklake_catalog::DeleteFileId,
    table_id: TableId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT f.delete_file_id, f.path, f.path_is_relative, \
                s.path, s.path_is_relative, t.path, t.path_is_relative \
         FROM {ducklake_delete_file} f \
         JOIN {ducklake_table} t USING(table_id) \
         JOIN {ducklake_schema} s USING(schema_id) \
         WHERE f.delete_file_id = ?1 AND f.table_id = ?2",
        vec![overwritten.0.into(), table_id.0.into()],
    )
}

pub(crate) fn insert_table_stats_for_create_table(
    table_id: TableId,
    record_count: i64,
    file_size_bytes: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_table_stats} VALUES (?1, ?2, ?2, ?3)",
        vec![
            table_id.0.into(),
            record_count.into(),
            file_size_bytes.into(),
        ],
    )
}

struct ColumnInsertValues {
    column_id: ColumnId,
    table_id: TableId,
    new_snapshot: i64,
    column_order: i64,
    column_name: String,
    column_type: String,
    initial_default: Option<String>,
    default_value: Option<String>,
    nulls_allowed: bool,
    parent_column: Option<ColumnId>,
    default_value_type: Option<String>,
    default_value_dialect: Option<String>,
}

pub(crate) fn insert_schema(
    schema_id: i64,
    schema_uuid: String,
    new_snapshot: i64,
    schema_name: &str,
    uuid_style: UuidParamStyle,
) -> CatalogQuery {
    CatalogQuery::new(
        format!(
            "INSERT INTO {{ducklake_schema}} VALUES (?1, {}, ?3, NULL, ?4, ?5, TRUE)",
            uuid_param("?2", uuid_style)
        ),
        vec![
            schema_id.into(),
            schema_uuid.into(),
            new_snapshot.into(),
            schema_name.into(),
            format!("{schema_name}/").into(),
        ],
    )
}

pub(crate) fn insert_table(
    table_id: i64,
    table_uuid: &str,
    new_snapshot: i64,
    schema_id: SchemaId,
    table_name: &str,
    path: Option<&str>,
    path_is_relative: bool,
    uuid_style: UuidParamStyle,
) -> CatalogQuery {
    CatalogQuery::new(
        format!(
            "INSERT INTO {{ducklake_table}} VALUES (?1, {}, ?3, NULL, ?4, ?5, ?6, ?7)",
            uuid_param("?2", uuid_style)
        ),
        vec![
            table_id.into(),
            table_uuid.into(),
            new_snapshot.into(),
            schema_id.0.into(),
            table_name.into(),
            CatalogParam::OptString(path.map(str::to_string)),
            path_is_relative.into(),
        ],
    )
}

pub(crate) fn insert_view(
    view_id: i64,
    view_uuid: String,
    new_snapshot: i64,
    schema_id: SchemaId,
    create: &crate::CreateViewCommit,
    uuid_style: UuidParamStyle,
) -> CatalogQuery {
    CatalogQuery::new(
        format!(
            "INSERT INTO {{ducklake_view}} VALUES (?1, {}, ?3, NULL, ?4, ?5, ?6, ?7, ?8)",
            uuid_param("?2", uuid_style)
        ),
        vec![
            view_id.into(),
            view_uuid.into(),
            new_snapshot.into(),
            schema_id.0.into(),
            create.view_name.as_str().into(),
            create.dialect.as_str().into(),
            create.sql.as_str().into(),
            create.column_aliases.as_deref().unwrap_or("").into(),
        ],
    )
}

fn insert_column_values(column: ColumnInsertValues) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_column} VALUES \
         (?1, ?2, NULL, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
        vec![
            column.column_id.0.into(),
            column.new_snapshot.into(),
            column.table_id.0.into(),
            column.column_order.into(),
            column.column_name.into(),
            column.column_type.into(),
            CatalogParam::OptString(column.initial_default),
            CatalogParam::OptString(column.default_value),
            column.nulls_allowed.into(),
            CatalogParam::OptI64(column.parent_column.map(|column_id| column_id.0)),
            CatalogParam::OptString(column.default_value_type),
            CatalogParam::OptString(column.default_value_dialect),
        ],
    )
}

pub(crate) fn column_types_at(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT column_id, column_type \
         FROM {ducklake_column} \
         WHERE table_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot)",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn table_column_stats(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT table_id, column_id, contains_null, contains_nan, min_value, max_value, extra_stats \
         FROM {ducklake_table_column_stats} WHERE table_id = ?1",
        vec![table_id.0.into()],
    )
}

pub(crate) fn table_column_stats_count(table_id: TableId, column_id: ColumnId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT COUNT(*) FROM {ducklake_table_column_stats} \
         WHERE table_id = ?1 AND column_id = ?2",
        vec![table_id.0.into(), column_id.0.into()],
    )
}

pub(crate) fn table_stats_present(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT 1 FROM {ducklake_table_stats} WHERE table_id = ?1",
        vec![table_id.0.into()],
    )
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ActiveTableStats {
    pub(crate) record_count: i64,
    pub(crate) next_row_id: i64,
    pub(crate) file_size_bytes: i64,
}

pub(crate) fn active_table_stats_from_data_files(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT \
             COALESCE(SUM(record_count), 0), \
             COALESCE(MAX(row_id_start + record_count), 0), \
             COALESCE(SUM(file_size_bytes), 0) \
         FROM {ducklake_data_file} \
         WHERE table_id = ?1 AND end_snapshot IS NULL",
        vec![table_id.0.into()],
    )
}

pub(crate) fn active_table_stats_value(row: &dyn EmbeddedRow) -> Result<ActiveTableStats> {
    Ok(ActiveTableStats {
        record_count: row.get_i64(0)?,
        next_row_id: row.get_i64(1)?,
        file_size_bytes: row.get_i64(2)?,
    })
}

pub(crate) fn insert_table_stats(
    table_id: TableId,
    record_count: i64,
    next_row_id: i64,
    file_size_bytes: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_table_stats} VALUES (?1, ?2, ?3, ?4)",
        vec![
            table_id.0.into(),
            record_count.into(),
            next_row_id.into(),
            file_size_bytes.into(),
        ],
    )
}

pub(crate) fn update_table_stats_for_append(
    table_id: TableId,
    record_count: i64,
    allocated_row_id_count: i64,
    file_size_bytes: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table_stats} \
         SET record_count = COALESCE(record_count, 0) + ?2, \
             next_row_id = COALESCE(next_row_id, 0) + ?3, \
             file_size_bytes = COALESCE(file_size_bytes, 0) + ?4 \
         WHERE table_id = ?1",
        vec![
            table_id.0.into(),
            record_count.into(),
            allocated_row_id_count.into(),
            file_size_bytes.into(),
        ],
    )
    .with_affected_count("SELECT COUNT(*) FROM {ducklake_table_stats} WHERE table_id = ?1")
}

pub(crate) fn update_table_stats_for_inlined_append(
    table_id: TableId,
    rows_appended: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table_stats} \
         SET record_count = COALESCE(record_count, 0) + ?2, \
             next_row_id = COALESCE(next_row_id, 0) + ?2 \
         WHERE table_id = ?1",
        vec![table_id.0.into(), rows_appended.into()],
    )
    .with_affected_count("SELECT COUNT(*) FROM {ducklake_table_stats} WHERE table_id = ?1")
}

pub(crate) fn insert_table_stats_for_inlined_append(
    table_id: TableId,
    rows_appended: i64,
    next_row_id: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_table_stats} VALUES (?1, ?2, ?3, 0)",
        vec![table_id.0.into(), rows_appended.into(), next_row_id.into()],
    )
}

pub(crate) fn update_table_stats_for_delete(table_id: TableId, deleted_rows: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table_stats} \
         SET record_count = CASE \
             WHEN COALESCE(record_count, 0) - ?2 > 0 THEN COALESCE(record_count, 0) - ?2 \
             ELSE 0 \
         END \
         WHERE table_id = ?1",
        vec![table_id.0.into(), deleted_rows.into()],
    )
}

pub(crate) fn update_table_stats_for_table_change(
    table_id: TableId,
    appended_rows: i64,
    deleted_rows: i64,
    appended_file_size_bytes: i64,
    allocated_row_id_count: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table_stats} \
         SET record_count = CASE \
                 WHEN COALESCE(record_count, 0) + ?2 - ?3 > 0 \
                 THEN COALESCE(record_count, 0) + ?2 - ?3 \
                 ELSE 0 \
             END, \
             next_row_id = COALESCE(next_row_id, 0) + ?5, \
             file_size_bytes = COALESCE(file_size_bytes, 0) + ?4 \
         WHERE table_id = ?1",
        vec![
            table_id.0.into(),
            appended_rows.into(),
            deleted_rows.into(),
            appended_file_size_bytes.into(),
            allocated_row_id_count.into(),
        ],
    )
    .with_affected_count("SELECT COUNT(*) FROM {ducklake_table_stats} WHERE table_id = ?1")
}

pub(crate) fn reset_table_stats_for_truncate(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table_stats} \
         SET record_count = 0, file_size_bytes = 0 \
         WHERE table_id = ?1",
        vec![table_id.0.into()],
    )
}

pub(crate) fn update_table_stats_for_maintenance(
    table_id: TableId,
    source_file_size_bytes: i64,
    output_file_size_bytes: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table_stats} \
         SET file_size_bytes = CASE \
             WHEN COALESCE(file_size_bytes, 0) - ?2 + ?3 > 0 \
             THEN COALESCE(file_size_bytes, 0) - ?2 + ?3 \
             ELSE 0 \
         END \
         WHERE table_id = ?1",
        vec![
            table_id.0.into(),
            source_file_size_bytes.into(),
            output_file_size_bytes.into(),
        ],
    )
}

pub(crate) fn end_active_table_data_files(table_id: TableId, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_data_file} \
         SET end_snapshot = ?1 \
         WHERE table_id = ?2 AND end_snapshot IS NULL",
        vec![new_snapshot.into(), table_id.0.into()],
    )
}

pub(crate) fn end_active_table_delete_files(table_id: TableId, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_delete_file} \
         SET end_snapshot = ?1 \
         WHERE table_id = ?2 AND end_snapshot IS NULL",
        vec![new_snapshot.into(), table_id.0.into()],
    )
}

pub(crate) fn end_active_data_files_by_id(
    table_id: TableId,
    source_ids: &[i64],
    new_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        format!(
            "UPDATE {{ducklake_data_file}} \
             SET end_snapshot = ?1 \
             WHERE table_id = ?2 \
               AND data_file_id IN ({}) \
               AND end_snapshot IS NULL",
            join_i64s(source_ids)
        ),
        vec![new_snapshot.into(), table_id.0.into()],
    )
}

pub(crate) fn end_active_delete_files_by_data_file_id(
    table_id: TableId,
    source_ids: &[i64],
    new_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        format!(
            "UPDATE {{ducklake_delete_file}} \
             SET end_snapshot = ?1 \
             WHERE table_id = ?2 \
               AND data_file_id IN ({}) \
               AND end_snapshot IS NULL",
            join_i64s(source_ids)
        ),
        vec![new_snapshot.into(), table_id.0.into()],
    )
}

pub(crate) fn table_column_contains_null(table_id: TableId, column_id: ColumnId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT contains_null \
         FROM {ducklake_table_column_stats} \
         WHERE table_id = ?1 AND column_id = ?2",
        vec![table_id.0.into(), column_id.0.into()],
    )
}

pub(crate) fn delete_table_column_stats(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "DELETE FROM {ducklake_table_column_stats} WHERE table_id = ?1",
        vec![table_id.0.into()],
    )
}

pub(crate) fn active_data_file_count(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT COUNT(*) \
         FROM {ducklake_data_file} \
         WHERE table_id = ?1 AND end_snapshot IS NULL",
        vec![table_id.0.into()],
    )
}

pub(crate) fn active_file_column_stats(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT fcs.data_file_id, fcs.table_id, fcs.column_id, fcs.column_size_bytes, \
                fcs.value_count, fcs.null_count, fcs.min_value, fcs.max_value, \
                fcs.contains_nan, fcs.extra_stats \
         FROM {ducklake_file_column_stats} fcs \
         JOIN {ducklake_data_file} df \
           ON df.data_file_id = fcs.data_file_id AND df.table_id = fcs.table_id \
         WHERE fcs.table_id = ?1 AND df.end_snapshot IS NULL",
        vec![table_id.0.into()],
    )
}

pub(crate) fn active_file_variant_stats(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT fvs.data_file_id, fvs.table_id, fvs.column_id, fvs.variant_path, \
                fvs.shredded_type, fvs.column_size_bytes, fvs.value_count, \
                fvs.null_count, fvs.min_value, fvs.max_value, fvs.contains_nan, \
                fvs.extra_stats \
         FROM {ducklake_file_variant_stats} fvs \
         JOIN {ducklake_data_file} df \
           ON df.data_file_id = fvs.data_file_id AND df.table_id = fvs.table_id \
         WHERE fvs.table_id = ?1 AND df.end_snapshot IS NULL",
        vec![table_id.0.into()],
    )
}

pub(crate) fn update_table_column_stats(stats: &TableColumnStats) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_table_column_stats} \
         SET contains_null = ?3, contains_nan = ?4, min_value = ?5, max_value = ?6, extra_stats = ?7 \
         WHERE table_id = ?1 AND column_id = ?2",
        table_column_stats_params(stats),
    )
}

pub(crate) fn insert_table_column_stats(stats: &TableColumnStats) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_table_column_stats} VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
        table_column_stats_params(stats),
    )
}

fn table_column_stats_params(stats: &TableColumnStats) -> Vec<CatalogParam> {
    vec![
        stats.table_id.0.into(),
        stats.column_id.0.into(),
        CatalogParam::OptBool(stats.contains_null),
        CatalogParam::OptBool(stats.contains_nan),
        CatalogParam::OptString(stats.min_value.clone()),
        CatalogParam::OptString(stats.max_value.clone()),
        CatalogParam::OptString(stats.extra_stats.clone()),
    ]
}

pub(crate) fn update_schema_versions_begin_after(
    new_begin_snapshot: i64,
    after_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_schema_versions} \
         SET begin_snapshot = ?1 \
         WHERE begin_snapshot > ?2",
        vec![new_begin_snapshot.into(), after_snapshot.into()],
    )
}

pub(crate) fn delete_scheduled_cleanup_files_from_next_file_id(next_file_id: i64) -> CatalogQuery {
    CatalogQuery::new(
        "DELETE FROM {ducklake_files_scheduled_for_deletion} WHERE data_file_id >= ?1",
        vec![next_file_id.into()],
    )
}

pub(crate) fn inlined_data_table_names(table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT table_name FROM {ducklake_inlined_data_tables} WHERE table_id = ?1",
        vec![table_id.0.into()],
    )
}

pub(crate) fn inlined_data_table_name(table_id: TableId, schema_version: i64) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT table_name \
         FROM {ducklake_inlined_data_tables} \
         WHERE table_id = ?1 AND schema_version = ?2",
        vec![table_id.0.into(), schema_version.into()],
    )
}

pub(crate) fn insert_inlined_data_table_ref(
    table_id: TableId,
    table_name: &str,
    schema_version: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_inlined_data_tables} VALUES (?1, ?2, ?3)",
        vec![
            table_id.0.into(),
            table_name.to_string().into(),
            schema_version.into(),
        ],
    )
}

impl From<i64> for CatalogParam {
    fn from(value: i64) -> Self {
        Self::I64(value)
    }
}

impl From<String> for CatalogParam {
    fn from(value: String) -> Self {
        Self::String(value)
    }
}

impl From<&str> for CatalogParam {
    fn from(value: &str) -> Self {
        Self::String(value.to_string())
    }
}

impl From<bool> for CatalogParam {
    fn from(value: bool) -> Self {
        Self::Bool(value)
    }
}

#[derive(Debug, Clone)]
pub(crate) struct CatalogQuery {
    sql: Cow<'static, str>,
    params: Vec<CatalogParam>,
    affected_count_sql: Option<Cow<'static, str>>,
}

impl CatalogQuery {
    fn new(sql: impl Into<Cow<'static, str>>, params: Vec<CatalogParam>) -> Self {
        Self {
            sql: sql.into(),
            params,
            affected_count_sql: None,
        }
    }

    fn with_affected_count(mut self, sql: impl Into<Cow<'static, str>>) -> Self {
        self.affected_count_sql = Some(sql.into());
        self
    }

    pub(crate) fn params(&self) -> &[CatalogParam] {
        &self.params
    }

    pub(crate) fn render_embedded(&self) -> String {
        render_table_refs(self.sql.as_ref(), TableRefStyle::Unqualified)
    }

    pub(crate) fn render_postgres(&self, metadata_schema: &str) -> String {
        let sql = render_table_refs(
            self.sql.as_ref(),
            TableRefStyle::PostgresQualified(metadata_schema),
        );
        render_numbered_placeholders(&sql, "$")
    }

    pub(crate) fn render_quack(&self) -> String {
        self.render_quack_with_metadata_catalog(None)
    }

    pub(crate) fn render_quack_with_metadata_catalog(
        &self,
        metadata_catalog: Option<&str>,
    ) -> String {
        render_quack_query_sql(self.sql.as_ref(), metadata_catalog, &self.params)
    }

    pub(crate) fn render_quack_affected_count_with_metadata_catalog(
        &self,
        metadata_catalog: Option<&str>,
    ) -> Option<String> {
        self.affected_count_sql
            .as_deref()
            .map(|sql| render_quack_query_sql(sql, metadata_catalog, &self.params))
    }
}

fn render_quack_query_sql(
    sql: &str,
    metadata_catalog: Option<&str>,
    params: &[CatalogParam],
) -> String {
    let style = metadata_catalog
        .map(TableRefStyle::Qualified)
        .unwrap_or(TableRefStyle::Unqualified);
    let sql = render_table_refs(sql, style);
    render_params_as_literals(&sql, params)
}

fn uuid_param(placeholder: &str, style: UuidParamStyle) -> String {
    match style {
        UuidParamStyle::Text => placeholder.to_string(),
        UuidParamStyle::CastAsUuid => format!("CAST({placeholder} AS UUID)"),
        UuidParamStyle::CastTextAsUuid => format!("CAST(CAST({placeholder} AS TEXT) AS UUID)"),
    }
}

enum TableRefStyle<'a> {
    Unqualified,
    Qualified(&'a str),
    PostgresQualified(&'a str),
}

pub(crate) fn table_id_exists(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT COUNT(*) FROM {ducklake_table} \
         WHERE table_id = ?1 AND begin_snapshot <= ?2 AND (end_snapshot IS NULL OR ?2 < end_snapshot)",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_column(
    table_id: TableId,
    column_id: ColumnId,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT column_id, begin_snapshot, end_snapshot, table_id, column_order, column_name, \
                column_type, initial_default, default_value, nulls_allowed, parent_column, \
                default_value_type, default_value_dialect \
         FROM {ducklake_column} \
         WHERE table_id = ?1 \
           AND column_id = ?2 \
           AND begin_snapshot <= ?3 \
           AND (end_snapshot IS NULL OR ?3 < end_snapshot)",
        vec![table_id.0.into(), column_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_columns(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT column_id, begin_snapshot, end_snapshot, table_id, column_order, column_name, \
                column_type, initial_default, default_value, nulls_allowed, parent_column, \
                default_value_type, default_value_dialect \
         FROM {ducklake_column} \
         WHERE table_id = ?1 \
           AND begin_snapshot <= ?2 \
           AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
         ORDER BY parent_column, column_order",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn active_column_refs(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT column_id, column_name, parent_column \
         FROM {ducklake_column} \
         WHERE table_id = ?1 AND begin_snapshot <= ?2 AND (end_snapshot IS NULL OR ?2 < end_snapshot)",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn conflicting_sibling_column_name(
    table_id: TableId,
    column_id: ColumnId,
    parent_column: Option<ColumnId>,
    new_column_name: &str,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT column_name FROM {ducklake_column} \
         WHERE table_id = ?1 \
           AND ((CAST(?5 AS BIGINT) IS NULL AND parent_column IS NULL) OR parent_column = ?5) \
           AND column_id <> ?2 \
           AND lower(column_name) = lower(?3) \
           AND begin_snapshot <= ?4 \
           AND (end_snapshot IS NULL OR ?4 < end_snapshot) \
         ORDER BY column_id LIMIT 1",
        vec![
            table_id.0.into(),
            column_id.0.into(),
            new_column_name.into(),
            snapshot.0.into(),
            CatalogParam::OptI64(parent_column.map(|column_id| column_id.0)),
        ],
    )
}

pub(crate) fn top_level_column_count(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT COUNT(*) \
         FROM {ducklake_column} \
         WHERE table_id = ?1 \
           AND parent_column IS NULL \
           AND begin_snapshot <= ?2 \
           AND (end_snapshot IS NULL OR ?2 < end_snapshot)",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn active_partition_reference_count(
    table_id: TableId,
    column_id: ColumnId,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT COUNT(*) \
         FROM {ducklake_partition_info} pi \
         JOIN {ducklake_partition_column} pc USING(partition_id) \
         WHERE pi.table_id = ?1 \
           AND pc.column_id = ?2 \
           AND pi.begin_snapshot <= ?3 \
           AND (pi.end_snapshot IS NULL OR ?3 < pi.end_snapshot)",
        vec![table_id.0.into(), column_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn active_sort_expressions(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT se.expression \
         FROM {ducklake_sort_info} si \
         JOIN {ducklake_sort_expression} se USING(sort_id) \
         WHERE si.table_id = ?1 \
           AND si.begin_snapshot <= ?2 \
           AND (si.end_snapshot IS NULL OR ?2 < si.end_snapshot) \
         ORDER BY si.sort_id, se.sort_key_index",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn active_partition_infos(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT partition_id, table_id, begin_snapshot, end_snapshot \
         FROM {ducklake_partition_info} \
         WHERE table_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
         ORDER BY partition_id",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn created_partition_infos_after(snapshot_id: i64) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT partition_id, table_id \
         FROM {ducklake_partition_info} \
         WHERE begin_snapshot > ?1 \
         ORDER BY begin_snapshot, partition_id",
        vec![snapshot_id.into()],
    )
}

pub(crate) fn partition_columns(partition_id: PartitionId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT partition_id, table_id, partition_key_index, column_id, transform \
         FROM {ducklake_partition_column} \
         WHERE partition_id = ?1 ORDER BY partition_key_index",
        vec![partition_id.0.into()],
    )
}

pub(crate) fn delete_partition_columns(partition_id: PartitionId) -> CatalogQuery {
    CatalogQuery::new(
        "DELETE FROM {ducklake_partition_column} WHERE partition_id = ?1",
        vec![partition_id.0.into()],
    )
}

pub(crate) fn delete_partition_info(partition_id: PartitionId) -> CatalogQuery {
    CatalogQuery::new(
        "DELETE FROM {ducklake_partition_info} WHERE partition_id = ?1",
        vec![partition_id.0.into()],
    )
}

pub(crate) fn reopen_partition_info(partition_id: PartitionId) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_partition_info} \
         SET end_snapshot = NULL \
         WHERE partition_id = ?1",
        vec![partition_id.0.into()],
    )
}

pub(crate) fn set_partition_info_end_snapshot(
    partition_id: PartitionId,
    end_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_partition_info} \
         SET end_snapshot = ?2 \
         WHERE partition_id = ?1",
        vec![partition_id.0.into(), end_snapshot.into()],
    )
}

pub(crate) fn close_partition(partition_id: PartitionId, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_partition_info} \
         SET end_snapshot = ?2 \
         WHERE partition_id = ?1 AND end_snapshot IS NULL",
        vec![partition_id.0.into(), new_snapshot.into()],
    )
    .with_affected_count(
        "SELECT COUNT(*) FROM {ducklake_partition_info} \
         WHERE partition_id = ?1 AND end_snapshot IS NULL",
    )
}

pub(crate) fn insert_partition_info(
    partition_id: PartitionId,
    table_id: TableId,
    new_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_partition_info} VALUES (?1, ?2, ?3, NULL)",
        vec![
            partition_id.0.into(),
            table_id.0.into(),
            new_snapshot.into(),
        ],
    )
}

pub(crate) fn insert_partition_column(
    partition_id: PartitionId,
    table_id: TableId,
    partition_key_index: i64,
    column: &PartitionColumnCommit,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_partition_column} VALUES (?1, ?2, ?3, ?4, ?5)",
        vec![
            partition_id.0.into(),
            table_id.0.into(),
            partition_key_index.into(),
            column.column_id.0.into(),
            partition_transform_string(&column.transform).into(),
        ],
    )
}

pub(crate) fn active_sort_infos(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT sort_id, table_id, begin_snapshot, end_snapshot \
         FROM {ducklake_sort_info} \
         WHERE table_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
         ORDER BY sort_id",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn sort_expressions(sort_id: SortId, table_id: TableId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT sort_id, table_id, sort_key_index, expression, dialect, sort_direction, null_order \
         FROM {ducklake_sort_expression} \
         WHERE sort_id = ?1 AND table_id = ?2 \
         ORDER BY sort_key_index",
        vec![sort_id.0.into(), table_id.0.into()],
    )
}

pub(crate) fn close_sort(sort_id: SortId, table_id: TableId, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_sort_info} \
         SET end_snapshot = ?2 \
         WHERE sort_id = ?1 AND table_id = ?3 AND end_snapshot IS NULL",
        vec![sort_id.0.into(), new_snapshot.into(), table_id.0.into()],
    )
    .with_affected_count(
        "SELECT COUNT(*) FROM {ducklake_sort_info} \
         WHERE sort_id = ?1 AND table_id = ?3 AND end_snapshot IS NULL",
    )
}

pub(crate) fn insert_sort_info(
    sort_id: SortId,
    table_id: TableId,
    new_snapshot: i64,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_sort_info} VALUES (?1, ?2, ?3, NULL)",
        vec![sort_id.0.into(), table_id.0.into(), new_snapshot.into()],
    )
}

pub(crate) fn insert_sort_expression(
    sort_id: SortId,
    table_id: TableId,
    sort_key_index: i64,
    expression: &SortExpressionCommit,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_sort_expression} VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
        vec![
            sort_id.0.into(),
            table_id.0.into(),
            sort_key_index.into(),
            expression.expression.as_str().into(),
            expression.dialect.as_str().into(),
            expression.sort_direction.as_str().into(),
            expression.null_order.as_str().into(),
        ],
    )
}

pub(crate) fn insert_sort_expression_row(expression: &SortExpression) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_sort_expression} VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
        vec![
            expression.sort_id.0.into(),
            expression.table_id.0.into(),
            expression.sort_key_index.into(),
            expression.expression.as_str().into(),
            CatalogParam::OptString(expression.dialect.clone()),
            CatalogParam::OptString(expression.sort_direction.clone()),
            CatalogParam::OptString(expression.null_order.clone()),
        ],
    )
}

pub(crate) fn visible_top_level_column_names(
    table_id: TableId,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT column_name FROM {ducklake_column} \
         WHERE table_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
         AND parent_column IS NULL",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_column_parent_refs(table_id: TableId, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT column_id, parent_column \
         FROM {ducklake_column} \
         WHERE table_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot)",
        vec![table_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_table_ids_for_schema(
    schema_id: SchemaId,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT table_id \
         FROM {ducklake_table} \
         WHERE schema_id = ?1 \
           AND begin_snapshot <= ?2 \
           AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
         ORDER BY table_id",
        vec![schema_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn close_comment_tag(object_id: i64, new_snapshot: i64) -> CatalogQuery {
    CatalogQuery::new(
        "UPDATE {ducklake_tag} \
         SET end_snapshot = ?1 \
         WHERE object_id = ?2 AND key = 'comment' AND end_snapshot IS NULL",
        vec![new_snapshot.into(), object_id.into()],
    )
}

pub(crate) fn insert_comment_tag(
    object_id: i64,
    new_snapshot: i64,
    comment: Option<String>,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_tag} VALUES (?1, ?2, NULL, 'comment', ?3)",
        vec![
            object_id.into(),
            new_snapshot.into(),
            CatalogParam::OptString(comment),
        ],
    )
}

pub(crate) fn snapshot_changes_after(snapshot_id: i64) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT changes_made \
         FROM {ducklake_snapshot_changes} \
         WHERE snapshot_id > ?1 ORDER BY snapshot_id",
        vec![snapshot_id.into()],
    )
}

pub(crate) fn delete_snapshot_changes_after(snapshot_id: i64) -> CatalogQuery {
    CatalogQuery::new(
        "DELETE FROM {ducklake_snapshot_changes} WHERE snapshot_id > ?1",
        vec![snapshot_id.into()],
    )
}

pub(crate) fn insert_squashed_snapshot_change(
    snapshot_id: i64,
    changes_made: &str,
) -> CatalogQuery {
    CatalogQuery::new(
        "INSERT INTO {ducklake_snapshot_changes} VALUES (?1, ?2, NULL, NULL, NULL)",
        vec![snapshot_id.into(), changes_made.into()],
    )
}

pub(crate) fn visible_schema_exact(schema_name: &str, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT schema_id, {schema_uuid_as_text}, begin_snapshot, end_snapshot, schema_name, path, path_is_relative \
         FROM {ducklake_schema} \
         WHERE schema_name = ?1 \
           AND begin_snapshot <= ?2 \
           AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
         ORDER BY schema_id",
        vec![schema_name.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_schema_case(schema_name: &str, snapshot: SnapshotId) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT schema_id, {schema_uuid_as_text}, begin_snapshot, end_snapshot, schema_name, path, path_is_relative \
         FROM {ducklake_schema} \
         WHERE lower(schema_name) = lower(?1) \
           AND begin_snapshot <= ?2 \
           AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
         ORDER BY schema_id",
        vec![schema_name.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_table_exact(
    schema_id: SchemaId,
    table_name: &str,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT table_id, {table_uuid_as_text}, begin_snapshot, end_snapshot, schema_id, table_name, path, path_is_relative \
         FROM {ducklake_table} \
         WHERE schema_id = ?1 \
           AND table_name = ?2 \
           AND begin_snapshot <= ?3 \
           AND (end_snapshot IS NULL OR ?3 < end_snapshot) \
         ORDER BY table_id",
        vec![schema_id.0.into(), table_name.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_tables_case(
    schema_id: SchemaId,
    table_name: &str,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT table_id, {table_uuid_as_text}, begin_snapshot, end_snapshot, schema_id, table_name, path, path_is_relative \
         FROM {ducklake_table} \
         WHERE schema_id = ?1 \
           AND lower(table_name) = lower(?2) \
           AND begin_snapshot <= ?3 \
           AND (end_snapshot IS NULL OR ?3 < end_snapshot) \
         ORDER BY table_id",
        vec![schema_id.0.into(), table_name.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_view_ids_for_schema(
    schema_id: SchemaId,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT view_id \
         FROM {ducklake_view} \
         WHERE schema_id = ?1 \
           AND begin_snapshot <= ?2 \
           AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
         ORDER BY view_id",
        vec![schema_id.0.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_view_exact(
    schema_id: SchemaId,
    view_name: &str,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT view_id, {view_uuid_as_text}, begin_snapshot, end_snapshot, schema_id, view_name, dialect, sql, column_aliases \
         FROM {ducklake_view} \
         WHERE schema_id = ?1 \
           AND view_name = ?2 \
           AND begin_snapshot <= ?3 \
           AND (end_snapshot IS NULL OR ?3 < end_snapshot) \
         ORDER BY view_id",
        vec![schema_id.0.into(), view_name.into(), snapshot.0.into()],
    )
}

pub(crate) fn visible_views_case(
    schema_id: SchemaId,
    view_name: &str,
    snapshot: SnapshotId,
) -> CatalogQuery {
    CatalogQuery::new(
        "SELECT view_id, {view_uuid_as_text}, begin_snapshot, end_snapshot, schema_id, view_name, dialect, sql, column_aliases \
         FROM {ducklake_view} \
         WHERE schema_id = ?1 \
           AND lower(view_name) = lower(?2) \
           AND begin_snapshot <= ?3 \
           AND (end_snapshot IS NULL OR ?3 < end_snapshot) \
         ORDER BY view_id",
        vec![schema_id.0.into(), view_name.into(), snapshot.0.into()],
    )
}

pub(crate) fn resolve_visible_schema(
    schema_name: &str,
    exact: Vec<SchemaEntry>,
    case_matches: Vec<SchemaEntry>,
) -> Result<Option<SchemaEntry>> {
    match exact.len() {
        1 => return Ok(exact.into_iter().next()),
        count if count > 1 => {
            return Err(StorageError::Backend(format!(
                "DuckLake visible schema lookup returned {count} exact rows for {schema_name}"
            )));
        }
        _ => {}
    }

    match case_matches.len() {
        0 => Ok(None),
        1 => Ok(case_matches.into_iter().next()),
        count => Err(StorageError::Backend(format!(
            "ambiguous DuckLake schema identifier {schema_name:?}: {count} visible schemas differ only by ASCII case"
        ))),
    }
}

pub(crate) fn resolve_visible_table(
    table_name: &str,
    exact: Vec<TableEntry>,
    case_matches: Vec<TableEntry>,
) -> Result<Option<TableEntry>> {
    match exact.len() {
        1 => return Ok(exact.into_iter().next()),
        count if count > 1 => {
            return Err(StorageError::Backend(format!(
                "DuckLake visible table lookup returned {count} exact rows for table {table_name}"
            )));
        }
        _ => {}
    }

    match case_matches.len() {
        0 => Ok(None),
        1 => Ok(case_matches.into_iter().next()),
        count => Err(StorageError::Backend(format!(
            "ambiguous DuckLake table identifier {table_name:?}: {count} visible tables differ only by ASCII case"
        ))),
    }
}

pub(crate) fn resolve_visible_view(
    view_name: &str,
    exact: Vec<ViewEntry>,
    case_matches: Vec<ViewEntry>,
) -> Result<Option<ViewEntry>> {
    match exact.len() {
        1 => return Ok(exact.into_iter().next()),
        count if count > 1 => {
            return Err(StorageError::Backend(format!(
                "DuckLake visible view lookup returned {count} exact rows for view {view_name}"
            )));
        }
        _ => {}
    }

    match case_matches.len() {
        0 => Ok(None),
        1 => Ok(case_matches.into_iter().next()),
        count => Err(StorageError::Backend(format!(
            "ambiguous DuckLake view identifier {view_name:?}: {count} visible views differ only by ASCII case"
        ))),
    }
}

pub(crate) fn i64_value(row: &dyn EmbeddedRow) -> Result<i64> {
    row.get_i64(0)
}

pub(crate) fn opt_i64_value(row: &dyn EmbeddedRow) -> Result<Option<i64>> {
    row.get_opt_i64(0)
}

pub(crate) fn opt_bool_value(row: &dyn EmbeddedRow) -> Result<Option<bool>> {
    row.get_opt_bool(0)
}

pub(crate) fn string_value(row: &dyn EmbeddedRow) -> Result<String> {
    row.get_string(0)
}

pub(crate) fn opt_string_value(row: &dyn EmbeddedRow) -> Result<Option<String>> {
    row.get_opt_string(0)
}

pub(crate) fn latest_snapshot_ids_value(
    row: &dyn EmbeddedRow,
) -> Result<crate::commit::CommitSnapshotIds> {
    Ok(crate::commit::CommitSnapshotIds {
        snapshot_id: row.get_i64(0)?,
        schema_version: row.get_i64(1)?,
        next_catalog_id: row.get_i64(2)?,
        next_file_id: row.get_i64(3)?,
    })
}

pub(crate) fn column_type_value(row: &dyn EmbeddedRow) -> Result<(ColumnId, DuckLakeType)> {
    let column_id = ColumnId(row.get_i64(0)?);
    let raw_type = row.get_string(1)?;
    let column_type =
        DuckLakeType::parse(&raw_type).map_err(|err| StorageError::Catalog(err.into()))?;
    Ok((column_id, column_type))
}

pub(crate) fn column_types_by_id(
    rows: impl IntoIterator<Item = (ColumnId, DuckLakeType)>,
) -> HashMap<ColumnId, DuckLakeType> {
    rows.into_iter().collect()
}

pub(crate) fn table_column_stats_by_column(
    rows: impl IntoIterator<Item = TableColumnStats>,
) -> HashMap<ColumnId, TableColumnStats> {
    rows.into_iter()
        .map(|stats| (stats.column_id, stats))
        .collect()
}

pub(crate) fn validate_set_not_null_stats_result(
    column: &ColumnEntry,
    has_table_stats: bool,
    contains_null: Option<bool>,
) -> Result<()> {
    if !has_table_stats {
        return Err(StorageError::Backend(format!(
            "Cannot SET NOT NULL on table {} - the table has transaction-local changes or no stats are available",
            column.table_id.0
        )));
    }
    match contains_null {
        Some(false) => Ok(()),
        Some(true) => Err(StorageError::Backend(format!(
            "Cannot SET NOT NULL on column {} - the column has NULL values",
            column.column_name
        ))),
        None => Err(StorageError::Backend(format!(
            "Cannot SET NOT NULL on table {} - no column stats are available",
            column.table_id.0
        ))),
    }
}

pub(crate) fn active_column_ref(
    row: &dyn EmbeddedRow,
) -> Result<(ColumnId, String, Option<ColumnId>)> {
    Ok((
        ColumnId(row.get_i64(0)?),
        row.get_string(1)?,
        row.get_opt_i64(2)?.map(ColumnId),
    ))
}

pub(crate) fn column_parent_ref(row: &dyn EmbeddedRow) -> Result<(ColumnId, Option<i64>)> {
    Ok((ColumnId(row.get_i64(0)?), row.get_opt_i64(1)?))
}

pub(crate) fn validate_partitioning_columns(
    table_id: TableId,
    columns: &[PartitionColumnCommit],
    visible_column_ids: impl IntoIterator<Item = ColumnId>,
) -> Result<()> {
    let mut seen = Vec::new();
    for column in columns {
        let key = (column.column_id, column.transform.clone());
        if seen.contains(&key) {
            return Err(StorageError::Backend(format!(
                "DuckLake partition spec has duplicate column {} transform {}",
                column.column_id.0,
                partition_transform_string(&column.transform)
            )));
        }
        seen.push(key);
    }

    let visible = visible_column_ids.into_iter().collect::<HashSet<_>>();
    for column in columns {
        if !visible.contains(&column.column_id) {
            return Err(StorageError::Catalog(
                ducklake_catalog::CatalogError::NotFound(format!(
                    "column {} in table {}",
                    column.column_id.0, table_id.0
                )),
            ));
        }
    }
    Ok(())
}

pub(crate) fn partition_transform_string(transform: &PartitionTransform) -> String {
    match transform {
        PartitionTransform::Identity => "identity".to_string(),
        PartitionTransform::Bucket(bucket_count) => format!("bucket({bucket_count})"),
        PartitionTransform::Year => "year".to_string(),
        PartitionTransform::Month => "month".to_string(),
        PartitionTransform::Day => "day".to_string(),
        PartitionTransform::Hour => "hour".to_string(),
    }
}

fn render_table_refs(sql: &str, style: TableRefStyle<'_>) -> String {
    let table = |name: &str| match style {
        TableRefStyle::Unqualified => quote_identifier(name),
        TableRefStyle::Qualified(catalog) => {
            format!("{}.{}", quote_identifier(catalog), quote_identifier(name))
        }
        TableRefStyle::PostgresQualified(schema) => {
            format!("{}.{}", quote_identifier(schema), quote_identifier(name))
        }
    };
    // Postgres stores *_uuid columns as native UUID, which tokio-postgres cannot
    // deserialize into a Rust String; cast them to text on read. Embedded and
    // Quack backends store/return them as text already, so leave them raw.
    let uuid_col = |name: &str| match style {
        TableRefStyle::Unqualified | TableRefStyle::Qualified(_) => name.to_string(),
        TableRefStyle::PostgresQualified(_) => format!("{name}::TEXT"),
    };

    sql.replace("{ducklake_metadata}", &table("ducklake_metadata"))
        .replace("{ducklake_schema}", &table("ducklake_schema"))
        .replace("{ducklake_column}", &table("ducklake_column"))
        .replace("{ducklake_snapshot}", &table("ducklake_snapshot"))
        .replace(
            "{ducklake_snapshot_changes}",
            &table("ducklake_snapshot_changes"),
        )
        .replace(
            "{ducklake_schema_versions}",
            &table("ducklake_schema_versions"),
        )
        .replace("{ducklake_table_stats}", &table("ducklake_table_stats"))
        .replace("{ducklake_data_file}", &table("ducklake_data_file"))
        .replace("{ducklake_delete_file}", &table("ducklake_delete_file"))
        .replace(
            "{ducklake_files_scheduled_for_deletion}",
            &table("ducklake_files_scheduled_for_deletion"),
        )
        .replace(
            "{ducklake_file_partition_value}",
            &table("ducklake_file_partition_value"),
        )
        .replace(
            "{ducklake_file_column_stats}",
            &table("ducklake_file_column_stats"),
        )
        .replace(
            "{ducklake_file_variant_stats}",
            &table("ducklake_file_variant_stats"),
        )
        .replace(
            "{ducklake_table_column_stats}",
            &table("ducklake_table_column_stats"),
        )
        .replace("{ducklake_table}", &table("ducklake_table"))
        .replace("{ducklake_view}", &table("ducklake_view"))
        .replace("{ducklake_tag}", &table("ducklake_tag"))
        .replace(
            "{ducklake_partition_info}",
            &table("ducklake_partition_info"),
        )
        .replace(
            "{ducklake_partition_column}",
            &table("ducklake_partition_column"),
        )
        .replace("{ducklake_sort_info}", &table("ducklake_sort_info"))
        .replace(
            "{ducklake_sort_expression}",
            &table("ducklake_sort_expression"),
        )
        .replace(
            "{ducklake_inlined_data_tables}",
            &table("ducklake_inlined_data_tables"),
        )
        .replace("{schema_uuid_as_text}", &uuid_col("schema_uuid"))
        .replace("{table_uuid_as_text}", &uuid_col("table_uuid"))
        .replace("{view_uuid_as_text}", &uuid_col("view_uuid"))
}

fn render_numbered_placeholders(sql: &str, prefix: &str) -> String {
    // Rewrite every `?N` placeholder to `<prefix>N` in a single pass so any number
    // of placeholders is handled. A fixed cap would silently leave larger indexes
    // (e.g. `?33`) untouched and ship invalid SQL. `?` and ASCII digits are
    // single-byte, so slicing at these offsets always lands on char boundaries.
    let bytes = sql.as_bytes();
    let mut out = String::with_capacity(sql.len() + prefix.len());
    let mut i = 0;
    let mut span_start = 0;
    while i < bytes.len() {
        if bytes[i] == b'?' && bytes.get(i + 1).is_some_and(u8::is_ascii_digit) {
            out.push_str(&sql[span_start..i]);
            out.push_str(prefix);
            i += 1;
            let digits_start = i;
            while i < bytes.len() && bytes[i].is_ascii_digit() {
                i += 1;
            }
            out.push_str(&sql[digits_start..i]);
            span_start = i;
        } else {
            i += 1;
        }
    }
    out.push_str(&sql[span_start..]);
    out
}

fn render_params_as_literals(sql: &str, params: &[CatalogParam]) -> String {
    let mut out = sql.to_string();
    for idx in (1..=params.len()).rev() {
        out = out.replace(&format!("?{idx}"), &catalog_param_literal(&params[idx - 1]));
    }
    out
}

fn catalog_param_literal(param: &CatalogParam) -> String {
    match param {
        CatalogParam::I64(value) => value.to_string(),
        CatalogParam::String(value) => sql_string(value),
        CatalogParam::OptI64(Some(value)) => value.to_string(),
        CatalogParam::OptI64(None) => "NULL".into(),
        CatalogParam::OptString(Some(value)) => sql_string(value),
        CatalogParam::OptString(None) => "NULL".into(),
        CatalogParam::OptBool(Some(value)) => {
            if *value {
                "TRUE".into()
            } else {
                "FALSE".into()
            }
        }
        CatalogParam::OptBool(None) => "NULL".into(),
        CatalogParam::Bool(value) => {
            if *value {
                "TRUE".into()
            } else {
                "FALSE".into()
            }
        }
    }
}

#[cfg(test)]
#[path = "queries/tests.rs"]
mod tests;
