//! SQLite DuckLake catalog backend.

use std::collections::HashSet;
use std::fs;
use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

#[cfg(test)]
use crate::{
    CleanupOldFilesOptions, ColumnDefaultCommit, CreateSchemaCommit, CreateTableCommit,
    CreateViewCommit, DropSchemaCommit, DropTableCommit, DropViewCommit, ExpireSnapshotsOptions,
    InlinedDataAppendCommit, InlinedDataDeletionCommit, MergeAdjacentCompactionCommit,
    RenameColumnCommit, RenameTableCommit,
};
use arrow::array::{Array, ArrayRef, Int64Array};
#[cfg(test)]
use arrow::array::{
    BinaryArray, Date32Array, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float32Array,
    Float64Array, Int8Array, Int16Array, IntervalMonthDayNanoArray, StringArray,
    Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
    TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt8Array,
    UInt16Array, UInt32Array, UInt64Array,
};
#[cfg(test)]
use arrow::datatypes::{IntervalMonthDayNano, IntervalUnit, i256};
use arrow::record_batch::RecordBatch;
#[cfg(test)]
use arrow_schema::TimeUnit;
#[cfg(test)]
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
#[cfg(test)]
use ducklake_catalog::InlinedDataTableRef;
#[cfg(test)]
use ducklake_catalog::PartitionTransform;
#[cfg(test)]
use ducklake_catalog::ViewId;
#[cfg(test)]
use ducklake_catalog::{ColumnId, FileColumnStats, FileVariantStats};
#[cfg(test)]
use ducklake_catalog::{DataFileId, MacroId, MacroImplId, PartitionId, SchemaId, SortId};
use ducklake_catalog::{SchemaField, SnapshotId, TableColumnStats, TableId};
use parquet::arrow::ArrowWriter;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use rusqlite::types::{Value, ValueRef};
use rusqlite::{Connection, OptionalExtension, Row, TransactionBehavior, params, params_from_iter};

use crate::embedded_rows::format_timestamp_micros;
use crate::inlined_values::InlinedSqlScalarRef;
#[cfg(test)]
use crate::inlined_values::{parse_inlined_date32, parse_inlined_timestamp};
use crate::{
    ActivePartitionSpec, CatalogBackend, DuckLakeCatalogConnectionOptions, PartitionColumnCommit,
    Result, StorageError, local_filesystem_path,
};

#[derive(Debug, Clone)]
pub struct SqliteCatalog {
    path: PathBuf,
    connection_options: DuckLakeCatalogConnectionOptions,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SqliteTransactionCleanupFile {
    pub path: String,
    pub path_is_relative: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct SqliteTransactionDeleteFileRewrite {
    path: PathBuf,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct SqliteTransactionFilePath {
    path: String,
    path_is_relative: bool,
    schema_path: Option<String>,
    schema_path_is_relative: bool,
    table_path: Option<String>,
    table_path_is_relative: bool,
}

impl SqliteCatalog {
    pub fn new(path: impl Into<PathBuf>) -> Self {
        Self {
            path: path.into(),
            connection_options: DuckLakeCatalogConnectionOptions::default(),
        }
    }

    impl_connection_options_builder_method!();

    impl_path_accessor_method!();

    async fn with_conn<T, F>(&self, f: F) -> Result<T>
    where
        T: Send + 'static,
        F: FnOnce(Connection) -> Result<T> + Send + 'static,
    {
        let path = self.path.clone();
        let busy_timeout_ms = self.connection_options.busy_timeout_ms;
        tokio::task::spawn_blocking(move || {
            let conn =
                Connection::open(path).map_err(|err| StorageError::Backend(err.to_string()))?;
            conn.busy_timeout(Duration::from_millis(busy_timeout_ms.unwrap_or(5_000)))
                .map_err(sql_err)?;
            f(conn)
        })
        .await
        .map_err(|err| StorageError::Backend(err.to_string()))?
    }

    impl_embedded_run_commit_request_method!(SqliteCommitOps, with_conn, |conn| {
        sqlite_write_transaction(conn)
    });

    pub async fn squash_transaction_snapshots(
        &self,
        initial_snapshot: SnapshotId,
    ) -> Result<Vec<SqliteTransactionCleanupFile>> {
        self.with_conn(move |mut conn| {
            sqlite_squash_transaction_snapshots(&mut conn, initial_snapshot)
        })
        .await
    }

    pub async fn transaction_staged_cleanup_files(
        &self,
        initial_snapshot: SnapshotId,
    ) -> Result<Vec<SqliteTransactionCleanupFile>> {
        self.with_conn(move |conn| {
            let initial_snapshot = initial_snapshot.0;
            let initial_next_file_id = sqlite_snapshot_next_file_id(&conn, initial_snapshot)?;
            sqlite_transaction_cleanup_files(&conn, initial_snapshot, initial_next_file_id)
        })
        .await
    }

    pub async fn transaction_created_cleanup_files(
        &self,
        initial_snapshot: SnapshotId,
    ) -> Result<Vec<SqliteTransactionCleanupFile>> {
        self.with_conn(move |conn| {
            sqlite_transaction_created_cleanup_files(&conn, initial_snapshot.0)
        })
        .await
    }

    pub async fn commit_view_comment(
        &self,
        schema_name: impl Into<String>,
        view_name: impl Into<String>,
        comment: Option<String>,
        commit_message: impl Into<String>,
    ) -> Result<SnapshotId> {
        let schema_name = schema_name.into();
        let view_name = view_name.into();
        let commit_message = commit_message.into();
        self.with_conn(move |mut conn| {
            if schema_name.trim().is_empty() {
                return Err(StorageError::Backend(
                    "DuckLake comment-view schema name is empty".into(),
                ));
            }
            if view_name.trim().is_empty() {
                return Err(StorageError::Backend(
                    "DuckLake comment-view view name is empty".into(),
                ));
            }

            let now_timestamp = format_timestamp_micros(crate::sql_utils::current_time_micros()?)?;
            let tx = sqlite_write_transaction(&mut conn)?;
            if !sqlite_table_exists(&tx, "ducklake_tag")? {
                return Err(StorageError::Backend(
                    "DuckLake catalog does not contain ducklake_tag".into(),
                ));
            }
            let latest = run_catalog_query(
                &tx,
                &crate::catalog_queries::latest_snapshot_ids(),
                crate::catalog_queries::latest_snapshot_ids_value,
            )?
            .into_iter()
            .next()
            .ok_or_else(|| StorageError::Backend("DuckLake commit found no snapshots".into()))?;
            let snapshot = SnapshotId(latest.snapshot_id);
            let exact_schema = run_catalog_query(
                &tx,
                &crate::catalog_queries::visible_schema_exact(&schema_name, snapshot),
                crate::embedded_reads::schema_entry,
            )?;
            let schema = if !exact_schema.is_empty() {
                crate::catalog_queries::resolve_visible_schema(
                    &schema_name,
                    exact_schema,
                    Vec::new(),
                )?
            } else {
                let case_matches = run_catalog_query(
                    &tx,
                    &crate::catalog_queries::visible_schema_case(&schema_name, snapshot),
                    crate::embedded_reads::schema_entry,
                )?;
                crate::catalog_queries::resolve_visible_schema(
                    &schema_name,
                    Vec::new(),
                    case_matches,
                )?
            }
            .ok_or_else(|| {
                StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
                    "schema {schema_name}"
                )))
            })?;
            let exact_view = run_catalog_query(
                &tx,
                &crate::catalog_queries::visible_view_exact(schema.schema_id, &view_name, snapshot),
                crate::embedded_reads::view_entry,
            )?;
            let view = if !exact_view.is_empty() {
                crate::catalog_queries::resolve_visible_view(&view_name, exact_view, Vec::new())?
            } else {
                let case_matches = run_catalog_query(
                    &tx,
                    &crate::catalog_queries::visible_views_case(
                        schema.schema_id,
                        &view_name,
                        snapshot,
                    ),
                    crate::embedded_reads::view_entry,
                )?;
                crate::catalog_queries::resolve_visible_view(&view_name, Vec::new(), case_matches)?
            }
            .ok_or_else(|| {
                StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
                    "view {schema_name}.{view_name}"
                )))
            })?;

            let new_snapshot = latest.snapshot_id + 1;
            execute_catalog_command(
                &tx,
                &crate::catalog_queries::close_comment_tag(view.view_id.0, new_snapshot),
            )?;
            execute_catalog_command(
                &tx,
                &crate::catalog_queries::insert_comment_tag(view.view_id.0, new_snapshot, comment),
            )?;

            execute_catalog_command(
                &tx,
                &crate::catalog_queries::insert_snapshot_with_timestamp_text(
                    new_snapshot,
                    now_timestamp,
                    latest.schema_version + 1,
                    latest.next_catalog_id,
                    latest.next_file_id,
                ),
            )?;

            if sqlite_table_exists(&tx, "ducklake_snapshot_changes")? {
                execute_catalog_command(
                    &tx,
                    &crate::catalog_queries::insert_snapshot_change(
                        new_snapshot,
                        &format!("altered_view:{}", view.view_id.0),
                        &commit_message,
                    ),
                )?;
            }

            tx.commit().map_err(sql_err)?;
            Ok(SnapshotId(new_snapshot))
        })
        .await
    }
}

fn sqlite_write_transaction(conn: &mut Connection) -> Result<rusqlite::Transaction<'_>> {
    conn.transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(sql_err)
}

fn sqlite_squash_transaction_snapshots(
    conn: &mut Connection,
    initial_snapshot: SnapshotId,
) -> Result<Vec<SqliteTransactionCleanupFile>> {
    let initial_snapshot = initial_snapshot.0;
    let Some(latest_snapshot) = run_catalog_query(
        conn,
        &crate::catalog_queries::max_snapshot_id(),
        crate::catalog_queries::opt_i64_value,
    )?
    .into_iter()
    .next()
    .flatten() else {
        return Ok(Vec::new());
    };
    if latest_snapshot <= initial_snapshot {
        return Ok(Vec::new());
    }

    let final_snapshot = initial_snapshot + 1;
    let initial_next_file_id = sqlite_snapshot_next_file_id(conn, initial_snapshot)?;
    let cleanup_files =
        sqlite_transaction_cleanup_files(conn, initial_snapshot, initial_next_file_id)?;
    let delete_files_to_rewrite =
        sqlite_transaction_delete_files_to_rewrite(conn, initial_snapshot)?;
    sqlite_rewrite_transaction_delete_files(
        &delete_files_to_rewrite,
        initial_snapshot,
        final_snapshot,
    )?;
    let latest: (Value, i64, i64, i64) = conn
        .query_row(
            "SELECT snapshot_time, schema_version, next_catalog_id, next_file_id \
             FROM ducklake_snapshot WHERE snapshot_id = ?1",
            [latest_snapshot],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .map_err(sql_err)?;
    let changes = sqlite_transaction_squashed_changes(conn, initial_snapshot)?;

    let tx = sqlite_write_transaction(conn)?;
    let squashed_partitioning =
        sqlite_plan_squashed_transaction_partitioning(&tx, initial_snapshot, latest_snapshot)?;
    for table in sqlite_transaction_versioned_tables(&tx)? {
        let begin_sql = format!("UPDATE {table} SET begin_snapshot = ?1 WHERE begin_snapshot > ?2");
        tx.execute(&begin_sql, params![final_snapshot, initial_snapshot])
            .map_err(sql_err)?;
        let end_sql = format!(
            "UPDATE {table} SET end_snapshot = ?1 \
             WHERE end_snapshot IS NOT NULL AND end_snapshot > ?2"
        );
        tx.execute(&end_sql, params![final_snapshot, initial_snapshot])
            .map_err(sql_err)?;
    }
    for table in ["ducklake_data_file", "ducklake_delete_file"] {
        if sqlite_table_exists(&tx, table)? {
            let partial_sql = format!(
                "UPDATE {table} SET partial_max = ?1 \
                 WHERE partial_max IS NOT NULL AND partial_max > ?2"
            );
            tx.execute(&partial_sql, params![final_snapshot, initial_snapshot])
                .map_err(sql_err)?;
        }
    }
    if sqlite_table_exists(&tx, "ducklake_schema_versions")? {
        execute_catalog_command(
            &tx,
            &crate::catalog_queries::update_schema_versions_begin_after(
                final_snapshot,
                initial_snapshot,
            ),
        )?;
    }
    if sqlite_table_exists(&tx, "ducklake_files_scheduled_for_deletion")? {
        execute_catalog_command(
            &tx,
            &crate::catalog_queries::delete_scheduled_cleanup_files_from_next_file_id(
                initial_next_file_id,
            ),
        )?;
    }
    if let Some(partitioning) = squashed_partitioning {
        sqlite_apply_squashed_transaction_partitioning(&tx, final_snapshot, partitioning)?;
    }

    execute_catalog_command(
        &tx,
        &crate::catalog_queries::delete_snapshots_after(initial_snapshot),
    )?;
    tx.execute(
        "INSERT INTO ducklake_snapshot VALUES (?1, ?2, ?3, ?4, ?5)",
        params![final_snapshot, latest.0, latest.1, latest.2, latest.3],
    )
    .map_err(sql_err)?;
    execute_catalog_command(
        &tx,
        &crate::catalog_queries::delete_snapshot_changes_after(initial_snapshot),
    )?;
    execute_catalog_command(
        &tx,
        &crate::catalog_queries::insert_squashed_snapshot_change(final_snapshot, &changes),
    )?;
    tx.commit().map_err(sql_err)?;

    Ok(cleanup_files)
}

#[derive(Debug)]
struct SqliteTransactionPartitionRow {
    partition_id: i64,
    table_id: i64,
}

fn sqlite_transaction_partition_row(
    row: &dyn crate::embedded_rows::EmbeddedRow,
) -> Result<SqliteTransactionPartitionRow> {
    Ok(SqliteTransactionPartitionRow {
        partition_id: row.get_i64(0)?,
        table_id: row.get_i64(1)?,
    })
}

#[derive(Debug)]
struct SqliteSquashedTransactionPartitioning {
    created_partition_ids: Vec<i64>,
    final_rows: Vec<SqliteSquashedTransactionPartitionRow>,
    close_initial_partition_ids: Vec<i64>,
    reopen_initial_partition_ids: Vec<i64>,
}

#[derive(Debug)]
struct SqliteSquashedTransactionPartitionRow {
    partition_id: i64,
    table_id: i64,
    columns: Vec<PartitionColumnCommit>,
}

fn sqlite_plan_squashed_transaction_partitioning(
    tx: &rusqlite::Transaction<'_>,
    initial_snapshot: i64,
    latest_snapshot: i64,
) -> Result<Option<SqliteSquashedTransactionPartitioning>> {
    if !sqlite_table_exists(tx, "ducklake_partition_info")?
        || !sqlite_table_exists(tx, "ducklake_partition_column")?
    {
        return Ok(None);
    }

    let created = run_catalog_query(
        tx,
        &crate::catalog_queries::created_partition_infos_after(initial_snapshot),
        sqlite_transaction_partition_row,
    )?;
    if created.is_empty() {
        return Ok(None);
    }

    let first_partition_id = created
        .iter()
        .map(|row| row.partition_id)
        .min()
        .ok_or_else(|| {
            StorageError::Backend("DuckLake transaction has no partition rows".into())
        })?;
    let created_partition_ids = created
        .iter()
        .map(|row| row.partition_id)
        .collect::<Vec<_>>();

    let mut seen = HashSet::with_capacity(created.len());
    let mut final_table_order = Vec::new();
    for row in created.iter().rev() {
        if seen.insert(row.table_id) {
            final_table_order.push(row.table_id);
        }
    }

    let mut final_rows = Vec::new();
    let mut close_initial_partition_ids = Vec::new();
    let mut reopen_initial_partition_ids = Vec::new();
    for table_id in final_table_order {
        let initial = sqlite_single_active_partition_spec(
            tx,
            TableId(table_id),
            SnapshotId(initial_snapshot),
        )?;
        let final_spec = sqlite_single_active_partition_spec(
            tx,
            TableId(table_id),
            SnapshotId(latest_snapshot),
        )?;
        let initial_columns = initial
            .as_ref()
            .map(|spec| spec.columns.clone())
            .unwrap_or_default();
        let final_columns = final_spec
            .as_ref()
            .map(|spec| spec.columns.clone())
            .unwrap_or_default();

        if initial_columns == final_columns {
            if let Some(initial) = initial {
                reopen_initial_partition_ids.push(initial.partition_id.0);
            }
            continue;
        }

        if let Some(initial) = &initial {
            close_initial_partition_ids.push(initial.partition_id.0);
        }
        if !final_columns.is_empty() || initial.is_some() {
            let inserted_count = i64::try_from(final_rows.len()).map_err(|_| {
                StorageError::Backend("DuckLake partition row count is out of range".into())
            })?;
            final_rows.push(SqliteSquashedTransactionPartitionRow {
                partition_id: first_partition_id + inserted_count,
                table_id,
                columns: final_columns,
            });
        }
    }

    Ok(Some(SqliteSquashedTransactionPartitioning {
        created_partition_ids,
        final_rows,
        close_initial_partition_ids,
        reopen_initial_partition_ids,
    }))
}

fn sqlite_single_active_partition_spec(
    tx: &rusqlite::Transaction<'_>,
    table_id: TableId,
    snapshot: SnapshotId,
) -> Result<Option<ActivePartitionSpec>> {
    let mut active = sqlite_active_partition_specs(tx, table_id, snapshot)?;
    if active.len() > 1 {
        return Err(StorageError::Backend(format!(
            "DuckLake table {} has {} active partition specs at snapshot {}",
            table_id.0,
            active.len(),
            snapshot.0
        )));
    }
    Ok(active.pop())
}

fn sqlite_apply_squashed_transaction_partitioning(
    tx: &rusqlite::Transaction<'_>,
    final_snapshot: i64,
    partitioning: SqliteSquashedTransactionPartitioning,
) -> Result<()> {
    for partition_id in partitioning.created_partition_ids {
        let partition_id = ducklake_catalog::PartitionId(partition_id);
        execute_catalog_command(
            tx,
            &crate::catalog_queries::delete_partition_columns(partition_id),
        )?;
        execute_catalog_command(
            tx,
            &crate::catalog_queries::delete_partition_info(partition_id),
        )?;
    }

    for partition_id in partitioning.reopen_initial_partition_ids {
        execute_catalog_command(
            tx,
            &crate::catalog_queries::reopen_partition_info(ducklake_catalog::PartitionId(
                partition_id,
            )),
        )?;
    }
    for partition_id in partitioning.close_initial_partition_ids {
        execute_catalog_command(
            tx,
            &crate::catalog_queries::set_partition_info_end_snapshot(
                ducklake_catalog::PartitionId(partition_id),
                final_snapshot,
            ),
        )?;
    }

    for row in partitioning.final_rows {
        let partition_id = ducklake_catalog::PartitionId(row.partition_id);
        let table_id = ducklake_catalog::TableId(row.table_id);
        execute_catalog_command(
            tx,
            &crate::catalog_queries::insert_partition_info(partition_id, table_id, final_snapshot),
        )?;
        for (idx, column) in row.columns.iter().enumerate() {
            let partition_key_index = i64::try_from(idx).map_err(|_| {
                StorageError::Backend("DuckLake partition key index is out of range".into())
            })?;
            execute_catalog_command(
                tx,
                &crate::catalog_queries::insert_partition_column(
                    partition_id,
                    table_id,
                    partition_key_index,
                    column,
                ),
            )?;
        }
    }

    Ok(())
}

fn sqlite_transaction_delete_files_to_rewrite(
    conn: &Connection,
    initial_snapshot: i64,
) -> Result<Vec<SqliteTransactionDeleteFileRewrite>> {
    if !sqlite_table_exists(conn, "ducklake_delete_file")? {
        return Ok(Vec::new());
    }

    let data_path = sqlite_transaction_data_path(conn)?;
    let mut stmt = conn
        .prepare(
            "SELECT f.path, f.path_is_relative, \
                    s.path, s.path_is_relative, \
                    t.path, t.path_is_relative \
             FROM ducklake_delete_file f \
             JOIN ducklake_table t USING(table_id) \
             JOIN ducklake_schema s USING(schema_id) \
             WHERE f.begin_snapshot > ?1 \
               AND f.end_snapshot IS NULL \
               AND lower(f.format) = 'parquet' \
               AND f.partial_max IS NOT NULL \
               AND f.partial_max > ?1",
        )
        .map_err(sql_err)?;
    let rows = collect_rows(
        &mut stmt,
        [initial_snapshot],
        sqlite_transaction_file_path_row,
    )?;
    rows.into_iter()
        .map(|row| {
            Ok(SqliteTransactionDeleteFileRewrite {
                path: sqlite_resolve_transaction_file_path(&data_path, &row)?,
            })
        })
        .collect()
}

fn sqlite_rewrite_transaction_delete_files(
    files: &[SqliteTransactionDeleteFileRewrite],
    initial_snapshot: i64,
    final_snapshot: i64,
) -> Result<()> {
    if files.is_empty() {
        return Ok(());
    }

    for file in files {
        sqlite_rewrite_delete_file_snapshot_ids(&file.path, initial_snapshot, final_snapshot)?;
    }
    Ok(())
}

fn sqlite_rewrite_delete_file_snapshot_ids(
    path: &Path,
    initial_snapshot: i64,
    final_snapshot: i64,
) -> Result<()> {
    let input = fs::File::open(path).map_err(|err| StorageError::Backend(err.to_string()))?;
    let builder = ParquetRecordBatchReaderBuilder::try_new(input)
        .map_err(|err| StorageError::Backend(err.to_string()))?;
    let reader = builder
        .build()
        .map_err(|err| StorageError::Backend(err.to_string()))?;

    let mut batches = Vec::new();
    let mut rewrote_any = false;
    for batch in reader {
        let batch = batch.map_err(|err| StorageError::Backend(err.to_string()))?;
        let Some(snapshot_idx) = batch
            .schema()
            .index_of("_ducklake_internal_snapshot_id")
            .ok()
        else {
            batches.push(batch);
            continue;
        };
        let snapshots = batch
            .column(snapshot_idx)
            .as_any()
            .downcast_ref::<Int64Array>()
            .ok_or_else(|| {
                StorageError::Backend("DuckLake delete file snapshot column must be int64".into())
            })?;
        let remapped = (0..snapshots.len())
            .map(|row| {
                if snapshots.is_null(row) {
                    None
                } else {
                    let snapshot = snapshots.value(row);
                    Some(if snapshot > initial_snapshot {
                        final_snapshot
                    } else {
                        snapshot
                    })
                }
            })
            .collect::<Vec<_>>();
        let mut columns = batch.columns().to_vec();
        columns[snapshot_idx] = Arc::new(Int64Array::from(remapped)) as ArrayRef;
        let batch = RecordBatch::try_new(Arc::clone(&batch.schema()), columns)
            .map_err(|err| StorageError::Backend(err.to_string()))?;
        batches.push(batch);
        rewrote_any = true;
    }

    if !rewrote_any {
        return Ok(());
    }
    let Some(schema) = batches.first().map(|batch| batch.schema()) else {
        return Ok(());
    };
    let mut output = Cursor::new(Vec::new());
    {
        let mut writer = ArrowWriter::try_new(&mut output, schema, None)
            .map_err(|err| StorageError::Backend(err.to_string()))?;
        for batch in &batches {
            writer
                .write(batch)
                .map_err(|err| StorageError::Backend(err.to_string()))?;
        }
        writer
            .close()
            .map_err(|err| StorageError::Backend(err.to_string()))?;
    }
    fs::write(path, output.into_inner()).map_err(|err| StorageError::Backend(err.to_string()))
}

fn sqlite_transaction_cleanup_files(
    conn: &Connection,
    initial_snapshot: i64,
    initial_next_file_id: i64,
) -> Result<Vec<SqliteTransactionCleanupFile>> {
    let data_path = sqlite_transaction_data_path(conn)?;
    let mut files = Vec::new();
    if sqlite_table_exists(conn, "ducklake_data_file")? {
        let mut stmt = conn
            .prepare(
                "SELECT f.path, f.path_is_relative, \
                        s.path, s.path_is_relative, \
                        t.path, t.path_is_relative \
                 FROM ducklake_data_file f \
                 JOIN ducklake_table t USING(table_id) \
                 JOIN ducklake_schema s USING(schema_id) \
                 WHERE f.begin_snapshot > ?1 AND f.end_snapshot IS NOT NULL",
            )
            .map_err(sql_err)?;
        let rows = collect_rows(
            &mut stmt,
            [initial_snapshot],
            sqlite_transaction_file_path_row,
        )?;
        files.extend(
            rows.into_iter()
                .map(|row| {
                    Ok(SqliteTransactionCleanupFile {
                        path: sqlite_resolve_transaction_file_path(&data_path, &row)?
                            .to_string_lossy()
                            .into_owned(),
                        path_is_relative: false,
                    })
                })
                .collect::<Result<Vec<_>>>()?,
        );
    }
    if sqlite_table_exists(conn, "ducklake_delete_file")? {
        let mut stmt = conn
            .prepare(
                "SELECT f.path, f.path_is_relative, \
                        s.path, s.path_is_relative, \
                        t.path, t.path_is_relative \
                 FROM ducklake_delete_file f \
                 JOIN ducklake_table t USING(table_id) \
                 JOIN ducklake_schema s USING(schema_id) \
                 WHERE f.begin_snapshot > ?1 AND f.end_snapshot IS NOT NULL",
            )
            .map_err(sql_err)?;
        let rows = collect_rows(
            &mut stmt,
            [initial_snapshot],
            sqlite_transaction_file_path_row,
        )?;
        files.extend(
            rows.into_iter()
                .map(|row| {
                    Ok(SqliteTransactionCleanupFile {
                        path: sqlite_resolve_transaction_file_path(&data_path, &row)?
                            .to_string_lossy()
                            .into_owned(),
                        path_is_relative: false,
                    })
                })
                .collect::<Result<Vec<_>>>()?,
        );
    }
    if sqlite_table_exists(conn, "ducklake_files_scheduled_for_deletion")? {
        let mut stmt = conn
            .prepare(
                "SELECT path, path_is_relative FROM ducklake_files_scheduled_for_deletion \
                 WHERE data_file_id >= ?1",
            )
            .map_err(sql_err)?;
        let scheduled = collect_rows(&mut stmt, [initial_next_file_id], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, bool>(1)?))
        })?;
        files.extend(
            scheduled
                .into_iter()
                .map(|(path, path_is_relative)| {
                    let resolved = if path_is_relative {
                        crate::join_cleanup_path_components(&data_path, &path)
                    } else {
                        path
                    };
                    Ok(SqliteTransactionCleanupFile {
                        path: local_filesystem_path(&resolved)?,
                        path_is_relative: false,
                    })
                })
                .collect::<Result<Vec<_>>>()?,
        );
    }

    let mut seen = HashSet::new();
    files.retain(|file| seen.insert((file.path.clone(), file.path_is_relative)));
    Ok(files)
}

fn sqlite_transaction_created_cleanup_files(
    conn: &Connection,
    initial_snapshot: i64,
) -> Result<Vec<SqliteTransactionCleanupFile>> {
    let data_path = sqlite_transaction_data_path(conn)?;
    let mut files = Vec::new();
    for table in ["ducklake_data_file", "ducklake_delete_file"] {
        if !sqlite_table_exists(conn, table)? {
            continue;
        }
        let sql = format!(
            "SELECT f.path, f.path_is_relative, \
                    s.path, s.path_is_relative, \
                    t.path, t.path_is_relative \
             FROM {table} f \
             JOIN ducklake_table t USING(table_id) \
             JOIN ducklake_schema s USING(schema_id) \
             WHERE f.begin_snapshot > ?1"
        );
        let mut stmt = conn.prepare(&sql).map_err(sql_err)?;
        let rows = collect_rows(
            &mut stmt,
            [initial_snapshot],
            sqlite_transaction_file_path_row,
        )?;
        files.extend(
            rows.into_iter()
                .map(|row| {
                    Ok(SqliteTransactionCleanupFile {
                        path: sqlite_resolve_transaction_file_path(&data_path, &row)?
                            .to_string_lossy()
                            .into_owned(),
                        path_is_relative: false,
                    })
                })
                .collect::<Result<Vec<_>>>()?,
        );
    }

    let mut seen = HashSet::new();
    files.retain(|file| seen.insert((file.path.clone(), file.path_is_relative)));
    Ok(files)
}

fn sqlite_snapshot_next_file_id(conn: &Connection, snapshot_id: i64) -> Result<i64> {
    run_catalog_query(
        conn,
        &crate::catalog_queries::snapshot_next_file_id(snapshot_id),
        crate::catalog_queries::i64_value,
    )?
    .into_iter()
    .next()
    .ok_or_else(|| {
        StorageError::Backend(format!(
            "DuckLake snapshot {snapshot_id} has no next_file_id"
        ))
    })
}

fn sqlite_transaction_data_path(conn: &Connection) -> Result<String> {
    global_metadata_value(conn, "data_path")?
        .ok_or_else(|| StorageError::Backend("ducklake_metadata.data_path missing".into()))
}

fn sqlite_transaction_file_path_row(row: &Row<'_>) -> rusqlite::Result<SqliteTransactionFilePath> {
    Ok(SqliteTransactionFilePath {
        path: row.get(0)?,
        path_is_relative: row.get(1)?,
        schema_path: row.get(2)?,
        schema_path_is_relative: row.get(3)?,
        table_path: row.get(4)?,
        table_path_is_relative: row.get(5)?,
    })
}

fn sqlite_resolve_transaction_file_path(
    data_path: &str,
    row: &SqliteTransactionFilePath,
) -> Result<PathBuf> {
    let resolved = crate::resolve_catalog_file_path(
        data_path,
        row.schema_path.as_deref(),
        row.schema_path_is_relative,
        row.table_path.as_deref(),
        row.table_path_is_relative,
        &row.path,
        row.path_is_relative,
    );
    Ok(PathBuf::from(local_filesystem_path(&resolved)?))
}

fn sqlite_transaction_squashed_changes(conn: &Connection, initial_snapshot: i64) -> Result<String> {
    let changes = run_catalog_query(
        conn,
        &crate::catalog_queries::snapshot_changes_after(initial_snapshot),
        crate::catalog_queries::string_value,
    )?;
    Ok(changes.join(","))
}

fn sqlite_transaction_versioned_tables(conn: &Connection) -> Result<Vec<String>> {
    let mut tables = Vec::new();
    for table in [
        "ducklake_schema",
        "ducklake_table",
        "ducklake_view",
        "ducklake_column",
        "ducklake_column_tag",
        "ducklake_tag",
        "ducklake_data_file",
        "ducklake_delete_file",
        "ducklake_partition_info",
        "ducklake_macro",
        "ducklake_sort_info",
    ] {
        if sqlite_table_exists(conn, table)? {
            tables.push(table.to_string());
        }
    }

    let mut stmt = conn
        .prepare(
            "SELECT name FROM sqlite_master \
             WHERE type = 'table' \
               AND (name LIKE 'ducklake_inlined_data_%' \
                    OR name LIKE 'ducklake_inlined_delete_%') \
               AND name <> 'ducklake_inlined_data_tables' \
             ORDER BY name",
        )
        .map_err(sql_err)?;
    let dynamic_tables = collect_rows(&mut stmt, [], |row| row.get::<_, String>(0))?;
    tables.extend(dynamic_tables);
    Ok(tables)
}

fn sqlite_initialize_ducklake(
    conn: &mut Connection,
    bootstrap: &crate::catalog_ddl::CatalogBootstrap,
) -> Result<()> {
    let ddl = crate::catalog_ddl::render_initialization_batch(None, bootstrap);
    let tx = sqlite_write_transaction(conn)?;
    tx.execute_batch(&ddl).map_err(sql_err)?;
    tx.commit().map_err(sql_err)?;
    Ok(())
}

struct SqliteCommitOps;

impl_embedded_sync_commit_ops!(
    SqliteCommitOps,
    tx_type = rusqlite::Transaction<'tx>,
    tx = rusqlite::Transaction<'_>,
    table_exists = sqlite_table_exists
);

impl_embedded_sync_table_data_commit_ops!(
    SqliteCommitOps,
    rusqlite::Transaction<'_>,
    dialect = crate::inlined_values::InlinedSqlDialect::Sqlite,
    next_row_id_for_append =
        crate::stats_engine::next_row_id_for_append_with_data_file_fallback_sync_query,
    insert_inlined_batches = sqlite_insert_inlined_batches,
    overwritten_cleanup_data_path = |_tx| Ok::<String, StorageError>(String::new()),
    write_inlined_file_deletions = self::write_inlined_file_deletions,
    apply_inlined_data_deletions = sqlite_apply_inlined_data_deletions,
    delete_flushed_inlined_data = sqlite_delete_flushed_inlined_data,
    delete_flushed_inlined_file_deletions = sqlite_delete_flushed_inlined_file_deletions,
    insert_table_stats_from_active_files = sqlite_insert_table_stats_from_active_files
);

impl_embedded_sync_maintenance_commit_ops!(
    SqliteCommitOps,
    rusqlite::Transaction<'_>,
    recompute_table_column_stats_from_active_files =
        recompute_sqlite_table_column_stats_from_active_files
);

impl_embedded_sync_catalog_object_commit_ops!(
    SqliteCommitOps,
    rusqlite::Transaction<'_>,
    uuid_style = crate::catalog_queries::UuidParamStyle::Text,
    data_path = sqlite_transaction_data_path,
    close_inlined = close_sqlite_inlined_data_rows_for_table
);

impl crate::commit::SyncCreateTableWithFilesCommitOps for SqliteCommitOps {}

impl_embedded_sync_metadata_commit_ops!(
    SqliteCommitOps,
    rusqlite::Transaction<'_>,
    rewrite_sort_expressions_for_rename = rewrite_sqlite_sort_expressions_for_rename
);

#[async_trait]
impl crate::catalog_initialization_methods::CatalogInitializationExecutor for SqliteCatalog {
    impl_catalog_initialization_info_methods!(
        catalog_type = "sqlite",
        default_data_path = |_this| None,
        snapshot_time_sql = |_this| {
            let now_timestamp = format_timestamp_micros(crate::sql_utils::current_time_micros()?)?;
            Ok(crate::catalog_ddl::sql_string(&now_timestamp))
        }
    );

    impl_sync_catalog_initialization_request!(
        with_session = with_conn,
        capture = |_this| {},
        exists = |conn| sqlite_table_exists(conn, "ducklake_metadata"),
        load = |conn| {
            crate::catalog_metadata::load_catalog_metadata_sync(|query, mapper| {
                run_catalog_query(conn, query, mapper)
            })
        },
        create = |conn, bootstrap| sqlite_initialize_ducklake(conn, &bootstrap),
    );
}

#[async_trait]
impl CatalogBackend for SqliteCatalog {
    impl_catalog_type_method!("sqlite");

    impl_embedded_metadata_method!(with_conn);

    impl_embedded_set_metadata_option_method!(with_conn, |conn| sqlite_write_transaction(conn));

    impl_catalog_initialization_method!();

    impl_catalog_backend_read_methods_direct!();

    impl_embedded_inlined_read_methods!(
        file_deletion_table_exists_sql = crate::catalog_table_exists::SQLITE_TABLE_PRESENT_SQL
    );

    impl_catalog_backend_commit_methods!();

    impl_embedded_catalog_maintenance_methods!(with_conn, |conn| sqlite_write_transaction(conn));
}

fn sqlite_insert_table_stats_from_active_files(conn: &Connection, table_id: TableId) -> Result<()> {
    crate::stats_engine::insert_table_stats_from_active_files_sync(
        table_id,
        |query| {
            run_catalog_query(
                conn,
                &query,
                crate::catalog_queries::active_table_stats_value,
            )?
            .into_iter()
            .next()
            .ok_or_else(|| {
                StorageError::Backend("DuckLake active table-stats query returned no rows".into())
            })
        },
        |query| execute_catalog_command(conn, &query),
    )
}

impl_embedded_insert_inlined_batches_helper!(
    fn sqlite_insert_inlined_batches,
    tx = rusqlite::Transaction<'_>,
    params_from_iter = params_from_iter,
    metadata_i64_value = |value| Value::Integer(value),
    metadata_null_value = Value::Null,
    placeholder = |_idx, _field| Ok("?".to_string()),
    value = sqlite_inlined_value,
);

fn sqlite_inlined_value(field: &SchemaField, array: &dyn Array, row: usize) -> Result<Value> {
    if matches!(
        field.ducklake_type,
        ducklake_catalog::DuckLakeType::Struct
            | ducklake_catalog::DuckLakeType::List
            | ducklake_catalog::DuckLakeType::Map
    ) {
        if array.is_null(row) {
            return Ok(Value::Null);
        }
        return crate::inlined_values::inlined_duckdb_literal_value(
            field,
            array,
            row,
            "SQLite inlined append",
            crate::inlined_values::InlinedIntervalFormat::PreserveNanoseconds,
        )
        .map(Value::Text);
    }
    crate::inlined_values::encode_inlined_scalar_value(
        field,
        array,
        row,
        "SQLite inlined append",
        crate::inlined_values::InlinedIntervalFormat::PreserveNanoseconds,
        &SqliteInlinedValueEncoder,
    )
}

struct SqliteInlinedValueEncoder;

impl_inlined_scalar_value_encoder!(
    impl SqliteInlinedValueEncoder,
    Value = Value,
    {
        null(_field) => Ok(Value::Null),
        boolean(value) => Ok(Value::Integer(i64::from(value))),
        int8(value) => Ok(Value::Integer(i64::from(value))),
        int16(value) => Ok(Value::Integer(i64::from(value))),
        int32(value) => Ok(Value::Integer(i64::from(value))),
        int64(value) => Ok(Value::Integer(value)),
        uint8(value) => Ok(Value::Integer(i64::from(value))),
        uint16(value) => Ok(Value::Integer(i64::from(value))),
        uint32(value) => Ok(Value::Integer(i64::from(value))),
        uint64(_field, value) => Ok(Value::Text(value.to_string())),
        float32(value) => Ok(Value::Text(value.to_string())),
        float64(value) => Ok(Value::Text(value.to_string())),
        text(_field, value) => Ok(Value::Text(value)),
        bytes(_field, value) => Ok(Value::Blob(value)),
    }
);

impl_embedded_inlined_mutation_helpers!(
    tx = rusqlite::Transaction<'_>,
    table_exists = sqlite_table_exists,
    apply_inlined_data_deletions = sqlite_apply_inlined_data_deletions,
    delete_flushed_inlined_data = sqlite_delete_flushed_inlined_data,
    delete_flushed_inlined_file_deletions = sqlite_delete_flushed_inlined_file_deletions,
    write_inlined_file_deletions = write_inlined_file_deletions,
    close_inlined_data_rows_for_table = close_sqlite_inlined_data_rows_for_table,
);

impl_embedded_sort_rename_rewrite_helper!(
    fn rewrite_sqlite_sort_expressions_for_rename,
    tx = rusqlite::Transaction<'_>,
    table_exists = sqlite_table_exists,
);

fn sqlite_active_partition_specs(
    tx: &rusqlite::Transaction<'_>,
    table_id: TableId,
    snapshot: SnapshotId,
) -> Result<Vec<ActivePartitionSpec>> {
    let partitions = run_catalog_query(
        tx,
        &crate::catalog_queries::active_partition_infos(table_id, snapshot),
        crate::embedded_reads::partition_info,
    )?;
    let mut specs = Vec::with_capacity(partitions.len());
    for partition in partitions {
        let columns = run_catalog_query(
            tx,
            &crate::catalog_queries::partition_columns(partition.partition_id),
            crate::embedded_reads::partition_column,
        )?
        .into_iter()
        .map(|column| PartitionColumnCommit {
            column_id: column.column_id,
            transform: column.transform,
        })
        .collect();
        specs.push(ActivePartitionSpec {
            partition_id: partition.partition_id,
            columns,
        });
    }
    Ok(specs)
}

fn recompute_sqlite_table_column_stats_from_active_files(
    conn: &Connection,
    table_id: TableId,
    snapshot: SnapshotId,
) -> Result<()> {
    crate::stats_engine::recompute_table_column_stats_from_active_files_sync(
        table_id,
        snapshot,
        |table| sqlite_table_exists(conn, table),
        |query| execute_catalog_command(conn, &query),
        |query| run_catalog_query(conn, &query, crate::catalog_queries::column_type_value),
        |query| {
            Ok(
                run_catalog_query(conn, &query, crate::catalog_queries::i64_value)?
                    .into_iter()
                    .next()
                    .unwrap_or(0),
            )
        },
        |query| run_catalog_query(conn, &query, crate::embedded_reads::file_column_stats),
        |query| run_catalog_query(conn, &query, crate::embedded_reads::file_variant_stats_row),
        |stats| upsert_sqlite_table_column_stats(conn, stats),
    )
}

fn upsert_sqlite_table_column_stats(conn: &Connection, stats: &TableColumnStats) -> Result<()> {
    crate::stats_engine::upsert_table_column_stats_by_update_sync(stats, |query| {
        execute_catalog_command(conn, &query)
    })
}

impl_embedded_query_helpers!(
    statement = rusqlite::Statement<'_>,
    params_trait = rusqlite::Params,
    row = Row<'_>,
    error = rusqlite::Error,
    params_from_iter = params_from_iter,
    catalog_params = sqlite_catalog_params
);

impl_embedded_async_read_query_executor!(impl SqliteCatalog);

impl_embedded_catalog_params!(
    fn sqlite_catalog_params -> Value {
        null = Value::Null,
        i64 = |value| Value::Integer(value),
        string = |value| Value::Text(value),
        bool = |value| Value::Integer(i64::from(value)),
    }
);

impl_embedded_maintenance_conn!(
    table_exists = sqlite_table_exists,
    catalog_params = sqlite_catalog_params,
    scheduled_cleanup_timestamp = bound_current_time,
);

fn sqlite_table_exists(conn: &Connection, table_name: &str) -> Result<bool> {
    conn.query_row(
        crate::catalog_table_exists::SQLITE_TABLE_EXISTS_SQL,
        params![table_name],
        |row| row.get(0),
    )
    .map_err(sql_err)
}

struct SqliteInlinedValueDecoder;

impl_embedded_inlined_data_reader!(
    dialect = crate::inlined_values::InlinedSqlDialect::Sqlite,
    projection = |field| {
        crate::inlined_values::inlined_read_projection_expr(
            field,
            crate::inlined_values::InlinedSqlDialect::Sqlite,
            crate::embedded_reads::quote_identifier,
        )
    },
    decoder = SqliteInlinedValueDecoder,
);

impl crate::inlined_values::InlinedSqlValueDecoder<ValueRef<'_>> for SqliteInlinedValueDecoder {
    impl_inlined_scalar_decoder_methods!(
        value = ValueRef<'_>,
        backend = "SQLite",
        scalar = sqlite_scalar,
    );
    impl_inlined_scalar_string_decoder_method!(
        value = ValueRef<'_>,
        backend = "SQLite",
        scalar = sqlite_scalar,
    );
}

fn sqlite_scalar(value: ValueRef<'_>) -> InlinedSqlScalarRef<'_> {
    match value {
        ValueRef::Null => InlinedSqlScalarRef::Null,
        ValueRef::Integer(value) => InlinedSqlScalarRef::Signed(i128::from(value)),
        ValueRef::Real(value) => InlinedSqlScalarRef::Float(value),
        ValueRef::Text(value) => InlinedSqlScalarRef::Text(value),
        ValueRef::Blob(value) => InlinedSqlScalarRef::Bytes(value),
    }
}

impl_embedded_sql_err!(fn sql_err, error = rusqlite::Error);

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