//! DuckLake catalog backends and file access.
//!
//! This crate owns the async backend boundary used by
//! `datafusion-ducklake-provider`. It adds catalog I/O, object-store path
//! handling, metadata commits, cleanup, and maintenance hooks on top of the
//! pure DuckLake metadata logic in `ducklake-catalog`.
//!
//! Backend drivers are feature-gated:
//!
//! - `sqlite` enables the bundled SQLite backend and is enabled by default.
//! - `duckdb` enables embedded DuckDB-file catalogs.
//! - `postgres` enables remote Postgres catalogs.
//! - `quack` enables the client-only DuckDB Quack protocol backend.
//! - `object-store-s3` enables S3-compatible object-store support through
//!   `object_store`.
//!
//! Remote-only builds can disable default features and enable `remote` without
//! linking SQLite or DuckDB native driver crates.

use std::collections::HashMap;

use arrow::record_batch::RecordBatch;
use async_trait::async_trait;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) use ducklake_catalog::TableSchema;
use ducklake_catalog::{
    CatalogError, CatalogMetadata, ColumnEntry, ColumnId, ColumnMapping, ColumnTagEntry, DataFile,
    DataFileId, DeleteFile, FileColumnStats, FilePartitionValue, FileScheduledForDeletion,
    FileVariantStats, InlinedDataTableRef, InlinedFileDeletion, MacroEntry, MacroId, MacroImpl,
    MacroImplId, MacroParameter, MappingId, NameMappingEntry, PartitionColumn, PartitionInfo,
    SUPPORTED_VERSION, SchemaEntry, SchemaVersion, Snapshot, SnapshotChange, SortExpression,
    SortId, SortInfo, TableColumnStats, TableEntry, TableStats, TagEntry, ViewEntry,
};

/// Errors surfaced by storage backends.
#[derive(Debug, thiserror::Error)]
pub enum StorageError {
    #[error(transparent)]
    Catalog(#[from] CatalogError),

    #[error("backend error: {0}")]
    Backend(String),

    #[error(
        "DuckLake {operation} conflict detected; expected snapshot {expected_snapshot}, found {found_snapshot}"
    )]
    Conflict {
        operation: String,
        expected_snapshot: i64,
        found_snapshot: i64,
    },

    #[error("unsupported DuckLake feature: {0}")]
    Unsupported(String),
}

impl StorageError {
    pub fn conflict(
        operation: impl Into<String>,
        expected_snapshot: ducklake_catalog::SnapshotId,
        found_snapshot: i64,
    ) -> Self {
        Self::Conflict {
            operation: operation.into(),
            expected_snapshot: expected_snapshot.0,
            found_snapshot,
        }
    }

    pub fn is_conflict(&self) -> bool {
        matches!(self, Self::Conflict { .. })
    }
}

pub type Result<T> = std::result::Result<T, StorageError>;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListedFile {
    pub data_file: String,
    pub data_file_size_bytes: i64,
    pub data_file_footer_size: Option<i64>,
    pub data_file_encryption_key: Option<String>,
    pub delete_file: Option<String>,
    pub delete_file_size_bytes: Option<i64>,
    pub delete_file_footer_size: Option<i64>,
    pub delete_file_encryption_key: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TableInfo {
    pub table_name: String,
    pub schema_id: ducklake_catalog::SchemaId,
    pub table_id: ducklake_catalog::TableId,
    pub table_uuid: String,
    pub record_count: i64,
    pub column_count: i64,
    pub file_count: i64,
    pub file_size_bytes: i64,
    pub delete_file_count: i64,
    pub delete_file_size_bytes: i64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListedCatalogPath {
    pub object_kind: String,
    pub schema_name: Option<String>,
    pub table_name: Option<String>,
    pub data_file_id: Option<DataFileId>,
    pub path: Option<String>,
    pub path_is_relative: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListedSnapshot {
    pub snapshot_id: ducklake_catalog::SnapshotId,
    pub snapshot_time: i64,
    pub schema_version: i64,
    pub next_catalog_id: i64,
    pub next_file_id: i64,
    pub changes_made: Option<String>,
    pub author: Option<String>,
    pub commit_message: Option<String>,
    pub commit_extra_info: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListedOption {
    pub option_name: String,
    pub description: Option<String>,
    pub value: String,
    pub scope: String,
    pub scope_entry: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListedTag {
    pub object_kind: String,
    pub object_id: i64,
    pub schema_name: Option<String>,
    pub object_name: Option<String>,
    pub column_id: Option<ColumnId>,
    pub column_name: Option<String>,
    pub key: String,
    pub value: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CatalogSettings {
    pub catalog_type: String,
    pub extension_version: String,
    pub data_path: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MetadataOptionScope {
    pub scope: Option<String>,
    pub scope_id: Option<i64>,
}

impl MetadataOptionScope {
    pub fn global() -> Self {
        Self {
            scope: None,
            scope_id: None,
        }
    }

    pub fn schema(schema_id: ducklake_catalog::SchemaId) -> Self {
        Self {
            scope: Some("schema".to_string()),
            scope_id: Some(schema_id.0),
        }
    }

    pub fn table(table_id: ducklake_catalog::TableId) -> Self {
        Self {
            scope: Some("table".to_string()),
            scope_id: Some(table_id.0),
        }
    }
}

/// Commit-conflict behavior for backend metadata writes.
///
/// The current implementation follows an abort policy: if the latest snapshot
/// no longer matches the writer's expected snapshot, the backend returns
/// [`StorageError::Conflict`] and does not retry automatically.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum CommitConflictPolicy {
    #[default]
    Abort,
}

#[path = "engine/table_data.rs"]
mod table_data;
pub use table_data::{
    AppendFile, DeleteFileCommit, FlushedInlinedDataCommit, FlushedInlinedFileDeletion,
    FlushedInlinedFileDeletionCommit, InlinedDataAppend, InlinedDataAppendCommit, InlinedDataBatch,
    InlinedDataDeletionCommit, InlinedFileDeletionCommit, TableChangeCommit,
};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) use table_data::{
    AppendFilePlan, AppendFilesSnapshotPlan, DeleteFilesSnapshotPlan, InlinedDataAppendPlan,
    TableChangeSnapshotPlan, plan_append_files_commit, plan_delete_files_commit,
    plan_table_change_commit, validate_inlined_append_rows,
};

#[path = "engine/catalog_objects.rs"]
mod catalog_objects;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) use catalog_objects::{
    CatalogObjectMutationPlan, CreateTableWithFilesCommitPlan, PlannedCreateTableWithFilesFile,
    ducklake_quote_identifier, ducklake_sql_string_literal, plan_create_schema_catalog_mutation,
    plan_create_table_catalog_mutation, plan_create_table_with_files_commit,
    plan_create_table_with_inlined_data_commit, plan_create_view_catalog_mutation,
    plan_drop_schema_catalog_mutation, plan_drop_table_catalog_mutation,
    plan_drop_view_catalog_mutation, plan_rename_table_catalog_mutation,
};
pub use catalog_objects::{
    CreateSchemaCommit, CreateTableColumn, CreateTableCommit, CreateTableWithFilesCommit,
    CreateTableWithInlinedDataCommit, CreateViewCommit, DropSchemaCommit, DropTableCommit,
    DropViewCommit, RenameTableCommit,
};

#[path = "engine/schema_evolution.rs"]
mod schema_evolution;
pub use schema_evolution::{
    AddColumn, AddColumnCommit, ColumnDefaultCommit, ColumnNullabilityCommit, ColumnTypeCommit,
    ColumnTypeTree, ColumnTypeTreeCommit, DropColumnCommit, RenameColumnCommit,
};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) use schema_evolution::{
    CatalogSnapshotIds, ColumnCloseRequest, ColumnSubtreeCloseRequest, DropColumnFacts,
    MetadataMutationPlan, PlannedColumnReplacement, PlannedColumnTypeChange,
    PlannedColumnTypeTreeChange, PlannedColumnTypeTreeRow, drop_column_sort_reference_lookup,
    plan_add_columns_commit, plan_column_default_mutation, plan_column_nullability_mutation,
    plan_column_type_mutation, plan_column_type_tree_mutation, plan_drop_column_mutation,
    plan_rename_column_mutation_start, validate_add_column_commit,
};

#[path = "engine/table_properties.rs"]
mod table_properties;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) use table_properties::{
    ActivePartitionSpec, ActiveSortSpec, PartitioningBatchPlan, PartitioningTableFacts,
    SortedByChangePlan, SortedByCommitPlan, TruncateTableCommitPlan, TruncateTableSnapshotPlan,
    deterministic_catalog_uuid, ensure_expected_snapshot, final_partitioning_changes,
    plan_partitioning_batch_commit, plan_sorted_by_commit, plan_truncate_table_commit,
    sort_expression_commit, sort_key_index, validate_create_table_columns,
    validate_sort_expression_columns,
};
pub use table_properties::{
    PartitionColumnCommit, PartitioningBatchCommit, PartitioningCommit, PartitioningTableCommit,
    SortExpressionCommit, SortedByCommit, TruncateTableCommit,
};

#[path = "engine/maintenance.rs"]
mod maintenance;
pub use maintenance::{
    CleanupFile, CleanupOldFilesOptions, ExpireSnapshotsOptions, MergeAdjacentCompactionCommit,
    RewriteDataFilesCommit,
};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) use maintenance::{
    CleanupMetadataFile, MaintenanceCommitPlan, plan_cleanup_metadata_file_ids,
    plan_cleanup_old_files, plan_expirable_snapshots, plan_merge_adjacent_compaction_commit,
    plan_merge_adjacent_compaction_sources, plan_remove_files_scheduled_for_deletion,
    plan_rewrite_data_files_commit, plan_rewrite_data_files_sources,
    plan_snapshot_expiration_prune,
};

#[derive(Debug, Clone, Default)]
pub struct DuckLakeCatalogConnectionOptions {
    pub data_path_override: Option<String>,
    pub metadata_options: HashMap<String, String>,
    pub busy_timeout_ms: Option<u64>,
}

impl DuckLakeCatalogConnectionOptions {
    pub fn apply_to_metadata(&self, mut metadata: CatalogMetadata) -> CatalogMetadata {
        if let Some(data_path) = &self.data_path_override {
            metadata.data_path = data_path.clone();
        }
        for (key, value) in &self.metadata_options {
            let has_persisted_global = metadata
                .raw
                .iter()
                .any(|kv| kv.key.eq_ignore_ascii_case(key) && kv.scope.is_none());
            if !has_persisted_global {
                metadata.raw.push(ducklake_catalog::MetadataKv {
                    key: key.clone(),
                    value: value.clone(),
                    scope: None,
                    scope_id: None,
                });
            }
        }
        metadata
    }
}

#[derive(Debug, Clone)]
pub struct DuckLakeCatalogInitOptions {
    pub data_path: Option<String>,
    pub create_if_not_exists: bool,
    pub automatic_migration: bool,
    pub override_data_path: bool,
    pub encrypted: Option<bool>,
    pub metadata_options: HashMap<String, String>,
}

impl Default for DuckLakeCatalogInitOptions {
    fn default() -> Self {
        Self {
            data_path: None,
            create_if_not_exists: true,
            automatic_migration: false,
            override_data_path: false,
            encrypted: None,
            metadata_options: HashMap::new(),
        }
    }
}

pub(crate) fn validate_loaded_ducklake_metadata(
    catalog_type: &str,
    metadata: &CatalogMetadata,
    options: &DuckLakeCatalogInitOptions,
) -> Result<CatalogMetadata> {
    if metadata.version != SUPPORTED_VERSION {
        if options.automatic_migration {
            return Err(StorageError::Backend(format!(
                "automatic DuckLake catalog migration from version {} is not implemented for {catalog_type} catalogs",
                metadata.version
            )));
        }
        return Err(StorageError::Backend(format!(
            "DuckLake catalog version mismatch: catalog version is {}, but this crate supports {}. To automatically migrate later, set AUTOMATIC_MIGRATION true when attaching.",
            metadata.version, SUPPORTED_VERSION
        )));
    }

    if let Some(explicit_data_path) = options.data_path.as_deref() {
        let explicit_data_path = ducklake_store_data_path(explicit_data_path);
        if !metadata.data_path.is_empty()
            && metadata.data_path != explicit_data_path
            && !options.override_data_path
        {
            return Err(StorageError::Backend(format!(
                "DuckLake catalog data_path is {:?}, but ATTACH specified {:?}; set OVERRIDE_DATA_PATH true to use a connection-local override",
                metadata.data_path, explicit_data_path
            )));
        }
    }
    Ok(metadata.clone())
}

/// Async catalog backend boundary.
///
/// Driver crates must stay inside their cfg-gated modules; this trait exposes
/// only pure DuckLake catalog structs.
#[async_trait]
pub trait CatalogBackend: Send + Sync {
    fn catalog_type(&self) -> &'static str {
        "unknown"
    }

    async fn metadata(&self) -> Result<CatalogMetadata>;
    async fn initialize_ducklake(
        &self,
        options: DuckLakeCatalogInitOptions,
    ) -> Result<CatalogMetadata> {
        match self.metadata().await {
            Ok(metadata) => {
                validate_loaded_ducklake_metadata(self.catalog_type(), &metadata, &options)
            }
            Err(err) if options.create_if_not_exists => Err(StorageError::Backend(format!(
                "DuckLake catalog creation is not implemented for {} catalogs: {err}",
                self.catalog_type()
            ))),
            Err(err) => Err(err),
        }
    }

    async fn snapshots(&self) -> Result<Vec<Snapshot>>;
    async fn snapshot_changes(&self) -> Result<Vec<SnapshotChange>>;
    async fn schema_versions(&self, table: ducklake_catalog::TableId)
    -> Result<Vec<SchemaVersion>>;
    async fn schemas_at(&self, snapshot: ducklake_catalog::SnapshotId) -> Result<Vec<SchemaEntry>>;
    async fn tables_at(
        &self,
        schema: ducklake_catalog::SchemaId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<TableEntry>>;
    async fn views_at(
        &self,
        schema: ducklake_catalog::SchemaId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<ViewEntry>>;
    async fn tags_at(&self, snapshot: ducklake_catalog::SnapshotId) -> Result<Vec<TagEntry>>;
    async fn column_tags_at(
        &self,
        table: ducklake_catalog::TableId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<ColumnTagEntry>>;
    async fn columns_at(
        &self,
        table: ducklake_catalog::TableId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<ColumnEntry>>;
    async fn data_files_at(
        &self,
        table: ducklake_catalog::TableId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<DataFile>>;
    async fn delete_files_at(
        &self,
        table: ducklake_catalog::TableId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<DeleteFile>>;
    async fn file_stats(&self, table: ducklake_catalog::TableId) -> Result<Vec<FileColumnStats>>;
    async fn file_variant_stats(
        &self,
        table: ducklake_catalog::TableId,
    ) -> Result<Vec<FileVariantStats>>;
    async fn table_stats(&self, table: ducklake_catalog::TableId) -> Result<Option<TableStats>>;
    async fn table_column_stats(
        &self,
        table: ducklake_catalog::TableId,
    ) -> Result<Vec<TableColumnStats>>;
    async fn partition_info_at(
        &self,
        table: ducklake_catalog::TableId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<PartitionInfo>>;
    async fn partition_columns(
        &self,
        partition: ducklake_catalog::PartitionId,
    ) -> Result<Vec<PartitionColumn>>;
    async fn file_partition_values(
        &self,
        table: ducklake_catalog::TableId,
    ) -> Result<Vec<FilePartitionValue>>;
    async fn files_scheduled_for_deletion(&self) -> Result<Vec<FileScheduledForDeletion>>;
    async fn column_mapping(
        &self,
        mapping: MappingId,
    ) -> Result<(ColumnMapping, Vec<NameMappingEntry>)>;
    async fn inlined_tables(
        &self,
        table: ducklake_catalog::TableId,
        schema_version: i64,
    ) -> Result<Vec<InlinedDataTableRef>>;
    async fn inlined_file_deletions_at(
        &self,
        table: ducklake_catalog::TableId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<InlinedFileDeletion>>;
    async fn macros_at(
        &self,
        schema: ducklake_catalog::SchemaId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<MacroEntry>>;
    async fn macro_impls(&self, macro_id: MacroId) -> Result<Vec<MacroImpl>>;
    async fn macro_parameters(
        &self,
        macro_id: MacroId,
        impl_id: MacroImplId,
    ) -> Result<Vec<MacroParameter>>;
    async fn sort_info_at(
        &self,
        table: ducklake_catalog::TableId,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<SortInfo>>;
    async fn sort_expressions(&self, sort_id: SortId) -> Result<Vec<SortExpression>>;
    async fn inlined_data(
        &self,
        table: &InlinedDataTableRef,
        schema: &ducklake_catalog::TableSchema,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<RecordBatch>>;

    async fn inlined_data_with_metadata(
        &self,
        table: &InlinedDataTableRef,
        schema: &ducklake_catalog::TableSchema,
        snapshot: ducklake_catalog::SnapshotId,
    ) -> Result<Vec<InlinedDataBatch>> {
        Ok(self
            .inlined_data(table, schema, snapshot)
            .await?
            .into_iter()
            .map(InlinedDataBatch::without_metadata)
            .collect())
    }

    fn commit_conflict_policy(&self) -> CommitConflictPolicy {
        CommitConflictPolicy::Abort
    }

    async fn commit_append_file(&self, file: AppendFile) -> Result<ducklake_catalog::SnapshotId> {
        self.commit_append_files(vec![file]).await
    }

    async fn commit_append_files(
        &self,
        _files: Vec<AppendFile>,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "append writes are not supported by this catalog backend".into(),
        ))
    }

    fn supports_inlined_appends(&self) -> bool {
        false
    }

    async fn commit_append_inlined_data(
        &self,
        _append: InlinedDataAppendCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "inlined append writes are not supported by this catalog backend".into(),
        ))
    }

    async fn commit_delete_files(
        &self,
        _files: Vec<DeleteFileCommit>,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "delete writes are not supported by this catalog backend".into(),
        ))
    }

    async fn commit_table_changes(
        &self,
        _changes: TableChangeCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "mixed table changes are not supported by this catalog backend".into(),
        ))
    }

    async fn commit_create_schema(
        &self,
        _create: CreateSchemaCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake schema creation is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_create_table(
        &self,
        _create: CreateTableCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake table creation is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_create_table_with_files(
        &self,
        create: CreateTableWithFilesCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        if create.files.is_empty() {
            return self.commit_create_table(create.create).await;
        }
        Err(StorageError::Unsupported(
            "DuckLake same-snapshot CREATE TABLE AS is not supported by this catalog backend"
                .into(),
        ))
    }

    async fn commit_create_table_with_inlined_data(
        &self,
        _create: CreateTableWithInlinedDataCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake same-snapshot CREATE TABLE AS with inlined data is not supported by this catalog backend"
                .into(),
        ))
    }

    async fn commit_create_view(
        &self,
        _create: CreateViewCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake view creation is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_drop_table(
        &self,
        _drop: DropTableCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake table drop is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_rename_table(
        &self,
        _rename: RenameTableCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake table rename is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_drop_view(
        &self,
        _drop: DropViewCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake view drop is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_drop_schema(
        &self,
        _drop: DropSchemaCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake schema drop is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_column_nullability(
        &self,
        _change: ColumnNullabilityCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake ALTER COLUMN SET/DROP NOT NULL is not supported by this catalog backend"
                .into(),
        ))
    }

    async fn commit_column_default(
        &self,
        _change: ColumnDefaultCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake ALTER COLUMN SET/DROP DEFAULT is not supported by this catalog backend"
                .into(),
        ))
    }

    async fn commit_column_type(
        &self,
        _change: ColumnTypeCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake ALTER COLUMN SET DATA TYPE is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_column_type_tree(
        &self,
        _change: ColumnTypeTreeCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake complex ALTER COLUMN SET DATA TYPE is not supported by this catalog backend"
                .into(),
        ))
    }

    async fn commit_rename_column(
        &self,
        _rename: RenameColumnCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake ALTER TABLE RENAME COLUMN is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_drop_column(
        &self,
        _drop: DropColumnCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake ALTER TABLE DROP COLUMN is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_add_columns(
        &self,
        _add: AddColumnCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake ALTER TABLE ADD COLUMN is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_partitioning(
        &self,
        partitioning: PartitioningCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        self.commit_partitioning_batch(PartitioningBatchCommit {
            expected_snapshot: partitioning.expected_snapshot,
            changes: vec![PartitioningTableCommit {
                table_id: partitioning.table_id,
                columns: partitioning.columns,
            }],
            commit_message: partitioning.commit_message,
        })
        .await
    }

    async fn commit_partitioning_batch(
        &self,
        _partitioning: PartitioningBatchCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake SET/RESET PARTITIONED BY is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_sorted_by(
        &self,
        _sorted_by: SortedByCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "DuckLake SET/RESET SORTED BY is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_truncate_table(
        &self,
        _truncate: TruncateTableCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "full-table delete/truncate is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_merge_adjacent_compaction(
        &self,
        _compaction: MergeAdjacentCompactionCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "merge-adjacent compaction is not supported by this catalog backend".into(),
        ))
    }

    async fn commit_rewrite_data_files(
        &self,
        _rewrite: RewriteDataFilesCommit,
    ) -> Result<ducklake_catalog::SnapshotId> {
        Err(StorageError::Unsupported(
            "rewrite-data-files compaction is not supported by this catalog backend".into(),
        ))
    }

    async fn cleanup_old_files(
        &self,
        _options: CleanupOldFilesOptions,
    ) -> Result<Vec<CleanupFile>> {
        Err(StorageError::Unsupported(
            "cleanup of old DuckLake files is not supported by this catalog backend".into(),
        ))
    }

    async fn remove_files_scheduled_for_deletion(
        &self,
        _data_file_ids: Vec<DataFileId>,
    ) -> Result<()> {
        Err(StorageError::Unsupported(
            "cleanup of DuckLake metadata queue rows is not supported by this catalog backend"
                .into(),
        ))
    }

    async fn expire_snapshots(&self, _options: ExpireSnapshotsOptions) -> Result<Vec<Snapshot>> {
        Err(StorageError::Unsupported(
            "DuckLake snapshot expiration is not supported by this catalog backend".into(),
        ))
    }

    async fn set_metadata_option(
        &self,
        _key: String,
        _value: String,
        _scope: MetadataOptionScope,
    ) -> Result<()> {
        Err(StorageError::Unsupported(
            "DuckLake metadata option updates are not supported by this catalog backend".into(),
        ))
    }
}

#[path = "engine/scan.rs"]
mod scan;
pub use scan::{
    catalog_database_size_bytes, catalog_settings, current_snapshot_id, list_catalog_paths,
    list_options, list_snapshots, list_table_files, list_table_info, list_tags, plan_table_scan,
};
#[cfg(test)]
pub(crate) use scan::{hive_partition_value_from_path, resolve_catalog_identifier};

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "engine/commit.rs"]
mod commit;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "engine/commit_engine.rs"]
mod commit_engine;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "catalog/ddl.rs"]
mod catalog_ddl;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "catalog/queries.rs"]
mod catalog_queries;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "catalog/metadata.rs"]
mod catalog_metadata;

#[path = "catalog/path.rs"]
mod catalog_path;
#[cfg(any(feature = "sqlite", feature = "quack"))]
pub(crate) use catalog_path::join_cleanup_path_components;
#[cfg(feature = "sqlite")]
pub(crate) use catalog_path::resolve_catalog_file_path;
pub use catalog_path::{
    LOCAL_OBJECT_STORE_URL, ResolvedObjectStorePath, ducklake_store_data_path,
    local_filesystem_path, path_is_within, resolve_object_store_path,
};
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) use catalog_path::{
    renamed_table_path_for_target_schema, resolve_cleanup_file_path,
    scheduled_cleanup_path_for_table_file,
};

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[macro_use]
#[path = "catalog/initialization_methods.rs"]
mod catalog_initialization_methods;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "engine/stats_engine.rs"]
mod stats_engine;

#[path = "engine/physical_schema.rs"]
mod physical_schema;
#[cfg(all(
    test,
    any(
        feature = "sqlite",
        feature = "duckdb",
        feature = "postgres",
        feature = "quack"
    )
))]
pub(crate) use physical_schema::physical_schema_from_arrow_schema_with_overrides;
pub use physical_schema::{
    create_table_columns_from_arrow_schema, physical_schema_for_file,
    physical_schema_from_arrow_schema,
};
pub(crate) use physical_schema::{legacy_name_fallback_mapping, reject_unsupported_files};

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "engine/sort_rewrite_engine.rs"]
mod sort_rewrite_engine;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) use sort_rewrite_engine::{
    rewrite_sort_expression_for_renamed_column, sort_expression_references_unqualified_column,
    sort_expression_unqualified_column_refs,
};

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[macro_use]
#[path = "metadata/prune_engine.rs"]
mod metadata_prune_engine;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
mod sql_utils;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "metadata/option.rs"]
mod metadata_option;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "catalog/table_exists.rs"]
mod catalog_table_exists;
#[cfg(all(
    test,
    any(
        feature = "sqlite",
        feature = "duckdb",
        feature = "postgres",
        feature = "quack"
    )
))]
#[path = "testing/test_support.rs"]
mod test_support;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "embedded/rows.rs"]
mod embedded_rows;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "embedded/reads.rs"]
mod embedded_reads;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "embedded/maintenance.rs"]
mod embedded_maintenance;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[macro_use]
#[path = "engine/inlined_values.rs"]
mod inlined_values;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[allow(unused_imports)]
pub(crate) use inlined_values::{
    InlinedIntervalFormat, InlinedSqlDialect, arrow_time_unit_nanos, duckdb_literal_to_json,
    format_date32, format_decimal_i128, format_decimal_i256, format_interval_month_day_nano,
    format_microsecond_interval_month_day_nano, format_time64, format_timestamp, format_uuid_bytes,
    hex_encode_bytes, inlined_arrow_field_from_schema, inlined_column_type,
    inlined_duckdb_literal_value, inlined_nested_json_values_to_array,
    inlined_nested_values_to_array, json_string, parse_hex_bytes, parse_interval_month_day_nano,
    parse_uuid_bytes, validate_inlined_append_for_dialect,
};

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[macro_use]
#[path = "catalog/read_methods.rs"]
mod catalog_read_methods;

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[macro_use]
#[path = "catalog/commit_methods.rs"]
mod catalog_commit_methods;

#[cfg(any(feature = "sqlite", feature = "duckdb"))]
#[macro_use]
#[path = "catalog/embedded/maintenance_methods.rs"]
mod catalog_embedded_maintenance_methods;

#[cfg(any(feature = "sqlite", feature = "duckdb"))]
#[macro_use]
#[path = "catalog/embedded/commit_methods.rs"]
mod catalog_embedded_commit_methods;

#[cfg(any(feature = "sqlite", feature = "duckdb"))]
#[macro_use]
#[path = "catalog/embedded/query_methods.rs"]
mod catalog_embedded_query_methods;

#[cfg(any(feature = "postgres", feature = "quack"))]
#[macro_use]
#[path = "catalog/async/commit_methods.rs"]
mod catalog_async_commit_methods;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[path = "catalog/async/query_methods.rs"]
mod catalog_async_query_methods;

#[cfg(any(feature = "postgres", feature = "quack"))]
#[macro_use]
#[path = "catalog/async/maintenance_methods.rs"]
mod catalog_async_maintenance_methods;

#[cfg(any(feature = "postgres", feature = "quack"))]
#[macro_use]
#[path = "catalog/async/metadata_methods.rs"]
mod catalog_async_metadata_methods;

#[cfg(any(feature = "postgres", feature = "quack"))]
#[macro_use]
#[path = "catalog/async/inlined_read_methods.rs"]
mod catalog_async_inlined_read_methods;

#[cfg(any(feature = "postgres", feature = "quack"))]
#[path = "catalog/async/inlined_mutation_methods.rs"]
mod catalog_async_inlined_mutation_methods;

#[cfg(feature = "sqlite")]
#[path = "backend/sqlite.rs"]
pub mod sqlite;

#[cfg(feature = "duckdb")]
#[path = "backend/duckdb_file.rs"]
pub mod duckdb_file;

#[cfg(feature = "postgres")]
#[path = "backend/postgres.rs"]
pub mod postgres;

#[cfg(feature = "quack")]
#[path = "backend/quack.rs"]
pub mod quack;
#[cfg(feature = "quack")]
#[path = "backend/quack_inlined.rs"]
mod quack_inlined;
#[cfg(feature = "quack")]
#[path = "backend/quack_protocol.rs"]
mod quack_protocol;

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