//! Shared passive catalog-read queries for DuckLake catalog backends.
//!
//! Each read surface is described once here as a [`ReadQuery`]: the SQL text
//! with DuckDB/SQLite-style `?N` placeholders, the `i64` parameters, and a row
//! mapper over the driver-bridging [`EmbeddedRow`] trait. The per-driver
//! executor owns statement preparation and row iteration; renderers adapt table
//! qualification, timestamp/UUID projections, and placeholder style.

#![allow(dead_code)]

use std::borrow::Cow;

use ducklake_catalog::{
    ColumnEntry, ColumnId, ColumnMapping, ColumnTagEntry, DataFile, DataFileId, DeleteFile,
    DeleteFileId, DeleteFormat, FileColumnStats, FilePartitionValue, FileScheduledForDeletion,
    FileVariantStats, InlinedDataTableRef, InlinedFileDeletion, MacroEntry, MacroId, MacroImpl,
    MacroImplId, MacroParameter, MappingId, NameMappingEntry, PartitionColumn, PartitionId,
    PartitionInfo, PartitionTransform, SchemaEntry, SchemaId, SchemaVersion, Snapshot,
    SnapshotChange, SnapshotId, SortExpression, SortId, SortInfo, TableColumnStats, TableEntry,
    TableId, TableStats, TagEntry, Validity, ViewEntry, ViewId,
};

use crate::Result;
pub(crate) use crate::ducklake_quote_identifier as quote_identifier;
use crate::embedded_rows::EmbeddedRow;

pub(crate) struct ReadQuery<T: 'static> {
    pub(crate) sql: Cow<'static, str>,
    /// Every embedded read parameter today is an `i64` id or snapshot value.
    /// If a non-integer parameter ever appears, widen this to a param enum.
    pub(crate) params: Vec<i64>,
    pub(crate) map: fn(&dyn EmbeddedRow) -> Result<T>,
}

impl<T> ReadQuery<T> {
    pub(crate) fn render_embedded(&self) -> String {
        render_read_sql(self.sql.as_ref(), ReadTableRefStyle::Unqualified)
    }

    #[cfg(feature = "postgres")]
    pub(crate) fn render_postgres(&self, metadata_schema: &str) -> String {
        let sql = render_read_sql(
            self.sql.as_ref(),
            ReadTableRefStyle::PostgresQualified(metadata_schema),
        );
        render_numbered_placeholders(&sql, "$")
    }

    #[cfg(feature = "quack")]
    pub(crate) fn render_quack(&self) -> String {
        self.render_quack_with_metadata_catalog(None)
    }

    #[cfg(feature = "quack")]
    pub(crate) fn render_quack_with_metadata_catalog(
        &self,
        metadata_catalog: Option<&str>,
    ) -> String {
        let style = metadata_catalog
            .map(ReadTableRefStyle::Qualified)
            .unwrap_or(ReadTableRefStyle::Unqualified);
        let sql = render_read_sql(self.sql.as_ref(), style);
        render_placeholders_as_literals(&sql, &self.params)
    }
}

pub(crate) fn snapshots() -> ReadQuery<Snapshot> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT snapshot_id, {snapshot_time_micros}, schema_version, next_catalog_id, next_file_id \
             FROM {ducklake_snapshot} ORDER BY snapshot_id",
        ),
        params: Vec::new(),
        map: snapshot,
    }
}

pub(crate) fn latest_snapshot_id(default: i64) -> ReadQuery<i64> {
    ReadQuery {
        sql: Cow::Borrowed("SELECT COALESCE(MAX(snapshot_id), ?1) FROM {ducklake_snapshot}"),
        params: vec![default],
        map: i64_value,
    }
}

pub(crate) fn snapshot_changes() -> ReadQuery<SnapshotChange> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT snapshot_id, changes_made, author, commit_message, commit_extra_info \
             FROM {ducklake_snapshot_changes} ORDER BY snapshot_id",
        ),
        params: Vec::new(),
        map: snapshot_change,
    }
}

pub(crate) fn schema_versions(table: TableId) -> ReadQuery<SchemaVersion> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT begin_snapshot, schema_version, table_id \
             FROM {ducklake_schema_versions} \
             WHERE table_id = ?1 ORDER BY begin_snapshot, schema_version",
        ),
        params: vec![table.0],
        map: schema_version,
    }
}

pub(crate) fn schemas_at(snapshot: SnapshotId) -> ReadQuery<SchemaEntry> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT schema_id, {schema_uuid_as_text}, begin_snapshot, end_snapshot, schema_name, path, path_is_relative \
             FROM {ducklake_schema} \
             WHERE ?1 >= begin_snapshot AND (end_snapshot IS NULL OR ?1 < end_snapshot) \
             ORDER BY schema_name",
        ),
        params: vec![snapshot.0],
        map: schema_entry,
    }
}

pub(crate) fn tables_at(schema: SchemaId, snapshot: SnapshotId) -> ReadQuery<TableEntry> {
    ReadQuery {
        sql: Cow::Borrowed(
            "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 ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
             ORDER BY table_name",
        ),
        params: vec![schema.0, snapshot.0],
        map: table_entry,
    }
}

pub(crate) fn views_at(schema: SchemaId, snapshot: SnapshotId) -> ReadQuery<ViewEntry> {
    ReadQuery {
        sql: Cow::Borrowed(
            "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 ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
             ORDER BY view_name",
        ),
        params: vec![schema.0, snapshot.0],
        map: view_entry,
    }
}

pub(crate) fn tags_at(snapshot: SnapshotId) -> ReadQuery<TagEntry> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT object_id, begin_snapshot, end_snapshot, key, value \
             FROM {ducklake_tag} \
             WHERE ?1 >= begin_snapshot AND (end_snapshot IS NULL OR ?1 < end_snapshot) \
             ORDER BY object_id, key",
        ),
        params: vec![snapshot.0],
        map: tag_entry,
    }
}

pub(crate) fn column_tags_at(table: TableId, snapshot: SnapshotId) -> ReadQuery<ColumnTagEntry> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT table_id, column_id, begin_snapshot, end_snapshot, key, value \
             FROM {ducklake_column_tag} \
             WHERE table_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
             ORDER BY column_id, key",
        ),
        params: vec![table.0, snapshot.0],
        map: column_tag_entry,
    }
}

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

pub(crate) fn snapshot(row: &dyn EmbeddedRow) -> Result<Snapshot> {
    Ok(Snapshot {
        snapshot_id: SnapshotId(row.get_i64(0)?),
        snapshot_time: row.get_time_micros(1)?,
        schema_version: row.get_i64(2)?,
        next_catalog_id: row.get_i64(3)?,
        next_file_id: row.get_i64(4)?,
    })
}

pub(crate) fn snapshot_change(row: &dyn EmbeddedRow) -> Result<SnapshotChange> {
    Ok(SnapshotChange {
        snapshot_id: SnapshotId(row.get_i64(0)?),
        changes_made: row.get_opt_string(1)?,
        author: row.get_opt_string(2)?,
        commit_message: row.get_opt_string(3)?,
        commit_extra_info: row.get_opt_string(4)?,
    })
}

pub(crate) fn schema_version(row: &dyn EmbeddedRow) -> Result<SchemaVersion> {
    Ok(SchemaVersion {
        begin_snapshot: SnapshotId(row.get_i64(0)?),
        schema_version: row.get_i64(1)?,
        table_id: row.get_opt_i64(2)?.map(TableId),
    })
}

pub(crate) fn schema_entry(row: &dyn EmbeddedRow) -> Result<SchemaEntry> {
    Ok(SchemaEntry {
        schema_id: SchemaId(row.get_i64(0)?),
        schema_uuid: row.get_string(1)?,
        validity: validity(row, 2, 3)?,
        schema_name: row.get_string(4)?,
        path: row.get_opt_string(5)?,
        path_is_relative: row.get_bool(6)?,
    })
}

pub(crate) fn table_entry(row: &dyn EmbeddedRow) -> Result<TableEntry> {
    Ok(TableEntry {
        table_id: TableId(row.get_i64(0)?),
        table_uuid: row.get_string(1)?,
        validity: validity(row, 2, 3)?,
        schema_id: SchemaId(row.get_i64(4)?),
        table_name: row.get_string(5)?,
        path: row.get_opt_string(6)?,
        path_is_relative: row.get_bool(7)?,
    })
}

pub(crate) fn view_entry(row: &dyn EmbeddedRow) -> Result<ViewEntry> {
    Ok(ViewEntry {
        view_id: ViewId(row.get_i64(0)?),
        view_uuid: row.get_string(1)?,
        validity: validity(row, 2, 3)?,
        schema_id: SchemaId(row.get_i64(4)?),
        view_name: row.get_string(5)?,
        dialect: row.get_opt_string(6)?,
        sql: row.get_string(7)?,
        column_aliases: row.get_opt_string(8)?,
    })
}

pub(crate) fn tag_entry(row: &dyn EmbeddedRow) -> Result<TagEntry> {
    Ok(TagEntry {
        object_id: row.get_i64(0)?,
        validity: validity(row, 1, 2)?,
        key: row.get_string(3)?,
        value: row.get_opt_string(4)?,
    })
}

pub(crate) fn column_tag_entry(row: &dyn EmbeddedRow) -> Result<ColumnTagEntry> {
    Ok(ColumnTagEntry {
        table_id: TableId(row.get_i64(0)?),
        column_id: ColumnId(row.get_i64(1)?),
        validity: validity(row, 2, 3)?,
        key: row.get_string(4)?,
        value: row.get_opt_string(5)?,
    })
}

pub(crate) fn columns_at(table: TableId, snapshot: SnapshotId) -> ReadQuery<ColumnEntry> {
    ReadQuery {
        sql: Cow::Borrowed(
            "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 ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
             ORDER BY parent_column, column_order",
        ),
        params: vec![table.0, snapshot.0],
        map: column_entry,
    }
}

pub(crate) fn data_files_at(table: TableId, snapshot: SnapshotId) -> ReadQuery<DataFile> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT data_file_id, table_id, begin_snapshot, end_snapshot, file_order, path, path_is_relative, \
                    file_format, record_count, file_size_bytes, footer_size, row_id_start, partition_id, \
                    encryption_key, mapping_id, partial_max \
             FROM {ducklake_data_file} \
             WHERE table_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
             ORDER BY file_order",
        ),
        params: vec![table.0, snapshot.0],
        map: data_file,
    }
}

pub(crate) fn delete_files_at(table: TableId, snapshot: SnapshotId) -> ReadQuery<DeleteFile> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT delete_file_id, table_id, begin_snapshot, end_snapshot, data_file_id, path, path_is_relative, \
                    format, delete_count, file_size_bytes, footer_size, encryption_key, partial_max \
             FROM {ducklake_delete_file} \
             WHERE table_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot)",
        ),
        params: vec![table.0, snapshot.0],
        map: delete_file,
    }
}

pub(crate) fn inlined_file_deletions(
    table_name: &str,
    snapshot: SnapshotId,
) -> ReadQuery<InlinedFileDeletion> {
    ReadQuery {
        sql: Cow::Owned(format!(
            "SELECT file_id, row_id, begin_snapshot FROM {{dynamic_table:{table_name}}} \
             WHERE begin_snapshot <= ?1 ORDER BY file_id, row_id"
        )),
        params: vec![snapshot.0],
        map: inlined_file_deletion,
    }
}

pub(crate) fn file_stats(table: TableId) -> ReadQuery<FileColumnStats> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT data_file_id, table_id, column_id, column_size_bytes, value_count, null_count, \
                    min_value, max_value, contains_nan, extra_stats \
             FROM {ducklake_file_column_stats} WHERE table_id = ?1",
        ),
        params: vec![table.0],
        map: file_column_stats,
    }
}

pub(crate) fn file_variant_stats(table: TableId) -> ReadQuery<FileVariantStats> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT data_file_id, table_id, column_id, variant_path, shredded_type, column_size_bytes, \
                    value_count, null_count, min_value, max_value, contains_nan, extra_stats \
             FROM {ducklake_file_variant_stats} WHERE table_id = ?1",
        ),
        params: vec![table.0],
        map: file_variant_stats_row,
    }
}

pub(crate) fn table_stats(table: TableId) -> ReadQuery<TableStats> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT table_id, record_count, next_row_id, file_size_bytes \
             FROM {ducklake_table_stats} WHERE table_id = ?1",
        ),
        params: vec![table.0],
        map: table_stats_row,
    }
}

pub(crate) fn table_column_stats(table: TableId) -> ReadQuery<TableColumnStats> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT table_id, column_id, contains_null, contains_nan, min_value, max_value, extra_stats \
             FROM {ducklake_table_column_stats} WHERE table_id = ?1",
        ),
        params: vec![table.0],
        map: table_column_stats_row,
    }
}

pub(crate) fn partition_info_at(table: TableId, snapshot: SnapshotId) -> ReadQuery<PartitionInfo> {
    ReadQuery {
        sql: Cow::Borrowed(
            "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",
        ),
        params: vec![table.0, snapshot.0],
        map: partition_info,
    }
}

pub(crate) fn partition_columns(partition: PartitionId) -> ReadQuery<PartitionColumn> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT partition_id, table_id, partition_key_index, column_id, transform \
             FROM {ducklake_partition_column} \
             WHERE partition_id = ?1 ORDER BY partition_key_index",
        ),
        params: vec![partition.0],
        map: partition_column,
    }
}

pub(crate) fn file_partition_values(table: TableId) -> ReadQuery<FilePartitionValue> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT data_file_id, table_id, partition_key_index, partition_value \
             FROM {ducklake_file_partition_value} \
             WHERE table_id = ?1 ORDER BY data_file_id, partition_key_index",
        ),
        params: vec![table.0],
        map: file_partition_value,
    }
}

pub(crate) fn files_scheduled_for_deletion() -> ReadQuery<FileScheduledForDeletion> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT data_file_id, path, path_is_relative, {schedule_start_micros} \
             FROM {ducklake_files_scheduled_for_deletion} ORDER BY schedule_start, data_file_id",
        ),
        params: Vec::new(),
        map: file_scheduled_for_deletion,
    }
}

pub(crate) fn column_mapping_head(mapping: MappingId) -> ReadQuery<ColumnMapping> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT mapping_id, table_id, type FROM {ducklake_column_mapping} WHERE mapping_id = ?1",
        ),
        params: vec![mapping.0],
        map: column_mapping_row,
    }
}

pub(crate) fn name_mappings(mapping: MappingId) -> ReadQuery<NameMappingEntry> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT mapping_id, column_id, source_name, target_field_id, parent_column, is_partition \
             FROM {ducklake_name_mapping} WHERE mapping_id = ?1",
        ),
        params: vec![mapping.0],
        map: name_mapping_entry,
    }
}

pub(crate) fn inlined_tables(
    table: TableId,
    schema_version: i64,
) -> ReadQuery<InlinedDataTableRef> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT table_id, table_name, schema_version \
             FROM {ducklake_inlined_data_tables} \
             WHERE table_id = ?1 AND schema_version = ?2",
        ),
        params: vec![table.0, schema_version],
        map: inlined_data_table_ref,
    }
}

pub(crate) fn macros_at(schema: SchemaId, snapshot: SnapshotId) -> ReadQuery<MacroEntry> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT schema_id, macro_id, macro_name, begin_snapshot, end_snapshot \
             FROM {ducklake_macro} \
             WHERE schema_id = ?1 AND ?2 >= begin_snapshot AND (end_snapshot IS NULL OR ?2 < end_snapshot) \
             ORDER BY macro_name",
        ),
        params: vec![schema.0, snapshot.0],
        map: macro_entry,
    }
}

pub(crate) fn macro_impls(macro_id: MacroId) -> ReadQuery<MacroImpl> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT macro_id, impl_id, dialect, sql, type \
             FROM {ducklake_macro_impl} \
             WHERE macro_id = ?1 ORDER BY impl_id",
        ),
        params: vec![macro_id.0],
        map: macro_impl,
    }
}

pub(crate) fn macro_parameters(
    macro_id: MacroId,
    impl_id: MacroImplId,
) -> ReadQuery<MacroParameter> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT macro_id, impl_id, column_id, parameter_name, parameter_type, default_value, default_value_type \
             FROM {ducklake_macro_parameters} \
             WHERE macro_id = ?1 AND impl_id = ?2 ORDER BY column_id",
        ),
        params: vec![macro_id.0, impl_id.0],
        map: macro_parameter,
    }
}

pub(crate) fn sort_info_at(table: TableId, snapshot: SnapshotId) -> ReadQuery<SortInfo> {
    ReadQuery {
        sql: Cow::Borrowed(
            "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",
        ),
        params: vec![table.0, snapshot.0],
        map: sort_info,
    }
}

pub(crate) fn sort_expressions(sort_id: SortId) -> ReadQuery<SortExpression> {
    ReadQuery {
        sql: Cow::Borrowed(
            "SELECT sort_id, table_id, sort_key_index, expression, dialect, sort_direction, null_order \
             FROM {ducklake_sort_expression} \
             WHERE sort_id = ?1 ORDER BY sort_key_index",
        ),
        params: vec![sort_id.0],
        map: sort_expression,
    }
}

pub(crate) fn column_entry(row: &dyn EmbeddedRow) -> Result<ColumnEntry> {
    Ok(ColumnEntry {
        column_id: ColumnId(row.get_i64(0)?),
        validity: validity(row, 1, 2)?,
        table_id: TableId(row.get_i64(3)?),
        column_order: row.get_i64(4)?,
        column_name: row.get_string(5)?,
        column_type: row.get_string(6)?,
        initial_default: row.get_opt_string(7)?,
        default_value: row.get_opt_string(8)?,
        nulls_allowed: row.get_bool(9)?,
        parent_column: row.get_opt_i64(10)?.map(ColumnId),
        default_value_type: row.get_opt_string(11)?,
        default_value_dialect: row.get_opt_string(12)?,
    })
}

pub(crate) fn data_file(row: &dyn EmbeddedRow) -> Result<DataFile> {
    Ok(DataFile {
        data_file_id: DataFileId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        validity: validity(row, 2, 3)?,
        file_order: row.get_opt_i64(4)?,
        path: row.get_string(5)?,
        path_is_relative: row.get_bool(6)?,
        file_format: row.get_string(7)?,
        record_count: row.get_i64(8)?,
        file_size_bytes: row.get_i64(9)?,
        footer_size: row.get_opt_i64(10)?,
        row_id_start: row.get_opt_i64(11)?,
        partition_id: row.get_opt_i64(12)?.map(PartitionId),
        encryption_key: row.get_opt_string(13)?,
        mapping_id: row.get_opt_i64(14)?.map(MappingId),
        partial_max: row.get_opt_i64(15)?.map(SnapshotId),
    })
}

pub(crate) fn delete_file(row: &dyn EmbeddedRow) -> Result<DeleteFile> {
    let format = row.get_string(7)?;
    Ok(DeleteFile {
        delete_file_id: DeleteFileId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        validity: validity(row, 2, 3)?,
        data_file_id: DataFileId(row.get_i64(4)?),
        path: row.get_string(5)?,
        path_is_relative: row.get_bool(6)?,
        format: parse_delete_format(&format),
        delete_count: row.get_i64(8)?,
        file_size_bytes: row.get_i64(9)?,
        footer_size: row.get_opt_i64(10)?,
        encryption_key: row.get_opt_string(11)?,
        partial_max: row.get_opt_i64(12)?.map(SnapshotId),
    })
}

pub(crate) fn inlined_file_deletion(row: &dyn EmbeddedRow) -> Result<InlinedFileDeletion> {
    Ok(InlinedFileDeletion {
        data_file_id: DataFileId(row.get_i64(0)?),
        row_id: row.get_i64(1)?,
        begin_snapshot: SnapshotId(row.get_i64(2)?),
    })
}

pub(crate) fn file_column_stats(row: &dyn EmbeddedRow) -> Result<FileColumnStats> {
    Ok(FileColumnStats {
        data_file_id: DataFileId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        column_id: ColumnId(row.get_i64(2)?),
        column_size_bytes: row.get_opt_i64(3)?,
        value_count: row.get_opt_i64(4)?,
        null_count: row.get_opt_i64(5)?,
        min_value: row.get_opt_string(6)?,
        max_value: row.get_opt_string(7)?,
        contains_nan: row.get_opt_bool(8)?,
        extra_stats: row.get_opt_string(9)?,
    })
}

pub(crate) fn file_variant_stats_row(row: &dyn EmbeddedRow) -> Result<FileVariantStats> {
    Ok(FileVariantStats {
        data_file_id: DataFileId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        column_id: ColumnId(row.get_i64(2)?),
        variant_path: row.get_string(3)?,
        shredded_type: row.get_string(4)?,
        column_size_bytes: row.get_opt_i64(5)?,
        value_count: row.get_opt_i64(6)?,
        null_count: row.get_opt_i64(7)?,
        min_value: row.get_opt_string(8)?,
        max_value: row.get_opt_string(9)?,
        contains_nan: row.get_opt_bool(10)?,
        extra_stats: row.get_opt_string(11)?,
    })
}

pub(crate) fn table_stats_row(row: &dyn EmbeddedRow) -> Result<TableStats> {
    Ok(TableStats {
        table_id: TableId(row.get_i64(0)?),
        record_count: row.get_i64(1)?,
        next_row_id: row.get_i64(2)?,
        file_size_bytes: row.get_i64(3)?,
    })
}

pub(crate) fn table_column_stats_row(row: &dyn EmbeddedRow) -> Result<TableColumnStats> {
    Ok(TableColumnStats {
        table_id: TableId(row.get_i64(0)?),
        column_id: ColumnId(row.get_i64(1)?),
        contains_null: row.get_opt_bool(2)?,
        contains_nan: row.get_opt_bool(3)?,
        min_value: row.get_opt_string(4)?,
        max_value: row.get_opt_string(5)?,
        extra_stats: row.get_opt_string(6)?,
    })
}

pub(crate) fn partition_info(row: &dyn EmbeddedRow) -> Result<PartitionInfo> {
    Ok(PartitionInfo {
        partition_id: PartitionId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        validity: validity(row, 2, 3)?,
    })
}

pub(crate) fn partition_column(row: &dyn EmbeddedRow) -> Result<PartitionColumn> {
    let transform = row.get_string(4)?;
    let transform = PartitionTransform::parse(&transform)?;
    Ok(PartitionColumn {
        partition_id: PartitionId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        partition_key_index: row.get_i64(2)?,
        column_id: ColumnId(row.get_i64(3)?),
        transform,
    })
}

pub(crate) fn file_partition_value(row: &dyn EmbeddedRow) -> Result<FilePartitionValue> {
    Ok(FilePartitionValue {
        data_file_id: DataFileId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        partition_key_index: row.get_i64(2)?,
        partition_value: row.get_opt_string(3)?,
    })
}

pub(crate) fn file_scheduled_for_deletion(
    row: &dyn EmbeddedRow,
) -> Result<FileScheduledForDeletion> {
    Ok(FileScheduledForDeletion {
        data_file_id: DataFileId(row.get_i64(0)?),
        path: row.get_string(1)?,
        path_is_relative: row.get_bool(2)?,
        schedule_start: row.get_time_micros(3)?,
    })
}

pub(crate) fn name_mapping_entry(row: &dyn EmbeddedRow) -> Result<NameMappingEntry> {
    Ok(NameMappingEntry {
        mapping_id: MappingId(row.get_i64(0)?),
        column_id: ColumnId(row.get_i64(1)?),
        source_name: row.get_string(2)?,
        target_field_id: ColumnId(row.get_i64(3)?),
        parent_column: row.get_opt_i64(4)?.map(ColumnId),
        is_partition: row.get_bool(5)?,
    })
}

pub(crate) fn macro_entry(row: &dyn EmbeddedRow) -> Result<MacroEntry> {
    Ok(MacroEntry {
        schema_id: SchemaId(row.get_i64(0)?),
        macro_id: MacroId(row.get_i64(1)?),
        macro_name: row.get_string(2)?,
        validity: validity(row, 3, 4)?,
    })
}

pub(crate) fn macro_impl(row: &dyn EmbeddedRow) -> Result<MacroImpl> {
    Ok(MacroImpl {
        macro_id: MacroId(row.get_i64(0)?),
        impl_id: MacroImplId(row.get_i64(1)?),
        dialect: row.get_string(2)?,
        sql: row.get_string(3)?,
        macro_type: row.get_string(4)?,
    })
}

pub(crate) fn macro_parameter(row: &dyn EmbeddedRow) -> Result<MacroParameter> {
    Ok(MacroParameter {
        macro_id: MacroId(row.get_i64(0)?),
        impl_id: MacroImplId(row.get_i64(1)?),
        column_id: row.get_i64(2)?,
        parameter_name: row.get_string(3)?,
        parameter_type: row.get_string(4)?,
        default_value: row.get_opt_string(5)?,
        default_value_type: row.get_opt_string(6)?,
    })
}

pub(crate) fn sort_info(row: &dyn EmbeddedRow) -> Result<SortInfo> {
    Ok(SortInfo {
        sort_id: SortId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        validity: validity(row, 2, 3)?,
    })
}

pub(crate) fn sort_expression(row: &dyn EmbeddedRow) -> Result<SortExpression> {
    Ok(SortExpression {
        sort_id: SortId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        sort_key_index: row.get_i64(2)?,
        expression: row.get_string(3)?,
        dialect: row.get_opt_string(4)?,
        sort_direction: row.get_opt_string(5)?,
        null_order: row.get_opt_string(6)?,
    })
}

pub(crate) fn column_mapping_row(row: &dyn EmbeddedRow) -> Result<ColumnMapping> {
    Ok(ColumnMapping {
        mapping_id: MappingId(row.get_i64(0)?),
        table_id: TableId(row.get_i64(1)?),
        mapping_type: row.get_string(2)?,
    })
}

pub(crate) fn inlined_data_table_ref(row: &dyn EmbeddedRow) -> Result<InlinedDataTableRef> {
    Ok(InlinedDataTableRef {
        table_id: TableId(row.get_i64(0)?),
        table_name: row.get_string(1)?,
        schema_version: row.get_i64(2)?,
    })
}

fn parse_delete_format(value: &str) -> DeleteFormat {
    if value.eq_ignore_ascii_case("parquet") {
        return DeleteFormat::Parquet;
    }
    if value.eq_ignore_ascii_case("puffin") {
        return DeleteFormat::Puffin;
    }
    DeleteFormat::Unknown(value.into())
}

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

fn render_read_sql(sql: &str, style: ReadTableRefStyle<'_>) -> String {
    let table = |name: &str| match style {
        ReadTableRefStyle::Unqualified => quote_identifier(name),
        ReadTableRefStyle::Qualified(catalog) => {
            format!("{}.{}", quote_identifier(catalog), quote_identifier(name))
        }
        ReadTableRefStyle::PostgresQualified(schema) => {
            format!("{}.{}", quote_identifier(schema), quote_identifier(name))
        }
    };

    let snapshot_time_micros = match style {
        ReadTableRefStyle::Unqualified | ReadTableRefStyle::Qualified(_) => "snapshot_time",
        ReadTableRefStyle::PostgresQualified(_) => {
            "(EXTRACT(EPOCH FROM snapshot_time) * 1000000)::BIGINT"
        }
    };
    let schedule_start_micros = match style {
        ReadTableRefStyle::Unqualified | ReadTableRefStyle::Qualified(_) => "schedule_start",
        ReadTableRefStyle::PostgresQualified(_) => {
            "(EXTRACT(EPOCH FROM schedule_start) * 1000000)::BIGINT"
        }
    };
    let uuid_expr = |column: &str| match style {
        ReadTableRefStyle::Unqualified | ReadTableRefStyle::Qualified(_) => column.to_string(),
        ReadTableRefStyle::PostgresQualified(_) => format!("{column}::TEXT"),
    };

    render_dynamic_table_refs(sql, table)
        .replace("{snapshot_time_micros}", snapshot_time_micros)
        .replace("{schedule_start_micros}", schedule_start_micros)
        .replace("{schema_uuid_as_text}", &uuid_expr("schema_uuid"))
        .replace("{table_uuid_as_text}", &uuid_expr("table_uuid"))
        .replace("{view_uuid_as_text}", &uuid_expr("view_uuid"))
        .replace(
            "{ducklake_schema_versions}",
            &table("ducklake_schema_versions"),
        )
        .replace("{ducklake_schema}", &table("ducklake_schema"))
        .replace("{ducklake_table}", &table("ducklake_table"))
        .replace("{ducklake_view}", &table("ducklake_view"))
        .replace("{ducklake_snapshot}", &table("ducklake_snapshot"))
        .replace(
            "{ducklake_snapshot_changes}",
            &table("ducklake_snapshot_changes"),
        )
        .replace("{ducklake_tag}", &table("ducklake_tag"))
        .replace("{ducklake_column_tag}", &table("ducklake_column_tag"))
        .replace("{ducklake_column}", &table("ducklake_column"))
        .replace("{ducklake_data_file}", &table("ducklake_data_file"))
        .replace("{ducklake_delete_file}", &table("ducklake_delete_file"))
        .replace(
            "{ducklake_file_column_stats}",
            &table("ducklake_file_column_stats"),
        )
        .replace(
            "{ducklake_file_variant_stats}",
            &table("ducklake_file_variant_stats"),
        )
        .replace("{ducklake_table_stats}", &table("ducklake_table_stats"))
        .replace(
            "{ducklake_table_column_stats}",
            &table("ducklake_table_column_stats"),
        )
        .replace(
            "{ducklake_partition_info}",
            &table("ducklake_partition_info"),
        )
        .replace(
            "{ducklake_partition_column}",
            &table("ducklake_partition_column"),
        )
        .replace(
            "{ducklake_file_partition_value}",
            &table("ducklake_file_partition_value"),
        )
        .replace(
            "{ducklake_files_scheduled_for_deletion}",
            &table("ducklake_files_scheduled_for_deletion"),
        )
        .replace(
            "{ducklake_column_mapping}",
            &table("ducklake_column_mapping"),
        )
        .replace("{ducklake_name_mapping}", &table("ducklake_name_mapping"))
        .replace(
            "{ducklake_inlined_data_tables}",
            &table("ducklake_inlined_data_tables"),
        )
        .replace(
            "{ducklake_macro_parameters}",
            &table("ducklake_macro_parameters"),
        )
        .replace("{ducklake_macro_impl}", &table("ducklake_macro_impl"))
        .replace("{ducklake_macro}", &table("ducklake_macro"))
        .replace("{ducklake_sort_info}", &table("ducklake_sort_info"))
        .replace(
            "{ducklake_sort_expression}",
            &table("ducklake_sort_expression"),
        )
}

fn render_dynamic_table_refs(sql: &str, table: impl Fn(&str) -> String) -> String {
    const PREFIX: &str = "{dynamic_table:";

    let mut out = String::with_capacity(sql.len());
    let mut remaining = sql;
    while let Some(start) = remaining.find(PREFIX) {
        out.push_str(&remaining[..start]);
        let after_prefix = &remaining[start + PREFIX.len()..];
        let Some(end) = after_prefix.find('}') else {
            out.push_str(&remaining[start..]);
            return out;
        };
        out.push_str(&table(&after_prefix[..end]));
        remaining = &after_prefix[end + 1..];
    }
    out.push_str(remaining);
    out
}

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
}

/// Render a descriptor's `?N` placeholders as integer literals for backends
/// without bind support (Quack). All read parameters are `i64`, so literal
/// substitution needs no escaping. Placeholders are replaced highest-index
/// first so `?1` never clobbers the prefix of `?10`.
#[cfg(feature = "quack")]
pub(crate) fn render_placeholders_as_literals(sql: &str, params: &[i64]) -> String {
    let mut sql = sql.to_string();
    for (idx, value) in params.iter().enumerate().rev() {
        sql = sql.replace(&format!("?{}", idx + 1), &value.to_string());
    }
    sql
}

fn validity(row: &dyn EmbeddedRow, begin_idx: usize, end_idx: usize) -> Result<Validity> {
    Ok(Validity {
        begin: SnapshotId(row.get_i64(begin_idx)?),
        end: row.get_opt_i64(end_idx)?.map(SnapshotId),
    })
}

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