//! client-side DuckDB-over-Quack catalog backend.
//!
//! This module intentionally models only the client boundary from
//! `external-projects/duckdb-quack` (`quack_uri`, `quack_client`, and
//! `quack_message`). The server and DuckDB storage-extension pieces are out of
//! scope for this crate.

use std::{
    collections::HashMap,
    env, fmt,
    str::FromStr,
    sync::{Arc, Mutex as StdMutex, OnceLock},
};

#[cfg(test)]
use crate::quack_protocol::*;
use crate::quack_protocol::{
    ConnectionRequest, QuackConnection, QuackHttpClient, QuackQueryResult, QuackRequest,
    QuackResponse, QuackRow, QuackUri, validate_token,
};
pub use crate::quack_protocol::{DEFAULT_PORT, DEFAULT_SSL, QUACK_VERSION};
#[cfg(test)]
use crate::{
    AddColumnCommit, CleanupOldFilesOptions, ColumnDefaultCommit, ColumnTypeCommit,
    ColumnTypeTreeCommit, CreateSchemaCommit, CreateTableCommit, CreateViewCommit,
    DropColumnCommit, DropSchemaCommit, DropTableCommit, DropViewCommit, ExpireSnapshotsOptions,
    FlushedInlinedDataCommit, FlushedInlinedFileDeletionCommit, InlinedDataDeletionCommit,
    InlinedFileDeletionCommit, MergeAdjacentCompactionCommit, RenameColumnCommit,
    RenameTableCommit, RewriteDataFilesCommit, SortedByCommit, TruncateTableCommit,
};
use arrow::array::Array;
use async_trait::async_trait;
#[cfg(test)]
use ducklake_catalog::PartitionTransform;
#[cfg(test)]
use ducklake_catalog::{ColumnId, DataFileId, FilePartitionValue, PartitionId};
use ducklake_catalog::{
    DuckLakeType, FileColumnStats, SchemaField, Snapshot, SnapshotId, TableColumnStats, TableId,
    TableSchema,
};

use crate::catalog_async_commit_methods::AsyncCatalogQueryCommitExecutor;
use crate::catalog_async_query_methods::{AsyncCatalogQueryExecutor, AsyncReadQueryExecutor};
use crate::embedded_rows::EmbeddedRow;
use crate::inlined_values::{
    InlinedSqlDialect, inlined_column_type, sql_cast_string_literal, sql_string_literal,
};
use crate::quack_inlined::quack_inlined_record_batch;
use crate::sql_utils::join_i64s;
use crate::{
    AppendFile, CatalogBackend, CleanupMetadataFile, DeleteFileCommit,
    DuckLakeCatalogConnectionOptions, InlinedDataAppend, InlinedDataAppendCommit, InlinedDataBatch,
    Result, StorageError, TableChangeCommit, ducklake_quote_identifier as quack_quote_identifier,
};

static QUACK_CATALOG_WRITE_LOCKS: OnceLock<
    StdMutex<HashMap<QuackWriteLockKey, Arc<tokio::sync::Mutex<()>>>>,
> = OnceLock::new();

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct QuackWriteLockKey {
    http_url: String,
    metadata_catalog: Option<String>,
}

#[derive(Clone)]
pub struct QuackCatalog {
    uri: QuackUri,
    token: String,
    metadata_catalog: Option<String>,
    commit_mode: QuackCommitMode,
    connection_options: DuckLakeCatalogConnectionOptions,
    client: QuackHttpClient,
    connection: Arc<tokio::sync::Mutex<Option<QuackConnection>>>,
    server_capabilities: Arc<tokio::sync::Mutex<Option<QuackServerCapabilities>>>,
}

pub const QUACK_COMMIT_MODE_ENV: &str = "DATAFUSION_DUCKLAKE_QUACK_COMMIT_MODE";

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum QuackCommitMode {
    #[default]
    Auto,
    Client,
    Server,
}

impl QuackCommitMode {
    fn from_env() -> Result<Self> {
        match env::var(QUACK_COMMIT_MODE_ENV) {
            Ok(value) if value.trim().is_empty() => Ok(Self::Auto),
            Ok(value) => value.parse(),
            Err(env::VarError::NotPresent) => Ok(Self::Auto),
            Err(err) => Err(StorageError::Backend(format!(
                "Failed to read {QUACK_COMMIT_MODE_ENV}: {err}"
            ))),
        }
    }

    fn requires_server_side(self) -> bool {
        matches!(self, Self::Server)
    }
}

impl FromStr for QuackCommitMode {
    type Err = StorageError;

    fn from_str(value: &str) -> Result<Self> {
        match value.trim().to_ascii_lowercase().as_str() {
            "auto" => Ok(Self::Auto),
            "client" | "client_side" | "client-side" => Ok(Self::Client),
            "server" | "server_side" | "server-side" | "server_required" | "server-required" => {
                Ok(Self::Server)
            }
            other => Err(StorageError::Backend(format!(
                "DuckLake Quack commit mode {other:?} is not supported; expected auto, client, server, or server_required"
            ))),
        }
    }
}

impl fmt::Display for QuackCommitMode {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Auto => f.write_str("auto"),
            Self::Client => f.write_str("client"),
            Self::Server => f.write_str("server"),
        }
    }
}

#[derive(Debug, Clone, Copy)]
struct QuackServerCapabilities {
    ducklake_commit: bool,
}

#[derive(Debug, Clone, Copy)]
enum QuackServerSideCommitKind {
    AppendFiles,
    AppendInlinedData,
    DeleteFiles,
    TableChanges,
    CreateTableWithInlinedData,
    TruncateTable,
    MergeAdjacentCompaction,
    RewriteDataFiles,
}

impl QuackServerSideCommitKind {
    fn for_request(request: &crate::commit_engine::CommitRequest) -> Option<Self> {
        match request {
            crate::commit_engine::CommitRequest::AppendFiles(_) => Some(Self::AppendFiles),
            crate::commit_engine::CommitRequest::AppendInlinedData(_) => {
                Some(Self::AppendInlinedData)
            }
            crate::commit_engine::CommitRequest::DeleteFiles(_) => Some(Self::DeleteFiles),
            crate::commit_engine::CommitRequest::TableChanges(_) => Some(Self::TableChanges),
            crate::commit_engine::CommitRequest::CreateTableWithInlinedData(_) => {
                Some(Self::CreateTableWithInlinedData)
            }
            crate::commit_engine::CommitRequest::TruncateTable(_) => Some(Self::TruncateTable),
            crate::commit_engine::CommitRequest::MergeAdjacentCompaction(_) => {
                Some(Self::MergeAdjacentCompaction)
            }
            crate::commit_engine::CommitRequest::RewriteDataFiles(_) => {
                Some(Self::RewriteDataFiles)
            }
            crate::commit_engine::CommitRequest::CreateSchema(_)
            | crate::commit_engine::CommitRequest::CreateTable(_)
            | crate::commit_engine::CommitRequest::CreateTableWithFiles(_)
            | crate::commit_engine::CommitRequest::CreateView(_)
            | crate::commit_engine::CommitRequest::DropTable(_)
            | crate::commit_engine::CommitRequest::RenameTable(_)
            | crate::commit_engine::CommitRequest::DropView(_)
            | crate::commit_engine::CommitRequest::DropSchema(_)
            | crate::commit_engine::CommitRequest::ColumnNullability(_)
            | crate::commit_engine::CommitRequest::ColumnDefault(_)
            | crate::commit_engine::CommitRequest::ColumnType(_)
            | crate::commit_engine::CommitRequest::ColumnTypeTree(_)
            | crate::commit_engine::CommitRequest::RenameColumn(_)
            | crate::commit_engine::CommitRequest::DropColumn(_)
            | crate::commit_engine::CommitRequest::AddColumns(_)
            | crate::commit_engine::CommitRequest::PartitioningBatch(_)
            | crate::commit_engine::CommitRequest::SortedBy(_) => None,
        }
    }
}

impl fmt::Display for QuackServerSideCommitKind {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::AppendFiles => f.write_str("append-file commits"),
            Self::AppendInlinedData => f.write_str("inlined append commits"),
            Self::DeleteFiles => f.write_str("delete-file commits"),
            Self::TableChanges => f.write_str("table-change commits"),
            Self::CreateTableWithInlinedData => f.write_str("create-table-as inlined-data commits"),
            Self::TruncateTable => f.write_str("truncate commits"),
            Self::MergeAdjacentCompaction => f.write_str("merge-adjacent compaction commits"),
            Self::RewriteDataFiles => f.write_str("rewrite-data-files commits"),
        }
    }
}

impl QuackCatalog {
    pub fn try_new(uri: impl Into<String>, token: impl Into<String>) -> Result<Self> {
        let uri = QuackUri::parse_for_client(uri)?;
        Self::try_new_with_uri(uri, token)
    }

    pub fn try_new_with_ssl(
        uri: impl Into<String>,
        token: impl Into<String>,
        ssl: bool,
    ) -> Result<Self> {
        let uri = QuackUri::parse_with_ssl(uri, ssl)?;
        Self::try_new_with_uri(uri, token)
    }

    fn try_new_with_uri(uri: QuackUri, token: impl Into<String>) -> Result<Self> {
        let token = token.into();
        validate_token(&token)?;
        Ok(Self {
            client: QuackHttpClient::new(uri.clone()),
            uri,
            token,
            metadata_catalog: None,
            commit_mode: QuackCommitMode::from_env()?,
            connection_options: DuckLakeCatalogConnectionOptions::default(),
            connection: Arc::new(tokio::sync::Mutex::new(None)),
            server_capabilities: Arc::new(tokio::sync::Mutex::new(None)),
        })
    }

    pub fn with_metadata_catalog(mut self, metadata_catalog: impl Into<String>) -> Result<Self> {
        let metadata_catalog = metadata_catalog.into().trim().to_string();
        if metadata_catalog.is_empty() {
            return Err(StorageError::Backend(
                "Quack metadata_catalog cannot be empty".into(),
            ));
        }
        self.metadata_catalog = Some(metadata_catalog);
        Ok(self)
    }

    pub fn with_commit_mode(mut self, mode: QuackCommitMode) -> Self {
        self.commit_mode = mode;
        self
    }

    impl_connection_options_builder_method!();

    pub fn uri(&self) -> &QuackUri {
        &self.uri
    }

    pub fn token(&self) -> &str {
        &self.token
    }

    pub fn metadata_catalog(&self) -> Option<&str> {
        self.metadata_catalog.as_deref()
    }

    pub fn commit_mode(&self) -> QuackCommitMode {
        self.commit_mode
    }

    pub fn connection_request(&self) -> QuackRequest {
        QuackRequest::Connection(ConnectionRequest::new(self.token.clone()))
    }

    pub async fn connect(&self) -> Result<QuackConnection> {
        self.cached_connection().await
    }

    async fn connect_with_client(&self, client: &QuackHttpClient) -> Result<QuackConnection> {
        let request = self.connection_request();
        let response = client.send(&request).await?;
        let QuackResponse::Connection(response) = response else {
            return Err(StorageError::Backend(
                "Quack connection request returned a non-connection response".into(),
            ));
        };
        QuackConnection::from_response(self.uri.clone(), response)
    }

    async fn cached_connection(&self) -> Result<QuackConnection> {
        let mut connection = self.connection.lock().await;
        if let Some(connection) = connection.as_ref() {
            return Ok(connection.clone());
        }

        let new_connection = self.connect_with_client(&self.client).await?;
        *connection = Some(new_connection.clone());
        Ok(new_connection)
    }

    async fn clear_cached_connection(&self) {
        *self.connection.lock().await = None;
    }

    pub async fn disconnect(&self) -> Result<()> {
        let connection = self.connection.lock().await.take();
        let Some(connection) = connection else {
            return Ok(());
        };
        self.client.send(&connection.disconnect_request()).await?;
        Ok(())
    }

    async fn query_sql(&self, sql: impl Into<String>) -> Result<QuackQueryResult> {
        let sql = self.qualify_metadata_sql(sql.into());
        let connection = self.cached_connection().await?;
        let result = self
            .query_sql_with_connection(&self.client, &connection, sql)
            .await;
        if result
            .as_ref()
            .is_err_and(quack_error_invalidates_connection)
        {
            self.clear_cached_connection().await;
        }
        result
    }

    async fn with_write_lock<T, F, Fut>(&self, f: F) -> Result<T>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<T>>,
    {
        let write_lock = quack_catalog_write_lock(self)?;
        let _write_guard = write_lock.lock().await;
        f().await
    }

    async fn run_unlocked_write_transaction<T, F>(
        &self,
        context: &str,
        operation: Option<&str>,
        expected_snapshot: Option<SnapshotId>,
        f: F,
    ) -> Result<T>
    where
        T: Send,
        F: for<'catalog> FnOnce(
            &'catalog Self,
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = Result<T>> + Send + 'catalog>,
        >,
    {
        self.query_sql("BEGIN TRANSACTION").await?;
        match f(self).await {
            Ok(value) => match self.query_sql("COMMIT").await {
                Ok(_) => Ok(value),
                Err(err) => {
                    quack_abort_write_transaction(self, context, operation, expected_snapshot, err)
                        .await
                }
            },
            Err(err) => {
                quack_abort_write_transaction(self, context, operation, expected_snapshot, err)
                    .await
            }
        }
    }

    async fn run_write_transaction<T, F>(
        &self,
        context: &str,
        operation: Option<&str>,
        expected_snapshot: Option<SnapshotId>,
        f: F,
    ) -> Result<T>
    where
        T: Send,
        F: for<'catalog> FnOnce(
            &'catalog Self,
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = Result<T>> + Send + 'catalog>,
        >,
    {
        self.with_write_lock(|| async {
            self.run_unlocked_write_transaction(context, operation, expected_snapshot, f)
                .await
        })
        .await
    }

    async fn run_commit_request(
        &self,
        request: crate::commit_engine::CommitRequest,
    ) -> Result<SnapshotId> {
        request.validate_preflight()?;
        if let Some(snapshot_id) = self.try_run_server_side_commit(&request).await? {
            return Ok(snapshot_id);
        }
        let abort = request.abort_context();
        self.run_write_transaction(
            abort.label,
            abort.operation,
            abort.expected_snapshot,
            |catalog| {
                Box::pin(async move {
                    crate::commit_engine::apply_async_commit_request(
                        &QuackCommitOps,
                        catalog,
                        &request,
                    )
                    .await
                })
            },
        )
        .await
    }

    async fn try_run_server_side_commit(
        &self,
        request: &crate::commit_engine::CommitRequest,
    ) -> Result<Option<SnapshotId>> {
        let Some(kind) = QuackServerSideCommitKind::for_request(request) else {
            return Ok(None);
        };
        if matches!(self.commit_mode, QuackCommitMode::Client) {
            return Ok(None);
        }
        if !self.server_capabilities().await?.ducklake_commit {
            if self.commit_mode.requires_server_side() {
                return Err(StorageError::Backend(format!(
                    "DuckLake Quack commit mode {} requires server-side ducklake_commit for {kind}, but the Quack server does not expose ducklake_commit",
                    self.commit_mode
                )));
            }
            return Ok(None);
        }
        match request {
            crate::commit_engine::CommitRequest::AppendFiles(files) => {
                if !quack_append_files_can_use_server_side_commit(files) {
                    if self.commit_mode.requires_server_side() {
                        return Err(StorageError::Backend(format!(
                            "DuckLake Quack commit mode {} requires server-side ducklake_commit for {kind}, but this commit contains metadata that cannot be staged safely yet",
                            self.commit_mode
                        )));
                    }
                    return Ok(None);
                }
                self.try_run_server_side_append_files(files).await
            }
            crate::commit_engine::CommitRequest::AppendInlinedData(append) => {
                self.try_run_server_side_append_inlined_data(append).await
            }
            crate::commit_engine::CommitRequest::DeleteFiles(files) => {
                self.try_run_server_side_delete_files(files).await
            }
            crate::commit_engine::CommitRequest::TableChanges(changes) => {
                self.try_run_server_side_table_changes(changes).await
            }
            crate::commit_engine::CommitRequest::CreateTableWithInlinedData(_) => {
                if self.commit_mode.requires_server_side() {
                    return Err(StorageError::Backend(format!(
                        "DuckLake Quack commit mode {} requires server-side ducklake_commit for {kind}, but ducklake_commit does not stage CREATE TABLE schema/table metadata for inlined CTAS",
                        self.commit_mode
                    )));
                }
                Ok(None)
            }
            _ => {
                if self.commit_mode.requires_server_side() {
                    return Err(StorageError::Backend(format!(
                        "DuckLake Quack commit mode {} requires server-side ducklake_commit for {kind}, but this commit kind is not staged by the Rust Quack backend yet",
                        self.commit_mode
                    )));
                }
                Ok(None)
            }
        }
    }

    async fn try_run_server_side_append_files(
        &self,
        files: &[AppendFile],
    ) -> Result<Option<SnapshotId>> {
        let append_plan = crate::plan_append_files_commit(files)?;
        let ids = self.load_server_side_snapshot_ids().await?;
        let metadata_tables = self.server_side_append_metadata_tables(files).await?;
        if !metadata_tables.can_stage(files) {
            return self.server_side_unavailable(
                QuackServerSideCommitKind::AppendFiles,
                "required optional metadata tables are missing",
            );
        }
        let next_row_id = self
            .server_side_next_row_id(append_plan.table_id, metadata_tables.table_stats)
            .await?;
        let snapshot_plan =
            append_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id, next_row_id, files)?;
        let metadata = self.metadata().await?;
        let metadata_schema_name = self.metadata_catalog.as_deref().unwrap_or("main");
        let batch = quack_staged_append_files_sql(QuackStagedAppendFilesSql {
            files,
            planned_files: &snapshot_plan.files,
            snapshot_ids: ids,
            data_path: &metadata.data_path,
            metadata_schema_name,
        })?;
        self.execute_server_side_commit_batch(batch).await
    }

    async fn try_run_server_side_append_inlined_data(
        &self,
        append: &InlinedDataAppendCommit,
    ) -> Result<Option<SnapshotId>> {
        crate::validate_inlined_append_rows(
            append.rows_appended,
            &append.batches,
            None,
            "DuckLake Quack server-side inlined append",
        )?;
        if !self
            .server_side_has_existing_inlined_table(append.table_id)
            .await?
        {
            return self.server_side_unavailable(
                QuackServerSideCommitKind::AppendInlinedData,
                "this commit requires creating a new inlined-data table",
            );
        }
        let ids = self.load_server_side_snapshot_ids().await?;
        if ids.snapshot_id != append.expected_snapshot.0 {
            return Err(StorageError::conflict(
                "append",
                append.expected_snapshot,
                ids.snapshot_id,
            ));
        }
        let has_table_stats = self.table_exists("ducklake_table_stats").await?;
        let row_id_start = self
            .server_side_next_row_id(append.table_id, has_table_stats)
            .await?
            .unwrap_or(0);
        let metadata = self.metadata().await?;
        let metadata_schema_name = self.metadata_catalog.as_deref().unwrap_or("main");
        let batch = quack_staged_append_inlined_data_sql(QuackStagedAppendInlinedDataSql {
            append,
            row_id_start,
            snapshot_ids: ids,
            data_path: &metadata.data_path,
            metadata_schema_name,
        })?;
        self.execute_server_side_commit_batch(batch).await
    }

    fn server_side_unavailable<T>(
        &self,
        kind: QuackServerSideCommitKind,
        reason: impl Into<String>,
    ) -> Result<Option<T>> {
        let reason = reason.into();
        if self.commit_mode.requires_server_side() {
            return Err(StorageError::Backend(format!(
                "DuckLake Quack commit mode {} requires server-side ducklake_commit for {kind}, but {reason}",
                self.commit_mode
            )));
        }
        Ok(None)
    }

    async fn try_run_server_side_delete_files(
        &self,
        files: &[DeleteFileCommit],
    ) -> Result<Option<SnapshotId>> {
        let delete_plan = crate::plan_delete_files_commit(files)?;
        let ids = self.load_server_side_snapshot_ids().await?;
        let snapshot_plan = delete_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id, files)?;
        let metadata = self.metadata().await?;
        let target_data_paths = self
            .active_data_file_paths(delete_plan.table_id, SnapshotId(ids.snapshot_id))
            .await?;
        let metadata_schema_name = self.metadata_catalog.as_deref().unwrap_or("main");
        let batch = quack_staged_delete_files_sql(QuackStagedDeleteFilesSql {
            files,
            snapshot_ids: ids,
            new_snapshot: snapshot_plan.new_snapshot,
            data_path: &metadata.data_path,
            metadata_schema_name,
            target_data_paths: &target_data_paths,
        })?;
        self.execute_server_side_commit_batch(batch).await
    }

    async fn try_run_server_side_table_changes(
        &self,
        changes: &TableChangeCommit,
    ) -> Result<Option<SnapshotId>> {
        if !quack_append_files_can_use_server_side_commit(&changes.append_files) {
            return self.server_side_unavailable(
                QuackServerSideCommitKind::TableChanges,
                "this commit contains file variant stats that cannot be staged safely yet",
            );
        }
        if !changes.inlined_data_appends.is_empty()
            && !self
                .server_side_has_existing_inlined_table(changes.table_id)
                .await?
        {
            return self.server_side_unavailable(
                QuackServerSideCommitKind::TableChanges,
                "this commit requires creating a new inlined-data table",
            );
        }
        if !changes.flushed_inlined_data.is_empty()
            || !changes.flushed_inlined_file_deletions.is_empty()
        {
            return self.server_side_unavailable(
                QuackServerSideCommitKind::TableChanges,
                "this commit contains flushed inlined rows that are not staged by the Rust Quack backend yet",
            );
        }

        let table_change_plan = crate::plan_table_change_commit(changes)?;
        let ids = self.load_server_side_snapshot_ids().await?;
        let metadata_tables = self
            .server_side_append_metadata_tables(&changes.append_files)
            .await?;
        if !metadata_tables.can_stage(&changes.append_files) {
            return self.server_side_unavailable(
                QuackServerSideCommitKind::TableChanges,
                "required optional metadata tables are missing",
            );
        }
        let next_row_id = self
            .server_side_next_row_id(changes.table_id, metadata_tables.table_stats)
            .await?;
        let snapshot_plan = table_change_plan.plan_snapshot(
            ids.snapshot_id,
            ids.next_file_id,
            next_row_id,
            changes,
        )?;
        let metadata = self.metadata().await?;
        let target_data_paths = self
            .active_data_file_paths(changes.table_id, SnapshotId(ids.snapshot_id))
            .await?;
        let metadata_schema_name = self.metadata_catalog.as_deref().unwrap_or("main");
        let batch = quack_staged_table_changes_sql(QuackStagedTableChangesSql {
            changes,
            snapshot_plan: &snapshot_plan,
            snapshot_ids: ids,
            data_path: &metadata.data_path,
            metadata_schema_name,
            target_data_paths: &target_data_paths,
        })?;
        self.execute_server_side_commit_batch(batch).await
    }

    async fn server_side_has_existing_inlined_table(&self, table_id: TableId) -> Result<bool> {
        if !self.table_exists("ducklake_inlined_data_tables").await? {
            return Ok(false);
        }
        Ok(!self
            .run_catalog_query(
                &crate::catalog_queries::inlined_data_table_names(table_id),
                crate::catalog_queries::string_value,
            )
            .await?
            .is_empty())
    }

    async fn active_data_file_paths(
        &self,
        table_id: TableId,
        snapshot_id: SnapshotId,
    ) -> Result<HashMap<i64, String>> {
        Ok(self
            .data_files_at(table_id, snapshot_id)
            .await?
            .into_iter()
            .map(|file| (file.data_file_id.0, file.path))
            .collect())
    }

    async fn execute_server_side_commit_batch(&self, batch: String) -> Result<Option<SnapshotId>> {
        self.with_write_lock(|| async move {
            let result = self.query_sql(batch).await?;
            let mut rows = result.typed_rows();
            let row = rows.next().ok_or_else(|| {
                StorageError::Backend("server-side ducklake_commit returned no rows".into())
            })?;
            let committed_snapshot_id = row.get_i64(0)?;
            let _committed_schema_version = row.get_i64(1)?;
            let had_flushes = row.get_opt_bool(2)?.unwrap_or(false);
            if had_flushes {
                self.query_sql("CALL quack_clear_cache()").await?;
            }
            Ok(Some(SnapshotId(committed_snapshot_id)))
        })
        .await
    }

    async fn load_server_side_snapshot_ids(&self) -> Result<crate::commit::CommitSnapshotIds> {
        self.run_catalog_query(
            &crate::catalog_queries::latest_snapshot_ids(),
            crate::catalog_queries::latest_snapshot_ids_value,
        )
        .await?
        .into_iter()
        .next()
        .ok_or_else(|| {
            StorageError::Backend("DuckLake server-side append found no catalog snapshot".into())
        })
    }

    async fn server_side_append_metadata_tables(
        &self,
        files: &[AppendFile],
    ) -> Result<QuackAppendMetadataTables> {
        Ok(QuackAppendMetadataTables {
            table_stats: self.table_exists("ducklake_table_stats").await?,
            file_column_stats: self.table_exists("ducklake_file_column_stats").await?,
            file_partition_values: files.iter().any(|file| !file.partition_values.is_empty())
                && self.table_exists("ducklake_file_partition_value").await?,
        })
    }

    async fn server_side_next_row_id(
        &self,
        table_id: TableId,
        has_table_stats: bool,
    ) -> Result<Option<i64>> {
        crate::stats_engine::next_row_id_for_append_from_table_stats_async_query(
            |query| async move {
                Ok(self
                    .run_catalog_query(&query, crate::catalog_queries::opt_i64_value)
                    .await?
                    .into_iter()
                    .next()
                    .flatten())
            },
            table_id,
            has_table_stats,
        )
        .await
    }

    async fn server_capabilities(&self) -> Result<QuackServerCapabilities> {
        let mut cached = self.server_capabilities.lock().await;
        if let Some(capabilities) = *cached {
            return Ok(capabilities);
        }
        let capabilities = self.probe_server_capabilities().await;
        let capabilities = capabilities.unwrap_or(QuackServerCapabilities {
            ducklake_commit: false,
        });
        *cached = Some(capabilities);
        Ok(capabilities)
    }

    async fn probe_server_capabilities(&self) -> Result<QuackServerCapabilities> {
        let result = self
            .query_sql(
                "SELECT 1 FROM duckdb_functions() WHERE function_name = 'ducklake_commit' LIMIT 1",
            )
            .await?;
        Ok(QuackServerCapabilities {
            ducklake_commit: result.typed_rows().next().is_some(),
        })
    }

    async fn query_sql_with_connection(
        &self,
        client: &QuackHttpClient,
        connection: &QuackConnection,
        sql: String,
    ) -> Result<QuackQueryResult> {
        let response = client
            .send_unvalidated(&connection.prepare_request(sql.clone()))
            .await
            .map_err(|err| StorageError::Backend(format!("Quack query {sql:?} failed: {err}")))?;
        let prepare = match response {
            QuackResponse::Prepare(prepare) => prepare,
            QuackResponse::Success(_) => return Ok(QuackQueryResult::empty()),
            QuackResponse::Error(error) => return Err(StorageError::Backend(error.message)),
            other => {
                return Err(StorageError::Backend(format!(
                    "Quack prepare request returned a {} response",
                    other.message_type().as_str()
                )));
            }
        };
        let column_names = prepare.result_names;
        let column_types = prepare.result_types;
        let result_uuid = prepare.result_uuid;
        let mut chunks = prepare.results;

        if prepare.needs_more_fetch {
            loop {
                let response = client
                    .send(&connection.fetch_request(result_uuid))
                    .await
                    .map_err(|err| {
                        StorageError::Backend(format!("Quack query {sql:?} fetch failed: {err}"))
                    })?;
                let QuackResponse::Fetch(fetch) = response else {
                    return Err(StorageError::Backend(
                        "Quack fetch request returned a non-fetch response".into(),
                    ));
                };
                if fetch.results.is_empty() {
                    break;
                }
                chunks.extend(fetch.results);
            }
        }

        QuackQueryResult::from_chunks(column_names, column_types, chunks)
    }

    async fn table_exists(&self, table_name: &str) -> Result<bool> {
        let catalog_filter = self
            .metadata_catalog
            .as_deref()
            .map(|catalog| format!(" AND table_catalog = {}", sql_string_literal(catalog)))
            .unwrap_or_default();
        let result = self
            .query_sql(crate::catalog_table_exists::quack_table_exists_count_sql(
                &sql_string_literal(table_name),
                &catalog_filter,
            ))
            .await?;
        let Some(row) = result.typed_rows().next() else {
            return Ok(false);
        };
        Ok(row.get_i64(0)? > 0)
    }

    fn qualify_metadata_sql(&self, sql: String) -> String {
        self.metadata_catalog
            .as_deref()
            .map(|catalog| quack_qualify_metadata_sql(&sql, catalog))
            .unwrap_or(sql)
    }

    fn metadata_table_identifier(&self, table_name: &str) -> String {
        let table = quack_quote_identifier(table_name);
        self.metadata_catalog
            .as_deref()
            .map(|catalog| format!("{}.{table}", quack_quote_identifier(catalog)))
            .unwrap_or(table)
    }

    fn dynamic_metadata_table_identifier(&self, table_name: &str) -> String {
        let Some(catalog) = self.metadata_catalog.as_deref() else {
            return quack_quote_identifier(table_name);
        };

        let quoted_catalog = quack_quote_identifier(catalog);
        let quoted_prefix = format!("{quoted_catalog}.");
        if let Some(table) = table_name.strip_prefix(&quoted_prefix)
            && is_quack_dynamic_metadata_table_identifier(table)
        {
            return format!("{quoted_catalog}.{}", quack_quote_identifier(table));
        }

        if is_quack_unquoted_identifier(catalog) {
            let bare_prefix = format!("{catalog}.");
            if let Some(table) = table_name.strip_prefix(&bare_prefix)
                && is_quack_dynamic_metadata_table_identifier(table)
            {
                return format!("{quoted_catalog}.{}", quack_quote_identifier(table));
            }
        }

        self.metadata_table_identifier(table_name)
    }
}

fn quack_error_invalidates_connection(err: &StorageError) -> bool {
    let StorageError::Backend(message) = err else {
        return false;
    };
    message.starts_with("Quack query ")
        && (message.contains("failed to send Quack request")
            || message.contains("failed to read Quack response body")
            || message.contains("Quack server returned HTTP status")
            || message.contains("truncated Quack DuckDB binary message")
            || message.contains("trailing bytes in Quack binary message"))
}

#[derive(Debug, Clone, Copy)]
struct QuackAppendMetadataTables {
    table_stats: bool,
    file_column_stats: bool,
    file_partition_values: bool,
}

impl QuackAppendMetadataTables {
    fn can_stage(self, files: &[AppendFile]) -> bool {
        files
            .iter()
            .all(|file| file.column_stats.is_empty() || self.file_column_stats)
            && files
                .iter()
                .all(|file| file.partition_values.is_empty() || self.file_partition_values)
    }
}

struct QuackStagedAppendFilesSql<'a> {
    files: &'a [AppendFile],
    planned_files: &'a [crate::AppendFilePlan],
    snapshot_ids: crate::commit::CommitSnapshotIds,
    data_path: &'a str,
    metadata_schema_name: &'a str,
}

struct QuackStagedDeleteFilesSql<'a> {
    files: &'a [DeleteFileCommit],
    snapshot_ids: crate::commit::CommitSnapshotIds,
    new_snapshot: SnapshotId,
    data_path: &'a str,
    metadata_schema_name: &'a str,
    target_data_paths: &'a HashMap<i64, String>,
}

struct QuackStagedAppendInlinedDataSql<'a> {
    append: &'a InlinedDataAppendCommit,
    row_id_start: i64,
    snapshot_ids: crate::commit::CommitSnapshotIds,
    data_path: &'a str,
    metadata_schema_name: &'a str,
}

struct QuackStagedTableChangesSql<'a> {
    changes: &'a TableChangeCommit,
    snapshot_plan: &'a crate::TableChangeSnapshotPlan,
    snapshot_ids: crate::commit::CommitSnapshotIds,
    data_path: &'a str,
    metadata_schema_name: &'a str,
    target_data_paths: &'a HashMap<i64, String>,
}

const QUACK_STAGED_TABLES: &[(&str, &str)] = &[
    (
        "ducklake_staged_commit",
        "commit_author VARCHAR, commit_message VARCHAR, commit_extra_info VARCHAR, \
         transaction_snapshot_id BIGINT, data_path VARCHAR, path_separator VARCHAR, \
         transaction_next_catalog_id UBIGINT, transaction_next_file_id UBIGINT",
    ),
    (
        "ducklake_staged_data_file",
        "data_file_id BIGINT, table_id BIGINT, file_order BIGINT, \
         path VARCHAR, path_is_relative BOOLEAN, file_format VARCHAR, \
         record_count BIGINT, file_size_bytes BIGINT, footer_size BIGINT, \
         row_id_start BIGINT, partition_id BIGINT, encryption_key VARCHAR, \
         mapping_id UBIGINT, partial_max BIGINT, begin_snapshot BIGINT, \
         compaction_id BIGINT",
    ),
    (
        "ducklake_staged_data_file_column_stats",
        "data_file_id BIGINT, table_id BIGINT, column_id BIGINT, \
         column_size_bytes BIGINT, has_num_values BOOLEAN, num_values BIGINT, \
         has_null_count BOOLEAN, null_count BIGINT, has_min BOOLEAN, min_value VARCHAR, \
         has_max BOOLEAN, max_value VARCHAR, has_contains_nan BOOLEAN, contains_nan BOOLEAN, \
         any_valid BOOLEAN, extra_stats VARCHAR",
    ),
    (
        "ducklake_staged_data_file_partition",
        "local_file_id BIGINT, partition_column_idx BIGINT, partition_value VARCHAR",
    ),
    (
        "ducklake_staged_delete_file",
        "table_id BIGINT, data_file_path VARCHAR, data_file_id BIGINT, \
         file_name VARCHAR, format VARCHAR, delete_count BIGINT, \
         file_size_bytes BIGINT, footer_size BIGINT, encryption_key VARCHAR, \
         begin_snapshot BIGINT, max_snapshot BIGINT, source VARCHAR, \
         overwrites_existing_delete BOOLEAN, overwrite_delete_file_id BIGINT, \
         overwrite_delete_file_path VARCHAR, attached_local_file_id BIGINT",
    ),
    (
        "ducklake_staged_inlined_data",
        "table_id BIGINT, has_preserved_row_ids BOOLEAN",
    ),
    (
        "ducklake_staged_inlined_row",
        "table_id BIGINT, row_order BIGINT, preserved_row_id BIGINT, value_literals VARCHAR",
    ),
    (
        "ducklake_staged_inlined_column_stats",
        "table_id BIGINT, column_id BIGINT, column_size_bytes BIGINT, \
         has_num_values BOOLEAN, num_values BIGINT, has_null_count BOOLEAN, \
         null_count BIGINT, has_min BOOLEAN, min_value VARCHAR, has_max BOOLEAN, \
         max_value VARCHAR, has_contains_nan BOOLEAN, contains_nan BOOLEAN, \
         any_valid BOOLEAN, extra_stats VARCHAR",
    ),
    (
        "ducklake_staged_inlined_delete",
        "table_id BIGINT, inlined_table_name VARCHAR, deleted_row_id BIGINT",
    ),
    (
        "ducklake_staged_inlined_file_delete",
        "table_id BIGINT, file_id BIGINT, deleted_row_id BIGINT",
    ),
    (
        "ducklake_staged_dropped_file",
        "path VARCHAR, data_file_id BIGINT",
    ),
    ("ducklake_staged_tables_deleted_from", "table_id BIGINT"),
    (
        "ducklake_staged_flushed_inlined",
        "inlined_table_name VARCHAR, schema_version BIGINT, flush_snapshot_id BIGINT",
    ),
    (
        "ducklake_staged_compaction",
        "compaction_id BIGINT, table_id BIGINT, compaction_type VARCHAR, \
         row_id_start BIGINT, output_local_file_id BIGINT",
    ),
    (
        "ducklake_staged_compaction_source",
        "compaction_id BIGINT, source_order BIGINT, source_data_file_id BIGINT, \
         source_path VARCHAR, source_row_count BIGINT, source_begin_snapshot BIGINT, \
         source_max_partial_file_snapshot BIGINT, source_inlined_delete_count BIGINT, \
         last_delete_file_id BIGINT, last_delete_path VARCHAR, \
         last_delete_row_count BIGINT, last_delete_end_snapshot BIGINT",
    ),
    (
        "ducklake_staged_name_map",
        "map_id UBIGINT, table_id BIGINT",
    ),
    (
        "ducklake_staged_name_map_entry",
        "map_id UBIGINT, entry_id BIGINT, parent_entry_id BIGINT, \
         source_name VARCHAR, target_field_id BIGINT, hive_partition BOOLEAN",
    ),
];

fn quack_append_files_can_use_server_side_commit(files: &[AppendFile]) -> bool {
    files.iter().all(|file| file.variant_stats.is_empty())
}

fn quack_staged_commit_prefix(
    snapshot_ids: crate::commit::CommitSnapshotIds,
    data_path: &str,
    commit_message: &str,
) -> String {
    let mut sql = String::new();
    for (table, columns) in QUACK_STAGED_TABLES {
        sql.push_str("CREATE TEMPORARY TABLE IF NOT EXISTS ");
        sql.push_str(table);
        sql.push('(');
        sql.push_str(columns);
        sql.push_str(");");
    }
    for (table, _) in QUACK_STAGED_TABLES {
        sql.push_str("TRUNCATE ");
        sql.push_str(table);
        sql.push(';');
    }

    sql.push_str(&format!(
        "INSERT INTO ducklake_staged_commit VALUES \
         (NULL, {}, NULL, {}, {}, '/', {}, {});",
        sql_string_literal(commit_message),
        snapshot_ids.snapshot_id,
        sql_string_literal(data_path),
        snapshot_ids.next_catalog_id,
        snapshot_ids.next_file_id,
    ));
    sql
}

fn quack_staged_commit_suffix(metadata_schema_name: &str, schema_version: i64) -> String {
    format!(
        "SELECT * FROM ducklake_commit({}, {}, \
         max_retry_count => 10, retry_wait_ms => 100, retry_backoff => 1.500000);",
        sql_string_literal(metadata_schema_name),
        schema_version,
    )
}

fn quack_staged_append_files_sql(input: QuackStagedAppendFilesSql<'_>) -> Result<String> {
    if input.files.len() != input.planned_files.len() {
        return Err(StorageError::Backend(format!(
            "DuckLake staged append planned {} files for {} input files",
            input.planned_files.len(),
            input.files.len()
        )));
    }

    let mut sql =
        quack_staged_commit_prefix(input.snapshot_ids, input.data_path, "datafusion append");

    for (idx, (file, plan)) in input
        .files
        .iter()
        .zip(input.planned_files.iter())
        .enumerate()
    {
        let local_file_id = i64::try_from(idx).map_err(|_| {
            StorageError::Backend("DuckLake staged append file index overflow".into())
        })?;
        sql.push_str(&quack_staged_data_file_sql(local_file_id, idx, file, plan)?);
        for stats in &file.column_stats {
            sql.push_str(&quack_staged_file_column_stats_sql(local_file_id, stats));
        }
        for partition in &file.partition_values {
            sql.push_str(&format!(
                "INSERT INTO ducklake_staged_data_file_partition VALUES ({}, {}, {});",
                local_file_id,
                partition.partition_key_index,
                quack_opt_string_sql(partition.partition_value.as_deref()),
            ));
        }
    }

    sql.push_str(&quack_staged_commit_suffix(
        input.metadata_schema_name,
        input.snapshot_ids.schema_version,
    ));
    Ok(sql)
}

fn quack_staged_delete_files_sql(input: QuackStagedDeleteFilesSql<'_>) -> Result<String> {
    let mut sql =
        quack_staged_commit_prefix(input.snapshot_ids, input.data_path, "datafusion delete");
    for file in input.files {
        let data_file_path = input
            .target_data_paths
            .get(&file.data_file_id.0)
            .ok_or_else(|| {
                StorageError::Backend(format!(
                    "DuckLake staged delete could not find active data file {}",
                    file.data_file_id.0
                ))
            })?;
        sql.push_str(&quack_staged_delete_file_sql(
            file,
            input.new_snapshot,
            data_file_path,
            None,
        )?);
    }
    sql.push_str(&quack_staged_commit_suffix(
        input.metadata_schema_name,
        input.snapshot_ids.schema_version,
    ));
    Ok(sql)
}

fn quack_staged_append_inlined_data_sql(
    input: QuackStagedAppendInlinedDataSql<'_>,
) -> Result<String> {
    let mut sql = quack_staged_commit_prefix(
        input.snapshot_ids,
        input.data_path,
        &input.append.commit_message,
    );
    sql.push_str(&quack_staged_inlined_append_rows_sql(
        input.append.table_id,
        &input.append.schema,
        &input.append.batches,
        None,
        input.append.rows_appended,
        input.row_id_start,
        "DuckLake staged Quack inlined append",
    )?);
    sql.push_str(&quack_staged_commit_suffix(
        input.metadata_schema_name,
        input.snapshot_ids.schema_version,
    ));
    Ok(sql)
}

fn quack_staged_table_changes_sql(input: QuackStagedTableChangesSql<'_>) -> Result<String> {
    if input.changes.append_files.len() != input.snapshot_plan.append_files.len() {
        return Err(StorageError::Backend(format!(
            "DuckLake staged table-change planned {} data files for {} input files",
            input.snapshot_plan.append_files.len(),
            input.changes.append_files.len()
        )));
    }
    if input.changes.inlined_data_appends.len() != input.snapshot_plan.inlined_data_appends.len() {
        return Err(StorageError::Backend(format!(
            "DuckLake staged table-change planned {} inlined appends for {} input appends",
            input.snapshot_plan.inlined_data_appends.len(),
            input.changes.inlined_data_appends.len()
        )));
    }

    let mut sql = quack_staged_commit_prefix(
        input.snapshot_ids,
        input.data_path,
        &input.changes.commit_message,
    );
    let mut local_file_id = 0_i64;

    for (idx, (file, plan)) in input
        .changes
        .append_files
        .iter()
        .zip(input.snapshot_plan.append_files.iter())
        .enumerate()
    {
        sql.push_str(&quack_staged_data_file_sql(local_file_id, idx, file, plan)?);
        for stats in &file.column_stats {
            sql.push_str(&quack_staged_file_column_stats_sql(local_file_id, stats));
        }
        for partition in &file.partition_values {
            sql.push_str(&format!(
                "INSERT INTO ducklake_staged_data_file_partition VALUES ({}, {}, {});",
                local_file_id,
                partition.partition_key_index,
                quack_opt_string_sql(partition.partition_value.as_deref()),
            ));
        }
        local_file_id = local_file_id.checked_add(1).ok_or_else(|| {
            StorageError::Backend("DuckLake staged table-change file id overflow".into())
        })?;
    }

    for file in &input.changes.delete_files {
        let data_file_path = input
            .target_data_paths
            .get(&file.data_file_id.0)
            .ok_or_else(|| {
                StorageError::Backend(format!(
                    "DuckLake staged table-change could not find active data file {}",
                    file.data_file_id.0
                ))
            })?;
        sql.push_str(&quack_staged_delete_file_sql(
            file,
            input.snapshot_plan.new_snapshot,
            data_file_path,
            None,
        )?);
    }

    for (append, plan) in input
        .changes
        .inlined_data_appends
        .iter()
        .zip(input.snapshot_plan.inlined_data_appends.iter())
    {
        sql.push_str(&quack_staged_inlined_append_sql(
            input.changes.table_id,
            append,
            plan,
        )?);
    }

    for deletion in &input.changes.inlined_data_deletions {
        for row_id in deletion.validated_row_ids()? {
            sql.push_str(&format!(
                "INSERT INTO ducklake_staged_inlined_delete VALUES ({}, {}, {});",
                input.changes.table_id.0,
                sql_string_literal(&deletion.table_name),
                row_id,
            ));
        }
    }

    for deletion in &input.changes.inlined_file_deletions {
        for row_id in deletion.validated_row_ids()? {
            sql.push_str(&format!(
                "INSERT INTO ducklake_staged_inlined_file_delete VALUES ({}, {}, {});",
                deletion.table_id.0, deletion.data_file_id.0, row_id,
            ));
        }
    }

    sql.push_str(&quack_staged_commit_suffix(
        input.metadata_schema_name,
        input.snapshot_ids.schema_version,
    ));
    Ok(sql)
}

fn quack_staged_inlined_append_sql(
    table_id: TableId,
    append: &InlinedDataAppend,
    plan: &crate::InlinedDataAppendPlan,
) -> Result<String> {
    quack_staged_inlined_append_rows_sql(
        table_id,
        &append.schema,
        &append.batches,
        append.row_ids.as_deref(),
        append.rows_appended,
        plan.row_id_start,
        "DuckLake staged Quack inlined append",
    )
}

fn quack_staged_inlined_append_rows_sql(
    table_id: TableId,
    schema: &TableSchema,
    batches: &[arrow::record_batch::RecordBatch],
    preserved_row_ids: Option<&[i64]>,
    rows_appended: i64,
    row_id_start: i64,
    context: &str,
) -> Result<String> {
    crate::validate_inlined_append_rows(rows_appended, batches, preserved_row_ids, context)?;
    let mut sql = format!(
        "INSERT INTO ducklake_staged_inlined_data VALUES ({}, {});",
        table_id.0,
        quack_sql_bool(preserved_row_ids.is_some()),
    );
    let mut row_order = 0_i64;
    let mut global_row_index = 0_usize;
    for batch in batches {
        if batch.num_columns() != schema.columns.len() {
            return Err(StorageError::Backend(
                "DuckLake staged Quack inlined append batch width does not match schema".into(),
            ));
        }
        for row in 0..batch.num_rows() {
            let mut values = Vec::with_capacity(batch.num_columns());
            for (field, array) in schema.columns.iter().zip(batch.columns()) {
                values.push(quack_inlined_sql_value(field, array.as_ref(), row)?);
            }
            let preserved_row_id = preserved_row_ids
                .map(|row_ids| row_ids[global_row_index])
                .unwrap_or_else(|| row_id_start + row_order);
            let preserved_row_id = preserved_row_ids
                .map(|_| quack_opt_i64_sql(Some(preserved_row_id)))
                .unwrap_or_else(|| "NULL".into());
            let tuple = format!("({})", values.join(", "));
            sql.push_str(&format!(
                "INSERT INTO ducklake_staged_inlined_row VALUES ({}, {}, {}, {});",
                table_id.0,
                row_order,
                preserved_row_id,
                sql_string_literal(&tuple),
            ));
            row_order = row_order.checked_add(1).ok_or_else(|| {
                StorageError::Backend(
                    "DuckLake staged Quack inlined append row order overflow".into(),
                )
            })?;
            global_row_index = global_row_index.checked_add(1).ok_or_else(|| {
                StorageError::Backend(
                    "DuckLake staged Quack inlined append row index overflow".into(),
                )
            })?;
        }
    }
    Ok(sql)
}

fn quack_staged_data_file_sql(
    local_file_id: i64,
    file_order: usize,
    file: &AppendFile,
    plan: &crate::AppendFilePlan,
) -> Result<String> {
    let file_order = i64::try_from(file_order)
        .map_err(|_| StorageError::Backend("DuckLake staged append file order overflow".into()))?;
    Ok(format!(
        "INSERT INTO ducklake_staged_data_file VALUES \
         ({}, {}, {}, {}, false, 'parquet', {}, {}, {}, {}, {}, {}, {}, {}, {}, NULL);",
        local_file_id,
        file.table_id.0,
        file_order,
        sql_string_literal(&file.path),
        file.record_count,
        file.file_size_bytes,
        quack_opt_i64_sql(file.footer_size),
        quack_opt_i64_sql(plan.row_id_start),
        quack_opt_i64_sql(file.partition_id.map(|id| id.0)),
        quack_opt_string_sql(file.encryption_key.as_deref()),
        quack_opt_i64_sql(file.mapping_id.map(|id| id.0)),
        quack_opt_i64_sql(file.partial_max.map(|snapshot| snapshot.0)),
        plan.begin_snapshot,
    ))
}

fn quack_staged_file_column_stats_sql(local_file_id: i64, stats: &FileColumnStats) -> String {
    format!(
        "INSERT INTO ducklake_staged_data_file_column_stats VALUES \
         ({}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {});",
        local_file_id,
        stats.table_id.0,
        stats.column_id.0,
        quack_opt_i64_sql(stats.column_size_bytes),
        quack_sql_bool(stats.value_count.is_some()),
        quack_opt_i64_sql(stats.value_count),
        quack_sql_bool(stats.null_count.is_some()),
        quack_opt_i64_sql(stats.null_count),
        quack_sql_bool(stats.min_value.is_some()),
        quack_opt_string_sql(stats.min_value.as_deref()),
        quack_sql_bool(stats.max_value.is_some()),
        quack_opt_string_sql(stats.max_value.as_deref()),
        quack_sql_bool(stats.contains_nan.is_some()),
        quack_opt_bool_sql(stats.contains_nan),
        quack_sql_bool(stats.value_count.unwrap_or(0) > stats.null_count.unwrap_or(0)),
        quack_opt_string_sql(stats.extra_stats.as_deref()),
    )
}

fn quack_staged_delete_file_sql(
    file: &DeleteFileCommit,
    new_snapshot: SnapshotId,
    data_file_path: &str,
    attached_local_file_id: Option<i64>,
) -> Result<String> {
    let begin_snapshot = file
        .begin_snapshot
        .map(|snapshot| snapshot.0)
        .unwrap_or(new_snapshot.0);
    let format = quack_delete_format_sql(&file.format)?;
    let overwrites_existing_delete = file.overwritten_delete_file_id.is_some();
    Ok(format!(
        "INSERT INTO ducklake_staged_delete_file VALUES \
         ({}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, 'REGULAR', {}, {}, NULL, {});",
        file.table_id.0,
        quack_opt_string_sql(Some(data_file_path)),
        file.data_file_id.0,
        quack_opt_string_sql(Some(&file.path)),
        quack_opt_string_sql(Some(format)),
        file.delete_count,
        file.file_size_bytes,
        quack_opt_i64_sql(file.footer_size),
        "NULL",
        begin_snapshot,
        quack_opt_i64_sql(file.partial_max.map(|snapshot| snapshot.0)),
        quack_sql_bool(overwrites_existing_delete),
        quack_opt_i64_sql(file.overwritten_delete_file_id.map(|id| id.0)),
        quack_opt_i64_sql(attached_local_file_id),
    ))
}

fn quack_delete_format_sql(format: &ducklake_catalog::DeleteFormat) -> Result<&'static str> {
    match format {
        ducklake_catalog::DeleteFormat::Parquet => Ok("parquet"),
        ducklake_catalog::DeleteFormat::Puffin => Ok("puffin"),
        ducklake_catalog::DeleteFormat::Unknown(format) => Err(StorageError::Unsupported(format!(
            "DuckLake delete file format {format} is not supported"
        ))),
    }
}

fn quack_opt_i64_sql(value: Option<i64>) -> String {
    value
        .map(|value| value.to_string())
        .unwrap_or_else(|| "NULL".into())
}

fn quack_opt_bool_sql(value: Option<bool>) -> String {
    value.map(quack_sql_bool).unwrap_or("NULL").to_string()
}

fn quack_opt_string_sql(value: Option<&str>) -> String {
    value
        .map(sql_string_literal)
        .unwrap_or_else(|| "NULL".into())
}

impl fmt::Debug for QuackCatalog {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("QuackCatalog")
            .field("uri", &self.uri)
            .field("token", &"<redacted>")
            .field("metadata_catalog", &self.metadata_catalog)
            .field("commit_mode", &self.commit_mode)
            .field("client", &self.client)
            .finish()
    }
}

#[async_trait]
impl AsyncReadQueryExecutor for QuackCatalog {
    fn read_query_render(&self) -> crate::catalog_async_query_methods::ReadQueryRender {
        crate::catalog_async_query_methods::ReadQueryRender::Quack
    }

    fn read_query_metadata_catalog(&self) -> Option<&str> {
        self.metadata_catalog()
    }

    async fn execute_rendered_read_query<T>(
        &self,
        sql: String,
        _params: Vec<i64>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send + 'static,
    {
        quack_collect_rows(self.query_sql(sql).await?, mapper)
    }
}

fn quack_catalog_write_lock(catalog: &QuackCatalog) -> Result<Arc<tokio::sync::Mutex<()>>> {
    let key = QuackWriteLockKey {
        http_url: catalog.uri.http_url().to_string(),
        metadata_catalog: catalog.metadata_catalog.clone(),
    };
    let locks = QUACK_CATALOG_WRITE_LOCKS.get_or_init(|| StdMutex::new(HashMap::new()));
    let mut locks = locks
        .lock()
        .map_err(|_| StorageError::Backend("Quack catalog write lock registry poisoned".into()))?;
    Ok(locks
        .entry(key)
        .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
        .clone())
}

fn quack_collect<T, F>(result: QuackQueryResult, mut mapper: F) -> Result<Vec<T>>
where
    F: FnMut(QuackRow<'_>) -> Result<T>,
{
    result.typed_rows().map(&mut mapper).collect()
}

/// Decode a Quack result set with a shared [`crate::embedded_reads`] mapper.
fn quack_collect_rows<T>(
    result: QuackQueryResult,
    map: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
) -> Result<Vec<T>> {
    crate::embedded_rows::collect_embedded_rows(result.typed_rows(), map)
}

async fn quack_execute_catalog_command(
    catalog: &QuackCatalog,
    query: &crate::catalog_queries::CatalogQuery,
) -> Result<usize> {
    let affected =
        match query.render_quack_affected_count_with_metadata_catalog(catalog.metadata_catalog()) {
            Some(count_sql) => Some(quack_single_i64(catalog, count_sql).await?),
            None => None,
        };
    let sql = query.render_quack_with_metadata_catalog(catalog.metadata_catalog());
    catalog.query_sql(sql).await?;
    usize::try_from(affected.unwrap_or(0)).map_err(|_| {
        StorageError::Backend("Quack affected-row probe returned a negative count".into())
    })
}

#[async_trait]
impl AsyncCatalogQueryExecutor for QuackCatalog {
    fn catalog_query_render(&self) -> crate::catalog_async_query_methods::CatalogQueryRender {
        crate::catalog_async_query_methods::CatalogQueryRender::Quack
    }

    fn catalog_query_metadata_catalog(&self) -> Option<&str> {
        self.metadata_catalog()
    }

    async fn execute_rendered_catalog_query<T>(
        &self,
        sql: String,
        _params: Vec<crate::catalog_queries::CatalogParam>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send,
    {
        quack_collect_rows(self.query_sql(sql).await?, mapper)
    }

    async fn execute_rendered_catalog_command(
        &self,
        sql: String,
        _params: Vec<crate::catalog_queries::CatalogParam>,
    ) -> Result<usize> {
        self.query_sql(sql).await?;
        Ok(0)
    }
}

#[async_trait]
impl AsyncCatalogQueryExecutor for &QuackCatalog {
    fn catalog_query_render(&self) -> crate::catalog_async_query_methods::CatalogQueryRender {
        (**self).catalog_query_render()
    }

    fn catalog_query_metadata_catalog(&self) -> Option<&str> {
        (**self).catalog_query_metadata_catalog()
    }

    async fn execute_rendered_catalog_query<T>(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send,
    {
        (**self)
            .execute_rendered_catalog_query(sql, params, mapper)
            .await
    }

    async fn execute_rendered_catalog_command(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
    ) -> Result<usize> {
        (**self).execute_rendered_catalog_command(sql, params).await
    }
}

fn quack_inlined_sql_value(field: &SchemaField, array: &dyn Array, row: usize) -> Result<String> {
    if matches!(
        field.ducklake_type,
        DuckLakeType::Struct | DuckLakeType::List | DuckLakeType::Map
    ) {
        return crate::inlined_values::inlined_duckdb_literal_value(
            field,
            array,
            row,
            "Quack inlined append",
            crate::inlined_values::InlinedIntervalFormat::MicrosecondsOnly("Quack"),
        );
    }
    crate::inlined_values::encode_inlined_scalar_value(
        field,
        array,
        row,
        "Quack inlined append",
        crate::inlined_values::InlinedIntervalFormat::MicrosecondsOnly("Quack"),
        &QuackInlinedValueEncoder,
    )
}

struct QuackInlinedValueEncoder;

impl_inlined_scalar_value_encoder!(
    impl QuackInlinedValueEncoder,
    Value = String,
    {
        null(_field) => Ok("NULL".into()),
        boolean(value) => Ok(quack_sql_bool(value).into()),
        int8(value) => Ok(value.to_string()),
        int16(value) => Ok(value.to_string()),
        int32(value) => Ok(value.to_string()),
        int64(value) => Ok(value.to_string()),
        uint8(value) => Ok(value.to_string()),
        uint16(value) => Ok(value.to_string()),
        uint32(value) => Ok(value.to_string()),
        uint64(field, value) => quack_inlined_casted_value(field, &value.to_string()),
        float32(value) => quack_inlined_float_value(value.is_finite(), value.to_string(), "FLOAT"),
        float64(value) => quack_inlined_float_value(value.is_finite(), value.to_string(), "DOUBLE"),
        text(field, value) => quack_inlined_text_value(field, value),
        bytes(_field, value) => Ok(format!(
            "from_hex({})",
            sql_string_literal(&crate::hex_encode_bytes(&value))
        )),
    }
);

fn quack_inlined_text_value(field: &SchemaField, value: String) -> Result<String> {
    match field.ducklake_type {
        DuckLakeType::Varchar | DuckLakeType::Json => Ok(sql_string_literal(&value)),
        DuckLakeType::Uuid => Ok(format!("CAST({} AS UUID)", sql_string_literal(&value))),
        _ => quack_inlined_casted_value(field, &value),
    }
}

fn quack_inlined_float_value(
    is_finite: bool,
    rendered: String,
    float_type: &str,
) -> Result<String> {
    if is_finite {
        Ok(rendered)
    } else {
        // NaN / +/-Infinity are not valid bare numeric literals in a DuckDB VALUES
        // list; render them as a typed cast instead (DuckDB parses 'NaN', 'inf',
        // '-inf' when given a float target type).
        Ok(sql_cast_string_literal(&rendered, float_type))
    }
}

fn quack_inlined_casted_value(field: &SchemaField, value: &str) -> Result<String> {
    Ok(sql_cast_string_literal(
        value,
        &inlined_column_type(field, InlinedSqlDialect::Quack)?,
    ))
}

fn quack_remove_scheduled_files_sql(data_file_ids: &[i64]) -> Option<String> {
    if data_file_ids.is_empty() {
        return None;
    }
    Some(format!(
        "DELETE FROM ducklake_files_scheduled_for_deletion WHERE data_file_id IN ({})",
        join_i64s(data_file_ids)
    ))
}

async fn quack_latest_snapshot_id(catalog: &QuackCatalog) -> Result<i64> {
    quack_single_i64(
        catalog,
        "SELECT COALESCE(MAX(snapshot_id), -9223372036854775808) FROM ducklake_snapshot".into(),
    )
    .await
}

async fn quack_abort_write_transaction<T>(
    catalog: &QuackCatalog,
    context: &str,
    operation: Option<&str>,
    expected_snapshot: Option<SnapshotId>,
    err: StorageError,
) -> Result<T> {
    let rollback = catalog.query_sql("ROLLBACK").await;
    if !err.is_conflict()
        && let (Some(operation), Some(expected_snapshot)) = (operation, expected_snapshot)
        && let Ok(found_snapshot) = quack_latest_snapshot_id(catalog).await
        && found_snapshot != expected_snapshot.0
    {
        return Err(StorageError::conflict(
            operation,
            expected_snapshot,
            found_snapshot,
        ));
    }
    if let Err(rollback_err) = rollback {
        return Err(StorageError::Backend(format!(
            "{context} failed: {err}; rollback failed: {rollback_err}"
        )));
    }
    Err(err)
}

struct QuackCommitOps;

#[async_trait]
impl crate::catalog_async_maintenance_methods::AsyncCleanupSqlExecutor for QuackCatalog {
    fn metadata_table_ref(&self, table: &str) -> String {
        table.to_string()
    }

    fn i64_placeholder(&self, index: usize) -> String {
        format!("?{index}")
    }

    async fn cleanup_table_exists(&self, table: &str) -> Result<bool> {
        self.table_exists(table).await
    }

    async fn cleanup_path_context_available(&self) -> Result<bool> {
        Ok(true)
    }

    async fn cleanup_data_path(&self) -> Result<String> {
        Ok(self.metadata().await?.data_path)
    }

    async fn query_i64_rows<T, F>(
        &self,
        sql: String,
        params: Vec<i64>,
        mut mapper: F,
    ) -> Result<Vec<T>>
    where
        T: Send + 'static,
        F: FnMut(&dyn crate::embedded_rows::EmbeddedRow) -> Result<T> + Send,
    {
        let sql = crate::embedded_reads::render_placeholders_as_literals(&sql, &params);
        let result = self.query_sql(sql).await?;
        quack_collect(result, |row| mapper(&row))
    }

    async fn insert_scheduled_cleanup_file(&self, file: CleanupMetadataFile) -> Result<()> {
        self.query_sql(
            crate::embedded_maintenance::insert_scheduled_cleanup_file_sql(
                &self.metadata_table_ref("ducklake_files_scheduled_for_deletion"),
                &file.id.to_string(),
                &sql_string_literal(&file.path),
                quack_sql_bool(file.path_is_relative),
                "CURRENT_TIMESTAMP",
            ),
        )
        .await?;
        Ok(())
    }
}

#[async_trait]
impl crate::catalog_async_inlined_mutation_methods::AsyncInlinedMutationExecutor for QuackCatalog {
    fn inlined_sql_dialect(&self) -> crate::inlined_values::InlinedSqlDialect {
        InlinedSqlDialect::Quack
    }

    fn inlined_metadata_table_identifier(&self, table_name: &str) -> String {
        self.metadata_table_identifier(table_name)
    }

    fn inlined_data_table_identifier(&self, table_name: &str) -> String {
        self.dynamic_metadata_table_identifier(table_name)
    }

    async fn inlined_table_exists(&self, table_name: String) -> Result<bool> {
        self.table_exists(&table_name).await
    }

    async fn execute_inlined_sql(&self, sql: String) -> Result<()> {
        self.query_sql(sql).await?;
        Ok(())
    }

    async fn query_inlined_i64(&self, sql: String) -> Result<i64> {
        quack_single_i64(self, sql).await
    }

    async fn query_optional_inlined_string(&self, sql: String) -> Result<Option<String>> {
        self.query_sql(sql)
            .await?
            .typed_rows()
            .next()
            .map(|row| row.get_string(0))
            .transpose()
    }
}

#[async_trait]
impl crate::catalog_async_inlined_mutation_methods::AsyncInlinedBatchInsertExecutor
    for QuackCatalog
{
    type Value = String;
    type InsertCommand = String;

    fn prepare_inlined_insert(&self, table_name: &str, _schema: &TableSchema) -> Result<String> {
        Ok(self.dynamic_metadata_table_identifier(table_name))
    }

    fn inlined_insert_i64_value(&self, value: i64) -> Self::Value {
        value.to_string()
    }

    fn inlined_insert_null_value(&self) -> Self::Value {
        "NULL".to_string()
    }

    fn inlined_insert_value(
        &self,
        field: &SchemaField,
        array: &dyn Array,
        row: usize,
    ) -> Result<Self::Value> {
        quack_inlined_sql_value(field, array, row)
    }

    async fn execute_prepared_inlined_insert(
        &self,
        command: &Self::InsertCommand,
        values: Vec<Self::Value>,
    ) -> Result<()> {
        self.query_sql(crate::inlined_values::inlined_insert_sql(
            command,
            &values.join(", "),
        ))
        .await?;
        Ok(())
    }
}

impl_async_catalog_query_commit_executor!(
    impl QuackCommitOps, QuackCatalog {
        execute = |_this, catalog, query| quack_execute_catalog_command(catalog, query),
        query = |_this, catalog, query, mapper| catalog.run_catalog_query(query, mapper),
        table_exists = |_this, catalog, table_name| catalog.table_exists(table_name),
    }
);

impl_async_commit_ops!(
    impl QuackCommitOps, QuackCatalog {
        latest_snapshot_ids = |this, catalog| crate::catalog_async_commit_methods::load_latest_snapshot_ids(this, catalog, "commit"),
        table_exists = |_this, catalog, table_name| catalog.table_exists(table_name),
        insert_snapshot = |_this, catalog, snapshot_id, schema_version, next_catalog_id, next_file_id| async {
            catalog
                .execute_catalog_command(&crate::catalog_queries::insert_snapshot_current_timestamp(
                    snapshot_id,
                    schema_version,
                    next_catalog_id,
                    next_file_id,
                ))
                .await?;
            Ok(())
        },
        insert_snapshot_change = |_this, catalog, snapshot_id, changes_made, commit_message| async {
            catalog
                .execute_catalog_command(&crate::catalog_queries::insert_snapshot_change(
                    snapshot_id,
                    changes_made,
                    commit_message,
                ))
                .await?;
            Ok(())
        },
    }
);

impl_async_table_data_commit_executor!(
    impl QuackCommitOps, QuackCatalog {
        dialect = crate::inlined_values::InlinedSqlDialect::Quack,
        upsert_table_column_stats = |_this, catalog, stats| {
            upsert_quack_table_column_stats(catalog, &stats)
        },
        inlined_executor = |_this, catalog| catalog,
        replace_overwritten_delete_file = |this, catalog, file, new_snapshot| async move {
            crate::catalog_async_commit_methods::replace_overwritten_delete_file(
                this,
                catalog,
                file,
                new_snapshot,
                || async move { Ok(catalog.metadata().await?.data_path) },
                |cleanup_row| async move {
                    crate::catalog_async_maintenance_methods::schedule_cleanup_metadata_files(
                        catalog,
                        &[cleanup_row],
                    )
                    .await
                },
            )
            .await
        },
        extra {}
    }
);

impl_async_maintenance_commit_executor!(
    impl QuackCommitOps, QuackCatalog {
        cleanup_executor = |_this, catalog| {
            crate::catalog_async_maintenance_methods::borrowed_cleanup_sql_executor(
                catalog,
            )
        },
        inlined_executor = |_this, catalog| catalog,
    }
);

impl_async_catalog_object_commit_ops! {
    impl QuackCommitOps,
    QuackCatalog,
    uuid_style = crate::catalog_queries::UuidParamStyle::CastAsUuid,
    inlined_executor = |_this, catalog| catalog,
    data_path = |_this, catalog| async move {
        Ok(catalog.metadata().await?.data_path)
    }
}

impl crate::commit::AsyncCreateTableWithFilesCommitOps<QuackCatalog> for QuackCommitOps {}

impl_async_partitioning_commit_ops! { impl QuackCommitOps, QuackCatalog }

impl_async_sorted_by_commit_ops! { impl QuackCommitOps, QuackCatalog }

impl_async_column_evolution_commit_ops! {
    impl QuackCommitOps, QuackCatalog,
    extra {}
}

async fn quack_single_i64(catalog: &QuackCatalog, sql: String) -> Result<i64> {
    catalog
        .query_sql(sql)
        .await?
        .typed_rows()
        .next()
        .ok_or_else(|| StorageError::Backend("Quack scalar query returned no rows".into()))?
        .get_i64(0)
}

async fn upsert_quack_table_column_stats(
    catalog: &QuackCatalog,
    stats: &TableColumnStats,
) -> Result<()> {
    crate::stats_engine::upsert_table_column_stats_by_count_async(
        stats,
        |query| async move {
            quack_single_i64(
                catalog,
                query.render_quack_with_metadata_catalog(catalog.metadata_catalog()),
            )
            .await
        },
        |query| async move {
            catalog.execute_catalog_command(&query).await?;
            Ok(())
        },
    )
    .await
}

async fn quack_delete_snapshots(catalog: &QuackCatalog, snapshots: &[Snapshot]) -> Result<()> {
    let snapshots = snapshots.to_vec();
    catalog
        .run_write_transaction("Quack snapshot expiration", None, None, |catalog| {
            Box::pin(async move {
                crate::catalog_async_maintenance_methods::prune_expired_snapshots(
                    catalog, &snapshots,
                )
                .await
            })
        })
        .await
}

impl_async_metadata_prune_executor!(
    impl QuackCatalog {
        table_exists = |this, table| QuackCatalog::table_exists(this, table),
        inlined_table_names = |this, table_ids| async {
            this.query_sql(
                crate::metadata_prune_engine::inlined_table_names_for_table_ids_sql(
                    &this.metadata_table_identifier("ducklake_inlined_data_tables"),
                    table_ids,
                ),
            )
            .await?
            .typed_rows()
            .map(|row| row.get_string(0))
            .collect()
        },
        drop_inlined_table = |this, table_name| async {
            this.query_sql(crate::metadata_prune_engine::drop_table_if_exists_sql(
                &this.metadata_table_identifier(table_name),
            ))
            .await?;
            Ok(())
        },
        snapshot_table_ref = |_this| "ducklake_snapshot".to_string(),
        macro_table_ref = |_this| "ducklake_macro".to_string(),
        macro_dependent_target_ref = |_this, table| quack_quote_identifier(table),
        column_mapping_table_ref = |_this| "ducklake_column_mapping".to_string(),
        name_mapping_target_ref = |_this| "ducklake_name_mapping".to_string(),
        delete_where = |this, table, _target_ref, predicate| async {
            if !QuackCatalog::table_exists(this, table).await? {
                return Ok(());
            }
            this.query_sql(crate::metadata_prune_engine::delete_where_sql(
                &this.metadata_table_identifier(table),
                None,
                predicate,
            ))
            .await?;
            Ok(())
        },
    }
);

fn quack_sql_bool(value: bool) -> &'static str {
    if value { "TRUE" } else { "FALSE" }
}

fn quack_qualify_metadata_sql(sql: &str, metadata_catalog: &str) -> String {
    let qualifier = format!("{}.", quack_quote_identifier(metadata_catalog));
    let mut output = String::with_capacity(sql.len() + qualifier.len() * 4);
    let mut chars = sql.char_indices().peekable();
    let mut last_end = 0;
    let mut in_string = false;
    let mut in_quoted_identifier = false;

    while let Some((idx, ch)) = chars.next() {
        if in_string {
            if ch == '\'' {
                if chars.peek().is_some_and(|(_, next)| *next == '\'') {
                    chars.next();
                } else {
                    in_string = false;
                }
            }
            continue;
        }

        if in_quoted_identifier {
            if ch == '"' {
                if chars.peek().is_some_and(|(_, next)| *next == '"') {
                    chars.next();
                } else {
                    in_quoted_identifier = false;
                }
            }
            continue;
        }

        if ch == '\'' {
            in_string = true;
            continue;
        }

        if ch == '"' {
            in_quoted_identifier = true;
            continue;
        }

        if !is_quack_sql_ident_start(ch) {
            continue;
        }

        let ident_start = idx;
        let mut ident_end = idx + ch.len_utf8();
        while let Some((next_idx, next_ch)) = chars.peek().copied() {
            if !is_quack_sql_ident_continue(next_ch) {
                break;
            }
            chars.next();
            ident_end = next_idx + next_ch.len_utf8();
        }

        let ident = &sql[ident_start..ident_end];
        if !is_quack_metadata_identifier(ident)
            || previous_non_ws_char(sql, ident_start) == Some('.')
            || next_non_ws_char(sql, ident_end) == Some('.')
        {
            continue;
        }

        output.push_str(&sql[last_end..ident_start]);
        output.push_str(&qualifier);
        output.push_str(ident);
        last_end = ident_end;
    }

    if last_end == 0 {
        sql.to_string()
    } else {
        output.push_str(&sql[last_end..]);
        output
    }
}

fn is_quack_sql_ident_start(ch: char) -> bool {
    ch == '_' || ch.is_ascii_alphabetic()
}

fn is_quack_sql_ident_continue(ch: char) -> bool {
    ch == '_' || ch.is_ascii_alphanumeric()
}

fn is_quack_unquoted_identifier(value: &str) -> bool {
    let mut chars = value.chars();
    let Some(first) = chars.next() else {
        return false;
    };
    is_quack_sql_ident_start(first) && chars.all(is_quack_sql_ident_continue)
}

fn is_quack_dynamic_metadata_table_identifier(value: &str) -> bool {
    is_quack_unquoted_identifier(value)
        && (value.starts_with("ducklake_inlined_data_")
            || value.starts_with("ducklake_inlined_delete_"))
}

fn previous_non_ws_char(sql: &str, before: usize) -> Option<char> {
    sql[..before].chars().rev().find(|ch| !ch.is_whitespace())
}

fn next_non_ws_char(sql: &str, after: usize) -> Option<char> {
    sql[after..].chars().find(|ch| !ch.is_whitespace())
}

fn is_quack_metadata_identifier(identifier: &str) -> bool {
    matches!(
        identifier,
        "ducklake_column"
            | "ducklake_column_mapping"
            | "ducklake_column_tag"
            | "ducklake_data_file"
            | "ducklake_delete_file"
            | "ducklake_file_column_stats"
            | "ducklake_file_partition_value"
            | "ducklake_file_variant_stats"
            | "ducklake_files_scheduled_for_deletion"
            | "ducklake_inlined_data_tables"
            | "ducklake_macro"
            | "ducklake_macro_impl"
            | "ducklake_macro_parameters"
            | "ducklake_metadata"
            | "ducklake_name_mapping"
            | "ducklake_partition_column"
            | "ducklake_partition_info"
            | "ducklake_schema"
            | "ducklake_schema_versions"
            | "ducklake_snapshot"
            | "ducklake_snapshot_changes"
            | "ducklake_sort_expression"
            | "ducklake_sort_info"
            | "ducklake_table"
            | "ducklake_table_column_stats"
            | "ducklake_table_stats"
            | "ducklake_tag"
            | "ducklake_view"
    ) || identifier.starts_with("ducklake_inlined_data_")
        || identifier.starts_with("ducklake_inlined_delete_")
}

impl_async_catalog_maintenance_executor!(
    impl QuackCatalog {
        load_cleanup_data_path = |this| async { Ok(this.metadata().await?.data_path) },
        load_scheduled_cleanup_files = |this| async { this.files_scheduled_for_deletion().await },
        delete_scheduled_cleanup_files = |this, data_file_ids| async {
            let Some(sql) = quack_remove_scheduled_files_sql(data_file_ids) else {
                return Ok(());
            };
            let write_lock = quack_catalog_write_lock(this)?;
            let _write_guard = write_lock.lock().await;
            this.query_sql(sql).await?;
            Ok(())
        },
        load_expirable_snapshots = |this, options| async {
            crate::catalog_async_maintenance_methods::load_expirable_snapshots_from_read_executor(
                this,
                options,
            )
            .await
        },
        prune_expired_snapshots = |this, snapshots| async {
            quack_delete_snapshots(this, snapshots).await
        },
    }
);

impl_async_catalog_metadata_executor!(
    impl QuackCatalog {
        connection_options = |this| &this.connection_options,
        replace_metadata_option = |this, key, value, scope| async {
            let sql =
                crate::metadata_option::render_set_metadata_option_literals(&key, &value, &scope);

            this.run_write_transaction(
                "Quack metadata option update",
                None,
                None,
                |catalog| Box::pin(async move {
                    catalog.query_sql(sql.delete).await?;
                    catalog.query_sql(sql.insert).await?;
                    Ok(())
                }),
            )
            .await
        },
    }
);

impl_async_inlined_read_executor!(
    impl QuackCatalog {
        file_deletion_table_exists = |this, table_name| async { this.table_exists(table_name).await },
        load_file_deletions = |this, table_name, snapshot| async {
            quack_collect_rows(
                this.query_sql(format!(
                    "SELECT file_id, row_id, begin_snapshot FROM {} \
                     WHERE begin_snapshot <= {} ORDER BY file_id, row_id",
                    this.metadata_table_identifier(table_name),
                    snapshot.0
                ))
                .await?,
                crate::embedded_reads::inlined_file_deletion,
            )
        },
        load_data = |this, table, schema, snapshot| async {
            if schema.columns.is_empty() {
                return Ok(Vec::new());
            }

            let projection = crate::inlined_values::inlined_read_projection_for_dialect(
                schema,
                crate::inlined_values::InlinedSqlDialect::Quack,
                quack_quote_identifier,
            )?;
            let table_identifier = this.dynamic_metadata_table_identifier(&table.table_name);
            let sql = crate::inlined_values::read_inlined_data_sql(
                &table_identifier,
                &projection,
                &snapshot.0.to_string(),
            );
            let result = this.query_sql(sql).await?;
            Ok(quack_inlined_record_batch(schema, &result.rows)?
                .into_iter()
                .collect())
        },
        load_data_with_metadata = |this, table, schema, snapshot| async {
            if schema.columns.is_empty() {
                return Ok(Vec::new());
            }

            let projection = crate::inlined_values::inlined_read_projection_for_dialect(
                schema,
                crate::inlined_values::InlinedSqlDialect::Quack,
                quack_quote_identifier,
            )?;
            let table_identifier = this.dynamic_metadata_table_identifier(&table.table_name);
            let sql = crate::inlined_values::read_inlined_data_with_metadata_sql(
                &table_identifier,
                &projection,
                &snapshot.0.to_string(),
            );
            let result = this.query_sql(sql).await?;
            if result.rows.is_empty() {
                return Ok(Vec::new());
            }

            let mut row_ids = Vec::with_capacity(result.rows.len());
            let mut begin_snapshots = Vec::with_capacity(result.rows.len());
            let mut user_rows = Vec::with_capacity(result.rows.len());
            for row in &result.rows {
                let typed = QuackRow { values: row };
                row_ids.push(typed.get_i64(0)?);
                begin_snapshots.push(SnapshotId(typed.get_i64(1)?));
                user_rows.push(row[2..].to_vec());
            }
            Ok(quack_inlined_record_batch(schema, &user_rows)?
                .map(|batch| InlinedDataBatch::with_metadata(batch, row_ids, begin_snapshots))
                .into_iter()
                .collect())
        },
    }
);

#[async_trait]
impl crate::catalog_initialization_methods::CatalogInitializationExecutor for QuackCatalog {
    impl_catalog_initialization_info_methods!(
        catalog_type = "quack",
        default_data_path = |_this| None,
        snapshot_time_sql = |_this| Ok("NOW()".into())
    );

    async fn run_initialization_request(
        &self,
        request: crate::catalog_initialization_methods::CatalogInitializationRequest,
    ) -> Result<crate::catalog_initialization_methods::CatalogInitializationResponse> {
        self.with_write_lock(|| async {
            request
                .execute_async(
                    || async { self.table_exists("ducklake_metadata").await },
                    || async {
                        crate::catalog_async_metadata_methods::load_catalog_metadata(self).await
                    },
                    |bootstrap| async move {
                        let metadata_catalog = self.metadata_catalog.clone();
                        self.run_unlocked_write_transaction(
                            "Quack catalog initialization",
                            None,
                            None,
                            |catalog| {
                                Box::pin(async move {
                                    if let Some(metadata_catalog) = metadata_catalog.as_deref() {
                                        catalog
                                            .query_sql(format!(
                                                "CREATE SCHEMA IF NOT EXISTS {}",
                                                quack_quote_identifier(metadata_catalog)
                                            ))
                                            .await?;
                                    }
                                    for statement in
                                        crate::catalog_ddl::render_initialization_statements(
                                            None, &bootstrap,
                                        )
                                    {
                                        catalog.query_sql(statement).await?;
                                    }
                                    Ok(())
                                })
                            },
                        )
                        .await
                    },
                )
                .await
        })
        .await
    }
}

#[async_trait]
impl CatalogBackend for QuackCatalog {
    impl_catalog_type_method!("quack");

    impl_async_catalog_metadata_methods!();

    impl_catalog_initialization_method!();

    impl_catalog_backend_read_methods_direct!();

    impl_async_catalog_maintenance_methods!();

    impl_async_inlined_read_methods!();

    impl_catalog_backend_commit_methods!();
}

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