uqa_storage/

catalog.rs

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Backend-neutral persistent catalog facade.
//!
//! The engine depends on this trait for table metadata, analyzers, models,
//! graph registries, and planner statistics. Concrete storage layers such as
//! `SQLite` or a future RocksDB-backed catalog implement it behind the same
//! object-safe boundary.

use serde::{Deserialize, Serialize};

use crate::backend::{StorageBackendError, StorageBackendResult};

mod cache_revisions;
mod graph_access;
pub mod graph_guards;
pub mod graph_identifiers;
pub mod graph_observations;
pub(crate) mod graph_snapshot;
mod identity;
mod record_revisions;
mod relation;
pub mod relation_acl;
mod relation_security;
mod schema;
mod sequence_security;
mod sequence_values;

pub use cache_revisions::CatalogCacheRevisions;
pub use graph_access::validate_graph_page;
pub use graph_access::{
    GraphEntityFilter, GraphEntityKind, GraphEntitySelector, MAX_GRAPH_ID_PAGE,
};
pub use identity::new_nonzero_catalog_identity;
pub use record_revisions::{CatalogRecordRef, CatalogRecordRevision};
pub use relation::RelationIdentity;
pub use relation_security::{BoundRelationSecurity, LegacyRelationSecurity, RelationSecurityRow};
pub use sequence_security::{BoundSequenceSecurity, LegacySequenceSecurity, SequenceSecurityRow};
pub use sequence_values::{
    sequence_value_allocation, sequence_value_reservation, SequenceLogResult,
    SequenceReservationResult, SequenceSetValueResult, SequenceValueAllocation,
    SequenceValuePosition, SequenceValueReservation,
};
mod table;

pub use schema::{BoundSchemaRow, SchemaAclEntry, SchemaPrivileges, SchemaRow};
pub use table::{TableAclEntry, TablePrivileges};

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RelationKind {
    Table,
    View,
    Sequence,
    ForeignTable,
    Index,
}

impl RelationKind {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Table => "table",
            Self::View => "view",
            Self::Sequence => "sequence",
            Self::ForeignTable => "foreign_table",
            Self::Index => "index",
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableSchema {
    pub relation: RelationIdentity,
    #[serde(flatten)]
    pub security: RelationSecurityRow,
    /// Stable logical relation identity. `CREATE TABLE` allocates a new value, while renames, schema changes, `TRUNCATE`, and reopen preserve it. A zero value marks a legacy catalog row that the engine upgrades during open.
    #[serde(default)]
    pub object_id: [u8; 16],
    /// Stable physical-storage identity. `CREATE TABLE` and `TRUNCATE` allocate a new value, while schema-only alterations and renames preserve it. A zero value marks a legacy catalog row that the engine upgrades during open.
    #[serde(default)]
    pub storage_generation: [u8; 16],
    pub analyzer_json: String,
    pub fts_fields: Vec<String>,
    pub vector_fields: Vec<VectorFieldSchema>,
    /// Serialized `Vec<uqa_sql::ast::ColumnDef>` capturing the schema
    /// columns (name, type, `auto_increment`, flags). Empty for
    /// tables created by the legacy code path before column tracking
    /// existed.
    #[serde(default)]
    pub columns_json: String,
    /// Serialized `uqa_sql::ast::TableConstraintSet`. Empty for catalogs
    /// created before durable table constraints were introduced.
    #[serde(default)]
    pub constraints_json: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VectorFieldSchema {
    pub field: String,
    pub dimensions: u32,
}

/// One row from graph-edge persistence, represented as a typed struct so the
/// catalog API stays explicit and clippy-clean.
#[derive(Debug, Clone)]
pub struct EdgeRow {
    pub edge_id: u64,
    pub source_id: u64,
    pub target_id: u64,
    pub label: String,
    pub properties_json: String,
}

/// One vertex in an atomic named-graph snapshot replacement.
#[derive(Debug, Clone)]
pub struct GraphVertexRow {
    pub vertex_id: u64,
    pub label: String,
    pub properties_json: String,
}

/// Complete persisted shape of one named graph. Catalog implementations
/// replace the graph membership and entity rows as one atomic operation.
#[derive(Debug, Clone)]
pub struct GraphSnapshot {
    pub vertices: Vec<GraphVertexRow>,
    pub edges: Vec<EdgeRow>,
    pub label_registry_json: String,
}

/// One row from the foreign-table registry.
#[derive(Debug, Clone)]
pub struct ForeignTableRow {
    pub relation: RelationIdentity,
    pub security: RelationSecurityRow,
    pub server_name: String,
    pub columns_json: String,
    pub options_json: String,
}

/// One foreign-server definition. Execution owns the versioned SQL metadata; absence identifies a legacy row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ForeignServerRow {
    pub name: String,
    pub fdw_type: String,
    pub options_json: String,
    pub metadata_json: Option<String>,
}

/// One durable view definition. `definition_json` contains a serialized
/// planner query plan, while ownership remains a typed catalog relation.
#[derive(Debug, Clone)]
pub struct ViewRow {
    pub relation: RelationIdentity,
    pub security: RelationSecurityRow,
    pub definition_json: String,
}

pub use uqa_core::catalog_index::CatalogIndexRow;

/// Values persisted into one column-statistics row.
#[derive(Debug, Clone, Copy)]
pub struct ColumnStatsInput<'a> {
    pub table_name: &'a str,
    pub column_name: &'a str,
    pub distinct_count: i64,
    pub null_count: i64,
    pub min_value: Option<&'a str>,
    pub max_value: Option<&'a str>,
    pub row_count: i64,
    pub histogram_json: &'a str,
    pub mcv_values_json: &'a str,
    pub mcv_frequencies_json: &'a str,
}

impl<'a> ColumnStatsInput<'a> {
    pub fn basic(
        table_name: &'a str,
        column_name: &'a str,
        distinct_count: i64,
        null_count: i64,
        min_value: Option<&'a str>,
        max_value: Option<&'a str>,
        row_count: i64,
    ) -> Self {
        Self {
            table_name,
            column_name,
            distinct_count,
            null_count,
            min_value,
            max_value,
            row_count,
            histogram_json: "[]",
            mcv_values_json: "[]",
            mcv_frequencies_json: "[]",
        }
    }
}

/// One row from persisted column statistics.
#[derive(Debug, Clone, PartialEq)]
pub struct ColumnStatsRow {
    pub column_name: String,
    pub distinct_count: i64,
    pub null_count: i64,
    pub min_value: Option<String>,
    pub max_value: Option<String>,
    pub row_count: i64,
    pub histogram_json: String,
    pub mcv_values_json: String,
    pub mcv_frequencies_json: String,
}

/// Durable SQL sequence state. Sequence allocation is implemented by the
/// catalog backend as one atomic mutation so independent engine sessions (and
/// independently opened engines) cannot return the same value.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SequenceOptions {
    pub data_type: String,
    /// `None` is accepted only while decoding a legacy row and is resolved from its increment direction by native sequence restoration.
    pub min_value: Option<i64>,
    /// `None` is accepted only while decoding a legacy row and is resolved from its increment direction by native sequence restoration.
    pub max_value: Option<i64>,
    pub cycle: bool,
    #[serde(default = "default_sequence_cache_size")]
    pub cache_size: i64,
}

const fn default_sequence_cache_size() -> i64 {
    1
}

impl Default for SequenceOptions {
    fn default() -> Self {
        Self {
            data_type: "bigint".into(),
            min_value: None,
            max_value: None,
            cycle: false,
            cache_size: default_sequence_cache_size(),
        }
    }
}

pub use uqa_core::catalog_sequence::{
    SequenceAclEntry, SequenceOwner, SequenceOwnerDependency, SequencePrivileges,
};

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SequenceRow {
    pub relation: RelationIdentity,
    /// Captured owner and ACL role identities, or a legacy row awaiting initial-open migration.
    #[serde(flatten)]
    pub security: SequenceSecurityRow,
    /// Stable identity of this sequence incarnation. Dropping and recreating the same qualified name must allocate a different value.
    #[serde(default)]
    pub object_id: [u8; 16],
    /// Changes when value-generation options or persistence replace the sequence's allocation state. Stable across ownership and name changes, value reservations, and `setval`.
    #[serde(default)]
    pub definition_generation: [u8; 16],
    pub start: i64,
    pub increment: i64,
    pub current: i64,
    /// False until the first allocation returns `current` verbatim.
    pub called: bool,
    /// Number of values whose advancement is already durable in the sequence log.
    #[serde(default)]
    pub log_count: i64,
    /// `PostgreSQL` `pg_class.relpersistence` code. Durable sequence rows accept only permanent (`p`) and unlogged (`u`) values.
    pub persistence: String,
    #[serde(default)]
    pub owner: Option<SequenceOwner>,
    #[serde(default)]
    pub options: SequenceOptions,
}

/// Engine-facing catalog facade for persistent metadata.
pub trait CatalogFacade: Send + Sync {
    /// Transaction model and database incarnation shared with the paired backend. Wrappers must forward this with the session affinity.
    fn transaction_model(&self) -> crate::StorageTransactionModel {
        crate::StorageTransactionModel::ProviderSerialized
    }

    /// Identity shared with the paired data backend's transaction context. Wrappers must delegate this when their underlying catalog reports an identity.
    fn transaction_affinity(&self) -> Option<crate::StorageSessionAffinity> {
        None
    }

    /// Prepare durable catalog storage inside the backend's owning initial-restore transaction. Already initialized catalogs may keep the default.
    fn initialize_storage(&self) -> StorageBackendResult<()> {
        Ok(())
    }

    /// Read transactional cache generations without loading schemas or column
    /// statistics. Providers without change tracking use conservative reloads.
    fn cache_revisions(&self) -> StorageBackendResult<Option<CatalogCacheRevisions>> {
        Ok(None)
    }

    /// Read selected catalog-record identities from one committed/private boundary, in request order and without loading table data. An outer `None` means this provider has no exact record tracking; an inner `None` means the selected record is absent. Callers must not interpret unavailable tracking as an unchanged object.
    fn record_revisions(
        &self,
        _records: &[CatalogRecordRef<'_>],
    ) -> StorageBackendResult<Option<Vec<Option<CatalogRecordRevision>>>> {
        Ok(None)
    }

    fn set_metadata(&self, key: &str, value: &str) -> StorageBackendResult<()>;
    /// Remove one metadata record, retaining its private deletion and write precondition until transaction end.
    fn delete_metadata(&self, key: &str) -> StorageBackendResult<()>;
    fn get_metadata(&self, key: &str) -> StorageBackendResult<Option<String>>;
    /// Read matching metadata keys from one catalog snapshot; the prefix is literal, including NUL and wildcard characters.
    fn metadata_with_prefix(&self, prefix: &str) -> StorageBackendResult<Vec<(String, String)>>;
    /// Replace one already bound relation or attribute ACL without rewriting its definition. The caller holds the relation lifetime and catalog tuple locks until transaction end.
    fn save_relation_acl(
        &self,
        relation: &RelationIdentity,
        column: Option<&str>,
        entry: &relation_acl::RelationAclTuple,
    ) -> StorageBackendResult<()> {
        if !self.transaction_model().is_versioned() {
            return relation_acl::save_serialized(self, relation, column, entry);
        }
        self.set_metadata(&relation_acl::key(relation, column), &entry.encode(column)?)
    }
    /// Whether this metadata key has a transaction-private replacement in the current session.
    fn metadata_has_private_changes(&self, _key: &str) -> StorageBackendResult<bool> {
        if self.transaction_model().is_versioned() {
            return Err(StorageBackendError::Other(
                "private metadata provenance is not supported by this catalog".into(),
            ));
        }
        Ok(false)
    }
    /// Persist evaluated maintenance state. Concurrent providers merge its counters under their existing record-publication boundary; serialized providers use an ordinary metadata replacement.
    fn save_statistics_maintenance(
        &self,
        table: &str,
        state: &crate::statistics_maintenance::StatisticsMaintenance,
    ) -> StorageBackendResult<()> {
        self.set_metadata(
            &crate::statistics_maintenance::StatisticsMaintenance::key(table),
            &serde_json::to_string(state)?,
        )
    }
    fn fts_storage_was_reset(&self) -> bool {
        false
    }

    /// Atomically migrate the former flat relation namespace into typed,
    /// schema-owned catalog objects. Implementations must reject normalized
    /// or cross-kind collisions instead of merging either object.
    fn migrate_relation_namespace(&self) -> StorageBackendResult<()>;

    fn save_schema_row(&self, schema: &SchemaRow) -> StorageBackendResult<()>;
    fn drop_schema(&self, name: &str) -> StorageBackendResult<()>;
    fn load_schema_rows(&self) -> StorageBackendResult<Vec<SchemaRow>>;

    /// Whether this exact schema has a private creation, replacement or deletion in the current session.
    fn schema_has_private_changes(&self, _name: &str) -> StorageBackendResult<bool> {
        if self.transaction_model().is_versioned() {
            return Err(StorageBackendError::Other(
                "private schema provenance is not supported by this catalog".into(),
            ));
        }
        Ok(false)
    }

    fn save_schema(&self, name: &str) -> StorageBackendResult<()> {
        self.save_schema_row(&SchemaRow::bootstrap(name))
    }

    fn load_schemas(&self) -> StorageBackendResult<Vec<String>> {
        Ok(self
            .load_schema_rows()?
            .into_iter()
            .map(|schema| schema.name().to_owned())
            .collect())
    }

    fn save_table(&self, schema: &TableSchema) -> StorageBackendResult<()>;
    fn load_tables(&self) -> StorageBackendResult<Vec<TableSchema>>;
    fn drop_table(&self, name: &str) -> StorageBackendResult<()>;
    /// Remove a table definition and every catalog/data row owned by it as
    /// one atomic catalog operation.
    fn drop_table_and_data(&self, name: &str) -> StorageBackendResult<()>;
    fn purge_table_data(&self, name: &str) -> StorageBackendResult<()>;
    fn rename_table_data(&self, from: &str, to: &str) -> StorageBackendResult<()>;
    fn drop_column_data(&self, table_name: &str, column_name: &str) -> StorageBackendResult<()>;
    fn rename_column_data(
        &self,
        table_name: &str,
        from: &str,
        to: &str,
    ) -> StorageBackendResult<()>;

    fn save_model(&self, name: &str, json: &str) -> StorageBackendResult<()>;
    fn load_models(&self) -> StorageBackendResult<Vec<(String, String)>>;
    fn load_model(&self, name: &str) -> StorageBackendResult<Option<String>>;
    fn drop_model(&self, name: &str) -> StorageBackendResult<()>;

    fn save_scoring_params(&self, name: &str, params_json: &str) -> StorageBackendResult<()>;
    fn load_scoring_params(&self, name: &str) -> StorageBackendResult<Option<String>>;
    fn load_all_scoring_params(&self) -> StorageBackendResult<Vec<(String, String)>>;
    fn drop_scoring_params(&self, name: &str) -> StorageBackendResult<()>;

    /// Create a sequence's definition and the value record of its definition generation, which starts at the value state of `sequence`.
    fn create_sequence_row(&self, sequence: &SequenceRow) -> StorageBackendResult<bool>;
    /// Replace a sequence's definition, ownership and security. A row that keeps its object identity and definition generation leaves the generation's value record, which only value operations move; a row with a new generation replaces the old generation's value record with one that starts at the value state of `sequence`.
    fn replace_sequence_row(&self, sequence: &SequenceRow) -> StorageBackendResult<bool>;
    /// Atomically move one sequence catalog row and its shared relation claim while preserving object identity and the generation's value record.
    fn rename_sequence_row(&self, from: &str, to: &str) -> StorageBackendResult<bool>;
    /// Remove a sequence's definition together with the value record of its generation.
    fn drop_sequence_row(&self, name: &str) -> StorageBackendResult<bool>;
    /// Sequence definitions, each with the value state of its generation's value record.
    fn load_sequence_rows(&self) -> StorageBackendResult<Vec<SequenceRow>>;

    /// Whether this sequence's definition has transaction-private changes, including creation, rename, deletion and definition replacement, which the transaction's own catalog reads must select.
    fn sequence_has_private_changes(
        &self,
        _relation: &RelationIdentity,
        _object_id: [u8; 16],
    ) -> StorageBackendResult<bool> {
        if self.transaction_model().is_versioned() {
            return Err(StorageBackendError::Other(
                "private sequence provenance is not supported by this catalog".into(),
            ));
        }
        Ok(false)
    }

    /// Whether the value record of this definition generation has transaction-private changes, which holds exactly when the current transaction created the generation. Values of such a generation are allocated in the transaction, since no other session can see it; the value record of a committed generation is moved only outside any transaction, as `PostgreSQL` moves a sequence's state whatever happens to the transaction that drew from it.
    fn sequence_value_has_private_changes(
        &self,
        _object_id: [u8; 16],
        _definition_generation: [u8; 16],
    ) -> StorageBackendResult<bool> {
        if self.transaction_model().is_versioned() {
            return Err(StorageBackendError::Other(
                "private sequence value provenance is not supported by this catalog".into(),
            ));
        }
        Ok(false)
    }

    /// Move the value state that earlier releases kept in sequence definitions into value records. Only an initial open runs this, inside its catalog preparation.
    fn migrate_sequence_values(&self) -> StorageBackendResult<()> {
        Ok(())
    }
    /// Reserve the next block of values from the value record of this generation of `object_id`, whatever name the sequence currently has. `DefinitionChanged` reports that another generation of the sequence has replaced this one, and `Missing` that the sequence no longer exists.
    fn reserve_sequence_values(
        &self,
        name: &str,
        object_id: [u8; 16],
        definition_generation: [u8; 16],
    ) -> StorageBackendResult<SequenceReservationResult>;
    /// Compatibility allocation API for callers that do not retain a session cache. A cached reservation's unused values are intentionally abandoned, just as when a `PostgreSQL` session disconnects.
    fn next_sequence_value(
        &self,
        name: &str,
        object_id: [u8; 16],
    ) -> StorageBackendResult<Option<i64>> {
        loop {
            let relation =
                RelationIdentity::from_legacy_name(name).map_err(StorageBackendError::Other)?;
            let Some(row) = self
                .load_sequence_rows()?
                .into_iter()
                .find(|row| row.relation == relation && row.object_id == object_id)
            else {
                return Ok(None);
            };
            match self.reserve_sequence_values(name, object_id, row.definition_generation)? {
                SequenceReservationResult::Reserved(reservation) => {
                    return Ok(Some(reservation.first_value));
                }
                SequenceReservationResult::DefinitionChanged => {}
                SequenceReservationResult::Missing => return Ok(None),
                SequenceReservationResult::Exhausted => {
                    return Err(StorageBackendError::Other(format!(
                        "sequence `{name}` exhausted"
                    )));
                }
            }
        }
    }
    /// Replace the value state in the value record of this generation of `object_id`, with the results `reserve_sequence_values` reports.
    fn set_sequence_value(
        &self,
        name: &str,
        object_id: [u8; 16],
        definition_generation: [u8; 16],
        value: i64,
        called: bool,
        log_count: i64,
    ) -> StorageBackendResult<SequenceSetValueResult>;
    /// Move the durable position in the value record of this generation of `object_id` to `logged` when its value and called flag still are `expected`. Nothing is written otherwise, and `Changed` reports the position the record holds. A caller that hands out values at or below a logged value records that value here before it hands any of them out.
    fn log_sequence_values(
        &self,
        name: &str,
        object_id: [u8; 16],
        definition_generation: [u8; 16],
        expected: (i64, bool),
        logged: SequenceValuePosition,
    ) -> StorageBackendResult<SequenceLogResult>;

    fn save_view(&self, view: &ViewRow) -> StorageBackendResult<()>;
    /// Atomically move one view catalog row and its shared relation claim.
    fn rename_view(
        &self,
        from: &RelationIdentity,
        to: &RelationIdentity,
    ) -> StorageBackendResult<bool>;
    fn drop_view(&self, relation: &RelationIdentity) -> StorageBackendResult<bool>;
    fn load_views(&self) -> StorageBackendResult<Vec<ViewRow>>;

    fn save_named_graph(&self, name: &str) -> StorageBackendResult<()>;
    /// Bind graph mutations to their definition and allocation generation while allowing independent data writers to share the definition. Concurrent backends and catalog wrappers must implement this capability.
    fn guard_graph_definition(&self, _graph: Option<&str>) -> StorageBackendResult<()> {
        Err(crate::StorageBackendError::Other(
            "graph definition guards are not supported".into(),
        ))
    }
    fn drop_named_graph(&self, name: &str) -> StorageBackendResult<()>;
    fn load_named_graphs(&self) -> StorageBackendResult<Vec<String>>;
    fn named_graph_exists(&self, name: &str) -> StorageBackendResult<bool>;

    /// Read one entity without loading any other graph payload.
    fn graph_vertex(&self, id: u64) -> StorageBackendResult<Option<GraphVertexRow>>;

    /// Visit requested vertices on the pinned view, including missing rows and
    /// duplicates in caller order. The internal callback must not reenter
    /// storage. A false result stops before the next identity is admitted.
    /// `None` declines the capability without invoking the callback; `Some`
    /// counts all invocations, including the one that stopped the read.
    fn for_each_graph_vertex_borrowed(
        &self,
        _ids: &[u64],
        _visit: &mut dyn FnMut(u64, Option<&GraphVertexRow>) -> bool,
    ) -> StorageBackendResult<Option<usize>> {
        Ok(None)
    }

    fn graph_edge(&self, id: u64) -> StorageBackendResult<Option<EdgeRow>>;

    /// Read a bounded, strictly ascending identity page from the pinned
    /// storage snapshot. Filters must be applied before the page limit.
    fn graph_entity_ids(
        &self,
        filter: GraphEntityFilter<'_>,
        after: Option<u64>,
        limit: usize,
    ) -> StorageBackendResult<Vec<u64>>;

    fn graph_entity_count(&self, filter: GraphEntityFilter<'_>) -> StorageBackendResult<u64>;
    fn graph_entity_max_id(&self, kind: GraphEntityKind) -> StorageBackendResult<Option<u64>>;
    fn graph_entity_memberships(
        &self,
        kind: GraphEntityKind,
        id: u64,
    ) -> StorageBackendResult<Vec<String>>;
    fn graph_has_membership(
        &self,
        kind: GraphEntityKind,
        id: u64,
        graph: &str,
    ) -> StorageBackendResult<bool>;
    /// Explicit bulk snapshot/export of one named graph. The caller pins
    /// the storage transaction. Engine startup, sessions, and query handles
    /// use direct access methods instead of retaining this owned copy.
    fn load_named_graph_snapshot(&self, name: &str) -> StorageBackendResult<Option<GraphSnapshot>> {
        graph_snapshot::load(self, name)
    }
    fn save_vertex(
        &self,
        vertex_id: u64,
        label: &str,
        properties_json: &str,
    ) -> StorageBackendResult<()>;
    fn delete_vertex(&self, vertex_id: u64) -> StorageBackendResult<()>;
    fn load_vertices(&self) -> StorageBackendResult<Vec<(u64, String, String)>>;
    fn save_edge(
        &self,
        edge_id: u64,
        source_id: u64,
        target_id: u64,
        label: &str,
        properties_json: &str,
    ) -> StorageBackendResult<()>;
    fn delete_edge(&self, edge_id: u64) -> StorageBackendResult<()>;
    fn load_edges(&self) -> StorageBackendResult<Vec<EdgeRow>>;
    fn save_graph_membership(
        &self,
        entity_type: &str,
        entity_id: u64,
        graph_name: &str,
    ) -> StorageBackendResult<()>;
    fn delete_graph_membership(
        &self,
        entity_type: &str,
        entity_id: u64,
        graph_name: &str,
    ) -> StorageBackendResult<()>;
    fn delete_graph_membership_for_graph(&self, graph_name: &str) -> StorageBackendResult<()>;
    fn load_graph_memberships(&self) -> StorageBackendResult<Vec<(String, u64, String)>>;
    fn purge_orphan_graph_entities(&self) -> StorageBackendResult<()>;
    fn replace_named_graph(
        &self,
        graph_name: &str,
        snapshot: &GraphSnapshot,
    ) -> StorageBackendResult<()>;
    fn drop_named_graph_data(&self, graph_name: &str) -> StorageBackendResult<()>;

    fn save_analyzer(&self, name: &str, config_json: &str) -> StorageBackendResult<()>;
    fn drop_analyzer(&self, name: &str) -> StorageBackendResult<()>;
    fn load_analyzers(&self) -> StorageBackendResult<Vec<(String, String)>>;

    /// Save a named definition's resolved descriptor with its diagnostic configuration in the same transaction.
    fn save_analyzer_revision(
        &self,
        _name: &str,
        _config_json: &str,
        _descriptor_json: &str,
    ) -> StorageBackendResult<()> {
        Err(StorageBackendError::Other(
            "durable analyzer descriptors are not supported by this catalog".into(),
        ))
    }

    /// Resolved named descriptors; legacy definitions appear only in `load_analyzers` until migrated.
    fn load_analyzer_descriptors(&self) -> StorageBackendResult<Vec<(String, String)>> {
        Ok(Vec::new())
    }

    /// Atomically replace a field's complete independent binding and its compatibility label. An empty label denotes the physical field's unnamed default.
    fn replace_table_field_analyzer_binding(
        &self,
        _table: &str,
        _field: &str,
        _phase: &str,
        _name: &str,
        _binding_json: &str,
    ) -> StorageBackendResult<()> {
        Err(StorageBackendError::Other(
            "durable analyzer bindings are not supported by this catalog".into(),
        ))
    }

    /// `(table, field, binding JSON)` rows. Each field has one complete binding envelope.
    fn load_table_field_analyzer_bindings(
        &self,
    ) -> StorageBackendResult<Vec<(String, String, String)>> {
        Ok(Vec::new())
    }

    fn save_table_field_analyzer(
        &self,
        table_name: &str,
        field: &str,
        phase: &str,
        analyzer_name: &str,
    ) -> StorageBackendResult<()>;
    fn replace_table_field_analyzer(
        &self,
        table_name: &str,
        field: &str,
        phase: &str,
        analyzer_name: &str,
    ) -> StorageBackendResult<()>;
    fn drop_table_field_analyzer_field(
        &self,
        table_name: &str,
        field: &str,
    ) -> StorageBackendResult<()>;
    fn drop_table_field_analyzers(&self, table_name: &str) -> StorageBackendResult<()>;
    fn load_table_field_analyzers(
        &self,
    ) -> StorageBackendResult<Vec<(String, String, String, String)>>;

    fn save_foreign_server(
        &self,
        name: &str,
        fdw_type: &str,
        options_json: &str,
    ) -> StorageBackendResult<()>;
    fn drop_foreign_server(&self, name: &str) -> StorageBackendResult<()>;
    fn load_foreign_servers(&self) -> StorageBackendResult<Vec<(String, String, String)>>;

    /// Atomically replace a complete foreign-server row. A catalog without metadata support must reject it instead of discarding the metadata.
    fn save_foreign_server_row(&self, row: &ForeignServerRow) -> StorageBackendResult<()> {
        if row.metadata_json.is_some() {
            return Err(StorageBackendError::Other(
                "foreign-server metadata is not supported by this catalog".into(),
            ));
        }
        self.save_foreign_server(&row.name, &row.fdw_type, &row.options_json)
    }

    /// Read definitions and their metadata from one catalog view. Legacy implementations expose absent metadata explicitly.
    fn load_foreign_server_rows(&self) -> StorageBackendResult<Vec<ForeignServerRow>> {
        Ok(self
            .load_foreign_servers()?
            .into_iter()
            .map(|(name, fdw_type, options_json)| ForeignServerRow {
                name,
                fdw_type,
                options_json,
                metadata_json: None,
            })
            .collect())
    }

    fn save_foreign_table(&self, row: &ForeignTableRow) -> StorageBackendResult<()>;
    /// Atomically move one foreign-table catalog row and its shared relation claim.
    fn rename_foreign_table(
        &self,
        from: &RelationIdentity,
        to: &RelationIdentity,
    ) -> StorageBackendResult<bool>;
    fn update_foreign_table_security(
        &self,
        relation: &RelationIdentity,
        security: &RelationSecurityRow,
    ) -> StorageBackendResult<bool>;
    fn drop_foreign_table(&self, relation: &RelationIdentity) -> StorageBackendResult<()>;
    fn load_foreign_tables(&self) -> StorageBackendResult<Vec<ForeignTableRow>>;

    fn save_catalog_index(
        &self,
        relation: &RelationIdentity,
        index_type: &str,
        table_name: &str,
        columns_json: &str,
        parameters_json: &str,
    ) -> StorageBackendResult<()> {
        self.save_catalog_index_row(&CatalogIndexRow {
            relation: relation.clone(),
            index_type: index_type.to_string(),
            table_name: table_name.to_string(),
            columns_json: columns_json.to_string(),
            parameters_json: parameters_json.to_string(),
            definition_json: None,
        })
    }
    fn save_catalog_index_row(&self, index: &CatalogIndexRow) -> StorageBackendResult<()>;
    fn drop_catalog_index(&self, relation: &RelationIdentity) -> StorageBackendResult<()>;
    fn drop_catalog_indexes_for_table(&self, table_name: &str) -> StorageBackendResult<()>;
    fn load_catalog_indexes(&self) -> StorageBackendResult<Vec<CatalogIndexRow>>;

    fn save_path_index(
        &self,
        graph_name: &str,
        label_sequences_json: &str,
    ) -> StorageBackendResult<()>;
    fn drop_path_index(&self, graph_name: &str) -> StorageBackendResult<()>;
    fn load_path_indexes(&self) -> StorageBackendResult<Vec<(String, String)>>;

    /// Durable reachability records, paged lexicographically by (source, target).
    /// Building calls clear, bounded saves, then finish inside one transaction.
    /// Graph mutations invalidate the materialization atomically; definitions remain.
    fn clear_path_index_data(&self, index: &str) -> StorageBackendResult<()>;
    fn save_path_index_pairs(
        &self,
        index: &str,
        sequence: &str,
        pairs: &[(u64, u64)],
    ) -> StorageBackendResult<()>;
    fn finish_path_index_data(
        &self,
        index: &str,
        graph: &str,
        definition: &str,
    ) -> StorageBackendResult<()>;
    fn path_index_data_is_current(
        &self,
        index: &str,
        definition: &str,
    ) -> StorageBackendResult<bool>;
    fn path_index_pairs(
        &self,
        index: &str,
        sequence: &str,
        after: Option<(u64, u64)>,
        limit: usize,
    ) -> StorageBackendResult<Vec<(u64, u64)>>;

    fn save_column_stats(&self, stats: ColumnStatsInput<'_>) -> StorageBackendResult<()>;
    /// Atomically replace the complete statistics snapshot for one table.
    /// Implementations must leave the prior snapshot intact if any row fails.
    fn replace_column_stats(
        &self,
        table_name: &str,
        stats: &[ColumnStatsInput<'_>],
    ) -> StorageBackendResult<()>;
    fn load_column_stats(&self, table_name: &str) -> StorageBackendResult<Vec<ColumnStatsRow>>;
    fn delete_column_stats(&self, table_name: &str) -> StorageBackendResult<()>;
}

#[cfg(test)]
mod tests {
    use super::RelationIdentity;

    #[test]
    fn relation_identity_rendering_is_reversible_and_collision_free() {
        let left = RelationIdentity::new("a.b", "c");
        let right = RelationIdentity::new("a", "b.c");
        assert_eq!(left.qualified_name(), "\"a.b\".c");
        assert_eq!(right.qualified_name(), "a.\"b.c\"");
        assert_ne!(left.qualified_name(), right.qualified_name());
        assert_eq!(
            RelationIdentity::from_legacy_name(&left.qualified_name()).unwrap(),
            left
        );
        assert_eq!(
            RelationIdentity::from_legacy_name(&right.qualified_name()).unwrap(),
            right
        );
    }

    #[test]
    fn relation_identity_preserves_quotes_and_unqualified_public_alias() {
        let quoted = RelationIdentity::new("public", "a\"b.c");
        assert_eq!(quoted.qualified_name(), "public.\"a\"\"b.c\"");
        assert_eq!(
            quoted.canonical_and_legacy_public_names(),
            vec![
                "public.\"a\"\"b.c\"".to_string(),
                "\"a\"\"b.c\"".to_string()
            ]
        );
        assert_eq!(
            RelationIdentity::from_legacy_name(&quoted.qualified_name()).unwrap(),
            quoted
        );
        assert_eq!(
            RelationIdentity::from_legacy_name("plain").unwrap(),
            RelationIdentity::new("public", "plain")
        );
        assert_eq!(
            RelationIdentity::new("app", "plain").canonical_and_legacy_public_names(),
            vec!["app.plain".to_string()]
        );
        assert_eq!(
            RelationIdentity::new("public", "Upper").canonical_and_legacy_public_names(),
            vec![
                "public.\"Upper\"".to_string(),
                "\"Upper\"".to_string(),
                "Upper".to_string()
            ]
        );
    }
}