datafusion-ducklake 0.8.0

//! Multicatalog manager: create, look up, and bootstrap DuckLake catalogs.
//!
//! See [`crate::metadata_writer_postgres::PostgresMetadataWriter`] for the writer
//! that binds to a `catalog_id` produced here.

use crate::Result;
use crate::maintenance::{
    CleanupCriteria, ExpireCriteria, ExpiredSnapshot, ScheduledFile, format_sql_timestamp,
};
use crate::metadata_writer_postgres::execute_ddl_statements;
use chrono::{DateTime, Utc};
use sqlx::AssertSqlSafe;
use sqlx::Row;
use sqlx::postgres::{PgPool, PgPoolOptions, PgRow};

/// SQL expression yielding a file path resolved relative to the catalog `data_path`
/// root (file → table → schema → data_path). An absolute path anywhere in the chain
/// short-circuits to that absolute path. Mirrors the SQLite writer's `RESOLVED_PATH`.
const PG_RESOLVED_PATH: &str = "CASE
    WHEN NOT df.path_is_relative THEN df.path
    WHEN NOT t.path_is_relative THEN t.path || '/' || df.path
    ELSE s.path || '/' || t.path || '/' || df.path
END";

/// Companion to [`PG_RESOLVED_PATH`]: true only when the whole chain is relative.
const PG_REL_FLAG: &str = "(df.path_is_relative AND t.path_is_relative AND s.path_is_relative)";

/// Bootstrap standard + multicatalog tables. Idempotent.
pub async fn initialize_multicatalog_schema(pool: &PgPool) -> Result<()> {
    execute_ddl_statements(
        pool,
        crate::metadata_writer_postgres::SQL_CREATE_STANDARD_TABLES,
    )
    .await?;
    execute_ddl_statements(
        pool,
        crate::metadata_writer_postgres::SQL_CREATE_MULTICATALOG_TABLES,
    )
    .await?;
    crate::metadata_writer_postgres::migrate_column_default_metadata(pool).await?;
    // Upgrade a pre-existing store's ducklake_column from the legacy single-row
    // column_id PK to the composite (table_id, column_id, begin_snapshot) PK so
    // versioned / type-promoted columns work on catalogs created before this
    // change. Idempotent (no-op once composite). This is the multicatalog
    // bootstrap path — runtimedb and the multicatalog tests call it — so the
    // migration MUST run here, not only in PostgresMetadataWriter::initialize_schema.
    crate::metadata_writer_postgres::migrate_ducklake_column_to_composite_pk(pool).await?;
    // Give every file row an explicit owner (NULL = this catalog's own) rather than
    // reading ownership off `path_is_relative`, and build the indexes that serve the
    // reference lookup. Additive and idempotent; MUST run here for the same reason as
    // the migration above — this is the bootstrap path callers actually use.
    crate::metadata_writer_postgres::migrate_file_ownership_column(pool).await?;
    Ok(())
}

/// Manages DuckLake catalogs within a shared metadata database.
///
/// `MulticatalogManager` is stateless: it just executes SQL against the pool.
/// Multiple managers against the same pool are safe.
#[derive(Debug, Clone)]
pub struct MulticatalogManager {
    pool: PgPool,
}

impl MulticatalogManager {
    /// Connect to PostgreSQL and initialize the multicatalog schema.
    pub async fn connect(connection_string: &str, max_connections: u32) -> Result<Self> {
        let pool = PgPoolOptions::new()
            .max_connections(max_connections)
            .connect(connection_string)
            .await?;
        initialize_multicatalog_schema(&pool).await?;
        Ok(Self::new(pool))
    }

    /// Construct a manager bound to a Postgres pool.
    pub fn new(pool: PgPool) -> Self {
        Self {
            pool,
        }
    }

    /// Build a catalog-scoped provider sharing this manager's pool.
    pub async fn provider(&self, catalog_id: i64) -> Result<crate::MulticatalogProvider> {
        crate::MulticatalogProvider::with_pool_and_id(self.pool.clone(), catalog_id).await
    }

    /// Build a catalog-scoped writer sharing this manager's pool.
    pub async fn writer(&self, catalog_id: i64) -> Result<crate::PostgresMetadataWriter> {
        crate::PostgresMetadataWriter::with_pool(self.pool.clone(), catalog_id).await
    }

    /// Create a catalog with the given name, returning its `catalog_id`.
    /// Idempotent. Concurrency-safety relies on the `UNIQUE (catalog_name)`
    /// constraint on `ducklake_catalog`.
    pub async fn create_catalog(&self, name: &str) -> Result<i64> {
        if name.trim().is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "Catalog name cannot be empty".to_string(),
            ));
        }

        // Fast path: already exists.
        if let Some(id) = self.find_catalog_id(name).await? {
            return Ok(id);
        }

        // Try to insert; if a concurrent writer beat us, fall back to lookup.
        let insert = sqlx::query(
            "INSERT INTO ducklake_catalog (catalog_name) VALUES ($1)
             ON CONFLICT (catalog_name) DO NOTHING
             RETURNING catalog_id",
        )
        .bind(name)
        .fetch_optional(&self.pool)
        .await?;

        if let Some(row) = insert {
            return Ok(row.try_get(0)?);
        }

        // Conflict: row exists from a concurrent writer. Look it up.
        self.find_catalog_id(name).await?.ok_or_else(|| {
            crate::DuckLakeError::Internal(format!(
                "Catalog '{}' insert conflicted but lookup found nothing — \
                 the row was inserted and then deleted concurrently",
                name
            ))
        })
    }

    /// Look up a catalog by name, returning `None` if not found.
    pub async fn find_catalog_id(&self, name: &str) -> Result<Option<i64>> {
        let row = sqlx::query("SELECT catalog_id FROM ducklake_catalog WHERE catalog_name = $1")
            .bind(name)
            .fetch_optional(&self.pool)
            .await?;

        match row {
            Some(r) => Ok(Some(r.try_get(0)?)),
            None => Ok(None),
        }
    }

    /// List all catalogs as (id, name) pairs ordered by id.
    pub async fn list_catalogs(&self) -> Result<Vec<(i64, String)>> {
        let rows = sqlx::query(
            "SELECT catalog_id, catalog_name FROM ducklake_catalog ORDER BY catalog_id",
        )
        .fetch_all(&self.pool)
        .await?;

        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let id: i64 = row.try_get(0)?;
            let name: String = row.try_get(1)?;
            out.push((id, name));
        }
        Ok(out)
    }

    /// Drop a catalog and every metadata row owned by it.
    ///
    /// Removes the catalog row, its mapping-table entries, and every
    /// entity reachable through those mappings: snapshots, schemas,
    /// tables, columns, data and delete file records, schema-version
    /// history. Idempotent — returns `false` if no catalog by that name
    /// exists.
    ///
    /// # Concurrency
    ///
    /// The catalog row is taken with `FOR UPDATE` before any deletion,
    /// so concurrent writers (which acquire the same lock in
    /// `lock_catalog`) serialise against the drop. A 30s `lock_timeout`
    /// bounds how long this waits for an in-progress writer to release
    /// the lock — matches
    /// [`crate::metadata_writer_postgres::DEFAULT_LOCK_TIMEOUT_MS`]. The
    /// entire teardown is one transaction; either everything is gone
    /// or nothing changes.
    ///
    /// # What's NOT cleaned up
    ///
    /// Data files on object storage are not removed here. The catalog's
    /// effective data path is retained so orphan cleanup can visit it after
    /// the metadata rows are gone. Callers needing synchronous cleanup must
    /// still arrange it around this metadata-only operation.
    pub async fn drop_catalog(&self, name: &str) -> Result<bool> {
        if name.trim().is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "Catalog name cannot be empty".to_string(),
            ));
        }

        let mut tx = self.pool.begin().await?;

        sqlx::query("SET LOCAL lock_timeout = 30000")
            .execute(&mut *tx)
            .await?;

        // Resolve catalog_id under FOR UPDATE. Concurrent writers (which
        // also take FOR UPDATE on this row in `lock_catalog`) serialise
        // against the drop. If the row has already been deleted by a
        // parallel drop, we treat this call as a no-op.
        let row = sqlx::query(
            "SELECT catalog_id, data_path
             FROM ducklake_catalog
             WHERE catalog_name = $1 FOR UPDATE",
        )
        .bind(name)
        .fetch_optional(&mut *tx)
        .await?;

        let (catalog_id, data_path): (i64, Option<String>) = match row {
            Some(r) => (r.try_get("catalog_id")?, r.try_get("data_path")?),
            None => {
                tx.commit().await?;
                return Ok(false);
            },
        };

        // Order matters: anything that reads through the catalog mapping
        // tables must run before we delete those mapping rows. Within
        // catalog-owned entities the order is rows-that-reference-
        // `table_id` first, then tables, then schemas, then snapshots.

        // Catalog-scoped child rows that reference `table_id`. Filtered
        // transitively through the schema map so we only touch rows
        // owned by this catalog.
        // Same set expire reclaims per dead table, plus the two per-file tables:
        // all are keyed on `table_id`, so they take the scoping this loop already
        // applies. Dropping a catalog without them orphans exactly what expire used
        // to orphan (#248), and on SQLite a reused data_file_id would then inherit
        // a dead file's stats and partition values.
        for child_table in [
            "ducklake_data_file",
            "ducklake_delete_file",
            "ducklake_file_column_stats",
            "ducklake_file_partition_value",
            "ducklake_column",
            "ducklake_schema_versions",
            "ducklake_table_stats",
            "ducklake_table_column_stats",
            "ducklake_partition_info",
            "ducklake_partition_column",
            "ducklake_sort_info",
            "ducklake_sort_expression",
        ] {
            sqlx::query(AssertSqlSafe(format!(
                "DELETE FROM {} WHERE table_id IN (
                    SELECT t.table_id FROM ducklake_table t
                    JOIN ducklake_catalog_schema_map m ON m.schema_id = t.schema_id
                    WHERE m.catalog_id = $1
                )",
                child_table
            )))
            .bind(catalog_id)
            .execute(&mut *tx)
            .await?;
        }

        // Tables and schemas owned by this catalog.
        sqlx::query(
            "DELETE FROM ducklake_table WHERE schema_id IN (
                SELECT schema_id FROM ducklake_catalog_schema_map WHERE catalog_id = $1
            )",
        )
        .bind(catalog_id)
        .execute(&mut *tx)
        .await?;

        sqlx::query(
            "DELETE FROM ducklake_schema WHERE schema_id IN (
                SELECT schema_id FROM ducklake_catalog_schema_map WHERE catalog_id = $1
            )",
        )
        .bind(catalog_id)
        .execute(&mut *tx)
        .await?;

        // Snapshots are catalog-scoped only via the snapshot map; they
        // carry no catalog_id of their own, so the map is the only path
        // to locate them.
        sqlx::query(
            "DELETE FROM ducklake_snapshot_changes WHERE snapshot_id IN (
                SELECT snapshot_id FROM ducklake_catalog_snapshot_map WHERE catalog_id = $1
            )",
        )
        .bind(catalog_id)
        .execute(&mut *tx)
        .await?;

        sqlx::query(
            "DELETE FROM ducklake_snapshot WHERE snapshot_id IN (
                SELECT snapshot_id FROM ducklake_catalog_snapshot_map WHERE catalog_id = $1
            )",
        )
        .bind(catalog_id)
        .execute(&mut *tx)
        .await?;

        // Now safe to remove the mapping rows — nothing reads through
        // them anymore.
        sqlx::query("DELETE FROM ducklake_catalog_schema_map WHERE catalog_id = $1")
            .bind(catalog_id)
            .execute(&mut *tx)
            .await?;
        sqlx::query("DELETE FROM ducklake_catalog_snapshot_map WHERE catalog_id = $1")
            .bind(catalog_id)
            .execute(&mut *tx)
            .await?;

        sqlx::query("DELETE FROM ducklake_files_scheduled_for_deletion WHERE catalog_id = $1")
            .bind(catalog_id)
            .execute(&mut *tx)
            .await?;

        if let Some(data_path) = data_path {
            sqlx::query(
                "INSERT INTO ducklake_dropped_data_path (data_path)
                 VALUES ($1)
                 ON CONFLICT (data_path) DO UPDATE SET dropped_at = NOW()",
            )
            .bind(data_path)
            .execute(&mut *tx)
            .await?;
        }

        // Finally the catalog row itself.
        sqlx::query("DELETE FROM ducklake_catalog WHERE catalog_id = $1")
            .bind(catalog_id)
            .execute(&mut *tx)
            .await?;

        tx.commit().await?;
        Ok(true)
    }

    /// Tombstone a single table inside a catalog at a new commit snapshot.
    ///
    /// Allocates a new snapshot (the "drop snapshot") and sets
    /// `end_snapshot = <drop snapshot>` on the currently-live rows for
    /// the named table in `ducklake_table`, `ducklake_column`,
    /// `ducklake_data_file`, and `ducklake_delete_file`. After commit,
    /// reads at any snapshot `>=` the drop snapshot will not see the
    /// table; reads at earlier snapshots are unaffected.
    ///
    /// Idempotent — returns `Ok(false)` when no live table by that
    /// `(catalog, schema, table)` triple exists. No snapshot is
    /// allocated in that case.
    ///
    /// # Concurrency
    ///
    /// The catalog row is taken with `FOR UPDATE` before any work, so
    /// concurrent writers (which acquire the same lock in
    /// `lock_catalog`) serialise against the drop. A 30s
    /// `lock_timeout` bounds how long this waits for an in-progress
    /// writer to release the lock — matches
    /// [`crate::metadata_writer_postgres::DEFAULT_LOCK_TIMEOUT_MS`].
    /// The entire teardown is one transaction; either the table is
    /// tombstoned at the new snapshot or nothing changes.
    ///
    /// # What's NOT cleaned up
    ///
    /// Data files on object storage are not removed here, and the
    /// tombstoned metadata rows remain in place under their
    /// `end_snapshot`. Physical reclamation is the job of a future
    /// expire-snapshots / cleanup-files vacuum (mirrors the official
    /// DuckLake design). Callers needing tighter storage cleanup
    /// should list the table's data file paths via
    /// [`crate::MetadataProvider::get_table_files_for_select`] BEFORE
    /// calling this and schedule their removal externally.
    pub async fn drop_table_in_catalog(
        &self,
        catalog_name: &str,
        schema_name: &str,
        table_name: &str,
    ) -> Result<bool> {
        if catalog_name.trim().is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "Catalog name cannot be empty".to_string(),
            ));
        }
        if schema_name.trim().is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "Schema name cannot be empty".to_string(),
            ));
        }
        if table_name.trim().is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "Table name cannot be empty".to_string(),
            ));
        }

        let mut tx = self.pool.begin().await?;

        sqlx::query("SET LOCAL lock_timeout = 30000")
            .execute(&mut *tx)
            .await?;

        // Resolve catalog_id under FOR UPDATE. Concurrent writers (which
        // also take FOR UPDATE on this row in `lock_catalog`) serialise
        // against the drop. An unknown catalog is a no-op — same
        // idempotency contract as `drop_catalog`.
        let catalog_id: i64 = match sqlx::query(
            "SELECT catalog_id FROM ducklake_catalog
             WHERE catalog_name = $1 FOR UPDATE",
        )
        .bind(catalog_name)
        .fetch_optional(&mut *tx)
        .await?
        {
            Some(r) => r.try_get(0)?,
            None => {
                tx.commit().await?;
                return Ok(false);
            },
        };

        // Resolve the live table_id by `(catalog, schema, table)`.
        // Filters on `end_snapshot IS NULL` on both schema and table so
        // an already-tombstoned table is treated as not-found and the
        // call becomes an idempotent no-op.
        let table_id: i64 = match sqlx::query(
            "SELECT t.table_id FROM ducklake_table t
             JOIN ducklake_schema s ON s.schema_id = t.schema_id
             JOIN ducklake_catalog_schema_map m ON m.schema_id = s.schema_id
             WHERE m.catalog_id = $1
               AND s.schema_name = $2
               AND s.end_snapshot IS NULL
               AND t.table_name = $3
               AND t.end_snapshot IS NULL",
        )
        .bind(catalog_id)
        .bind(schema_name)
        .bind(table_name)
        .fetch_optional(&mut *tx)
        .await?
        {
            Some(r) => r.try_get(0)?,
            None => {
                tx.commit().await?;
                return Ok(false);
            },
        };

        // Allocate the drop snapshot and register it under this catalog.
        // Every snapshot has a row in `ducklake_catalog_snapshot_map` —
        // same discipline as `MetadataWriter::create_snapshot` and
        // `begin_write_transaction`.
        //
        // DROP TABLE is a DDL change (the set of live tables in the
        // catalog shrinks by one), so the drop snapshot bumps
        // `schema_version` — `prev_max + 1` per the per-catalog dense
        // contract that `begin_write_transaction` enforces for DDL
        // commits. Computed AFTER the snapshot is inserted, filtered
        // strictly less than its own id, so concurrent writers (which
        // hold the same catalog FOR UPDATE lock we do) can't slip a
        // row in to invalidate the dense allocation.
        let drop_snapshot: i64 = sqlx::query(
            "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
             VALUES (CURRENT_TIMESTAMP, 0) RETURNING snapshot_id",
        )
        .fetch_one(&mut *tx)
        .await?
        .try_get(0)?;

        sqlx::query(
            "INSERT INTO ducklake_catalog_snapshot_map (catalog_id, snapshot_id)
             VALUES ($1, $2)",
        )
        .bind(catalog_id)
        .bind(drop_snapshot)
        .execute(&mut *tx)
        .await?;

        // Bump per-catalog dense schema_version: prior MAX + 1 (DDL).
        // Falls back to 1 when this is the catalog's first commit ever,
        // mirroring `begin_write_transaction`'s "no prior snapshot ⇒ v1"
        // guard so the schema_version stays monotone-from-1.
        let prev_max: i64 = sqlx::query(
            "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
             JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
             WHERE m.catalog_id = $1 AND s.snapshot_id < $2",
        )
        .bind(catalog_id)
        .bind(drop_snapshot)
        .fetch_one(&mut *tx)
        .await?
        .try_get(0)?;
        let new_schema_version = if prev_max == 0 {
            1
        } else {
            prev_max + 1
        };
        sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
            .bind(new_schema_version)
            .bind(drop_snapshot)
            .execute(&mut *tx)
            .await?;

        // Tombstone the table row and every currently-live child row
        // keyed by `table_id`. The `end_snapshot IS NULL` guard makes
        // each UPDATE a no-op for rows already tombstoned at an earlier
        // snapshot (e.g. data files superseded by a prior REPLACE).
        //
        // `ducklake_schema_versions` rows are not tombstoned — that
        // table has no `end_snapshot` column. They become unreferenced
        // after the table is fully expired; future vacuum reclaims
        // them. This mirrors the official DuckLake's `DropTables`
        // (which leaves `ducklake_schema_versions` to vacuum likewise).
        for child_table in
            ["ducklake_table", "ducklake_column", "ducklake_data_file", "ducklake_delete_file"]
        {
            sqlx::query(AssertSqlSafe(format!(
                "UPDATE {} SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
                child_table
            )))
            .bind(drop_snapshot)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
        }

        sqlx::query(
            "INSERT INTO ducklake_snapshot_changes
                 (snapshot_id, changes_made, author, commit_message, commit_extra_info)
             VALUES ($1, $2, NULL, NULL, NULL)",
        )
        .bind(drop_snapshot)
        .bind(format!("dropped_table:{table_id}"))
        .execute(&mut *tx)
        .await?;
        tx.commit().await?;
        Ok(true)
    }

    /// Read the legacy global `data_path` fallback.
    ///
    /// Use [`Self::get_data_path_in_catalog`] for a catalog-scoped root.
    pub async fn get_data_path(&self) -> Result<String> {
        let row =
            sqlx::query("SELECT value FROM ducklake_metadata WHERE key = $1 AND scope IS NULL")
                .bind("data_path")
                .fetch_optional(&self.pool)
                .await?;
        match row {
            Some(r) => Ok(r.try_get(0)?),
            None => Err(crate::DuckLakeError::InvalidConfig(
                "Missing required catalog metadata: 'data_path' not configured.".to_string(),
            )),
        }
    }

    /// Read the `data_path` for one catalog.
    ///
    /// Before the writer migration adds the catalog column, readers use the legacy global value.
    pub async fn get_data_path_in_catalog(&self, catalog_name: &str) -> Result<String> {
        let path: Option<String> =
            if crate::multicatalog_provider::catalog_has_data_path(&self.pool).await? {
                sqlx::query_scalar("SELECT data_path FROM ducklake_catalog WHERE catalog_name = $1")
                    .bind(catalog_name)
                    .fetch_optional(&self.pool)
                    .await?
                    .flatten()
            } else {
                sqlx::query_scalar(
                    "SELECT metadata.value
                 FROM ducklake_catalog AS catalog
                 CROSS JOIN LATERAL (
                     SELECT value FROM ducklake_metadata
                     WHERE key = 'data_path' AND scope IS NULL LIMIT 1
                 ) AS metadata
                 WHERE catalog.catalog_name = $1",
                )
                .bind(catalog_name)
                .fetch_optional(&self.pool)
                .await?
            };
        path.ok_or_else(|| {
            crate::DuckLakeError::InvalidConfig(format!(
                "Missing required catalog metadata: 'data_path' not configured for catalog {catalog_name:?}."
            ))
        })
    }

    /// List every live, legacy, or dropped catalog root that may need orphan cleanup.
    pub async fn list_data_paths(&self) -> Result<Vec<String>> {
        let paths = if crate::multicatalog_provider::catalog_has_data_path(&self.pool).await? {
            sqlx::query_scalar(
                "SELECT data_path FROM (
                     SELECT value AS data_path
                     FROM ducklake_metadata
                     WHERE key = 'data_path' AND scope IS NULL
                     UNION
                     SELECT data_path FROM ducklake_catalog
                     UNION
                     SELECT data_path FROM ducklake_dropped_data_path
                 ) AS paths
                 WHERE data_path IS NOT NULL
                 ORDER BY data_path",
            )
            .fetch_all(&self.pool)
            .await?
        } else {
            sqlx::query_scalar(
                "SELECT value FROM ducklake_metadata
                 WHERE key = 'data_path' AND scope IS NULL",
            )
            .fetch_all(&self.pool)
            .await?
        };
        Ok(paths)
    }

    pub(crate) async fn current_timestamp(&self) -> Result<DateTime<Utc>> {
        Ok(sqlx::query_scalar("SELECT NOW()")
            .fetch_one(&self.pool)
            .await?)
    }

    pub(crate) async fn clear_dropped_data_paths(
        &self,
        data_paths: &[String],
        dropped_before: DateTime<Utc>,
    ) -> Result<()> {
        sqlx::query(
            "DELETE FROM ducklake_dropped_data_path
             WHERE data_path::TEXT = ANY($1) AND dropped_at <= $2",
        )
        .bind(data_paths)
        .bind(dropped_before)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Expire snapshots within a single catalog and GC the metadata they leave behind.
    ///
    /// The multicatalog mirror of [`crate::metadata_writer_sqlite::SqliteMetadataWriter::expire_snapshots`].
    /// The most recent snapshot *of this catalog* is never expired. After deleting the
    /// chosen snapshot rows (and their `ducklake_catalog_snapshot_map` rows), every table /
    /// data file / delete file no longer reachable by any surviving snapshot of this catalog
    /// is removed and its physical path recorded in `ducklake_files_scheduled_for_deletion`
    /// (tagged with `catalog_id`) for later deletion via
    /// [`crate::maintenance::cleanup_old_files_in_catalog`]. Returns the expired snapshots.
    ///
    /// # Concurrency
    ///
    /// Takes the catalog row `FOR UPDATE` (30s `lock_timeout`) before any work, serialising
    /// against in-flight writers and other maintenance — same discipline as
    /// [`Self::drop_table_in_catalog`].
    ///
    /// # Catalog scoping
    ///
    /// Because Postgres is multi-catalog (one shared `ducklake_snapshot` /
    /// `ducklake_data_file`, partitioned by the `*_map` tables), every reachability check
    /// is restricted to *this* catalog's snapshots. Otherwise another catalog's snapshot
    /// sitting in the `[begin_snapshot, end_snapshot)` range of a file would wrongly keep it
    /// alive forever.
    pub async fn expire_snapshots_in_catalog(
        &self,
        catalog_name: &str,
        criteria: ExpireCriteria,
    ) -> Result<Vec<ExpiredSnapshot>> {
        let mut tx = self.pool.begin().await?;
        sqlx::query("SET LOCAL lock_timeout = 30000")
            .execute(&mut *tx)
            .await?;

        let catalog_id: i64 = match sqlx::query(
            "SELECT catalog_id FROM ducklake_catalog WHERE catalog_name = $1 FOR UPDATE",
        )
        .bind(catalog_name)
        .fetch_optional(&mut *tx)
        .await?
        {
            Some(r) => r.try_get(0)?,
            None => {
                tx.commit().await?;
                return Ok(Vec::new());
            },
        };

        // Most recent snapshot OF THIS CATALOG — never expirable.
        let most_recent: Option<i64> = sqlx::query(
            "SELECT MAX(s.snapshot_id) FROM ducklake_snapshot s
             JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
             WHERE m.catalog_id = $1",
        )
        .bind(catalog_id)
        .fetch_one(&mut *tx)
        .await?
        .try_get(0)?;
        let most_recent = match most_recent {
            Some(id) => id,
            None => {
                tx.commit().await?;
                return Ok(Vec::new());
            },
        };

        // 1. Resolve the candidate snapshots (catalog-scoped, never the most recent).
        let candidate_rows: Vec<PgRow> = match &criteria {
            ExpireCriteria::Versions(versions) => {
                let ids: Vec<i64> = versions
                    .iter()
                    .copied()
                    .filter(|&v| v != most_recent)
                    .collect();
                sqlx::query(
                    "SELECT s.snapshot_id, s.snapshot_time::text FROM ducklake_snapshot s
                     JOIN ducklake_catalog_snapshot_map m
                       ON m.snapshot_id = s.snapshot_id AND m.catalog_id = $1
                     WHERE s.snapshot_id <> $2 AND s.snapshot_id = ANY($3)
                     ORDER BY s.snapshot_id",
                )
                .bind(catalog_id)
                .bind(most_recent)
                .bind(&ids)
                .fetch_all(&mut *tx)
                .await?
            },
            ExpireCriteria::OlderThan(ts) => {
                sqlx::query(
                    "SELECT s.snapshot_id, s.snapshot_time::text FROM ducklake_snapshot s
                     JOIN ducklake_catalog_snapshot_map m
                       ON m.snapshot_id = s.snapshot_id AND m.catalog_id = $1
                     WHERE s.snapshot_id <> $2 AND s.snapshot_time < $3::timestamp
                     ORDER BY s.snapshot_id",
                )
                .bind(catalog_id)
                .bind(most_recent)
                .bind(format_sql_timestamp(ts))
                .fetch_all(&mut *tx)
                .await?
            },
        };
        let candidates: Vec<ExpiredSnapshot> = candidate_rows
            .into_iter()
            .map(|r| {
                Ok(ExpiredSnapshot {
                    snapshot_id: r.try_get(0)?,
                    snapshot_time: r.try_get(1)?,
                })
            })
            .collect::<Result<_>>()?;
        if candidates.is_empty() {
            tx.commit().await?;
            return Ok(Vec::new());
        }
        let expire_ids: Vec<i64> = candidates.iter().map(|s| s.snapshot_id).collect();

        // 2. Delete the snapshots: map rows first, then the (now catalog-private) rows.
        sqlx::query(
            "DELETE FROM ducklake_catalog_snapshot_map
             WHERE catalog_id = $1 AND snapshot_id = ANY($2)",
        )
        .bind(catalog_id)
        .bind(&expire_ids)
        .execute(&mut *tx)
        .await?;
        sqlx::query("DELETE FROM ducklake_snapshot WHERE snapshot_id = ANY($1)")
            .bind(&expire_ids)
            .execute(&mut *tx)
            .await?;
        // Official deletes ducklake_snapshot_changes in the same loop as
        // ducklake_snapshot (DeleteSnapshots). We write one row per commit, so
        // leaving them behind orphans at commit rate rather than at expiry rate.
        sqlx::query("DELETE FROM ducklake_snapshot_changes WHERE snapshot_id = ANY($1)")
            .bind(&expire_ids)
            .execute(&mut *tx)
            .await?;

        // 3. Tables in this catalog no longer covered by any surviving snapshot and with
        //    no surviving live version.
        let dead_tables: Vec<i64> = sqlx::query(
            "SELECT t.table_id FROM ducklake_table t
             JOIN ducklake_catalog_schema_map csm
               ON csm.schema_id = t.schema_id AND csm.catalog_id = $1
             WHERE t.end_snapshot IS NOT NULL AND NOT EXISTS (
                 SELECT 1 FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m
                   ON m.snapshot_id = s.snapshot_id AND m.catalog_id = $1
                 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 s
                       JOIN ducklake_catalog_snapshot_map m
                         ON m.snapshot_id = s.snapshot_id AND m.catalog_id = $1
                       WHERE s.snapshot_id >= t2.begin_snapshot
                         AND s.snapshot_id < t2.end_snapshot)))",
        )
        .bind(catalog_id)
        .fetch_all(&mut *tx)
        .await?
        .into_iter()
        .map(|r| r.try_get::<i64, _>(0))
        .collect::<std::result::Result<_, _>>()?;

        // 4. Data files orphaned (in this catalog): schedule paths, then drop the rows.
        //    `= ANY($2)` with an empty array is simply false — no special-casing needed.
        let dead_data_files = sqlx::query(AssertSqlSafe(format!(
            "SELECT df.data_file_id, {PG_RESOLVED_PATH} AS resolved_path, {PG_REL_FLAG} AS rel,
                    df.owner_catalog_id IS NULL AS owned
             FROM ducklake_data_file df
             JOIN ducklake_table t ON t.table_id = df.table_id
             JOIN ducklake_schema s ON s.schema_id = t.schema_id
             JOIN ducklake_catalog_schema_map csm
               ON csm.schema_id = s.schema_id AND csm.catalog_id = $1
             WHERE (df.table_id = ANY($2)) OR (df.end_snapshot IS NOT NULL AND NOT EXISTS (
                 SELECT 1 FROM ducklake_snapshot ss
                 JOIN ducklake_catalog_snapshot_map m
                   ON m.snapshot_id = ss.snapshot_id AND m.catalog_id = $1
                 WHERE ss.snapshot_id >= df.begin_snapshot AND ss.snapshot_id < df.end_snapshot))"
        )))
        .bind(catalog_id)
        .bind(&dead_tables)
        .fetch_all(&mut *tx)
        .await?;
        let data_file_ids = schedule_pg_files(&mut tx, catalog_id, dead_data_files).await?;
        sqlx::query("DELETE FROM ducklake_data_file WHERE data_file_id = ANY($1)")
            .bind(&data_file_ids)
            .execute(&mut *tx)
            .await?;
        // Per-file metadata dies with its file. Official deletes these in the same
        // group as ducklake_data_file, keyed on the same ids; the compaction path
        // here already does the same for the sources it retires. Without this the
        // rows are unreachable (every read joins through ducklake_data_file) but
        // never reclaimed, so the stats table grows without bound and is scanned
        // in full on the planning path of every query.
        // ducklake_file_variant_stats is the fourth table official deletes in this
        // group. It is not in this schema, so it is omitted rather than skipped at
        // runtime. IF THAT TABLE IS EVER ADDED, ADD IT HERE TOO.
        //
        // That instruction is the whole point of this block: omitting exactly one
        // per-file table here is how ducklake_file_column_stats came to hold rows
        // for files deleted long ago. Nothing reads them — every read joins through
        // ducklake_data_file — and nothing collects them, because expire only cleans
        // what it retires in its own transaction. The table grows without bound and
        // is read while planning every query, so the cost lands on queries that
        // touch none of it.
        for table in ["ducklake_file_column_stats", "ducklake_file_partition_value"] {
            sqlx::query(AssertSqlSafe(format!(
                "DELETE FROM {table} WHERE data_file_id = ANY($1)"
            )))
            .bind(&data_file_ids)
            .execute(&mut *tx)
            .await?;
        }

        // 5. Delete files orphaned by the data files above, a dead table, or no surviving
        //    snapshot.
        let dead_delete_files = sqlx::query(AssertSqlSafe(format!(
            "SELECT df.delete_file_id, {PG_RESOLVED_PATH} AS resolved_path, {PG_REL_FLAG} AS rel,
                    df.owner_catalog_id IS NULL AS owned
             FROM ducklake_delete_file df
             JOIN ducklake_table t ON t.table_id = df.table_id
             JOIN ducklake_schema s ON s.schema_id = t.schema_id
             JOIN ducklake_catalog_schema_map csm
               ON csm.schema_id = s.schema_id AND csm.catalog_id = $1
             WHERE (df.data_file_id = ANY($2)) OR (df.table_id = ANY($3))
                OR (df.end_snapshot IS NOT NULL AND NOT EXISTS (
                    SELECT 1 FROM ducklake_snapshot ss
                    JOIN ducklake_catalog_snapshot_map m
                      ON m.snapshot_id = ss.snapshot_id AND m.catalog_id = $1
                    WHERE ss.snapshot_id >= df.begin_snapshot AND ss.snapshot_id < df.end_snapshot))"
        )))
        .bind(catalog_id)
        .bind(&data_file_ids)
        .bind(&dead_tables)
        .fetch_all(&mut *tx)
        .await?;
        let delete_file_ids = schedule_pg_files(&mut tx, catalog_id, dead_delete_files).await?;
        sqlx::query("DELETE FROM ducklake_delete_file WHERE delete_file_id = ANY($1)")
            .bind(&delete_file_ids)
            .execute(&mut *tx)
            .await?;

        // 6. Reclaim per-table metadata for fully-expired tables. This is official's
        //    twelve-table list from DeleteSnapshots.
        //    Official's list is twelve (ducklake_metadata_manager.cpp, DeleteSnapshots).
        //    ducklake_column_tag, ducklake_inlined_data_tables and
        //    ducklake_column_mapping are omitted because this schema does not have
        //    them. IF ANY OF THE THREE IS EVER ADDED, ADD IT HERE TOO — official
        //    reclaims it, and a table-scoped table this loop misses orphans forever.
        //    ducklake_table_stats is in it because this writer
        //    creates and maintains it — the comment that used to say otherwise was
        //    wrong — and the SQLite expire path already reclaimed it.
        for table in [
            "ducklake_table",
            "ducklake_table_stats",
            "ducklake_table_column_stats",
            "ducklake_partition_info",
            "ducklake_partition_column",
            "ducklake_column",
            "ducklake_sort_info",
            "ducklake_sort_expression",
            "ducklake_schema_versions",
        ] {
            sqlx::query(AssertSqlSafe(format!(
                "DELETE FROM {table} WHERE table_id = ANY($1)"
            )))
            .bind(&dead_tables)
            .execute(&mut *tx)
            .await?;
        }

        // 7. Reclaim schemas no longer covered by a surviving snapshot of this catalog,
        //    along with their catalog_schema_map rows (no orphan maps).
        let dead_schemas: Vec<i64> = sqlx::query(
            "SELECT s.schema_id FROM ducklake_schema s
             JOIN ducklake_catalog_schema_map csm
               ON csm.schema_id = s.schema_id AND csm.catalog_id = $1
             WHERE s.end_snapshot IS NOT NULL AND NOT EXISTS (
                 SELECT 1 FROM ducklake_snapshot ss
                 JOIN ducklake_catalog_snapshot_map m
                   ON m.snapshot_id = ss.snapshot_id AND m.catalog_id = $1
                 WHERE ss.snapshot_id >= s.begin_snapshot AND ss.snapshot_id < s.end_snapshot)",
        )
        .bind(catalog_id)
        .fetch_all(&mut *tx)
        .await?
        .into_iter()
        .map(|r| r.try_get::<i64, _>(0))
        .collect::<std::result::Result<_, _>>()?;
        sqlx::query(
            "DELETE FROM ducklake_catalog_schema_map WHERE catalog_id = $1 AND schema_id = ANY($2)",
        )
        .bind(catalog_id)
        .bind(&dead_schemas)
        .execute(&mut *tx)
        .await?;
        sqlx::query("DELETE FROM ducklake_schema WHERE schema_id = ANY($1)")
            .bind(&dead_schemas)
            .execute(&mut *tx)
            .await?;

        tx.commit().await?;
        Ok(candidates)
    }

    /// List files scheduled for physical deletion for `catalog_name`, optionally filtered
    /// by schedule time. Unknown catalog ⇒ empty list.
    pub(crate) async fn list_scheduled_for_deletion_in_catalog(
        &self,
        catalog_name: &str,
        criteria: &CleanupCriteria,
    ) -> Result<Vec<ScheduledFile>> {
        let catalog_id = match self.find_catalog_id(catalog_name).await? {
            Some(id) => id,
            None => return Ok(Vec::new()),
        };
        let rows = match criteria {
            CleanupCriteria::All => {
                sqlx::query(
                    "SELECT data_file_id, path, path_is_relative, scheduled_by_owner
                 FROM ducklake_files_scheduled_for_deletion WHERE catalog_id = $1",
                )
                .bind(catalog_id)
                .fetch_all(&self.pool)
                .await?
            },
            CleanupCriteria::OlderThan(ts) => {
                sqlx::query(
                    "SELECT data_file_id, path, path_is_relative, scheduled_by_owner
                 FROM ducklake_files_scheduled_for_deletion
                 WHERE catalog_id = $1 AND schedule_start < $2::timestamptz",
                )
                .bind(catalog_id)
                .bind(format_sql_timestamp(ts))
                .fetch_all(&self.pool)
                .await?
            },
        };
        rows.into_iter()
            .map(|r| {
                Ok(ScheduledFile {
                    data_file_id: r.try_get(0)?,
                    path: r.try_get(1)?,
                    path_is_relative: r.try_get(2)?,
                    scheduled_by_owner: r.try_get(3)?,
                })
            })
            .collect()
    }

    /// Remove scheduled-deletion bookkeeping rows for `catalog_name` after their objects
    /// are gone. Unknown catalog ⇒ no-op.
    ///
    /// Keyed by `path`, not by `data_file_id`: data files and delete files are two
    /// independent identity sequences that share this table's `data_file_id` column, so
    /// a data file and a delete file routinely collide on one id. That was harmless
    /// while every listed row was deleted in the same pass, but a deferred row (one an
    /// absolute reference still names) now stays behind while its id-twin is reclaimed —
    /// and removing by id would take the deferred row with it, losing the owner's only
    /// record that the object is still meant to be reclaimed.
    pub(crate) async fn remove_scheduled_paths_in_catalog(
        &self,
        catalog_name: &str,
        paths: &[String],
    ) -> Result<()> {
        if paths.is_empty() {
            return Ok(());
        }
        let catalog_id = match self.find_catalog_id(catalog_name).await? {
            Some(id) => id,
            None => return Ok(()),
        };
        sqlx::query(
            "DELETE FROM ducklake_files_scheduled_for_deletion
             WHERE catalog_id = $1 AND path = ANY($2)",
        )
        .bind(catalog_id)
        .bind(paths)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Every catalog's `(catalog_id, data_path)`, for a caller that must reason about
    /// where OTHER catalogs' own files live. Catalogs with no `data_path` configured
    /// are omitted: nothing can be said about where their files sit.
    pub(crate) async fn list_catalog_data_paths(&self) -> Result<Vec<(i64, String)>> {
        let rows = sqlx::query(
            "SELECT catalog_id, data_path::TEXT FROM ducklake_catalog
             WHERE data_path IS NOT NULL",
        )
        .fetch_all(&self.pool)
        .await?;
        rows.into_iter()
            .map(|r| Ok((r.try_get::<i64, _>(0)?, r.try_get::<String, _>(1)?)))
            .collect()
    }

    /// Every path a `ducklake_data_file` or `ducklake_delete_file` row in ANY catalog
    /// references — one naming another catalog as its owner — exactly as stored.
    ///
    /// Deliberately unscoped: such a row references a file another catalog owns (see
    /// `schedule_pg_files`), so the question "does anyone still read this object" has
    /// to look across catalogs.
    ///
    /// Returns the rows rather than answering a membership question about a caller's
    /// list, because the caller must compare *canonical object-store keys* and SQL
    /// equality cannot. Two spellings of one path — a doubled separator, a leading
    /// slash — name the same object and read identically, so a stored row matched by
    /// string equality against a probe list is matched only if the caller happened to
    /// guess its spelling. A probe that misses reads as "nobody references this" and
    /// deletes a live file, which is the failure this whole rule exists to prevent.
    /// Selection is therefore left broad and the comparison done by the caller in
    /// Rust, the way [`crate::maintenance::delete_orphaned_files_multicatalog`] has
    /// always compared against real object locations.
    ///
    /// The partial indexes `idx_data_file_owner_catalog` /
    /// `idx_delete_file_owner_catalog` serve this as an index-only scan, so the
    /// scanned set is bounded by how many cross-catalog references exist rather than
    /// by the size of the file tables.
    ///
    /// A catalog's OWN absolute-path file is deliberately absent: it is not a
    /// reference, and including it would make the catalog defer its own reclaim
    /// against itself forever.
    pub async fn all_reference_paths(&self) -> Result<Vec<String>> {
        let rows = sqlx::query(
            "SELECT path FROM ducklake_data_file WHERE owner_catalog_id IS NOT NULL
             UNION
             SELECT path FROM ducklake_delete_file WHERE owner_catalog_id IS NOT NULL",
        )
        .fetch_all(&self.pool)
        .await?;
        rows.into_iter()
            .map(|r| Ok(r.try_get::<String, _>(0)?))
            .collect()
    }

    /// Push the scheduled rows for `paths` in `catalog_name` back to "scheduled
    /// now", so a `CleanupCriteria::OlderThan` sweep does not re-examine them
    /// until a full grace period has passed again. Used for files that are
    /// still referenced by another catalog. Matches on `path`: the
    /// `data_file_id` column holds delete-file ids too, so ids alone collide.
    /// Unknown catalog ⇒ no-op.
    pub(crate) async fn defer_scheduled_in_catalog(
        &self,
        catalog_name: &str,
        paths: &[String],
    ) -> Result<()> {
        if paths.is_empty() {
            return Ok(());
        }
        let catalog_id = match self.find_catalog_id(catalog_name).await? {
            Some(id) => id,
            None => return Ok(()),
        };
        sqlx::query(
            "UPDATE ducklake_files_scheduled_for_deletion SET schedule_start = NOW()
             WHERE catalog_id = $1 AND path = ANY($2)",
        )
        .bind(catalog_id)
        .bind(paths)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Every physical file referenced by a catalog whose effective `data_path`
    /// matches one of `data_paths`. Canonical-root cleanup must preserve aliases.
    ///
    /// Reference rows are collected from EVERY catalog, not only those in the swept
    /// group. Such a row names a file another catalog owns, and nothing constrains the
    /// referrer to share the owner's root: a catalog on a sibling `data_path`
    /// referencing a file under the swept one is in neither the group nor its
    /// nested-prefix merge, so scoping this arm would make its references invisible
    /// and the sweep would delete a file that catalog is reading. The rows are cheap
    /// to include — the partial indexes on `owner_catalog_id` serve them — and a path
    /// outside the swept root simply never matches a listed object, so widening the
    /// arm cannot preserve anything that should have been reclaimed.
    ///
    /// This arm covers references only, not every absolutely-spelled row. A catalog's
    /// OWN absolute-path file is already covered by the scoped arms whenever its
    /// catalog is in the swept group, because [`PG_RESOLVED_PATH`] short-circuits to
    /// the stored path. What is left uncovered is a catalog writing its own file into
    /// a DIFFERENT catalog's `data_path` — interleaving two catalogs' objects in one
    /// tree with nothing recording whose is whose — which this layout does not
    /// support.
    pub(crate) async fn list_referenced_paths_in_data_paths(
        &self,
        data_paths: &[String],
    ) -> Result<Vec<(String, bool)>> {
        let q = format!(
            "WITH root_catalog AS (
                 SELECT catalog_id FROM ducklake_catalog
                 WHERE data_path::TEXT = ANY($1)
             )
             SELECT {PG_RESOLVED_PATH} AS p, {PG_REL_FLAG} AS rel
             FROM ducklake_data_file df
             JOIN ducklake_table t ON t.table_id = df.table_id
             JOIN ducklake_schema s ON s.schema_id = t.schema_id
             JOIN ducklake_catalog_schema_map m ON m.schema_id = s.schema_id
             WHERE m.catalog_id IN (SELECT catalog_id FROM root_catalog)
             UNION ALL
             SELECT {PG_RESOLVED_PATH} AS p, {PG_REL_FLAG} AS rel
             FROM ducklake_delete_file df
             JOIN ducklake_table t ON t.table_id = df.table_id
             JOIN ducklake_schema s ON s.schema_id = t.schema_id
             JOIN ducklake_catalog_schema_map m ON m.schema_id = s.schema_id
             WHERE m.catalog_id IN (SELECT catalog_id FROM root_catalog)
             UNION ALL
             SELECT path AS p, path_is_relative AS rel
             FROM ducklake_files_scheduled_for_deletion
             WHERE catalog_id IN (SELECT catalog_id FROM root_catalog)
             UNION ALL
             SELECT path AS p, FALSE AS rel
             FROM ducklake_data_file WHERE owner_catalog_id IS NOT NULL
             UNION ALL
             SELECT path AS p, FALSE AS rel
             FROM ducklake_delete_file WHERE owner_catalog_id IS NOT NULL"
        );
        let rows = sqlx::query(AssertSqlSafe(q.as_str()))
            .bind(data_paths)
            .fetch_all(&self.pool)
            .await?;
        rows.into_iter()
            .map(|r| Ok((r.try_get::<String, _>(0)?, r.try_get::<bool, _>(1)?)))
            .collect()
    }
}

/// Schedule the physical files behind dead `(id, resolved_path, rel, owned)` rows
/// and return EVERY dead id, scheduled or not, so the caller deletes all the rows.
///
/// A catalog owns a file unless the row names another catalog as its owner
/// (`ducklake_data_file.owner_catalog_id`, or the delete-file equivalent). A row that
/// does is a *reference* to a file another catalog owns — a database fork registers
/// the source's files this way instead of copying them. Such a row is dropped like
/// any other dead row, but its object is never scheduled: the owner reclaims it
/// through its own expire, and `cleanup_old_files_in_catalog` holds that reclaim back
/// for as long as any reference row still points at the path.
///
/// The test is ownership, NOT how the path is spelled. Both the resolved chain flag
/// (`rel`) and the file row's own `path_is_relative` go false for a catalog's own
/// file — under an absolute schema or table path, or when
/// [`crate::DuckLakeTableWriter::begin_write_to_path`] put the file outside
/// `data_path` — and reading either as "someone else's" strands the object with
/// nothing left in the catalog naming it.
async fn schedule_pg_files(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    catalog_id: i64,
    rows: Vec<PgRow>,
) -> Result<Vec<i64>> {
    let mut ids = Vec::with_capacity(rows.len());
    for row in rows {
        let id: i64 = row.try_get(0)?;
        let path: String = row.try_get(1)?;
        let rel: bool = row.try_get(2)?;
        let owned: bool = row.try_get(3)?;
        ids.push(id);
        if !owned {
            continue;
        }
        // `scheduled_by_owner` marks this row as one the ownership rule produced, so
        // cleanup need not fall back to the layout check it applies to rows written
        // before the rule existed. It never exempts a row from the check that refuses
        // an object inside ANOTHER catalog's layout, which holds unconditionally.
        sqlx::query(
            "INSERT INTO ducklake_files_scheduled_for_deletion
                 (catalog_id, data_file_id, path, path_is_relative, schedule_start,
                  scheduled_by_owner)
             VALUES ($1, $2, $3, $4, NOW(), TRUE)",
        )
        .bind(catalog_id)
        .bind(id)
        .bind(&path)
        .bind(rel)
        .execute(&mut **tx)
        .await?;
    }
    Ok(ids)
}