alopex-embedded 0.8.0

//! Catalog API 向けの公開型定義。

use std::collections::{BTreeMap, BTreeSet, HashMap};

use alopex_cluster::{NodeId, SchemaApplyEvidence, SchemaApplyState, SchemaManifest};
use alopex_core::{KVStore, KVTransaction};
use alopex_sql::ast::ddl::{DataType, IndexMethod, VectorMetric};
use alopex_sql::catalog::persistent::{CatalogMeta, IndexFqn, NamespaceMeta, TableFqn};
use alopex_sql::catalog::{
    Catalog, CatalogOverlay, ColumnMetadata, Compression, IndexMetadata, RowIdMode, StorageOptions,
    StorageType, TableMetadata,
};
use alopex_sql::planner::types::ResolvedType;
use alopex_sql::{DataSourceFormat, TableType};
use serde::{Deserialize, Serialize};
use sha2::Digest;

use crate::{Database, Error, Result, Transaction, TxnMode};

/// Catalog 情報（公開 API 返却用）。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CatalogInfo {
    /// Catalog 名。
    pub name: String,
    /// コメント。
    pub comment: Option<String>,
    /// ストレージルート。
    pub storage_root: Option<String>,
}

impl From<CatalogMeta> for CatalogInfo {
    fn from(value: CatalogMeta) -> Self {
        Self {
            name: value.name,
            comment: value.comment,
            storage_root: value.storage_root,
        }
    }
}

/// Namespace 情報（公開 API 返却用）。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NamespaceInfo {
    /// Namespace 名。
    pub name: String,
    /// 所属 Catalog 名。
    pub catalog_name: String,
    /// コメント。
    pub comment: Option<String>,
    /// ストレージルート。
    pub storage_root: Option<String>,
}

impl From<NamespaceMeta> for NamespaceInfo {
    fn from(value: NamespaceMeta) -> Self {
        Self {
            name: value.name,
            catalog_name: value.catalog_name,
            comment: value.comment,
            storage_root: value.storage_root,
        }
    }
}

/// カラム情報（公開 API 返却用）。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ColumnInfo {
    /// カラム名。
    pub name: String,
    /// データ型（例: "INTEGER", "TEXT", "VECTOR(128, COSINE)"）。
    pub data_type: String,
    /// NULL 許可。
    pub nullable: bool,
    /// 主キーの一部かどうか。
    pub is_primary_key: bool,
    /// コメント。
    pub comment: Option<String>,
}

/// ストレージ設定情報（TableInfo 返却用）。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StorageInfo {
    /// ストレージ種別（"row" | "columnar"）。
    pub storage_type: String,
    /// 圧縮方式（"none" | "lz4" | "zstd"）。
    pub compression: String,
}

impl Default for StorageInfo {
    fn default() -> Self {
        Self {
            storage_type: "row".to_string(),
            compression: "none".to_string(),
        }
    }
}

impl From<&StorageOptions> for StorageInfo {
    fn from(value: &StorageOptions) -> Self {
        let storage_type = match value.storage_type {
            StorageType::Row => "row",
            StorageType::Columnar => "columnar",
        };
        let compression = match value.compression {
            Compression::None => "none",
            Compression::Lz4 => "lz4",
            Compression::Zstd => "zstd",
        };
        Self {
            storage_type: storage_type.to_string(),
            compression: compression.to_string(),
        }
    }
}

/// テーブル情報（公開 API 返却用）。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TableInfo {
    /// テーブル名。
    pub name: String,
    /// 所属 Catalog 名。
    pub catalog_name: String,
    /// 所属 Namespace 名。
    pub namespace_name: String,
    /// テーブル ID。
    pub table_id: u32,
    /// テーブル種別。
    pub table_type: TableType,
    /// カラム情報。
    pub columns: Vec<ColumnInfo>,
    /// 主キー。
    pub primary_key: Option<Vec<String>>,
    /// ストレージロケーション。
    pub storage_location: Option<String>,
    /// データソース形式。
    pub data_source_format: DataSourceFormat,
    /// ストレージ設定。
    pub storage_options: StorageInfo,
    /// コメント。
    pub comment: Option<String>,
    /// カスタムプロパティ。
    pub properties: HashMap<String, String>,
}

impl From<&TableMetadata> for TableInfo {
    fn from(value: &TableMetadata) -> Self {
        let primary_key = value.primary_key.clone();
        let columns = value
            .columns
            .iter()
            .map(|column| ColumnInfo {
                name: column.name.clone(),
                data_type: resolved_type_to_string(&column.data_type),
                nullable: !column.not_null,
                is_primary_key: column.primary_key
                    || primary_key
                        .as_ref()
                        .map(|keys| keys.iter().any(|name| name == &column.name))
                        .unwrap_or(false),
                comment: None,
            })
            .collect();
        let storage_options = if value.storage_options == StorageOptions::default() {
            StorageInfo::default()
        } else {
            StorageInfo::from(&value.storage_options)
        };

        Self {
            name: value.name.clone(),
            catalog_name: value.catalog_name.clone(),
            namespace_name: value.namespace_name.clone(),
            table_id: value.table_id,
            table_type: value.table_type,
            columns,
            primary_key,
            storage_location: value.storage_location.clone(),
            data_source_format: value.data_source_format,
            storage_options,
            comment: value.comment.clone(),
            properties: value.properties.clone(),
        }
    }
}

impl From<TableMetadata> for TableInfo {
    fn from(value: TableMetadata) -> Self {
        Self::from(&value)
    }
}

/// インデックス情報（公開 API 返却用）。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IndexInfo {
    /// インデックス名。
    pub name: String,
    /// インデックス ID。
    pub index_id: u32,
    /// 所属 Catalog 名。
    pub catalog_name: String,
    /// 所属 Namespace 名。
    pub namespace_name: String,
    /// 対象テーブル名。
    pub table_name: String,
    /// 対象カラム名。
    pub columns: Vec<String>,
    /// インデックス方式（"btree" | "hnsw"）。
    pub method: String,
    /// ユニーク制約。
    pub is_unique: bool,
}

impl From<&IndexMetadata> for IndexInfo {
    fn from(value: &IndexMetadata) -> Self {
        let method = match value.method {
            Some(IndexMethod::BTree) | None => "btree",
            Some(IndexMethod::Hnsw) => "hnsw",
        };
        Self {
            name: value.name.clone(),
            index_id: value.index_id,
            catalog_name: value.catalog_name.clone(),
            namespace_name: value.namespace_name.clone(),
            table_name: value.table.clone(),
            columns: value.columns.clone(),
            method: method.to_string(),
            is_unique: value.unique,
        }
    }
}

impl From<IndexMetadata> for IndexInfo {
    fn from(value: IndexMetadata) -> Self {
        Self::from(&value)
    }
}

/// Catalog 作成リクエスト。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateCatalogRequest {
    /// Catalog 名。
    pub name: String,
    /// コメント。
    pub comment: Option<String>,
    /// ストレージルート。
    pub storage_root: Option<String>,
}

impl CreateCatalogRequest {
    /// 必須フィールドを指定して作成する。
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            comment: None,
            storage_root: None,
        }
    }

    /// コメントを指定する。
    pub fn with_comment(mut self, comment: impl Into<String>) -> Self {
        self.comment = Some(comment.into());
        self
    }

    /// ストレージルートを指定する。
    pub fn with_storage_root(mut self, storage_root: impl Into<String>) -> Self {
        self.storage_root = Some(storage_root.into());
        self
    }

    /// 必須フィールドを検証して返す。
    pub fn build(self) -> Result<Self> {
        validate_required(&self.name, "catalog 名")?;
        Ok(self)
    }
}

/// Namespace 作成リクエスト。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateNamespaceRequest {
    /// 所属 Catalog 名。
    pub catalog_name: String,
    /// Namespace 名。
    pub name: String,
    /// コメント。
    pub comment: Option<String>,
    /// ストレージルート。
    pub storage_root: Option<String>,
}

impl CreateNamespaceRequest {
    /// 必須フィールドを指定して作成する。
    pub fn new(catalog_name: impl Into<String>, name: impl Into<String>) -> Self {
        Self {
            catalog_name: catalog_name.into(),
            name: name.into(),
            comment: None,
            storage_root: None,
        }
    }

    /// コメントを指定する。
    pub fn with_comment(mut self, comment: impl Into<String>) -> Self {
        self.comment = Some(comment.into());
        self
    }

    /// ストレージルートを指定する。
    pub fn with_storage_root(mut self, storage_root: impl Into<String>) -> Self {
        self.storage_root = Some(storage_root.into());
        self
    }

    /// 必須フィールドを検証して返す。
    pub fn build(self) -> Result<Self> {
        validate_required(&self.catalog_name, "catalog 名")?;
        validate_required(&self.name, "namespace 名")?;
        Ok(self)
    }
}

/// テーブル作成リクエスト。
#[derive(Debug, Clone)]
pub struct CreateTableRequest {
    /// Catalog 名（既定: "default"）。
    pub catalog_name: String,
    /// Namespace 名（既定: "default"）。
    pub namespace_name: String,
    /// テーブル名。
    pub name: String,
    /// スキーマ。
    pub schema: Option<Vec<ColumnDefinition>>,
    /// テーブル種別（既定: Managed）。
    pub table_type: TableType,
    /// データソース形式（None の場合は Alopex）。
    pub data_source_format: Option<DataSourceFormat>,
    /// 主キー。
    pub primary_key: Option<Vec<String>>,
    /// ストレージルート。
    pub storage_root: Option<String>,
    /// ストレージオプション。
    pub storage_options: Option<StorageOptions>,
    /// コメント。
    pub comment: Option<String>,
    /// カスタムプロパティ（None の場合は空の HashMap）。
    pub properties: Option<HashMap<String, String>>,
}

impl CreateTableRequest {
    /// 必須フィールドを指定して作成する。
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            catalog_name: "default".to_string(),
            namespace_name: "default".to_string(),
            name: name.into(),
            schema: None,
            table_type: TableType::Managed,
            data_source_format: None,
            primary_key: None,
            storage_root: None,
            storage_options: None,
            comment: None,
            properties: None,
        }
    }

    /// Catalog 名を指定する。
    pub fn with_catalog_name(mut self, catalog_name: impl Into<String>) -> Self {
        self.catalog_name = catalog_name.into();
        self
    }

    /// Namespace 名を指定する。
    pub fn with_namespace_name(mut self, namespace_name: impl Into<String>) -> Self {
        self.namespace_name = namespace_name.into();
        self
    }

    /// スキーマを指定する。
    pub fn with_schema(mut self, schema: Vec<ColumnDefinition>) -> Self {
        self.schema = Some(schema);
        self
    }

    /// テーブル種別を指定する。
    pub fn with_table_type(mut self, table_type: TableType) -> Self {
        self.table_type = table_type;
        self
    }

    /// データソース形式を指定する。
    pub fn with_data_source_format(mut self, data_source_format: DataSourceFormat) -> Self {
        self.data_source_format = Some(data_source_format);
        self
    }

    /// 主キーを指定する。
    pub fn with_primary_key(mut self, primary_key: Vec<String>) -> Self {
        self.primary_key = Some(primary_key);
        self
    }

    /// ストレージルートを指定する。
    pub fn with_storage_root(mut self, storage_root: impl Into<String>) -> Self {
        self.storage_root = Some(storage_root.into());
        self
    }

    /// ストレージオプションを指定する。
    pub fn with_storage_options(mut self, storage_options: StorageOptions) -> Self {
        self.storage_options = Some(storage_options);
        self
    }

    /// コメントを指定する。
    pub fn with_comment(mut self, comment: impl Into<String>) -> Self {
        self.comment = Some(comment.into());
        self
    }

    /// カスタムプロパティを指定する。
    pub fn with_properties(mut self, properties: HashMap<String, String>) -> Self {
        self.properties = Some(properties);
        self
    }

    /// 必須フィールドを検証して返す。
    pub fn build(mut self) -> Result<Self> {
        validate_required(&self.catalog_name, "catalog 名")?;
        validate_required(&self.namespace_name, "namespace 名")?;
        validate_required(&self.name, "table 名")?;

        if self.table_type == TableType::Managed && self.schema.is_none() {
            return Err(Error::SchemaRequired);
        }
        if self.table_type == TableType::External && self.storage_root.is_none() {
            return Err(Error::StorageRootRequired);
        }

        if self.data_source_format.is_none() {
            self.data_source_format = Some(DataSourceFormat::Alopex);
        }
        if self.properties.is_none() {
            self.properties = Some(HashMap::new());
        }
        Ok(self)
    }
}

/// カラム定義。
#[derive(Debug, Clone)]
pub struct ColumnDefinition {
    /// カラム名。
    pub name: String,
    /// データ型。
    pub data_type: DataType,
    /// NULL 許可（既定: true）。
    pub nullable: bool,
    /// コメント。
    pub comment: Option<String>,
}

impl ColumnDefinition {
    /// 必須フィールドを指定して作成する。
    pub fn new(name: impl Into<String>, data_type: DataType) -> Self {
        Self {
            name: name.into(),
            data_type,
            nullable: true,
            comment: None,
        }
    }

    /// NULL 許可を指定する。
    pub fn with_nullable(mut self, nullable: bool) -> Self {
        self.nullable = nullable;
        self
    }

    /// コメントを指定する。
    pub fn with_comment(mut self, comment: impl Into<String>) -> Self {
        self.comment = Some(comment.into());
        self
    }
}

impl Database {
    /// Catalog 一覧を取得する。
    pub fn list_catalogs(&self) -> Result<Vec<CatalogInfo>> {
        let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
        Ok(catalog
            .list_catalogs()
            .into_iter()
            .map(CatalogInfo::from)
            .collect())
    }

    /// Catalog を取得する。
    pub fn get_catalog(&self, name: &str) -> Result<CatalogInfo> {
        let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
        let meta = catalog
            .get_catalog(name)
            .ok_or_else(|| Error::CatalogNotFound(name.to_string()))?;
        Ok(meta.into())
    }

    /// Namespace 一覧を取得する。
    pub fn list_namespaces(&self, catalog_name: &str) -> Result<Vec<NamespaceInfo>> {
        let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
        ensure_catalog_exists(&*catalog, catalog_name)?;
        Ok(catalog
            .list_namespaces(catalog_name)
            .into_iter()
            .map(NamespaceInfo::from)
            .collect())
    }

    /// Namespace を取得する。
    pub fn get_namespace(&self, catalog_name: &str, namespace_name: &str) -> Result<NamespaceInfo> {
        let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
        ensure_catalog_exists(&*catalog, catalog_name)?;
        let meta = catalog
            .get_namespace(catalog_name, namespace_name)
            .ok_or_else(|| {
                Error::NamespaceNotFound(catalog_name.to_string(), namespace_name.to_string())
            })?;
        Ok(meta.into())
    }

    /// テーブル一覧を取得する。
    pub fn list_tables(&self, catalog_name: &str, namespace_name: &str) -> Result<Vec<TableInfo>> {
        let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
        ensure_namespace_exists(&*catalog, catalog_name, namespace_name)?;
        let namespace = catalog
            .get_namespace(catalog_name, namespace_name)
            .ok_or_else(|| {
                Error::NamespaceNotFound(catalog_name.to_string(), namespace_name.to_string())
            })?;

        let overlay = CatalogOverlay::new();
        let tables = catalog.list_tables_in_txn(catalog_name, namespace_name, &overlay);
        Ok(tables
            .into_iter()
            .map(|table| {
                let info = TableInfo::from(table);
                apply_storage_location(info, namespace.storage_root.as_deref())
            })
            .collect())
    }

    /// デフォルト catalog/namespace のテーブル一覧を取得する。
    pub fn list_tables_simple(&self) -> Result<Vec<TableInfo>> {
        self.list_tables("default", "default")
    }

    /// テーブル情報を取得する。
    pub fn get_table_info(
        &self,
        catalog_name: &str,
        namespace_name: &str,
        table_name: &str,
    ) -> Result<TableInfo> {
        let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
        ensure_namespace_exists(&*catalog, catalog_name, namespace_name)?;
        let namespace = catalog
            .get_namespace(catalog_name, namespace_name)
            .ok_or_else(|| {
                Error::NamespaceNotFound(catalog_name.to_string(), namespace_name.to_string())
            })?;

        let overlay = CatalogOverlay::new();
        let tables = catalog.list_tables_in_txn(catalog_name, namespace_name, &overlay);
        let table = tables
            .into_iter()
            .find(|table| table.name == table_name)
            .ok_or_else(|| {
                Error::TableNotFound(table_full_name(catalog_name, namespace_name, table_name))
            })?;

        let info = TableInfo::from(table);
        Ok(apply_storage_location(
            info,
            namespace.storage_root.as_deref(),
        ))
    }

    /// デフォルト catalog/namespace のテーブル情報を取得する。
    pub fn get_table_info_simple(&self, table_name: &str) -> Result<TableInfo> {
        self.get_table_info("default", "default", table_name)
    }

    /// テーブル情報をキャッシュから取得する（キャッシュミス時は DB ルックアップ）。
    ///
    /// パフォーマンス最適化のため、キャッシュを使用してテーブルメタデータを取得する。
    /// DDL 操作（テーブル作成/削除など）が発生するとキャッシュは自動的に無効化される。
    pub fn get_table_info_cached(
        &self,
        catalog_name: &str,
        namespace_name: &str,
        table_name: &str,
    ) -> Result<crate::CachedTableInfo> {
        // Check cache first
        if let Some(cached) = self.get_cached_table_info(catalog_name, namespace_name, table_name) {
            return Ok(cached);
        }

        // Cache miss: fetch from database and cache the result
        let info = self.get_table_info(catalog_name, namespace_name, table_name)?;
        let cached = crate::CachedTableInfo {
            storage_location: info.storage_location.clone(),
            format: format!("{:?}", info.data_source_format).to_uppercase(),
        };
        self.cache_table_info(catalog_name, namespace_name, table_name, cached.clone());
        Ok(cached)
    }

    /// インデックス一覧を取得する。
    pub fn list_indexes(
        &self,
        catalog_name: &str,
        namespace_name: &str,
        table_name: &str,
    ) -> Result<Vec<IndexInfo>> {
        let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
        ensure_namespace_exists(&*catalog, catalog_name, namespace_name)?;
        ensure_table_exists(&*catalog, catalog_name, namespace_name, table_name)?;

        let overlay = CatalogOverlay::new();
        let fqn = TableFqn::new(catalog_name, namespace_name, table_name);
        let indexes = catalog.list_indexes_in_txn(&fqn, &overlay);
        Ok(indexes.into_iter().map(IndexInfo::from).collect())
    }

    /// デフォルト catalog/namespace のインデックス一覧を取得する。
    pub fn list_indexes_simple(&self, table_name: &str) -> Result<Vec<IndexInfo>> {
        self.list_indexes("default", "default", table_name)
    }

    /// インデックス情報を取得する。
    pub fn get_index_info(
        &self,
        catalog_name: &str,
        namespace_name: &str,
        table_name: &str,
        index_name: &str,
    ) -> Result<IndexInfo> {
        let indexes = self.list_indexes(catalog_name, namespace_name, table_name)?;
        indexes
            .into_iter()
            .find(|index| index.name == index_name)
            .ok_or_else(|| {
                Error::IndexNotFound(index_full_name(
                    catalog_name,
                    namespace_name,
                    table_name,
                    index_name,
                ))
            })
    }

    /// デフォルト catalog/namespace のインデックス情報を取得する。
    pub fn get_index_info_simple(&self, table_name: &str, index_name: &str) -> Result<IndexInfo> {
        self.get_index_info("default", "default", table_name, index_name)
    }

    /// Catalog を作成する。
    ///
    /// # Examples
    ///
    /// ```
    /// use alopex_embedded::{CreateCatalogRequest, Database};
    ///
    /// let db = Database::new();
    /// let catalog = db.create_catalog(CreateCatalogRequest::new("main")).unwrap();
    /// assert_eq!(catalog.name, "main");
    /// ```
    pub fn create_catalog(&self, request: CreateCatalogRequest) -> Result<CatalogInfo> {
        let request = request.build()?;
        let mut catalog = self.sql_catalog.write().expect("catalog lock poisoned");
        if catalog.get_catalog(&request.name).is_some() {
            return Err(Error::CatalogAlreadyExists(request.name));
        }
        let meta = CatalogMeta {
            name: request.name,
            comment: request.comment,
            storage_root: request.storage_root,
        };
        catalog
            .create_catalog(meta.clone())
            .map_err(|err| Error::Sql(err.into()))?;
        self.invalidate_table_info_cache();
        Ok(meta.into())
    }

    /// Catalog を削除する。
    pub fn delete_catalog(&self, name: &str, force: bool) -> Result<()> {
        if name == "default" {
            return Err(Error::CannotDeleteDefault("catalog".to_string()));
        }
        let mut catalog = self.sql_catalog.write().expect("catalog lock poisoned");
        ensure_catalog_exists(&*catalog, name)?;

        if !force {
            let namespaces = catalog.list_namespaces(name);
            let has_non_default = namespaces.iter().any(|ns| ns.name != "default");
            let has_tables = namespaces.iter().any(|ns| {
                let overlay = CatalogOverlay::new();
                !catalog
                    .list_tables_in_txn(name, &ns.name, &overlay)
                    .is_empty()
            });
            if has_non_default || has_tables {
                return Err(Error::CatalogNotEmpty(name.to_string()));
            }
        }

        catalog
            .delete_catalog(name)
            .map_err(|err| Error::Sql(err.into()))?;
        self.invalidate_table_info_cache();
        Ok(())
    }

    /// Namespace を作成する。
    ///
    /// # Examples
    ///
    /// ```
    /// use alopex_embedded::{CreateCatalogRequest, CreateNamespaceRequest, Database};
    ///
    /// let db = Database::new();
    /// db.create_catalog(CreateCatalogRequest::new("main")).unwrap();
    /// let namespace = db
    ///     .create_namespace(CreateNamespaceRequest::new("main", "analytics"))
    ///     .unwrap();
    /// assert_eq!(namespace.catalog_name, "main");
    /// assert_eq!(namespace.name, "analytics");
    /// ```
    pub fn create_namespace(&self, request: CreateNamespaceRequest) -> Result<NamespaceInfo> {
        let request = request.build()?;
        let mut catalog = self.sql_catalog.write().expect("catalog lock poisoned");
        let catalog_meta = catalog
            .get_catalog(&request.catalog_name)
            .ok_or_else(|| Error::CatalogNotFound(request.catalog_name.clone()))?;
        if catalog
            .get_namespace(&request.catalog_name, &request.name)
            .is_some()
        {
            return Err(Error::NamespaceAlreadyExists(
                request.catalog_name,
                request.name,
            ));
        }

        let storage_root = request
            .storage_root
            .or_else(|| catalog_meta.storage_root.clone());
        let meta = NamespaceMeta {
            name: request.name,
            catalog_name: request.catalog_name,
            comment: request.comment,
            storage_root,
        };
        catalog
            .create_namespace(meta.clone())
            .map_err(|err| Error::Sql(err.into()))?;
        self.invalidate_table_info_cache();
        Ok(meta.into())
    }

    /// Namespace を削除する。
    pub fn delete_namespace(
        &self,
        catalog_name: &str,
        namespace_name: &str,
        force: bool,
    ) -> Result<()> {
        if namespace_name == "default" {
            return Err(Error::CannotDeleteDefault("namespace".to_string()));
        }
        let mut catalog = self.sql_catalog.write().expect("catalog lock poisoned");
        ensure_namespace_exists(&*catalog, catalog_name, namespace_name)?;

        let overlay = CatalogOverlay::new();
        let tables = catalog.list_tables_in_txn(catalog_name, namespace_name, &overlay);
        if !force && !tables.is_empty() {
            return Err(Error::NamespaceNotEmpty(
                catalog_name.to_string(),
                namespace_name.to_string(),
            ));
        }

        if force {
            let store = catalog.store().clone();
            let mut txn = store.begin(TxnMode::ReadWrite).map_err(Error::Core)?;
            for table in &tables {
                catalog
                    .persist_drop_table(&mut txn, &TableFqn::from(table))
                    .map_err(|err| Error::Sql(err.into()))?;
            }
            txn.commit_self().map_err(Error::Core)?;

            let mut overlay = CatalogOverlay::new();
            for table in tables {
                overlay.drop_table(&TableFqn::from(&table));
            }
            catalog.apply_overlay(overlay);
        }

        catalog
            .delete_namespace(catalog_name, namespace_name)
            .map_err(|err| Error::Sql(err.into()))?;
        self.invalidate_table_info_cache();
        Ok(())
    }

    /// テーブルを作成する。
    ///
    /// # Examples
    ///
    /// ```
    /// use alopex_embedded::{
    ///     ColumnDefinition, CreateCatalogRequest, CreateNamespaceRequest, CreateTableRequest,
    ///     Database,
    /// };
    /// use alopex_sql::ast::ddl::DataType;
    ///
    /// let db = Database::new();
    /// db.create_catalog(CreateCatalogRequest::new("default")).unwrap();
    /// db.create_namespace(CreateNamespaceRequest::new("default", "default"))
    ///     .unwrap();
    ///
    /// let schema = vec![ColumnDefinition::new("id", DataType::Integer)];
    /// let table = db
    ///     .create_table(CreateTableRequest::new("users").with_schema(schema))
    ///     .unwrap();
    /// assert_eq!(table.name, "users");
    /// ```
    pub fn create_table(&self, request: CreateTableRequest) -> Result<TableInfo> {
        let request = request.build()?;
        let mut catalog = self.sql_catalog.write().expect("catalog lock poisoned");

        ensure_namespace_exists(&*catalog, &request.catalog_name, &request.namespace_name)?;
        ensure_table_absent(
            &*catalog,
            &request.catalog_name,
            &request.namespace_name,
            &request.name,
        )?;

        if request.table_type == TableType::Managed && request.storage_root.is_some() {
            eprintln!("警告: managed テーブルの storage_root は無視されます");
        }

        let table_id = catalog.next_table_id();
        let primary_key = request.primary_key.clone();
        let columns = build_columns(request.schema.clone(), primary_key.as_ref())?;

        let storage_options = request.storage_options.unwrap_or_else(|| StorageOptions {
            compression: Compression::None,
            ..StorageOptions::default()
        });

        let namespace = catalog.get_namespace(&request.catalog_name, &request.namespace_name);
        let storage_location = resolve_storage_location(
            &request.table_type,
            request.storage_root.as_deref(),
            namespace.as_ref(),
            &request.name,
        )?;

        let mut table = TableMetadata::new(&request.name, columns).with_table_id(table_id);
        table.catalog_name = request.catalog_name.clone();
        table.namespace_name = request.namespace_name.clone();
        table.primary_key = primary_key;
        table.storage_options = storage_options;
        table.table_type = request.table_type;
        table.data_source_format = request
            .data_source_format
            .unwrap_or(DataSourceFormat::Alopex);
        table.storage_location = storage_location;
        table.comment = request.comment;
        table.properties = request.properties.unwrap_or_default();

        let store = catalog.store().clone();
        let mut txn = store.begin(TxnMode::ReadWrite).map_err(Error::Core)?;
        catalog
            .persist_create_table(&mut txn, &table)
            .map_err(|err| Error::Sql(err.into()))?;
        txn.commit_self().map_err(Error::Core)?;

        let mut overlay = CatalogOverlay::new();
        overlay.add_table(TableFqn::from(&table), table.clone());
        catalog.apply_overlay(overlay);
        drop(catalog); // Release lock before invalidating cache
        self.invalidate_table_info_cache();

        let info = TableInfo::from(table);
        let namespace_root = namespace.and_then(|ns| ns.storage_root);
        Ok(apply_storage_location(info, namespace_root.as_deref()))
    }

    /// デフォルト catalog/namespace のテーブルを作成する。
    pub fn create_table_simple(
        &self,
        name: &str,
        schema: Vec<ColumnDefinition>,
    ) -> Result<TableInfo> {
        self.create_table(CreateTableRequest::new(name).with_schema(schema))
    }

    /// テーブルを削除する。
    pub fn delete_table(
        &self,
        catalog_name: &str,
        namespace_name: &str,
        table_name: &str,
    ) -> Result<()> {
        let mut catalog = self.sql_catalog.write().expect("catalog lock poisoned");
        ensure_namespace_exists(&*catalog, catalog_name, namespace_name)?;
        let table = find_table_metadata(&*catalog, catalog_name, namespace_name, table_name)?
            .ok_or_else(|| {
                Error::TableNotFound(table_full_name(catalog_name, namespace_name, table_name))
            })?;

        let store = catalog.store().clone();
        let mut txn = store.begin(TxnMode::ReadWrite).map_err(Error::Core)?;
        catalog
            .persist_drop_table(&mut txn, &TableFqn::from(&table))
            .map_err(|err| Error::Sql(err.into()))?;
        txn.commit_self().map_err(Error::Core)?;

        let mut overlay = CatalogOverlay::new();
        overlay.drop_table(&TableFqn::from(&table));
        catalog.apply_overlay(overlay);
        drop(catalog); // Release lock before invalidating cache
        self.invalidate_table_info_cache();
        Ok(())
    }

    /// デフォルト catalog/namespace のテーブルを削除する。
    pub fn delete_table_simple(&self, name: &str) -> Result<()> {
        self.delete_table("default", "default", name)
    }
}

impl<'a> Transaction<'a> {
    /// Catalog 一覧を取得する（オーバーレイ反映）。
    pub fn list_catalogs(&self) -> Result<Vec<CatalogInfo>> {
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        Ok(catalog
            .list_catalogs_in_txn(self.catalog_overlay())
            .into_iter()
            .map(CatalogInfo::from)
            .collect())
    }

    /// Catalog を取得する（オーバーレイ反映）。
    pub fn get_catalog(&self, name: &str) -> Result<CatalogInfo> {
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        let meta = catalog
            .get_catalog_in_txn(name, self.catalog_overlay())
            .ok_or_else(|| Error::CatalogNotFound(name.to_string()))?;
        Ok(meta.clone().into())
    }

    /// Namespace 一覧を取得する（オーバーレイ反映）。
    pub fn list_namespaces(&self, catalog_name: &str) -> Result<Vec<NamespaceInfo>> {
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        ensure_catalog_exists_in_txn(&*catalog, self.catalog_overlay(), catalog_name)?;
        Ok(catalog
            .list_namespaces_in_txn(catalog_name, self.catalog_overlay())
            .into_iter()
            .map(NamespaceInfo::from)
            .collect())
    }

    /// Namespace を取得する（オーバーレイ反映）。
    pub fn get_namespace(&self, catalog_name: &str, namespace_name: &str) -> Result<NamespaceInfo> {
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        ensure_catalog_exists_in_txn(&*catalog, self.catalog_overlay(), catalog_name)?;
        let meta = catalog
            .get_namespace_in_txn(catalog_name, namespace_name, self.catalog_overlay())
            .ok_or_else(|| {
                Error::NamespaceNotFound(catalog_name.to_string(), namespace_name.to_string())
            })?;
        Ok(meta.clone().into())
    }

    /// テーブル一覧を取得する（オーバーレイ反映）。
    pub fn list_tables(&self, catalog_name: &str, namespace_name: &str) -> Result<Vec<TableInfo>> {
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        ensure_namespace_exists_in_txn(
            &*catalog,
            self.catalog_overlay(),
            catalog_name,
            namespace_name,
        )?;
        let namespace = catalog
            .get_namespace_in_txn(catalog_name, namespace_name, self.catalog_overlay())
            .cloned()
            .ok_or_else(|| {
                Error::NamespaceNotFound(catalog_name.to_string(), namespace_name.to_string())
            })?;
        let tables =
            catalog.list_tables_in_txn(catalog_name, namespace_name, self.catalog_overlay());
        Ok(tables
            .into_iter()
            .map(|table| {
                let info = TableInfo::from(table);
                apply_storage_location(info, namespace.storage_root.as_deref())
            })
            .collect())
    }

    /// テーブル情報を取得する（オーバーレイ反映）。
    pub fn get_table_info(
        &self,
        catalog_name: &str,
        namespace_name: &str,
        table_name: &str,
    ) -> Result<TableInfo> {
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        ensure_namespace_exists_in_txn(
            &*catalog,
            self.catalog_overlay(),
            catalog_name,
            namespace_name,
        )?;
        let namespace = catalog
            .get_namespace_in_txn(catalog_name, namespace_name, self.catalog_overlay())
            .cloned()
            .ok_or_else(|| {
                Error::NamespaceNotFound(catalog_name.to_string(), namespace_name.to_string())
            })?;

        let tables =
            catalog.list_tables_in_txn(catalog_name, namespace_name, self.catalog_overlay());
        let table = tables
            .into_iter()
            .find(|table| table.name == table_name)
            .ok_or_else(|| {
                Error::TableNotFound(table_full_name(catalog_name, namespace_name, table_name))
            })?;
        let info = TableInfo::from(table);
        Ok(apply_storage_location(
            info,
            namespace.storage_root.as_deref(),
        ))
    }

    /// Catalog を作成する（オーバーレイ反映）。
    ///
    /// # Examples
    ///
    /// ```
    /// use alopex_embedded::{CreateCatalogRequest, Database, TxnMode};
    ///
    /// let db = Database::new();
    /// let mut txn = db.begin(TxnMode::ReadWrite).unwrap();
    /// let catalog = txn.create_catalog(CreateCatalogRequest::new("main")).unwrap();
    /// assert_eq!(catalog.name, "main");
    /// ```
    pub fn create_catalog(&mut self, request: CreateCatalogRequest) -> Result<CatalogInfo> {
        ensure_write_mode(self)?;
        let request = request.build()?;
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        if catalog
            .get_catalog_in_txn(&request.name, self.catalog_overlay())
            .is_some()
        {
            return Err(Error::CatalogAlreadyExists(request.name));
        }

        let meta = CatalogMeta {
            name: request.name,
            comment: request.comment,
            storage_root: request.storage_root,
        };
        self.catalog_overlay_mut().add_catalog(meta.clone());
        self.catalog_modified = true;
        Ok(meta.into())
    }

    /// Catalog を削除する（オーバーレイ反映）。
    pub fn delete_catalog(&mut self, name: &str, force: bool) -> Result<()> {
        ensure_write_mode(self)?;
        if name == "default" {
            return Err(Error::CannotDeleteDefault("catalog".to_string()));
        }
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        ensure_catalog_exists_in_txn(&*catalog, self.catalog_overlay(), name)?;

        if !force {
            let namespaces = catalog.list_namespaces_in_txn(name, self.catalog_overlay());
            let has_non_default = namespaces.iter().any(|ns| ns.name != "default");
            let has_tables = namespaces.iter().any(|ns| {
                !catalog
                    .list_tables_in_txn(name, &ns.name, self.catalog_overlay())
                    .is_empty()
            });
            if has_non_default || has_tables {
                return Err(Error::CatalogNotEmpty(name.to_string()));
            }
        }

        if force {
            self.catalog_overlay_mut().drop_cascade_catalog(name);
        } else {
            self.catalog_overlay_mut().drop_catalog(name);
        }
        self.catalog_modified = true;
        Ok(())
    }

    /// Namespace を作成する（オーバーレイ反映）。
    pub fn create_namespace(&mut self, request: CreateNamespaceRequest) -> Result<NamespaceInfo> {
        ensure_write_mode(self)?;
        let request = request.build()?;
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        let catalog_meta = catalog
            .get_catalog_in_txn(&request.catalog_name, self.catalog_overlay())
            .ok_or_else(|| Error::CatalogNotFound(request.catalog_name.clone()))?;
        if catalog
            .get_namespace_in_txn(&request.catalog_name, &request.name, self.catalog_overlay())
            .is_some()
        {
            return Err(Error::NamespaceAlreadyExists(
                request.catalog_name,
                request.name,
            ));
        }

        let storage_root = request
            .storage_root
            .or_else(|| catalog_meta.storage_root.clone());
        let meta = NamespaceMeta {
            name: request.name,
            catalog_name: request.catalog_name,
            comment: request.comment,
            storage_root,
        };
        self.catalog_overlay_mut().add_namespace(meta.clone());
        self.catalog_modified = true;
        Ok(meta.into())
    }

    /// Namespace を削除する（オーバーレイ反映）。
    pub fn delete_namespace(
        &mut self,
        catalog_name: &str,
        namespace_name: &str,
        force: bool,
    ) -> Result<()> {
        ensure_write_mode(self)?;
        if namespace_name == "default" {
            return Err(Error::CannotDeleteDefault("namespace".to_string()));
        }
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        ensure_namespace_exists_in_txn(
            &*catalog,
            self.catalog_overlay(),
            catalog_name,
            namespace_name,
        )?;

        let tables =
            catalog.list_tables_in_txn(catalog_name, namespace_name, self.catalog_overlay());
        if !force && !tables.is_empty() {
            return Err(Error::NamespaceNotEmpty(
                catalog_name.to_string(),
                namespace_name.to_string(),
            ));
        }

        if force {
            self.catalog_overlay_mut()
                .drop_cascade_namespace(catalog_name, namespace_name);
        } else {
            self.catalog_overlay_mut()
                .drop_namespace(catalog_name, namespace_name);
        }
        self.catalog_modified = true;
        Ok(())
    }

    /// テーブルを作成する（オーバーレイ反映）。
    ///
    /// # Examples
    ///
    /// ```
    /// use alopex_embedded::{
    ///     ColumnDefinition, CreateCatalogRequest, CreateNamespaceRequest, CreateTableRequest,
    ///     Database, TxnMode,
    /// };
    /// use alopex_sql::ast::ddl::DataType;
    ///
    /// let db = Database::new();
    /// let mut txn = db.begin(TxnMode::ReadWrite).unwrap();
    /// txn.create_catalog(CreateCatalogRequest::new("main")).unwrap();
    /// txn.create_namespace(CreateNamespaceRequest::new("main", "default"))
    ///     .unwrap();
    ///
    /// let schema = vec![ColumnDefinition::new("id", DataType::Integer)];
    /// let table = txn
    ///     .create_table(
    ///         CreateTableRequest::new("events")
    ///             .with_catalog_name("main")
    ///             .with_namespace_name("default")
    ///             .with_schema(schema),
    ///     )
    ///     .unwrap();
    /// assert_eq!(table.name, "events");
    /// ```
    pub fn create_table(&mut self, request: CreateTableRequest) -> Result<TableInfo> {
        ensure_write_mode(self)?;
        let request = request.build()?;

        let mut catalog = self.db.sql_catalog.write().expect("catalog lock poisoned");
        ensure_namespace_exists_in_txn(
            &*catalog,
            self.catalog_overlay(),
            &request.catalog_name,
            &request.namespace_name,
        )?;
        ensure_table_absent_in_txn(
            &*catalog,
            self.catalog_overlay(),
            &request.catalog_name,
            &request.namespace_name,
            &request.name,
        )?;

        if request.table_type == TableType::Managed && request.storage_root.is_some() {
            eprintln!("警告: managed テーブルの storage_root は無視されます");
        }

        let table_id = catalog.next_table_id();
        let primary_key = request.primary_key.clone();
        let columns = build_columns(request.schema.clone(), primary_key.as_ref())?;

        let storage_options = request.storage_options.unwrap_or_else(|| StorageOptions {
            compression: Compression::None,
            ..StorageOptions::default()
        });

        let namespace = catalog
            .get_namespace_in_txn(
                &request.catalog_name,
                &request.namespace_name,
                self.catalog_overlay(),
            )
            .cloned();
        let storage_location = resolve_storage_location(
            &request.table_type,
            request.storage_root.as_deref(),
            namespace.as_ref(),
            &request.name,
        )?;

        let mut table = TableMetadata::new(&request.name, columns).with_table_id(table_id);
        table.catalog_name = request.catalog_name.clone();
        table.namespace_name = request.namespace_name.clone();
        table.primary_key = primary_key;
        table.storage_options = storage_options;
        table.table_type = request.table_type;
        table.data_source_format = request
            .data_source_format
            .unwrap_or(DataSourceFormat::Alopex);
        table.storage_location = storage_location;
        table.comment = request.comment;
        table.properties = request.properties.unwrap_or_default();

        self.catalog_overlay_mut()
            .add_table(TableFqn::from(&table), table.clone());
        self.catalog_modified = true;
        let info = TableInfo::from(table);
        let namespace_root = namespace.and_then(|ns| ns.storage_root);
        Ok(apply_storage_location(info, namespace_root.as_deref()))
    }

    /// テーブルを削除する（オーバーレイ反映）。
    pub fn delete_table(
        &mut self,
        catalog_name: &str,
        namespace_name: &str,
        table_name: &str,
    ) -> Result<()> {
        ensure_write_mode(self)?;
        let catalog = self.db.sql_catalog.read().expect("catalog lock poisoned");
        ensure_namespace_exists_in_txn(
            &*catalog,
            self.catalog_overlay(),
            catalog_name,
            namespace_name,
        )?;
        let table = find_table_metadata_in_txn(
            &*catalog,
            self.catalog_overlay(),
            catalog_name,
            namespace_name,
            table_name,
        )?
        .ok_or_else(|| {
            Error::TableNotFound(table_full_name(catalog_name, namespace_name, table_name))
        })?;

        self.catalog_overlay_mut()
            .drop_table(&TableFqn::from(&table));
        self.catalog_modified = true;
        Ok(())
    }
}

fn validate_required(value: &str, label: &str) -> Result<()> {
    if value.trim().is_empty() {
        return Err(Error::Core(alopex_core::Error::InvalidFormat(format!(
            "{label}が未指定です"
        ))));
    }
    Ok(())
}

fn ensure_catalog_exists<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    name: &str,
) -> Result<()> {
    if catalog.get_catalog(name).is_none() {
        return Err(Error::CatalogNotFound(name.to_string()));
    }
    Ok(())
}

fn ensure_catalog_exists_in_txn<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    overlay: &CatalogOverlay,
    name: &str,
) -> Result<()> {
    if catalog.get_catalog_in_txn(name, overlay).is_none() {
        return Err(Error::CatalogNotFound(name.to_string()));
    }
    Ok(())
}

fn ensure_namespace_exists<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    catalog_name: &str,
    namespace_name: &str,
) -> Result<()> {
    ensure_catalog_exists(catalog, catalog_name)?;
    if catalog
        .get_namespace(catalog_name, namespace_name)
        .is_none()
    {
        return Err(Error::NamespaceNotFound(
            catalog_name.to_string(),
            namespace_name.to_string(),
        ));
    }
    Ok(())
}

fn ensure_namespace_exists_in_txn<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    overlay: &CatalogOverlay,
    catalog_name: &str,
    namespace_name: &str,
) -> Result<()> {
    ensure_catalog_exists_in_txn(catalog, overlay, catalog_name)?;
    if catalog
        .get_namespace_in_txn(catalog_name, namespace_name, overlay)
        .is_none()
    {
        return Err(Error::NamespaceNotFound(
            catalog_name.to_string(),
            namespace_name.to_string(),
        ));
    }
    Ok(())
}

fn ensure_table_exists<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    catalog_name: &str,
    namespace_name: &str,
    table_name: &str,
) -> Result<()> {
    let Some(table) = find_table_metadata(catalog, catalog_name, namespace_name, table_name)?
    else {
        return Err(Error::TableNotFound(table_full_name(
            catalog_name,
            namespace_name,
            table_name,
        )));
    };
    let _ = table;
    Ok(())
}

fn ensure_table_absent<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    catalog_name: &str,
    namespace_name: &str,
    table_name: &str,
) -> Result<()> {
    if find_table_metadata(catalog, catalog_name, namespace_name, table_name)?.is_some() {
        return Err(Error::TableAlreadyExists(table_full_name(
            catalog_name,
            namespace_name,
            table_name,
        )));
    }
    Ok(())
}

fn ensure_table_absent_in_txn<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    overlay: &CatalogOverlay,
    catalog_name: &str,
    namespace_name: &str,
    table_name: &str,
) -> Result<()> {
    if find_table_metadata_in_txn(catalog, overlay, catalog_name, namespace_name, table_name)?
        .is_some()
    {
        return Err(Error::TableAlreadyExists(table_full_name(
            catalog_name,
            namespace_name,
            table_name,
        )));
    }
    Ok(())
}

fn find_table_metadata<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    catalog_name: &str,
    namespace_name: &str,
    table_name: &str,
) -> Result<Option<TableMetadata>> {
    let overlay = CatalogOverlay::new();
    let tables = catalog.list_tables_in_txn(catalog_name, namespace_name, &overlay);
    Ok(tables.into_iter().find(|table| table.name == table_name))
}

fn find_table_metadata_in_txn<S: alopex_core::kv::KVStore>(
    catalog: &alopex_sql::catalog::PersistentCatalog<S>,
    overlay: &CatalogOverlay,
    catalog_name: &str,
    namespace_name: &str,
    table_name: &str,
) -> Result<Option<TableMetadata>> {
    let tables = catalog.list_tables_in_txn(catalog_name, namespace_name, overlay);
    Ok(tables.into_iter().find(|table| table.name == table_name))
}

fn table_full_name(catalog_name: &str, namespace_name: &str, table_name: &str) -> String {
    format!("{catalog_name}.{namespace_name}.{table_name}")
}

fn index_full_name(
    catalog_name: &str,
    namespace_name: &str,
    table_name: &str,
    index_name: &str,
) -> String {
    format!("{catalog_name}.{namespace_name}.{table_name}.{index_name}")
}

fn apply_storage_location(mut info: TableInfo, namespace_root: Option<&str>) -> TableInfo {
    if info.storage_location.is_none() && info.table_type == TableType::Managed {
        if let Some(root) = namespace_root {
            info.storage_location = Some(format!("{root}/{}", info.name));
        }
    }
    info
}

fn resolve_storage_location(
    table_type: &TableType,
    request_storage_root: Option<&str>,
    namespace: Option<&NamespaceMeta>,
    table_name: &str,
) -> Result<Option<String>> {
    match table_type {
        TableType::Managed => Ok(namespace
            .and_then(|ns| ns.storage_root.as_deref())
            .map(|root| format!("{root}/{table_name}"))),
        TableType::External => {
            let storage_root = request_storage_root
                .map(|root| root.to_string())
                .ok_or(Error::StorageRootRequired)?;
            Ok(Some(storage_root))
        }
    }
}

fn build_columns(
    schema: Option<Vec<ColumnDefinition>>,
    primary_key: Option<&Vec<String>>,
) -> Result<Vec<ColumnMetadata>> {
    let Some(schema) = schema else {
        return Ok(Vec::new());
    };

    let mut columns = Vec::with_capacity(schema.len());
    for definition in schema {
        validate_required(&definition.name, "column 名")?;
        let mut column = ColumnMetadata::new(
            definition.name.clone(),
            ResolvedType::from_ast(&definition.data_type),
        )
        .with_not_null(!definition.nullable);
        if primary_key
            .map(|keys| keys.iter().any(|key| key == &definition.name))
            .unwrap_or(false)
        {
            column = column.with_primary_key(true).with_not_null(true);
        }
        columns.push(column);
    }

    if let Some(keys) = primary_key {
        let missing: Vec<String> = keys
            .iter()
            .filter(|key| !columns.iter().any(|col| col.name == **key))
            .cloned()
            .collect();
        if !missing.is_empty() {
            return Err(Error::Core(alopex_core::Error::InvalidFormat(format!(
                "主キーが見つかりません: {}",
                missing.join(", ")
            ))));
        }
    }

    Ok(columns)
}

fn ensure_write_mode(txn: &Transaction<'_>) -> Result<()> {
    let mode = txn.txn_mode()?;
    if mode != TxnMode::ReadWrite {
        return Err(Error::TxnReadOnly);
    }
    Ok(())
}

fn resolved_type_to_string(resolved_type: &ResolvedType) -> String {
    match resolved_type {
        ResolvedType::Integer => "INTEGER".to_string(),
        ResolvedType::BigInt => "BIGINT".to_string(),
        ResolvedType::Float => "FLOAT".to_string(),
        ResolvedType::Double => "DOUBLE".to_string(),
        ResolvedType::Text => "TEXT".to_string(),
        ResolvedType::Blob => "BLOB".to_string(),
        ResolvedType::Boolean => "BOOLEAN".to_string(),
        ResolvedType::Timestamp => "TIMESTAMP".to_string(),
        ResolvedType::Vector { dimension, metric } => {
            let metric = match metric {
                VectorMetric::Cosine => "COSINE",
                VectorMetric::L2 => "L2",
                VectorMetric::Inner => "INNER",
            };
            format!("VECTOR({dimension}, {metric})")
        }
        ResolvedType::Null => "NULL".to_string(),
    }
}

/// The only catalog representation accepted in a schema manifest.  It is a
/// versioned structural document, never user-supplied SQL or Rust-private
/// catalog persistence bytes.
pub const CATALOG_MANIFEST_DELTA_FORMAT: &str = "alopex.catalog.snapshot.v1";

const CATALOG_MANIFEST_VERSION_KEY: &[u8] = b"__alopex/schema-manifest-version/v1";

/// Stable, versioned public representation of the catalog carried by a
/// schema manifest.  The snapshot is additive: an existing object must be
/// identical, while a missing object is created atomically with the reported
/// catalog version.  Destructive schema replacement is intentionally outside
/// this Phase 1 control path.
#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogManifestDelta {
    pub format_version: u32,
    pub catalog_version: u64,
    #[serde(default)]
    pub catalogs: Vec<CatalogManifestCatalog>,
    #[serde(default)]
    pub namespaces: Vec<CatalogManifestNamespace>,
    pub tables: Vec<CatalogManifestTable>,
    pub indexes: Vec<CatalogManifestIndex>,
}

#[allow(missing_docs)]
impl CatalogManifestDelta {
    pub const FORMAT_VERSION: u32 = 1;

    pub fn encode(&self) -> std::result::Result<Vec<u8>, serde_json::Error> {
        serde_json::to_vec(self)
    }

    pub fn decode(bytes: &[u8]) -> std::result::Result<Self, serde_json::Error> {
        serde_json::from_slice(bytes)
    }
}

/// Catalog metadata in [`CatalogManifestDelta`].
#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogManifestCatalog {
    pub name: String,
    pub comment: Option<String>,
    pub storage_root: Option<String>,
}

impl From<&CatalogMeta> for CatalogManifestCatalog {
    fn from(meta: &CatalogMeta) -> Self {
        Self {
            name: meta.name.clone(),
            comment: meta.comment.clone(),
            storage_root: meta.storage_root.clone(),
        }
    }
}

impl From<&CatalogManifestCatalog> for CatalogMeta {
    fn from(manifest: &CatalogManifestCatalog) -> Self {
        Self {
            name: manifest.name.clone(),
            comment: manifest.comment.clone(),
            storage_root: manifest.storage_root.clone(),
        }
    }
}

/// Namespace metadata in [`CatalogManifestDelta`].
#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogManifestNamespace {
    pub catalog_name: String,
    pub name: String,
    pub comment: Option<String>,
    pub storage_root: Option<String>,
}

impl From<&NamespaceMeta> for CatalogManifestNamespace {
    fn from(meta: &NamespaceMeta) -> Self {
        Self {
            catalog_name: meta.catalog_name.clone(),
            name: meta.name.clone(),
            comment: meta.comment.clone(),
            storage_root: meta.storage_root.clone(),
        }
    }
}

impl From<&CatalogManifestNamespace> for NamespaceMeta {
    fn from(manifest: &CatalogManifestNamespace) -> Self {
        Self {
            catalog_name: manifest.catalog_name.clone(),
            name: manifest.name.clone(),
            comment: manifest.comment.clone(),
            storage_root: manifest.storage_root.clone(),
        }
    }
}

/// Table definition in [`CatalogManifestDelta`].
#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogManifestTable {
    pub table_id: u32,
    pub catalog_name: String,
    pub namespace_name: String,
    pub name: String,
    pub table_type: CatalogManifestTableType,
    pub data_source_format: CatalogManifestDataSourceFormat,
    pub columns: Vec<CatalogManifestColumn>,
    pub primary_key: Option<Vec<String>>,
    pub storage: CatalogManifestStorage,
    pub storage_location: Option<String>,
    pub comment: Option<String>,
    pub properties: BTreeMap<String, String>,
}

impl CatalogManifestTable {
    fn fqn(&self) -> TableFqn {
        TableFqn::new(&self.catalog_name, &self.namespace_name, &self.name)
    }

    fn to_metadata(&self) -> TableMetadata {
        let columns = self
            .columns
            .iter()
            .map(CatalogManifestColumn::to_metadata)
            .collect();
        let mut table = TableMetadata::new(self.name.clone(), columns).with_table_id(self.table_id);
        table.catalog_name = self.catalog_name.clone();
        table.namespace_name = self.namespace_name.clone();
        table.table_type = self.table_type.into();
        table.data_source_format = self.data_source_format.into();
        table.primary_key = self.primary_key.clone();
        table.storage_options = self.storage.to_options();
        table.storage_location = self.storage_location.clone();
        table.comment = self.comment.clone();
        table.properties = self.properties.clone().into_iter().collect();
        table
    }
}

impl From<&TableMetadata> for CatalogManifestTable {
    fn from(table: &TableMetadata) -> Self {
        Self {
            table_id: table.table_id,
            catalog_name: table.catalog_name.clone(),
            namespace_name: table.namespace_name.clone(),
            name: table.name.clone(),
            table_type: table.table_type.into(),
            data_source_format: table.data_source_format.into(),
            columns: table
                .columns
                .iter()
                .map(CatalogManifestColumn::from)
                .collect(),
            primary_key: table.primary_key.clone(),
            storage: CatalogManifestStorage::from(&table.storage_options),
            storage_location: table.storage_location.clone(),
            comment: table.comment.clone(),
            properties: table
                .properties
                .iter()
                .map(|(key, value)| (key.clone(), value.clone()))
                .collect(),
        }
    }
}

/// Column definition in [`CatalogManifestTable`].  SQL default expressions
/// are deliberately excluded because the persistent catalog does not retain
/// them as a portable catalog contract.
#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogManifestColumn {
    pub name: String,
    pub data_type: CatalogManifestDataType,
    pub not_null: bool,
    pub primary_key: bool,
    pub unique: bool,
}

impl CatalogManifestColumn {
    fn to_metadata(&self) -> ColumnMetadata {
        ColumnMetadata::new(self.name.clone(), self.data_type.clone().into())
            .with_not_null(self.not_null)
            .with_primary_key(self.primary_key)
            .with_unique(self.unique)
    }
}

impl From<&ColumnMetadata> for CatalogManifestColumn {
    fn from(column: &ColumnMetadata) -> Self {
        Self {
            name: column.name.clone(),
            data_type: (&column.data_type).into(),
            not_null: column.not_null,
            primary_key: column.primary_key,
            unique: column.unique,
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CatalogManifestTableType {
    Managed,
    External,
}

impl From<TableType> for CatalogManifestTableType {
    fn from(value: TableType) -> Self {
        match value {
            TableType::Managed => Self::Managed,
            TableType::External => Self::External,
        }
    }
}

impl From<CatalogManifestTableType> for TableType {
    fn from(value: CatalogManifestTableType) -> Self {
        match value {
            CatalogManifestTableType::Managed => Self::Managed,
            CatalogManifestTableType::External => Self::External,
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CatalogManifestDataSourceFormat {
    Alopex,
    Parquet,
    Delta,
}

impl From<DataSourceFormat> for CatalogManifestDataSourceFormat {
    fn from(value: DataSourceFormat) -> Self {
        match value {
            DataSourceFormat::Alopex => Self::Alopex,
            DataSourceFormat::Parquet => Self::Parquet,
            DataSourceFormat::Delta => Self::Delta,
        }
    }
}

impl From<CatalogManifestDataSourceFormat> for DataSourceFormat {
    fn from(value: CatalogManifestDataSourceFormat) -> Self {
        match value {
            CatalogManifestDataSourceFormat::Alopex => Self::Alopex,
            CatalogManifestDataSourceFormat::Parquet => Self::Parquet,
            CatalogManifestDataSourceFormat::Delta => Self::Delta,
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum CatalogManifestDataType {
    Integer,
    BigInt,
    Float,
    Double,
    Text,
    Blob,
    Boolean,
    Timestamp,
    Vector {
        dimension: u32,
        metric: CatalogManifestVectorMetric,
    },
    Null,
}

impl From<&ResolvedType> for CatalogManifestDataType {
    fn from(value: &ResolvedType) -> Self {
        match value {
            ResolvedType::Integer => Self::Integer,
            ResolvedType::BigInt => Self::BigInt,
            ResolvedType::Float => Self::Float,
            ResolvedType::Double => Self::Double,
            ResolvedType::Text => Self::Text,
            ResolvedType::Blob => Self::Blob,
            ResolvedType::Boolean => Self::Boolean,
            ResolvedType::Timestamp => Self::Timestamp,
            ResolvedType::Vector { dimension, metric } => Self::Vector {
                dimension: *dimension,
                metric: (*metric).into(),
            },
            ResolvedType::Null => Self::Null,
        }
    }
}

impl From<CatalogManifestDataType> for ResolvedType {
    fn from(value: CatalogManifestDataType) -> Self {
        match value {
            CatalogManifestDataType::Integer => Self::Integer,
            CatalogManifestDataType::BigInt => Self::BigInt,
            CatalogManifestDataType::Float => Self::Float,
            CatalogManifestDataType::Double => Self::Double,
            CatalogManifestDataType::Text => Self::Text,
            CatalogManifestDataType::Blob => Self::Blob,
            CatalogManifestDataType::Boolean => Self::Boolean,
            CatalogManifestDataType::Timestamp => Self::Timestamp,
            CatalogManifestDataType::Vector { dimension, metric } => Self::Vector {
                dimension,
                metric: metric.into(),
            },
            CatalogManifestDataType::Null => Self::Null,
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CatalogManifestVectorMetric {
    Cosine,
    L2,
    Inner,
}

impl From<VectorMetric> for CatalogManifestVectorMetric {
    fn from(value: VectorMetric) -> Self {
        match value {
            VectorMetric::Cosine => Self::Cosine,
            VectorMetric::L2 => Self::L2,
            VectorMetric::Inner => Self::Inner,
        }
    }
}

impl From<CatalogManifestVectorMetric> for VectorMetric {
    fn from(value: CatalogManifestVectorMetric) -> Self {
        match value {
            CatalogManifestVectorMetric::Cosine => Self::Cosine,
            CatalogManifestVectorMetric::L2 => Self::L2,
            CatalogManifestVectorMetric::Inner => Self::Inner,
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogManifestStorage {
    pub storage_type: CatalogManifestStorageType,
    pub compression: CatalogManifestCompression,
    pub row_group_size: u32,
    pub row_id_mode: CatalogManifestRowIdMode,
}

impl CatalogManifestStorage {
    fn to_options(&self) -> StorageOptions {
        StorageOptions {
            storage_type: self.storage_type.into(),
            compression: self.compression.into(),
            row_group_size: self.row_group_size,
            row_id_mode: self.row_id_mode.into(),
        }
    }
}

impl From<&StorageOptions> for CatalogManifestStorage {
    fn from(value: &StorageOptions) -> Self {
        Self {
            storage_type: value.storage_type.into(),
            compression: value.compression.into(),
            row_group_size: value.row_group_size,
            row_id_mode: value.row_id_mode.into(),
        }
    }
}

macro_rules! catalog_manifest_enum {
    ($manifest:ident, $native:ident, { $($variant:ident),+ $(,)? }) => {
        #[allow(missing_docs)]
        #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
        #[serde(rename_all = "snake_case")]
        pub enum $manifest { $($variant),+ }

        impl From<$native> for $manifest {
            fn from(value: $native) -> Self {
                match value { $($native::$variant => Self::$variant),+ }
            }
        }

        impl From<$manifest> for $native {
            fn from(value: $manifest) -> Self {
                match value { $($manifest::$variant => Self::$variant),+ }
            }
        }
    };
}

catalog_manifest_enum!(CatalogManifestStorageType, StorageType, { Row, Columnar });
catalog_manifest_enum!(CatalogManifestCompression, Compression, { None, Lz4, Zstd });
catalog_manifest_enum!(CatalogManifestRowIdMode, RowIdMode, { None, Direct });

/// Index definition in [`CatalogManifestDelta`].
#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogManifestIndex {
    pub index_id: u32,
    pub catalog_name: String,
    pub namespace_name: String,
    pub name: String,
    pub table: String,
    pub columns: Vec<String>,
    pub column_indices: Vec<usize>,
    pub unique: bool,
    pub method: Option<CatalogManifestIndexMethod>,
    pub options: Vec<(String, String)>,
}

impl CatalogManifestIndex {
    fn fqn(&self) -> IndexFqn {
        IndexFqn::new(
            &self.catalog_name,
            &self.namespace_name,
            &self.table,
            &self.name,
        )
    }

    fn to_metadata(&self) -> IndexMetadata {
        let mut index = IndexMetadata::new(
            self.index_id,
            self.name.clone(),
            self.table.clone(),
            self.columns.clone(),
        )
        .with_column_indices(self.column_indices.clone())
        .with_unique(self.unique)
        .with_options(self.options.clone());
        index.catalog_name = self.catalog_name.clone();
        index.namespace_name = self.namespace_name.clone();
        if let Some(method) = self.method {
            index = index.with_method(method.into());
        }
        index
    }
}

impl From<&IndexMetadata> for CatalogManifestIndex {
    fn from(index: &IndexMetadata) -> Self {
        Self {
            index_id: index.index_id,
            catalog_name: index.catalog_name.clone(),
            namespace_name: index.namespace_name.clone(),
            name: index.name.clone(),
            table: index.table.clone(),
            columns: index.columns.clone(),
            column_indices: index.column_indices.clone(),
            unique: index.unique,
            method: index.method.map(Into::into),
            options: index.options.clone(),
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CatalogManifestIndexMethod {
    BTree,
    Hnsw,
}

impl From<IndexMethod> for CatalogManifestIndexMethod {
    fn from(value: IndexMethod) -> Self {
        match value {
            IndexMethod::BTree => Self::BTree,
            IndexMethod::Hnsw => Self::Hnsw,
        }
    }
}

impl From<CatalogManifestIndexMethod> for IndexMethod {
    fn from(value: CatalogManifestIndexMethod) -> Self {
        match value {
            CatalogManifestIndexMethod::BTree => Self::BTree,
            CatalogManifestIndexMethod::Hnsw => Self::Hnsw,
        }
    }
}

impl Database {
    /// Exports the currently visible SQL catalog as a versioned manifest
    /// document.  This captures catalog metadata only; it does not transport
    /// user SQL or table data.
    pub fn export_catalog_manifest_delta(&self, catalog_version: u64) -> Result<Vec<u8>> {
        let catalog = self.sql_catalog.read().expect("catalog lock poisoned");
        let mut catalogs = catalog
            .list_catalogs()
            .iter()
            .map(CatalogManifestCatalog::from)
            .collect::<Vec<_>>();
        catalogs.sort_by(|left, right| left.name.cmp(&right.name));
        let mut namespaces = catalog
            .list_catalogs()
            .iter()
            .flat_map(|meta| {
                catalog
                    .list_namespaces(&meta.name)
                    .iter()
                    .map(CatalogManifestNamespace::from)
                    .collect::<Vec<_>>()
            })
            .collect::<Vec<_>>();
        namespaces.sort_by(|left, right| {
            (&left.catalog_name, &left.name).cmp(&(&right.catalog_name, &right.name))
        });
        let mut tables = catalog
            .list_tables()
            .iter()
            .map(CatalogManifestTable::from)
            .collect::<Vec<_>>();
        tables.sort_by(|left, right| {
            (&left.catalog_name, &left.namespace_name, &left.name).cmp(&(
                &right.catalog_name,
                &right.namespace_name,
                &right.name,
            ))
        });
        let mut indexes = catalog
            .list_tables()
            .iter()
            .flat_map(|table| {
                catalog
                    .get_indexes_for_table(&table.name)
                    .into_iter()
                    .filter(|index| {
                        index.catalog_name == table.catalog_name
                            && index.namespace_name == table.namespace_name
                    })
                    .map(CatalogManifestIndex::from)
                    .collect::<Vec<_>>()
            })
            .collect::<Vec<_>>();
        indexes.sort_by(|left, right| {
            (
                &left.catalog_name,
                &left.namespace_name,
                &left.table,
                &left.name,
            )
                .cmp(&(
                    &right.catalog_name,
                    &right.namespace_name,
                    &right.table,
                    &right.name,
                ))
        });
        CatalogManifestDelta {
            format_version: CatalogManifestDelta::FORMAT_VERSION,
            catalog_version,
            catalogs,
            namespaces,
            tables,
            indexes,
        }
        .encode()
        .map_err(catalog_manifest_encoding_error)
    }

    /// Returns the catalog version that was atomically recorded with the last
    /// successfully applied schema manifest.  `None` is an untouched local
    /// catalog, which compatibility checks treat as version zero.
    pub fn catalog_manifest_version(&self) -> Result<Option<u64>> {
        let mut txn = self.store.begin(TxnMode::ReadOnly).map_err(Error::Core)?;
        let stored = txn
            .get(&CATALOG_MANIFEST_VERSION_KEY.to_vec())
            .map_err(Error::Core)?;
        txn.rollback_self().map_err(Error::Core)?;
        match stored {
            None => Ok(None),
            Some(bytes) if bytes.len() == std::mem::size_of::<u64>() => {
                let mut encoded = [0u8; std::mem::size_of::<u64>()];
                encoded.copy_from_slice(&bytes);
                Ok(Some(u64::from_be_bytes(encoded)))
            }
            Some(_) => Err(Error::Core(alopex_core::Error::InvalidFormat(
                "invalid stored schema manifest catalog version".to_string(),
            ))),
        }
    }

    /// Applies a committed management manifest to this member's local SQL
    /// catalog.  The returned evidence is suitable for
    /// `SchemaApplyEvidenceAdapter`; it reports `Applied` only after the
    /// catalog overlay and manifest version commit together.
    pub fn apply_schema_manifest(
        &self,
        member: impl Into<NodeId>,
        manifest: &SchemaManifest,
    ) -> SchemaApplyEvidence {
        let member = member.into();
        if manifest.catalog_delta_format != CATALOG_MANIFEST_DELTA_FORMAT {
            return apply_evidence(
                manifest,
                member,
                SchemaApplyState::Incompatible,
                None,
                false,
                "unsupported catalog manifest format",
            );
        }
        let actual_checksum = format!("{:x}", sha2::Sha256::digest(&manifest.catalog_delta));
        if actual_checksum != manifest.checksum {
            return apply_evidence(
                manifest,
                member,
                SchemaApplyState::Failed,
                None,
                false,
                "catalog manifest checksum mismatch",
            );
        }
        let delta = match CatalogManifestDelta::decode(&manifest.catalog_delta) {
            Ok(delta) => delta,
            Err(_) => {
                return apply_evidence(
                    manifest,
                    member,
                    SchemaApplyState::Incompatible,
                    None,
                    false,
                    "catalog manifest payload is not a supported structural document",
                );
            }
        };
        if let Err(detail) = validate_catalog_manifest_delta(&delta) {
            return apply_evidence(
                manifest,
                member,
                SchemaApplyState::Incompatible,
                None,
                false,
                detail,
            );
        }
        if delta.catalog_version != manifest.schema_version {
            return apply_evidence(
                manifest,
                member,
                SchemaApplyState::Incompatible,
                None,
                false,
                "catalog document version does not match the committed schema version",
            );
        }
        let current_version = match self.catalog_manifest_version() {
            Ok(version) => version.unwrap_or(0),
            Err(_) => {
                return apply_evidence(
                    manifest,
                    member,
                    SchemaApplyState::Failed,
                    None,
                    false,
                    "local catalog manifest version could not be read",
                );
            }
        };
        if current_version < manifest.compatibility.minimum_catalog_version
            || current_version > manifest.compatibility.maximum_catalog_version
        {
            return apply_evidence(
                manifest,
                member,
                SchemaApplyState::Incompatible,
                Some(current_version),
                false,
                "local catalog version is outside the manifest compatibility range",
            );
        }
        match self.apply_catalog_manifest_delta(&delta) {
            Ok(()) => apply_evidence(
                manifest,
                member,
                SchemaApplyState::Applied,
                Some(delta.catalog_version),
                true,
                "",
            ),
            Err(detail) => apply_evidence(
                manifest,
                member,
                SchemaApplyState::Failed,
                Some(current_version),
                false,
                detail,
            ),
        }
    }

    fn apply_catalog_manifest_delta(
        &self,
        delta: &CatalogManifestDelta,
    ) -> std::result::Result<(), String> {
        let mut txn = self
            .store
            .begin(TxnMode::ReadWrite)
            .map_err(|error| error.to_string())?;
        let mut catalog = self
            .sql_catalog
            .write()
            .map_err(|_| "catalog lock poisoned".to_string())?;
        let mut overlay = CatalogOverlay::new();

        for manifest_catalog in &delta.catalogs {
            match catalog.get_catalog(&manifest_catalog.name) {
                Some(existing) if CatalogManifestCatalog::from(&existing) == *manifest_catalog => {}
                Some(_) => {
                    return Err(format!(
                        "local catalog {} differs from the manifest",
                        manifest_catalog.name
                    ));
                }
                None => overlay.add_catalog(manifest_catalog.into()),
            }
        }
        for manifest_namespace in &delta.namespaces {
            match catalog.get_namespace(&manifest_namespace.catalog_name, &manifest_namespace.name)
            {
                Some(existing)
                    if CatalogManifestNamespace::from(&existing) == *manifest_namespace => {}
                Some(_) => {
                    return Err(format!(
                        "local namespace {}.{} differs from the manifest",
                        manifest_namespace.catalog_name, manifest_namespace.name
                    ));
                }
                None => overlay.add_namespace(manifest_namespace.into()),
            }
        }

        for table in &delta.tables {
            let existing = catalog.list_tables().into_iter().find(|candidate| {
                candidate.name == table.name
                    && candidate.catalog_name == table.catalog_name
                    && candidate.namespace_name == table.namespace_name
            });
            match existing {
                Some(existing) if CatalogManifestTable::from(&existing) == *table => {}
                Some(_) => {
                    return Err(format!(
                        "local table {}.{}.{} differs from the manifest",
                        table.catalog_name, table.namespace_name, table.name
                    ));
                }
                None => overlay.add_table(table.fqn(), table.to_metadata()),
            }
        }

        for index in &delta.indexes {
            let existing = catalog.get_index(&index.name).filter(|candidate| {
                candidate.catalog_name == index.catalog_name
                    && candidate.namespace_name == index.namespace_name
                    && candidate.table == index.table
            });
            match existing {
                Some(existing) if CatalogManifestIndex::from(existing) == *index => {}
                Some(_) => {
                    return Err(format!(
                        "local index {}.{}.{} differs from the manifest",
                        index.catalog_name, index.namespace_name, index.name
                    ));
                }
                None => overlay.add_index(index.fqn(), index.to_metadata()),
            }
        }

        catalog
            .persist_overlay(&mut txn, &overlay)
            .map_err(|error| error.to_string())?;
        txn.put(
            CATALOG_MANIFEST_VERSION_KEY.to_vec(),
            delta.catalog_version.to_be_bytes().to_vec(),
        )
        .map_err(|error| error.to_string())?;
        txn.commit_self().map_err(|error| error.to_string())?;
        catalog.apply_overlay(overlay);
        drop(catalog);
        self.invalidate_table_info_cache();
        self.hnsw_cache
            .write()
            .map_err(|_| "HNSW cache lock poisoned".to_string())?
            .clear();
        Ok(())
    }
}

fn catalog_manifest_encoding_error(error: serde_json::Error) -> Error {
    Error::Core(alopex_core::Error::InvalidFormat(format!(
        "could not encode catalog manifest: {error}"
    )))
}

fn validate_catalog_manifest_delta(
    delta: &CatalogManifestDelta,
) -> std::result::Result<(), &'static str> {
    if delta.format_version != CatalogManifestDelta::FORMAT_VERSION {
        return Err("unsupported catalog manifest document version");
    }
    let mut catalogs = BTreeSet::new();
    if delta
        .catalogs
        .iter()
        .any(|catalog| catalog.name.trim().is_empty() || !catalogs.insert(catalog.name.as_str()))
    {
        return Err("catalog manifest contains an invalid or duplicate catalog");
    }
    let mut namespaces = BTreeSet::new();
    if delta.namespaces.iter().any(|namespace| {
        namespace.catalog_name.trim().is_empty()
            || namespace.name.trim().is_empty()
            || !namespaces.insert((namespace.catalog_name.as_str(), namespace.name.as_str()))
    }) {
        return Err("catalog manifest contains an invalid or duplicate namespace");
    }
    let mut tables = BTreeSet::new();
    for table in &delta.tables {
        if table.name.trim().is_empty()
            || table.catalog_name.trim().is_empty()
            || table.namespace_name.trim().is_empty()
            || table.columns.is_empty()
            || !tables.insert((
                table.catalog_name.as_str(),
                table.namespace_name.as_str(),
                table.name.as_str(),
            ))
        {
            return Err("catalog manifest contains an invalid or duplicate table");
        }
        let mut columns = BTreeSet::new();
        if table
            .columns
            .iter()
            .any(|column| column.name.trim().is_empty() || !columns.insert(column.name.as_str()))
        {
            return Err("catalog manifest contains an invalid or duplicate column");
        }
    }
    let mut indexes = BTreeSet::new();
    for index in &delta.indexes {
        let table_key = (
            index.catalog_name.as_str(),
            index.namespace_name.as_str(),
            index.table.as_str(),
        );
        if index.name.trim().is_empty()
            || index.columns.is_empty()
            || !tables.contains(&table_key)
            || !indexes.insert((
                index.catalog_name.as_str(),
                index.namespace_name.as_str(),
                index.table.as_str(),
                index.name.as_str(),
            ))
        {
            return Err("catalog manifest contains an invalid, duplicate, or orphaned index");
        }
    }
    Ok(())
}

fn apply_evidence(
    manifest: &SchemaManifest,
    member: NodeId,
    state: SchemaApplyState,
    catalog_version: Option<u64>,
    compatibility_verified: bool,
    detail: impl Into<String>,
) -> SchemaApplyEvidence {
    let detail = detail.into();
    SchemaApplyEvidence {
        manifest_id: manifest.id.clone(),
        member,
        state,
        catalog_version,
        checksum: (state == SchemaApplyState::Applied).then(|| manifest.checksum.clone()),
        compatibility_verified,
        failure_detail: (!detail.is_empty()).then_some(detail),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Database, TxnMode};
    use alopex_sql::catalog::{ColumnMetadata, RowIdMode};
    use alopex_sql::ExecutionResult;

    #[test]
    fn storage_info_default_is_row_none() {
        let info = StorageInfo::default();
        assert_eq!(info.storage_type, "row");
        assert_eq!(info.compression, "none");
    }

    #[test]
    fn column_definition_defaults_to_nullable() {
        let column = ColumnDefinition::new("id", DataType::Integer);
        assert!(column.nullable);
        assert!(column.comment.is_none());

        let column = column.with_nullable(false).with_comment("ID");
        assert!(!column.nullable);
        assert_eq!(column.comment.as_deref(), Some("ID"));
    }

    #[test]
    fn create_catalog_request_builder_validates_name() {
        let err = CreateCatalogRequest::new("").build().unwrap_err();
        assert!(matches!(err, Error::Core(_)));

        let request = CreateCatalogRequest::new("main")
            .with_comment("メイン")
            .with_storage_root("/data")
            .build()
            .unwrap();
        assert_eq!(request.name, "main");
        assert_eq!(request.comment.as_deref(), Some("メイン"));
        assert_eq!(request.storage_root.as_deref(), Some("/data"));
    }

    #[test]
    fn create_namespace_request_builder_validates_fields() {
        let err = CreateNamespaceRequest::new("", "default")
            .build()
            .unwrap_err();
        assert!(matches!(err, Error::Core(_)));

        let request = CreateNamespaceRequest::new("main", "analytics")
            .with_comment("分析")
            .build()
            .unwrap();
        assert_eq!(request.catalog_name, "main");
        assert_eq!(request.name, "analytics");
        assert_eq!(request.comment.as_deref(), Some("分析"));
    }

    #[test]
    fn create_table_request_defaults_and_validation() {
        let schema = vec![ColumnDefinition::new("id", DataType::Integer)];

        let request = CreateTableRequest::new("users")
            .with_schema(schema.clone())
            .build()
            .unwrap();
        assert_eq!(request.catalog_name, "default");
        assert_eq!(request.namespace_name, "default");
        assert_eq!(request.table_type, TableType::Managed);
        assert_eq!(request.data_source_format, Some(DataSourceFormat::Alopex));
        assert_eq!(request.properties.as_ref().unwrap().len(), 0);

        let err = CreateTableRequest::new("users").build().unwrap_err();
        assert!(matches!(err, Error::SchemaRequired));

        let err = CreateTableRequest::new("ext")
            .with_table_type(TableType::External)
            .build()
            .unwrap_err();
        assert!(matches!(err, Error::StorageRootRequired));

        let request = CreateTableRequest::new("ext")
            .with_table_type(TableType::External)
            .with_storage_root("/external")
            .build()
            .unwrap();
        assert_eq!(request.storage_root.as_deref(), Some("/external"));
        assert_eq!(request.data_source_format, Some(DataSourceFormat::Alopex));
        assert!(request.properties.as_ref().unwrap().is_empty());
    }

    #[test]
    fn table_info_converts_from_metadata() {
        let mut table = TableMetadata::new(
            "users",
            vec![
                ColumnMetadata::new("id", ResolvedType::Integer).with_primary_key(true),
                ColumnMetadata::new("name", ResolvedType::Text),
            ],
        )
        .with_table_id(42);
        table.catalog_name = "main".to_string();
        table.namespace_name = "default".to_string();
        table.primary_key = Some(vec!["id".to_string()]);
        table.storage_options = StorageOptions {
            storage_type: StorageType::Columnar,
            compression: Compression::Zstd,
            row_group_size: 1024,
            row_id_mode: RowIdMode::Direct,
        };

        let info = TableInfo::from(table);
        assert_eq!(info.name, "users");
        assert_eq!(info.table_id, 42);
        assert_eq!(info.catalog_name, "main");
        assert_eq!(info.namespace_name, "default");
        assert_eq!(info.columns.len(), 2);
        assert_eq!(info.columns[0].data_type, "INTEGER");
        assert!(info.columns[0].is_primary_key);
        assert_eq!(info.storage_options.storage_type, "columnar");
        assert_eq!(info.storage_options.compression, "zstd");
    }

    #[test]
    fn table_info_defaults_storage_options_to_row_none() {
        let table = TableMetadata::new(
            "logs",
            vec![ColumnMetadata::new("id", ResolvedType::Integer)],
        );
        let info = TableInfo::from(table);
        assert_eq!(info.storage_options.storage_type, "row");
        assert_eq!(info.storage_options.compression, "none");
    }

    #[test]
    fn index_info_converts_from_metadata() {
        let mut index = IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
            .with_unique(true)
            .with_method(IndexMethod::Hnsw);
        index.catalog_name = "main".to_string();
        index.namespace_name = "default".to_string();

        let info = IndexInfo::from(index);
        assert_eq!(info.name, "idx_users_id");
        assert_eq!(info.table_name, "users");
        assert_eq!(info.method, "hnsw");
        assert!(info.is_unique);
    }

    fn ensure_default_catalog_and_namespace(db: &Database) {
        let _ = db.create_catalog(CreateCatalogRequest::new("default"));
        let _ = db.create_namespace(CreateNamespaceRequest::new("default", "default"));
    }

    fn manifest_from_delta(delta: Vec<u8>, version: u64) -> SchemaManifest {
        SchemaManifest {
            id: alopex_cluster::SchemaManifestId::new("manifest-1"),
            parent_id: None,
            schema_version: version,
            catalog_delta_format: CATALOG_MANIFEST_DELTA_FORMAT.to_string(),
            checksum: format!("{:x}", sha2::Sha256::digest(&delta)),
            catalog_delta: delta,
            compatibility: alopex_cluster::SchemaCompatibility {
                minimum_catalog_version: 0,
                maximum_catalog_version: version,
            },
            owner: alopex_cluster::NodeId::new("node-a"),
            created_at_epoch: 3,
        }
    }

    #[test]
    fn verified_manifest_apply_makes_sql_catalog_and_reported_version_agree() {
        let source = Database::new();
        ensure_default_catalog_and_namespace(&source);
        source
            .execute_sql("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);")
            .unwrap();
        source
            .execute_sql("CREATE INDEX idx_users_name ON users (name);")
            .unwrap();
        let manifest = manifest_from_delta(source.export_catalog_manifest_delta(7).unwrap(), 7);

        let target = Database::new();
        let evidence = target.apply_schema_manifest("node-b", &manifest);

        assert_eq!(evidence.state, SchemaApplyState::Applied);
        assert_eq!(evidence.catalog_version, Some(7));
        assert_eq!(
            evidence.checksum.as_deref(),
            Some(manifest.checksum.as_str())
        );
        assert!(evidence.compatibility_verified);
        assert_eq!(target.catalog_manifest_version().unwrap(), Some(7));
        assert!(matches!(
            target
                .execute_sql("INSERT INTO users (id, name) VALUES (1, 'alice');")
                .unwrap(),
            ExecutionResult::RowsAffected(1)
        ));
        assert_eq!(
            target.get_table_info_simple("users").unwrap().table_id,
            source.get_table_info_simple("users").unwrap().table_id
        );
        let indexes = target.list_indexes_simple("users").unwrap();
        assert!(indexes.iter().any(|index| index.name == "idx_users_name"));
    }

    #[test]
    fn corrupted_or_incompatible_catalog_never_returns_applied_evidence() {
        let source = Database::new();
        ensure_default_catalog_and_namespace(&source);
        source
            .execute_sql("CREATE TABLE users (id INTEGER PRIMARY KEY);")
            .unwrap();
        let mut corrupt = manifest_from_delta(source.export_catalog_manifest_delta(4).unwrap(), 4);
        corrupt.checksum = "wrong".to_string();
        let target = Database::new();
        let evidence = target.apply_schema_manifest("node-b", &corrupt);
        assert_eq!(evidence.state, SchemaApplyState::Failed);
        assert_eq!(target.catalog_manifest_version().unwrap(), None);
        assert!(target
            .execute_sql("INSERT INTO users (id) VALUES (1);")
            .is_err());

        let manifest = manifest_from_delta(source.export_catalog_manifest_delta(4).unwrap(), 4);
        let mismatched = Database::new();
        ensure_default_catalog_and_namespace(&mismatched);
        mismatched
            .execute_sql("CREATE TABLE users (id TEXT PRIMARY KEY);")
            .unwrap();
        let evidence = mismatched.apply_schema_manifest("node-b", &manifest);
        assert_eq!(evidence.state, SchemaApplyState::Failed);
        assert_eq!(mismatched.catalog_manifest_version().unwrap(), None);
    }

    #[test]
    fn database_catalog_and_namespace_crud() {
        let db = Database::new();

        let catalog = db
            .create_catalog(CreateCatalogRequest::new("main"))
            .unwrap();
        assert_eq!(catalog.name, "main");

        let namespace = db
            .create_namespace(CreateNamespaceRequest::new("main", "analytics"))
            .unwrap();
        assert_eq!(namespace.catalog_name, "main");
        assert_eq!(namespace.name, "analytics");

        let list = db.list_namespaces("main").unwrap();
        assert_eq!(list.len(), 1);

        let err = db.delete_catalog("main", false).unwrap_err();
        assert!(matches!(err, Error::CatalogNotEmpty(_)));

        db.delete_catalog("main", true).unwrap();

        let err = db.get_catalog("main").unwrap_err();
        assert!(matches!(err, Error::CatalogNotFound(_)));
    }

    #[test]
    fn cannot_delete_default_catalog_or_namespace() {
        let db = Database::new();
        ensure_default_catalog_and_namespace(&db);

        let err = db.delete_catalog("default", true).unwrap_err();
        assert!(matches!(err, Error::CannotDeleteDefault(_)));

        let err = db.delete_namespace("default", "default", true).unwrap_err();
        assert!(matches!(err, Error::CannotDeleteDefault(_)));

        let mut txn = db.begin(TxnMode::ReadWrite).unwrap();
        let err = txn.delete_catalog("default", true).unwrap_err();
        assert!(matches!(err, Error::CannotDeleteDefault(_)));

        let err = txn
            .delete_namespace("default", "default", true)
            .unwrap_err();
        assert!(matches!(err, Error::CannotDeleteDefault(_)));
    }

    #[test]
    fn database_table_crud_and_simple_helpers() {
        let db = Database::new();
        ensure_default_catalog_and_namespace(&db);

        let schema = vec![ColumnDefinition::new("id", DataType::Integer)];
        let info = db.create_table_simple("users", schema).unwrap();
        assert_eq!(info.catalog_name, "default");
        assert_eq!(info.namespace_name, "default");
        assert_eq!(info.table_type, TableType::Managed);
        assert_eq!(info.data_source_format, DataSourceFormat::Alopex);
        assert_eq!(info.storage_options.storage_type, "row");
        assert_eq!(info.storage_options.compression, "none");

        let tables = db.list_tables_simple().unwrap();
        assert_eq!(tables.len(), 1);

        let info = db.get_table_info_simple("users").unwrap();
        assert_eq!(info.name, "users");

        let err = db
            .create_table_simple(
                "users",
                vec![ColumnDefinition::new("id", DataType::Integer)],
            )
            .unwrap_err();
        assert!(matches!(err, Error::TableAlreadyExists(_)));

        db.delete_table_simple("users").unwrap();
        assert!(db.list_tables_simple().unwrap().is_empty());
    }

    #[test]
    fn database_index_read_helpers() {
        let db = Database::new();
        ensure_default_catalog_and_namespace(&db);

        let schema = vec![ColumnDefinition::new("id", DataType::Integer)];
        db.create_table_simple("users", schema).unwrap();

        let result = db
            .execute_sql("CREATE INDEX idx_users_id ON users (id);")
            .unwrap();
        assert!(matches!(result, ExecutionResult::Success));

        let indexes = db.list_indexes_simple("users").unwrap();
        assert_eq!(indexes.len(), 1);
        assert_eq!(indexes[0].name, "idx_users_id");
        assert_eq!(indexes[0].method, "btree");

        let index = db.get_index_info_simple("users", "idx_users_id").unwrap();
        assert_eq!(index.table_name, "users");
    }

    #[test]
    fn transaction_overlay_visibility_and_commit() {
        let db = Database::new();
        let mut txn = db.begin(TxnMode::ReadWrite).unwrap();

        txn.create_catalog(CreateCatalogRequest::new("main"))
            .unwrap();
        txn.create_namespace(CreateNamespaceRequest::new("main", "default"))
            .unwrap();

        let schema = vec![ColumnDefinition::new("id", DataType::Integer)];
        txn.create_table(
            CreateTableRequest::new("events")
                .with_catalog_name("main")
                .with_namespace_name("default")
                .with_schema(schema),
        )
        .unwrap();

        let tables = txn.list_tables("main", "default").unwrap();
        assert_eq!(tables.len(), 1);

        txn.commit().unwrap();

        let info = db.get_table_info("main", "default", "events").unwrap();
        assert_eq!(info.name, "events");
    }

    #[test]
    fn transaction_commit_persists_overlay_to_store() {
        let db = Database::new();
        let mut txn = db.begin(TxnMode::ReadWrite).unwrap();

        txn.create_catalog(CreateCatalogRequest::new("main"))
            .unwrap();
        txn.create_namespace(CreateNamespaceRequest::new("main", "default"))
            .unwrap();

        let schema = vec![ColumnDefinition::new("id", DataType::Integer)];
        txn.create_table(
            CreateTableRequest::new("events")
                .with_catalog_name("main")
                .with_namespace_name("default")
                .with_schema(schema),
        )
        .unwrap();

        txn.commit().unwrap();

        let reloaded = alopex_sql::catalog::PersistentCatalog::load(db.store.clone()).unwrap();
        assert!(reloaded.get_catalog("main").is_some());
        assert!(reloaded.get_namespace("main", "default").is_some());
        assert!(reloaded.table_exists("events"));
    }

    #[test]
    fn transaction_rollback_discards_overlay() {
        let db = Database::new();
        let mut txn = db.begin(TxnMode::ReadWrite).unwrap();

        txn.create_catalog(CreateCatalogRequest::new("main"))
            .unwrap();
        txn.create_namespace(CreateNamespaceRequest::new("main", "default"))
            .unwrap();

        let schema = vec![ColumnDefinition::new("id", DataType::Integer)];
        txn.create_table(
            CreateTableRequest::new("staging")
                .with_catalog_name("main")
                .with_namespace_name("default")
                .with_schema(schema),
        )
        .unwrap();

        txn.rollback().unwrap();

        let err = db.get_table_info("main", "default", "staging").unwrap_err();
        assert!(matches!(err, Error::CatalogNotFound(_)));
    }

    #[test]
    fn transaction_readonly_rejects_ddl() {
        let db = Database::new();
        let mut txn = db.begin(TxnMode::ReadOnly).unwrap();
        let err = txn
            .create_catalog(CreateCatalogRequest::new("main"))
            .unwrap_err();
        assert!(matches!(err, Error::TxnReadOnly));
    }
}