//! Shared catalog-maintenance SQL for DuckDB-dialect catalog backends.
//!
//! Cleanup of scheduled files, snapshot expiration with metadata pruning,
//! and compaction source/delete-row lookups use identical SQL on SQLite,
//! DuckDB-file, and Quack. The synchronous maintenance orchestration also lives
//! here for the embedded pair, generic over a small [`MaintenanceConn`] driver
//! surface. Quack reuses the SQL builders but keeps its async transaction,
//! protocol decoding, and execution loops local.

#![allow(dead_code)]

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
use std::fs;
#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
use std::io::ErrorKind;

use ducklake_catalog::DeleteFileId;
#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
use ducklake_catalog::{Snapshot, TableId};

use crate::commit::MaintenanceSourceFile;
#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
use crate::embedded_reads::{self, ReadQuery, quote_identifier};
use crate::embedded_rows::EmbeddedRow;
use crate::sql_utils::join_i64s;
#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
use crate::{
    CleanupFile, CleanupOldFilesOptions, DeleteFileCommit, ExpireSnapshotsOptions, StorageError,
    local_filesystem_path, plan_cleanup_metadata_file_ids, plan_cleanup_old_files,
    plan_expirable_snapshots, plan_snapshot_expiration_prune,
};
use crate::{CleanupMetadataFile, Result, scheduled_cleanup_path_for_table_file};

/// Driver operations the shared maintenance flows need. Implemented for
/// `rusqlite::Connection` and `duckdb::Connection`; transactions participate
/// through their `Deref` to the connection type (pass `&*tx`).
#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait MaintenanceConn {
    fn table_exists(&self, table: &str) -> Result<bool>;
    fn execute_sql(&self, sql: &str) -> Result<()>;
    fn execute_catalog_query(&self, query: &crate::catalog_queries::CatalogQuery) -> Result<usize>;
    fn query_rows<T, F>(&self, sql: &str, params: &[i64], map: F) -> Result<Vec<T>>
    where
        F: FnMut(&dyn EmbeddedRow) -> Result<T>;
    fn metadata_value(&self, key: &str) -> Result<Option<String>>;
    /// Dialect-specific: SQLite binds a Rust-formatted timestamp, DuckDB
    /// uses `CURRENT_TIMESTAMP`.
    fn insert_scheduled_cleanup_files(&self, files: &[CleanupMetadataFile]) -> Result<()>;
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn run_query<T>(conn: &impl MaintenanceConn, query: &ReadQuery<T>) -> Result<Vec<T>> {
    let sql = query.render_embedded();
    conn.query_rows(&sql, &query.params, query.map)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn planned_cleanup_files(
    conn: &impl MaintenanceConn,
    options: CleanupOldFilesOptions,
) -> Result<Vec<CleanupFile>> {
    let data_path = conn.metadata_value("data_path")?.unwrap_or_default();
    let rows = run_query(conn, &embedded_reads::files_scheduled_for_deletion())?;
    plan_cleanup_old_files(options, &data_path, &rows)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn remove_scheduled_files_from_disk(files: &[CleanupFile]) -> Result<()> {
    let local_paths = files
        .iter()
        .map(|file| local_filesystem_path(&file.path))
        .collect::<Result<Vec<_>>>()?;
    for (file, local_path) in files.iter().zip(local_paths) {
        match fs::remove_file(&local_path) {
            Ok(()) => {}
            Err(err) if err.kind() == ErrorKind::NotFound => {}
            Err(err) => {
                return Err(StorageError::Backend(format!(
                    "failed to remove DuckLake scheduled file {}: {err}",
                    file.path
                )));
            }
        }
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn replace_overwritten_delete_file<LoadCleanup>(
    conn: &impl MaintenanceConn,
    file: &DeleteFileCommit,
    new_snapshot: i64,
    load_cleanup: LoadCleanup,
) -> Result<()>
where
    LoadCleanup: FnMut(DeleteFileId, TableId) -> Result<CleanupMetadataFile>,
{
    crate::commit::replace_overwritten_delete_file_sync(
        file,
        new_snapshot,
        |file, new_snapshot| {
            let query = crate::catalog_queries::close_overwritten_delete_file(file, new_snapshot)
                .ok_or_else(|| {
                StorageError::Backend("DuckLake overwrite missing delete file id".into())
            })?;
            conn.execute_catalog_query(&query)
        },
        load_cleanup,
        |file| {
            let query =
                crate::catalog_queries::delete_overwritten_delete_file(file).ok_or_else(|| {
                    StorageError::Backend("DuckLake overwrite missing delete file id".into())
                })?;
            conn.execute_catalog_query(&query)
        },
        |cleanup_row| schedule_cleanup_files(conn, &[cleanup_row]),
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn expirable_snapshots(
    conn: &impl MaintenanceConn,
    options: &ExpireSnapshotsOptions,
) -> Result<Vec<Snapshot>> {
    let latest = conn
        .query_rows(
            "SELECT MAX(snapshot_id) FROM ducklake_snapshot",
            &[],
            |row| row.get_opt_i64(0),
        )?
        .into_iter()
        .next()
        .flatten()
        .unwrap_or(i64::MIN);
    let snapshots = run_query(conn, &embedded_reads::snapshots())?;
    plan_expirable_snapshots(options, latest, snapshots)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn prune_expired_snapshots(
    conn: &impl MaintenanceConn,
    snapshots: &[Snapshot],
) -> Result<()> {
    let snapshot_ids = plan_snapshot_expiration_prune(snapshots, Vec::new(), &[], &[]).snapshot_ids;

    crate::metadata_prune_engine::delete_by_ids_sync(
        conn,
        "ducklake_snapshot",
        "snapshot_id",
        &snapshot_ids,
    )?;
    crate::metadata_prune_engine::delete_by_ids_sync(
        conn,
        "ducklake_snapshot_changes",
        "snapshot_id",
        &snapshot_ids,
    )?;

    let deleted_table_ids = tables_unreferenced_by_snapshots(conn)?;
    let deleted_file_rows = data_files_unreferenced_by_snapshots(conn, &deleted_table_ids)?;
    schedule_cleanup_files(conn, &deleted_file_rows)?;
    let deleted_data_file_ids = plan_cleanup_metadata_file_ids(&deleted_file_rows);
    crate::metadata_prune_engine::delete_file_metadata_sync(conn, &deleted_data_file_ids)?;

    let deleted_delete_file_rows =
        delete_files_unreferenced_by_snapshots(conn, &deleted_table_ids, &deleted_data_file_ids)?;
    schedule_cleanup_files(conn, &deleted_delete_file_rows)?;
    let prune_plan = plan_snapshot_expiration_prune(
        snapshots,
        deleted_table_ids,
        &deleted_file_rows,
        &deleted_delete_file_rows,
    );
    crate::metadata_prune_engine::delete_by_ids_sync(
        conn,
        "ducklake_delete_file",
        "delete_file_id",
        &prune_plan.deleted_delete_file_ids,
    )?;

    crate::metadata_prune_engine::delete_table_metadata_sync(conn, &prune_plan.deleted_table_ids)?;
    crate::metadata_prune_engine::delete_unreferenced_versioned_metadata_sync(conn)?;
    crate::metadata_prune_engine::delete_unreferenced_dependent_metadata_sync(conn)?;
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn tables_unreferenced_by_snapshots(conn: &impl MaintenanceConn) -> Result<Vec<i64>> {
    if !conn.table_exists("ducklake_table")? {
        return Ok(Vec::new());
    }
    conn.query_rows(tables_unreferenced_by_snapshots_sql(), &[], |row| {
        row.get_i64(0)
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn tables_unreferenced_by_snapshots_sql() -> &'static str {
    "SELECT DISTINCT table_id \
     FROM ducklake_table t \
     WHERE end_snapshot IS NOT NULL \
       AND NOT EXISTS ( \
           SELECT 1 FROM ducklake_snapshot s \
           WHERE s.snapshot_id >= t.begin_snapshot AND s.snapshot_id < t.end_snapshot \
       ) \
       AND NOT EXISTS ( \
           SELECT 1 FROM ducklake_table t2 \
           WHERE t2.table_id = t.table_id \
             AND (t2.end_snapshot IS NULL OR EXISTS ( \
                 SELECT 1 FROM ducklake_snapshot s2 \
                 WHERE s2.snapshot_id >= t2.begin_snapshot AND s2.snapshot_id < t2.end_snapshot \
             )) \
       ) \
     ORDER BY table_id"
}

pub(crate) fn tables_unreferenced_by_snapshots_sql_with_refs(
    table_ref: &str,
    snapshot_ref: &str,
) -> String {
    format!(
        "SELECT DISTINCT t.table_id \
         FROM {table_ref} t \
         WHERE t.end_snapshot IS NOT NULL \
           AND NOT EXISTS ( \
               SELECT 1 FROM {snapshot_ref} s \
               WHERE s.snapshot_id >= t.begin_snapshot AND s.snapshot_id < t.end_snapshot \
           ) \
           AND NOT EXISTS ( \
               SELECT 1 FROM {table_ref} t2 \
               WHERE t2.table_id = t.table_id \
                 AND (t2.end_snapshot IS NULL OR EXISTS ( \
                     SELECT 1 FROM {snapshot_ref} s2 \
                     WHERE s2.snapshot_id >= t2.begin_snapshot AND s2.snapshot_id < t2.end_snapshot \
                 )) \
           ) \
         ORDER BY t.table_id"
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn data_files_unreferenced_by_snapshots(
    conn: &impl MaintenanceConn,
    deleted_table_ids: &[i64],
) -> Result<Vec<CleanupMetadataFile>> {
    if !conn.table_exists("ducklake_data_file")? {
        return Ok(Vec::new());
    }
    let data_path = conn.metadata_value("data_path")?.unwrap_or_default();
    conn.query_rows(
        &data_files_unreferenced_by_snapshots_sql(deleted_table_ids),
        &[],
        |row| cleanup_file_row_with_table_path(row, &data_path),
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn data_files_unreferenced_by_snapshots_sql(deleted_table_ids: &[i64]) -> String {
    data_files_unreferenced_by_snapshots_sql_with_refs(
        deleted_table_ids,
        "ducklake_data_file",
        "ducklake_table",
        "ducklake_schema",
        "ducklake_snapshot",
    )
}

pub(crate) fn data_files_unreferenced_by_snapshots_sql_with_refs(
    deleted_table_ids: &[i64],
    data_file_ref: &str,
    table_ref: &str,
    schema_ref: &str,
    snapshot_ref: &str,
) -> String {
    let table_filter = aliased_table_id_or_filter("df", deleted_table_ids);
    format!(
        "SELECT df.data_file_id, df.path, df.path_is_relative, \
                sc.path, sc.path_is_relative, t.path, t.path_is_relative \
         FROM {data_file_ref} df \
         JOIN {table_ref} t ON t.table_id = df.table_id \
          AND t.begin_snapshot = ( \
              SELECT MAX(t2.begin_snapshot) FROM {table_ref} t2 \
              WHERE t2.table_id = df.table_id \
          ) \
         JOIN {schema_ref} sc ON sc.schema_id = t.schema_id \
          AND sc.begin_snapshot = ( \
              SELECT MAX(sc2.begin_snapshot) FROM {schema_ref} sc2 \
              WHERE sc2.schema_id = t.schema_id \
          ) \
         WHERE {table_filter}(df.end_snapshot IS NOT NULL AND NOT EXISTS ( \
             SELECT 1 FROM {snapshot_ref} sn \
             WHERE sn.snapshot_id >= df.begin_snapshot \
               AND sn.snapshot_id < df.end_snapshot \
         )) \
         ORDER BY df.data_file_id"
    )
}

pub(crate) fn data_files_unreferenced_by_snapshots_without_table_path_sql(
    deleted_table_ids: &[i64],
    data_file_ref: &str,
    snapshot_ref: &str,
) -> String {
    let table_filter = aliased_table_id_or_filter("df", deleted_table_ids);
    format!(
        "SELECT df.data_file_id, df.path, df.path_is_relative \
         FROM {data_file_ref} df \
         WHERE {table_filter}(df.end_snapshot IS NOT NULL AND NOT EXISTS ( \
             SELECT 1 FROM {snapshot_ref} sn \
             WHERE sn.snapshot_id >= df.begin_snapshot \
               AND sn.snapshot_id < df.end_snapshot \
         )) \
         ORDER BY df.data_file_id"
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn delete_files_unreferenced_by_snapshots(
    conn: &impl MaintenanceConn,
    deleted_table_ids: &[i64],
    deleted_data_file_ids: &[i64],
) -> Result<Vec<CleanupMetadataFile>> {
    if !conn.table_exists("ducklake_delete_file")? {
        return Ok(Vec::new());
    }
    let data_path = conn.metadata_value("data_path")?.unwrap_or_default();
    conn.query_rows(
        &delete_files_unreferenced_by_snapshots_sql(deleted_table_ids, deleted_data_file_ids),
        &[],
        |row| cleanup_file_row_with_table_path(row, &data_path),
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_files_unreferenced_by_snapshots_sql(
    deleted_table_ids: &[i64],
    deleted_data_file_ids: &[i64],
) -> String {
    delete_files_unreferenced_by_snapshots_sql_with_refs(
        deleted_table_ids,
        deleted_data_file_ids,
        "ducklake_delete_file",
        "ducklake_table",
        "ducklake_schema",
        "ducklake_snapshot",
    )
}

pub(crate) fn delete_files_unreferenced_by_snapshots_sql_with_refs(
    deleted_table_ids: &[i64],
    deleted_data_file_ids: &[i64],
    delete_file_ref: &str,
    table_ref: &str,
    schema_ref: &str,
    snapshot_ref: &str,
) -> String {
    let table_filter = aliased_table_id_or_filter("df", deleted_table_ids);
    let data_file_filter = if deleted_data_file_ids.is_empty() {
        String::new()
    } else {
        format!(
            "df.data_file_id IN ({}) OR ",
            join_i64s(deleted_data_file_ids)
        )
    };
    format!(
        "SELECT df.delete_file_id, df.path, df.path_is_relative, \
                sc.path, sc.path_is_relative, t.path, t.path_is_relative \
         FROM {delete_file_ref} df \
         JOIN {table_ref} t ON t.table_id = df.table_id \
          AND t.begin_snapshot = ( \
              SELECT MAX(t2.begin_snapshot) FROM {table_ref} t2 \
              WHERE t2.table_id = df.table_id \
          ) \
         JOIN {schema_ref} sc ON sc.schema_id = t.schema_id \
          AND sc.begin_snapshot = ( \
              SELECT MAX(sc2.begin_snapshot) FROM {schema_ref} sc2 \
              WHERE sc2.schema_id = t.schema_id \
          ) \
         WHERE {table_filter}{data_file_filter}(df.end_snapshot IS NOT NULL AND NOT EXISTS ( \
             SELECT 1 FROM {snapshot_ref} sn \
             WHERE sn.snapshot_id >= df.begin_snapshot \
               AND sn.snapshot_id < df.end_snapshot \
         )) \
         ORDER BY df.delete_file_id"
    )
}

pub(crate) fn delete_files_unreferenced_by_snapshots_without_table_path_sql(
    deleted_table_ids: &[i64],
    deleted_data_file_ids: &[i64],
    delete_file_ref: &str,
    snapshot_ref: &str,
) -> String {
    let table_filter = aliased_table_id_or_filter("df", deleted_table_ids);
    let data_file_filter = if deleted_data_file_ids.is_empty() {
        String::new()
    } else {
        format!(
            "df.data_file_id IN ({}) OR ",
            join_i64s(deleted_data_file_ids)
        )
    };
    format!(
        "SELECT df.delete_file_id, df.path, df.path_is_relative \
         FROM {delete_file_ref} df \
         WHERE {table_filter}{data_file_filter}(df.end_snapshot IS NOT NULL AND NOT EXISTS ( \
             SELECT 1 FROM {snapshot_ref} sn \
             WHERE sn.snapshot_id >= df.begin_snapshot \
               AND sn.snapshot_id < df.end_snapshot \
         )) \
         ORDER BY df.delete_file_id"
    )
}

pub(crate) fn cleanup_file_row_with_table_path(
    row: &dyn EmbeddedRow,
    data_path: &str,
) -> Result<CleanupMetadataFile> {
    cleanup_file_path_row(row)?.into_cleanup_file(data_path)
}

pub(crate) struct CleanupFilePathRow {
    pub(crate) id: i64,
    pub(crate) file_path: String,
    pub(crate) file_path_is_relative: bool,
    pub(crate) schema_path: Option<String>,
    pub(crate) schema_path_is_relative: bool,
    pub(crate) table_path: Option<String>,
    pub(crate) table_path_is_relative: bool,
}

impl CleanupFilePathRow {
    pub(crate) fn into_cleanup_file(self, data_path: &str) -> Result<CleanupMetadataFile> {
        let (path, path_is_relative) = scheduled_cleanup_path_for_table_file(
            data_path,
            self.schema_path.as_deref(),
            self.schema_path_is_relative,
            self.table_path.as_deref(),
            self.table_path_is_relative,
            &self.file_path,
            self.file_path_is_relative,
        );
        Ok(CleanupMetadataFile {
            id: self.id,
            path,
            path_is_relative,
        })
    }
}

pub(crate) fn cleanup_file_path_row(row: &dyn EmbeddedRow) -> Result<CleanupFilePathRow> {
    Ok(CleanupFilePathRow {
        id: row.get_i64(0)?,
        file_path: row.get_string(1)?,
        file_path_is_relative: row.get_bool(2)?,
        schema_path: row.get_opt_string(3)?,
        schema_path_is_relative: row.get_bool(4)?,
        table_path: row.get_opt_string(5)?,
        table_path_is_relative: row.get_bool(6)?,
    })
}

pub(crate) fn single_overwritten_cleanup_row(
    rows: Vec<CleanupFilePathRow>,
    overwritten: DeleteFileId,
) -> Result<CleanupFilePathRow> {
    let count = rows.len();
    rows.into_iter()
        .next()
        .filter(|_| count == 1)
        .ok_or_else(|| {
            crate::StorageError::Backend(format!(
                "DuckLake overwrite expected one delete file {}, found {count} rows",
                overwritten.0
            ))
        })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn load_overwritten_cleanup_file_sync<Query>(
    overwritten: DeleteFileId,
    table_id: TableId,
    data_path: &str,
    query: Query,
) -> Result<CleanupMetadataFile>
where
    Query: FnOnce(
        &crate::catalog_queries::CatalogQuery,
        fn(&dyn EmbeddedRow) -> Result<CleanupFilePathRow>,
    ) -> Result<Vec<CleanupFilePathRow>>,
{
    single_overwritten_cleanup_row(
        query(
            &crate::catalog_queries::overwritten_delete_file_cleanup_row(overwritten, table_id),
            cleanup_file_path_row,
        )?,
        overwritten,
    )?
    .into_cleanup_file(data_path)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn compaction_source_rows(
    conn: &impl MaintenanceConn,
    table_id: TableId,
    snapshot: i64,
    source_ids: &[i64],
) -> Result<Vec<MaintenanceSourceFile>> {
    let data_path = conn.metadata_value("data_path")?.unwrap_or_default();
    conn.query_rows(
        &compaction_source_rows_sql(source_ids),
        &[table_id.0, snapshot],
        |row| compaction_source_row_with_table_path(row, &data_path),
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn compaction_source_rows_sql(source_ids: &[i64]) -> String {
    compaction_source_rows_sql_with_refs(
        source_ids,
        "ducklake_data_file",
        "ducklake_table",
        "ducklake_schema",
        "?1",
        "?2",
    )
}

pub(crate) fn compaction_source_rows_sql_with_refs(
    source_ids: &[i64],
    data_file_ref: &str,
    table_ref: &str,
    schema_ref: &str,
    table_id_expr: &str,
    snapshot_expr: &str,
) -> String {
    format!(
        "SELECT f.data_file_id, f.path, f.path_is_relative, f.file_size_bytes, \
                s.path, s.path_is_relative, t.path, t.path_is_relative \
         FROM {data_file_ref} f \
         JOIN {table_ref} t USING(table_id) \
         JOIN {schema_ref} s USING(schema_id) \
         WHERE f.table_id = {table_id_expr} \
           AND f.data_file_id IN ({}) \
           AND {snapshot_expr} >= f.begin_snapshot \
           AND (f.end_snapshot IS NULL OR {snapshot_expr} < f.end_snapshot) \
           AND {snapshot_expr} >= t.begin_snapshot \
           AND (t.end_snapshot IS NULL OR {snapshot_expr} < t.end_snapshot) \
           AND {snapshot_expr} >= s.begin_snapshot \
           AND (s.end_snapshot IS NULL OR {snapshot_expr} < s.end_snapshot) \
         ORDER BY f.data_file_id",
        join_i64s(source_ids)
    )
}

pub(crate) fn compaction_source_rows_without_table_path_sql(
    source_ids: &[i64],
    data_file_ref: &str,
    table_id_expr: &str,
    snapshot_expr: &str,
) -> String {
    format!(
        "SELECT f.data_file_id, f.path, f.path_is_relative, f.file_size_bytes \
         FROM {data_file_ref} f \
         WHERE f.table_id = {table_id_expr} \
           AND f.data_file_id IN ({}) \
           AND {snapshot_expr} >= f.begin_snapshot \
           AND (f.end_snapshot IS NULL OR {snapshot_expr} < f.end_snapshot) \
         ORDER BY f.data_file_id",
        join_i64s(source_ids)
    )
}

pub(crate) fn compaction_source_row_without_table_path(
    row: &dyn EmbeddedRow,
) -> Result<MaintenanceSourceFile> {
    Ok(MaintenanceSourceFile {
        id: row.get_i64(0)?,
        cleanup_path: row.get_string(1)?,
        cleanup_path_is_relative: row.get_bool(2)?,
        file_size_bytes: row.get_i64(3)?,
    })
}

pub(crate) fn compaction_source_row_with_table_path(
    row: &dyn EmbeddedRow,
    data_path: &str,
) -> Result<MaintenanceSourceFile> {
    let file_path = row.get_string(1)?;
    let file_path_is_relative = row.get_bool(2)?;
    let schema_path = row.get_opt_string(4)?;
    let schema_path_is_relative = row.get_bool(5)?;
    let table_path = row.get_opt_string(6)?;
    let table_path_is_relative = row.get_bool(7)?;
    let (cleanup_path, cleanup_path_is_relative) = scheduled_cleanup_path_for_table_file(
        data_path,
        schema_path.as_deref(),
        schema_path_is_relative,
        table_path.as_deref(),
        table_path_is_relative,
        &file_path,
        file_path_is_relative,
    );
    Ok(MaintenanceSourceFile {
        id: row.get_i64(0)?,
        file_size_bytes: row.get_i64(3)?,
        cleanup_path,
        cleanup_path_is_relative,
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn compaction_delete_rows(
    conn: &impl MaintenanceConn,
    latest_snapshot: i64,
    source_ids: &[i64],
) -> Result<Vec<CleanupMetadataFile>> {
    if !conn.table_exists("ducklake_delete_file")? {
        return Ok(Vec::new());
    }
    let data_path = conn.metadata_value("data_path")?.unwrap_or_default();
    conn.query_rows(
        &compaction_delete_rows_sql(source_ids),
        &[latest_snapshot],
        |row| cleanup_file_row_with_table_path(row, &data_path),
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn compaction_delete_rows_sql(source_ids: &[i64]) -> String {
    compaction_delete_rows_sql_with_refs(
        source_ids,
        "ducklake_delete_file",
        "ducklake_table",
        "ducklake_schema",
        "?1",
    )
}

pub(crate) fn compaction_delete_rows_sql_with_refs(
    source_ids: &[i64],
    delete_file_ref: &str,
    table_ref: &str,
    schema_ref: &str,
    snapshot_expr: &str,
) -> String {
    format!(
        "SELECT f.delete_file_id, f.path, f.path_is_relative, \
                s.path, s.path_is_relative, t.path, t.path_is_relative \
         FROM {delete_file_ref} f \
         JOIN {table_ref} t USING(table_id) \
         JOIN {schema_ref} s USING(schema_id) \
         WHERE f.data_file_id IN ({}) \
           AND {snapshot_expr} >= f.begin_snapshot \
           AND (f.end_snapshot IS NULL OR {snapshot_expr} < f.end_snapshot) \
           AND {snapshot_expr} >= t.begin_snapshot \
           AND (t.end_snapshot IS NULL OR {snapshot_expr} < t.end_snapshot) \
           AND {snapshot_expr} >= s.begin_snapshot \
           AND (s.end_snapshot IS NULL OR {snapshot_expr} < s.end_snapshot) \
         ORDER BY f.delete_file_id",
        join_i64s(source_ids)
    )
}

pub(crate) fn compaction_delete_rows_without_table_path_sql(
    source_ids: &[i64],
    delete_file_ref: &str,
    snapshot_expr: &str,
) -> String {
    format!(
        "SELECT f.delete_file_id, f.path, f.path_is_relative \
         FROM {delete_file_ref} f \
         WHERE f.data_file_id IN ({}) \
           AND {snapshot_expr} >= f.begin_snapshot \
           AND (f.end_snapshot IS NULL OR {snapshot_expr} < f.end_snapshot) \
         ORDER BY f.delete_file_id",
        join_i64s(source_ids)
    )
}

pub(crate) fn cleanup_file_row_without_table_path(
    row: &dyn EmbeddedRow,
) -> Result<CleanupMetadataFile> {
    Ok(CleanupMetadataFile {
        id: row.get_i64(0)?,
        path: row.get_string(1)?,
        path_is_relative: row.get_bool(2)?,
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_inlined_file_deletions(
    conn: &impl MaintenanceConn,
    table_id: TableId,
    source_ids: &[i64],
) -> Result<()> {
    let table = format!("ducklake_inlined_delete_{}", table_id.0);
    if !conn.table_exists(&table)? {
        return Ok(());
    }
    conn.execute_sql(
        &crate::inlined_values::delete_inlined_file_deletions_for_files_sql(
            &quote_identifier(&table),
            source_ids,
        ),
    )
}

pub(crate) fn insert_scheduled_cleanup_file_sql(
    table: &str,
    id_expr: &str,
    path_expr: &str,
    path_is_relative_expr: &str,
    timestamp_expr: &str,
) -> String {
    format!(
        "INSERT INTO {table} VALUES ({id_expr}, {path_expr}, {path_is_relative_expr}, {timestamp_expr})"
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn schedule_cleanup_files(
    conn: &impl MaintenanceConn,
    files: &[CleanupMetadataFile],
) -> Result<()> {
    if files.is_empty() || !conn.table_exists("ducklake_files_scheduled_for_deletion")? {
        return Ok(());
    }
    conn.insert_scheduled_cleanup_files(files)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_file_metadata(
    conn: &impl MaintenanceConn,
    data_file_ids: &[i64],
) -> Result<()> {
    crate::metadata_prune_engine::delete_file_metadata_sync(conn, data_file_ids)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_by_ids(
    conn: &impl MaintenanceConn,
    table: &str,
    column: &str,
    ids: &[i64],
) -> Result<()> {
    crate::metadata_prune_engine::delete_by_ids_sync(conn, table, column, ids)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
impl<T> crate::metadata_prune_engine::SyncMetadataPruneExecutor for T
where
    T: MaintenanceConn + ?Sized,
{
    fn table_exists(&self, table: &str) -> Result<bool> {
        MaintenanceConn::table_exists(self, table)
    }

    fn inlined_table_names_for_table_ids(&self, table_ids: &[i64]) -> Result<Vec<String>> {
        self.query_rows(
            &format!(
                "SELECT table_name FROM ducklake_inlined_data_tables WHERE table_id IN ({})",
                join_i64s(table_ids)
            ),
            &[],
            |row| row.get_string(0),
        )
    }

    fn drop_inlined_table(&self, table_name: &str) -> Result<()> {
        self.execute_sql(&format!(
            "DROP TABLE IF EXISTS {}",
            quote_identifier(table_name)
        ))
    }

    fn snapshot_table_ref(&self) -> String {
        "ducklake_snapshot".to_string()
    }

    fn macro_table_ref(&self) -> String {
        "ducklake_macro".to_string()
    }

    fn delete_where(&self, table: &str, _target_ref: Option<&str>, predicate: &str) -> Result<()> {
        if !MaintenanceConn::table_exists(self, table)? {
            return Ok(());
        }
        self.execute_sql(&format!("DELETE FROM {table} WHERE {predicate}"))
    }
}

fn aliased_table_id_or_filter(alias: &str, table_ids: &[i64]) -> String {
    if table_ids.is_empty() {
        String::new()
    } else {
        format!("{alias}.table_id IN ({}) OR ", join_i64s(table_ids))
    }
}

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