cqlite-core 0.15.0

//! Schema Registry for Centralized Schema Management
//!
//! This module provides a centralized registry for managing table schemas, UDTs,
//! and other schema-related information with support for schema discovery,
//! validation, caching, and version management.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::SystemTime;

use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;

use crate::{
    platform::Platform,
    schema::{
        discovery::{SchemaDiscoveryConfig, SchemaDiscoveryEngine, SchemaInfo},
        CqlType, TableSchema, UdtRegistry,
    },
    types::{ComparatorType, UdtTypeDef},
    Config, Error, Result,
};

/// Configuration for schema registry
#[derive(Debug, Clone)]
pub struct SchemaRegistryConfig {
    /// Enable automatic schema discovery
    pub enable_auto_discovery: bool,
    /// Enable schema caching
    pub enable_caching: bool,
    /// Cache TTL in seconds
    pub cache_ttl_seconds: u64,
    /// Enable schema versioning
    pub enable_versioning: bool,
    /// Maximum versions to keep per schema
    pub max_versions_per_schema: usize,
    /// Enable schema validation
    pub enable_validation: bool,
    /// Auto-refresh schemas on SSTable changes
    pub auto_refresh_on_changes: bool,
    /// Discovery configuration
    pub discovery_config: SchemaDiscoveryConfig,
}

impl Default for SchemaRegistryConfig {
    fn default() -> Self {
        Self {
            enable_auto_discovery: true,
            enable_caching: true,
            cache_ttl_seconds: 3600, // 1 hour
            enable_versioning: true,
            max_versions_per_schema: 5,
            enable_validation: true,
            auto_refresh_on_changes: false, // Disabled by default for performance
            discovery_config: SchemaDiscoveryConfig::default(),
        }
    }
}

/// Centralized schema registry
#[derive(Debug)]
pub struct SchemaRegistry {
    /// Configuration
    config: SchemaRegistryConfig,
    /// Platform abstraction
    _platform: Arc<Platform>,
    /// Core configuration
    _core_config: Config,
    /// Registered table schemas by keyspace.table
    schemas: Arc<RwLock<HashMap<String, SchemaEntry>>>,
    /// UDT registry for managing user-defined types
    udt_registry: Arc<RwLock<UdtRegistry>>,
    /// Schema discovery engine
    discovery_engine: Arc<SchemaDiscoveryEngine>,
    /// Schema validator
    validator: Arc<SchemaValidator>,
    /// Schema version history
    version_history: Arc<RwLock<HashMap<String, Vec<SchemaVersion>>>>,
    /// Per-table update-serialization locks (issue #1710).
    ///
    /// Maps `table_id` -> an async mutex whose guard is held across the whole
    /// `register_schema` sequence {decide `existed` -> upsert schema -> append
    /// version-history} for that table. This serializes concurrent updates of
    /// the SAME table so the `schemas` map and `version_history` can never
    /// diverge (e.g. registry ends with task A's schema while history records
    /// task B's as newest), WITHOUT holding the global `schemas` write lock
    /// across the `create_schema_version(...).await`.
    ///
    /// Lock ordering (deadlock-free): the per-table lock is ALWAYS acquired
    /// BEFORE the `schemas`/`version_history` locks; those inner locks are only
    /// ever held briefly and never across an `.await`. The `std::sync::Mutex`
    /// guarding this map itself is held only long enough to clone the per-table
    /// `Arc` — never across an `.await`.
    update_locks: std::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
}

/// Derived parsing state precomputed once at schema registration (issue #1709).
///
/// [`SchemaRegistry::get_parsing_context`] is on the read path (called per
/// Index.db lookup via `key_digest`). Re-deriving the comparators — which are
/// CONSTANT for a registered schema — on every call cost four `TableSchema`
/// deep clones and one `CqlType::parse` per column per call. We instead compute
/// these once when the [`SchemaEntry`] is created and hand them out as `Arc`
/// clones (a pointer bump). The state lives ON the entry so it is invalidated
/// atomically whenever the entry is replaced.
#[derive(Debug, Clone)]
struct DerivedParsingState {
    /// The complete table schema, shared.
    schema: Arc<TableSchema>,
    /// Comparators for partition key components, in key order.
    partition_comparators: Arc<Vec<ComparatorType>>,
    /// Comparators for clustering key components, in key order.
    clustering_comparators: Arc<Vec<ComparatorType>>,
    /// Comparators for all columns, keyed by column name.
    column_comparators: Arc<HashMap<String, ComparatorType>>,
}

impl DerivedParsingState {
    /// Compute the derived comparators for a schema. Runs `CqlType::parse` per
    /// key/column exactly once, at registration time.
    fn compute(schema: &TableSchema) -> Result<Self> {
        let mut partition_comparators = Vec::new();
        for key_column in schema.ordered_partition_keys() {
            let cql_type = CqlType::parse(&key_column.data_type)?;
            partition_comparators.push(ComparatorType::from_cql_type(&cql_type)?);
        }

        let mut clustering_comparators = Vec::new();
        for key_column in schema.ordered_clustering_keys() {
            let cql_type = CqlType::parse(&key_column.data_type)?;
            clustering_comparators.push(ComparatorType::from_cql_type(&cql_type)?);
        }

        let mut column_comparators = HashMap::new();
        for column in &schema.columns {
            let cql_type = CqlType::parse(&column.data_type)?;
            column_comparators.insert(
                column.name.clone(),
                ComparatorType::from_cql_type(&cql_type)?,
            );
        }

        Ok(Self {
            schema: Arc::new(schema.clone()),
            partition_comparators: Arc::new(partition_comparators),
            clustering_comparators: Arc::new(clustering_comparators),
            column_comparators: Arc::new(column_comparators),
        })
    }

    /// Build a [`ParsingContext`] by cloning the shared `Arc`s (pointer bumps,
    /// no deep clone, no parse).
    fn to_parsing_context(&self) -> ParsingContext {
        ParsingContext {
            schema: self.schema.clone(),
            partition_comparators: self.partition_comparators.clone(),
            clustering_comparators: self.clustering_comparators.clone(),
            column_comparators: self.column_comparators.clone(),
        }
    }
}

/// Schema entry in the registry
#[derive(Debug, Clone)]
struct SchemaEntry {
    /// The table schema
    schema: TableSchema,
    /// Precomputed derived parsing state (comparators), or `None` if it could
    /// not be derived at registration (e.g. a malformed type string). When
    /// `None`, `get_parsing_context` recomputes on demand, preserving the exact
    /// original error behavior without deep-cloning the schema.
    derived: Option<DerivedParsingState>,
    /// Extended schema information if available
    extended_info: Option<SchemaInfo>,
    /// When the schema was registered/updated
    registered_at: SystemTime,
    /// Source of the schema
    source: SchemaSource,
    /// Validation status
    validation_status: SchemaValidationStatus,
    /// Associated SSTable files
    _associated_files: Vec<PathBuf>,
}

/// Source of schema information
#[derive(Debug, Clone)]
pub enum SchemaSource {
    /// Discovered from SSTable files
    Discovered(Vec<PathBuf>),
    /// Loaded from external definition
    External(PathBuf),
    /// Parsed from CQL DDL
    Cql(String),
    /// Manually registered
    Manual,
}

/// Schema validation status
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SchemaValidationStatus {
    /// Schema is valid
    Valid,
    /// Schema has warnings but is usable
    ValidWithWarnings,
    /// Schema is invalid
    Invalid,
    /// Not yet validated
    NotValidated,
}

/// Schema version information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaVersion {
    /// Version number
    pub version: u32,
    /// When this version was created
    pub created_at: SystemTime,
    /// Schema at this version
    pub schema: TableSchema,
    /// Changes from previous version
    pub changes: Vec<SchemaChange>,
    /// Source of this version
    pub source: String,
}

/// Schema change description
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaChange {
    /// Type of change
    pub change_type: SchemaChangeType,
    /// Component affected
    pub component: String,
    /// Description of the change
    pub description: String,
    /// Old value (if applicable)
    pub old_value: Option<String>,
    /// New value (if applicable)
    pub new_value: Option<String>,
}

/// Types of schema changes
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum SchemaChangeType {
    /// Column added
    ColumnAdded,
    /// Column removed
    ColumnRemoved,
    /// Column type changed
    ColumnTypeChanged,
    /// Column renamed
    ColumnRenamed,
    /// Index added
    IndexAdded,
    /// Index removed
    IndexRemoved,
    /// UDT added
    UdtAdded,
    /// UDT modified
    UdtModified,
    /// UDT removed
    UdtRemoved,
    /// Table option changed
    TableOptionChanged,
}

/// Schema validation report
#[derive(Debug, Clone)]
pub struct ValidationReport {
    /// Table identifier
    pub table_id: String,
    /// Overall validation status
    pub status: SchemaValidationStatus,
    /// Validation errors
    pub errors: Vec<ValidationError>,
    /// Validation warnings
    pub warnings: Vec<ValidationWarning>,
    /// Recommendations
    pub recommendations: Vec<String>,
    /// Validation timestamp
    pub validated_at: SystemTime,
}

/// Validation error details
#[derive(Debug, Clone)]
pub struct ValidationError {
    /// Error code
    pub code: String,
    /// Error message
    pub message: String,
    /// Affected component
    pub component: Option<String>,
    /// Severity level
    pub severity: ErrorSeverity,
}

/// Error severity levels
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ErrorSeverity {
    Critical,
    High,
    Medium,
    Low,
}

/// Validation warning details
#[derive(Debug, Clone)]
pub struct ValidationWarning {
    /// Warning code
    pub code: String,
    /// Warning message
    pub message: String,
    /// Affected component
    pub component: Option<String>,
}

/// Schema search query
#[derive(Debug, Clone)]
pub struct SchemaQuery {
    /// Keyspace filter (optional)
    pub keyspace: Option<String>,
    /// Table name pattern (supports wildcards)
    pub table_pattern: Option<String>,
    /// Include schemas with specific source types
    pub source_types: Option<Vec<SchemaSource>>,
    /// Include only validated schemas
    pub validated_only: bool,
    /// Include version history
    pub include_history: bool,
}

impl SchemaRegistry {
    /// Create a new schema registry
    pub async fn new(
        config: SchemaRegistryConfig,
        platform: Arc<Platform>,
        core_config: Config,
    ) -> Result<Self> {
        let discovery_engine = Arc::new(
            SchemaDiscoveryEngine::new(
                config.discovery_config.clone(),
                platform.clone(),
                core_config.clone(),
            )
            .await?,
        );

        let validator = Arc::new(SchemaValidator::new());
        let udt_registry = Arc::new(RwLock::new(UdtRegistry::new()));

        Ok(Self {
            config,
            _platform: platform,
            _core_config: core_config,
            schemas: Arc::new(RwLock::new(HashMap::new())),
            udt_registry,
            discovery_engine,
            validator,
            version_history: Arc::new(RwLock::new(HashMap::new())),
            update_locks: std::sync::Mutex::new(HashMap::new()),
        })
    }

    /// Fetch (creating if absent) the per-table update-serialization lock.
    ///
    /// The map's `std::sync::Mutex` is held only long enough to clone the
    /// per-table `Arc<tokio::sync::Mutex<()>>` and is NEVER held across an
    /// `.await`. On poisoning we recover the inner map rather than panic
    /// (no `unwrap()`/`expect()` in library code).
    fn table_update_lock(&self, table_id: &str) -> Arc<tokio::sync::Mutex<()>> {
        let mut locks = self
            .update_locks
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        locks
            .entry(table_id.to_string())
            .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
            .clone()
    }

    /// Discover and register schema from SSTable files
    pub async fn discover_schema(
        &self,
        keyspace: &str,
        table: &str,
        sstable_files: &[PathBuf],
    ) -> Result<TableSchema> {
        if !self.config.enable_auto_discovery {
            return Err(Error::Schema("Auto-discovery is disabled".to_string()));
        }

        // Use discovery engine to analyze SSTable files
        let schema_info = self
            .discovery_engine
            .discover_schema(keyspace, table, sstable_files)
            .await?;

        // Convert to TableSchema format for compatibility
        let table_schema = self.convert_schema_info_to_table_schema(&schema_info)?;

        // Register the discovered schema
        self.register_discovered_schema(
            table_schema.clone(),
            Some(schema_info),
            sstable_files.to_vec(),
        )
        .await?;

        Ok(table_schema)
    }

    /// Register a schema from external source
    pub async fn register_schema(&self, schema: TableSchema, source: SchemaSource) -> Result<()> {
        let table_id = format!("{}.{}", schema.keyspace, schema.table);

        // Serialize the WHOLE update for this table (issue #1710).
        //
        // The upsert of `schemas` and the append to `version_history` cannot be
        // performed under one held lock: `create_schema_version` awaits the
        // independent `version_history` lock, and holding the `schemas` write
        // guard across that `.await` was the lock-order hazard fixed earlier.
        // But if the upsert and the history-append are not serialized *together*
        // per table, two concurrent updates of the SAME table with DIFFERENT
        // schemas can interleave so the registry ends with one task's schema
        // while history records the other's as newest (registry/history
        // divergence).
        //
        // Fix: hold a per-table async mutex across the entire
        // {decide existed -> upsert -> append version} sequence below. It is
        // acquired FIRST — before any `schemas`/`version_history` lock — and
        // those inner locks are still taken only briefly and never across an
        // `.await`, so there is no lock-order cycle (no deadlock). Two
        // concurrent `register_schema` calls for the same table thus run
        // strictly one-after-another and stay consistent; calls for different
        // tables still proceed in parallel.
        let table_lock = self.table_update_lock(&table_id);
        let _update_guard = table_lock.lock().await;

        // Validate schema if validation is enabled
        let validation_status = if self.config.enable_validation {
            match self.validator.validate_table_schema(&schema).await {
                Ok(_) => SchemaValidationStatus::Valid,
                Err(_) => SchemaValidationStatus::Invalid,
            }
        } else {
            SchemaValidationStatus::NotValidated
        };

        // Create schema entry, precomputing the derived comparators once so the
        // read-path `get_parsing_context` is a pointer bump (issue #1709). A
        // malformed type leaves `derived = None`; the error then surfaces on the
        // context path exactly as before (never fails registration).
        let derived = DerivedParsingState::compute(&schema).ok();
        let entry = SchemaEntry {
            schema: schema.clone(),
            derived,
            extended_info: None,
            registered_at: SystemTime::now(),
            source,
            validation_status,
            _associated_files: Vec::new(),
        };

        // Store in registry.
        //
        // The existence check and the insert MUST be a single write-locked
        // critical section: two concurrent FIRST registrations of the same
        // table must not both observe "absent". A check-then-act split (read
        // lock to check, later write lock to insert) is a TOCTOU race — both
        // callers would skip `create_schema_version`, then one insert would
        // overwrite the other, silently dropping the second registration's
        // update-ness (a lost version). Here we capture `existed` and insert
        // atomically under one write guard.
        //
        // The guard is DROPPED before any `.await`: `create_schema_version`
        // acquires the independent `version_history` lock and must never run
        // while the `schemas` guard is held — that combination was the latent
        // lock-order hazard on the cold registration path (issue #1710).
        let existed = {
            let mut schemas = self.schemas.write().await;
            let existed = schemas.contains_key(&table_id);
            schemas.insert(table_id.clone(), entry);
            existed
        }; // `schemas` write guard released here — no `.await` occurred under it.

        if self.config.enable_versioning && existed {
            self.create_schema_version(&table_id, &schema).await?;
        }

        Ok(())
    }

    /// Get schema by keyspace and table name
    pub async fn get_schema(&self, keyspace: &str, table: &str) -> Result<TableSchema> {
        let table_id = format!("{}.{}", keyspace, table);
        let schemas = self.schemas.read().await;

        match schemas.get(&table_id) {
            Some(entry) => {
                // Check if schema is still valid (cache TTL)
                if self.is_entry_expired(entry) {
                    drop(schemas); // Release read lock
                    return self.refresh_schema(keyspace, table).await;
                }
                #[cfg(test)]
                crate::schema::work_counters::record_schema_clone();
                Ok(entry.schema.clone())
            }
            None => {
                drop(schemas); // Release read lock
                               // Try to discover schema if auto-discovery is enabled
                if self.config.enable_auto_discovery {
                    self.auto_discover_schema(keyspace, table).await
                } else {
                    Err(Error::Schema(format!(
                        "Schema not found: {}.{}",
                        keyspace, table
                    )))
                }
            }
        }
    }

    /// Get extended schema information
    pub async fn get_schema_info(&self, keyspace: &str, table: &str) -> Result<Option<SchemaInfo>> {
        let table_id = format!("{}.{}", keyspace, table);
        let schemas = self.schemas.read().await;

        match schemas.get(&table_id) {
            Some(entry) => Ok(entry.extended_info.clone()),
            None => Ok(None),
        }
    }

    /// List all registered schemas
    pub async fn list_schemas(&self, query: Option<SchemaQuery>) -> Result<Vec<TableSchema>> {
        let schemas = self.schemas.read().await;
        let mut results = Vec::new();

        for (_table_id, entry) in schemas.iter() {
            // Apply query filters if provided
            if let Some(ref q) = query {
                if !self.matches_query(&entry.schema, q) {
                    continue;
                }
            }

            results.push(entry.schema.clone());
        }

        // Sort by keyspace, then table name
        results.sort_by(|a, b| {
            a.keyspace
                .cmp(&b.keyspace)
                .then_with(|| a.table.cmp(&b.table))
        });

        Ok(results)
    }

    /// Find a schema by table name, optionally scoped to a keyspace, cloning
    /// only the matched schema (issue #1708).
    ///
    /// This is the single freshness-owning lookup that [`crate::schema::SchemaManager`]
    /// delegates to, so the registry — not the manager — owns freshness:
    ///
    /// * A **fresh** matched entry is returned by value with exactly ONE
    ///   `TableSchema` deep clone (issue #1587).
    /// * An **expired** matched entry is NOT served stale: it is routed through
    ///   the SAME TTL-refresh seam [`Self::get_schema`] uses for the matched
    ///   entry ([`Self::refresh_schema`]), and the refreshed schema (or that
    ///   seam's error) is returned. This closes the roborev Medium where an
    ///   entry expired in the registry was served stale indefinitely through the
    ///   manager.
    /// * **No match** returns `Ok(None)` with NO auto-discovery and NO
    ///   fabrication (issue #1710): a table nobody registered must surface as
    ///   not-found (the manager maps `None` to its own not-found error), never a
    ///   synthesized schema.
    pub async fn find_schema_by_table(
        &self,
        keyspace: &Option<String>,
        table: &str,
    ) -> Result<Option<TableSchema>> {
        // Resolve the matched entry under a read guard. Clone the schema ONLY
        // when it is fresh (the single #1587 clone); when it is EXPIRED, capture
        // just its identity so the guard can be dropped BEFORE running the
        // refresh seam (no `.await` is ever held under the read guard).
        enum Matched {
            Fresh(TableSchema),
            Expired { keyspace: String, table: String },
        }

        let matched = {
            let schemas = self.schemas.read().await;

            // Exact `keyspace.table` match first when a keyspace is provided.
            let mut hit = keyspace
                .as_ref()
                .and_then(|ks| schemas.get(&format!("{}.{}", ks, table)));

            // Otherwise match any registered schema by table name (keyspace-aware).
            if hit.is_none() {
                hit = schemas.values().find(|entry| {
                    crate::schema::table_name_matches(
                        &Some(entry.schema.keyspace.clone()),
                        &entry.schema.table,
                        keyspace,
                        table,
                    )
                });
            }

            match hit {
                None => None,
                Some(entry) if self.is_entry_expired(entry) => Some(Matched::Expired {
                    keyspace: entry.schema.keyspace.clone(),
                    table: entry.schema.table.clone(),
                }),
                Some(entry) => Some(Matched::Fresh(entry.schema.clone())),
            }
        }; // read guard dropped here — no `.await` occurred under it.

        match matched {
            // No match: no fabrication, no auto-discovery (issue #1710).
            None => Ok(None),
            Some(Matched::Fresh(schema)) => Ok(Some(schema)),
            // Expired: delegate freshness through the same refresh seam
            // `get_schema` uses for the matched entry — never serve the stale
            // copy.
            Some(Matched::Expired { keyspace, table }) => {
                self.refresh_schema(&keyspace, &table).await.map(Some)
            }
        }
    }

    /// Validate a schema
    #[allow(dead_code)]
    pub async fn validate_schema(&self, keyspace: &str, table: &str) -> Result<ValidationReport> {
        let schema = self.get_schema(keyspace, table).await?;
        let table_id = format!("{}.{}", keyspace, table);

        // Perform comprehensive validation
        let mut errors = Vec::new();
        let mut warnings = Vec::new();
        let mut recommendations = Vec::new();

        // Basic schema structure validation
        if let Err(e) = schema.validate() {
            errors.push(ValidationError {
                code: "SCHEMA_INVALID".to_string(),
                message: e.to_string(),
                component: None,
                severity: ErrorSeverity::Critical,
            });
        }

        // UDT validation
        self.validate_schema_udts(&schema, &mut errors, &mut warnings)
            .await;

        // Column type validation
        self.validate_column_types(&schema, &mut errors, &mut warnings)
            .await;

        // Performance recommendations
        self.generate_performance_recommendations(&schema, &mut recommendations)
            .await;

        // Determine overall status
        let status = if !errors.is_empty() {
            SchemaValidationStatus::Invalid
        } else if !warnings.is_empty() {
            SchemaValidationStatus::ValidWithWarnings
        } else {
            SchemaValidationStatus::Valid
        };

        // Update validation status in registry
        {
            let mut schemas = self.schemas.write().await;
            if let Some(entry) = schemas.get_mut(&table_id) {
                entry.validation_status = status.clone();
            }
        }

        Ok(ValidationReport {
            table_id,
            status,
            errors,
            warnings,
            recommendations,
            validated_at: SystemTime::now(),
        })
    }

    /// Get schema version history
    pub async fn get_schema_history(
        &self,
        keyspace: &str,
        table: &str,
    ) -> Result<Vec<SchemaVersion>> {
        if !self.config.enable_versioning {
            return Err(Error::Schema("Schema versioning is disabled".to_string()));
        }

        let table_id = format!("{}.{}", keyspace, table);
        let history = self.version_history.read().await;

        Ok(history.get(&table_id).cloned().unwrap_or_default())
    }

    /// Remove schema from registry
    pub async fn remove_schema(&self, keyspace: &str, table: &str) -> Result<()> {
        let table_id = format!("{}.{}", keyspace, table);

        {
            let mut schemas = self.schemas.write().await;
            schemas.remove(&table_id);
        }

        // Also remove from version history if versioning is enabled
        if self.config.enable_versioning {
            let mut history = self.version_history.write().await;
            history.remove(&table_id);
        }

        Ok(())
    }

    /// Generate CQL CREATE statement for schema
    pub async fn generate_cql(&self, keyspace: &str, table: &str) -> Result<String> {
        // First try to get extended schema info for better CQL generation
        if let Some(schema_info) = self.get_schema_info(keyspace, table).await? {
            return self.discovery_engine.generate_cql(&schema_info).await;
        }

        // Fallback to basic TableSchema CQL generation
        let schema = self.get_schema(keyspace, table).await?;
        Ok(self.generate_basic_cql(&schema))
    }

    /// Export schema as JSON
    #[cfg(feature = "experimental")]
    pub async fn export_schema_json(&self, keyspace: &str, table: &str) -> Result<String> {
        self.export_schema_json_with_config(
            keyspace,
            table,
            &crate::schema::json_exporter::JsonExportConfig::default(),
        )
        .await
    }

    #[cfg(not(feature = "experimental"))]
    pub async fn export_schema_json(&self, _keyspace: &str, _table: &str) -> Result<String> {
        Err(crate::error::Error::unsupported_format(
            "JSON export requires experimental feature",
        ))
    }

    /// Export schema as JSON with custom configuration
    #[cfg(feature = "experimental")]
    pub async fn export_schema_json_with_config(
        &self,
        keyspace: &str,
        table: &str,
        config: &crate::schema::json_exporter::JsonExportConfig,
    ) -> Result<String> {
        // Try extended schema info first
        if let Some(schema_info) = self.get_schema_info(keyspace, table).await? {
            return self
                .discovery_engine
                .export_json_with_config(&schema_info, config)
                .await;
        }

        // Fallback to basic TableSchema JSON
        let schema = self.get_schema(keyspace, table).await?;
        let exporter = crate::schema::json_exporter::JsonExporter::with_config(config.clone());
        exporter.export_table_schema(&schema)
    }

    #[cfg(not(feature = "experimental"))]
    pub async fn export_schema_json_with_config<T>(
        &self,
        _keyspace: &str,
        _table: &str,
        _config: &T,
    ) -> Result<String> {
        Err(crate::error::Error::unsupported_format(
            "JSON export requires experimental feature",
        ))
    }

    /// Export schema as compact JSON (minimal format)
    #[cfg(feature = "experimental")]
    pub async fn export_schema_json_compact(&self, keyspace: &str, table: &str) -> Result<String> {
        let config = crate::schema::json_exporter::JsonExportConfig {
            format_variant: crate::schema::json_exporter::JsonFormat::Compact,
            include_metadata: false,
            include_performance_metrics: false,
            include_type_details: false,
            pretty_format: false,
            ..Default::default()
        };
        self.export_schema_json_with_config(keyspace, table, &config)
            .await
    }

    #[cfg(not(feature = "experimental"))]
    pub async fn export_schema_json_compact(
        &self,
        _keyspace: &str,
        _table: &str,
    ) -> Result<String> {
        Err(crate::error::Error::unsupported_format(
            "JSON export requires experimental feature",
        ))
    }

    /// Export schema for API documentation (OpenAPI-compatible format)
    #[cfg(feature = "experimental")]
    pub async fn export_schema_json_openapi(&self, keyspace: &str, table: &str) -> Result<String> {
        let config = crate::schema::json_exporter::JsonExportConfig {
            format_variant: crate::schema::json_exporter::JsonFormat::OpenApi,
            include_documentation: true,
            include_type_details: true,
            include_metadata: false,
            ..Default::default()
        };
        self.export_schema_json_with_config(keyspace, table, &config)
            .await
    }

    #[cfg(not(feature = "experimental"))]
    pub async fn export_schema_json_openapi(
        &self,
        _keyspace: &str,
        _table: &str,
    ) -> Result<String> {
        Err(crate::error::Error::unsupported_format(
            "JSON export requires experimental feature",
        ))
    }

    /// Export schema for data pipeline tools
    #[cfg(feature = "experimental")]
    pub async fn export_schema_json_pipeline(&self, keyspace: &str, table: &str) -> Result<String> {
        let config = crate::schema::json_exporter::JsonExportConfig {
            format_variant: crate::schema::json_exporter::JsonFormat::DataPipeline,
            include_type_details: true,
            include_table_options: false,
            include_performance_metrics: true,
            ..Default::default()
        };
        self.export_schema_json_with_config(keyspace, table, &config)
            .await
    }

    #[cfg(not(feature = "experimental"))]
    pub async fn export_schema_json_pipeline(
        &self,
        _keyspace: &str,
        _table: &str,
    ) -> Result<String> {
        Err(crate::error::Error::unsupported_format(
            "JSON export requires experimental feature",
        ))
    }

    /// Export multiple schemas as a JSON collection
    #[cfg(feature = "experimental")]
    pub async fn export_multiple_schemas_json(
        &self,
        schema_infos: &[SchemaInfo],
    ) -> Result<String> {
        let exporter = crate::schema::json_exporter::JsonExporter::new();
        exporter.export_multiple_schemas(schema_infos)
    }

    #[cfg(not(feature = "experimental"))]
    pub async fn export_multiple_schemas_json(
        &self,
        _schema_infos: &[SchemaInfo],
    ) -> Result<String> {
        Err(crate::error::Error::unsupported_format(
            "JSON export requires experimental feature",
        ))
    }

    /// Export all schemas in a keyspace as JSON collection
    #[cfg(feature = "experimental")]
    pub async fn export_keyspace_schemas_json(&self, keyspace: &str) -> Result<String> {
        let mut schema_infos = Vec::new();

        // Get all schemas in the keyspace
        for (_table_id, entry) in self.schemas.read().await.iter() {
            if entry.schema.keyspace == keyspace {
                // Try to get extended schema info
                if let Ok(Some(schema_info)) = self
                    .get_schema_info(&entry.schema.keyspace, &entry.schema.table)
                    .await
                {
                    schema_infos.push(schema_info);
                }
            }
        }

        if schema_infos.is_empty() {
            return Err(Error::NotFound(format!(
                "No schemas found in keyspace '{}'",
                keyspace
            )));
        }

        self.export_multiple_schemas_json(&schema_infos).await
    }

    #[cfg(not(feature = "experimental"))]
    pub async fn export_keyspace_schemas_json(&self, _keyspace: &str) -> Result<String> {
        Err(crate::error::Error::unsupported_format(
            "JSON export requires experimental feature",
        ))
    }

    /// Register UDT in the registry
    pub async fn register_udt(&self, udt_def: UdtTypeDef) -> Result<()> {
        let mut registry = self.udt_registry.write().await;
        registry.register_udt(udt_def);
        Ok(())
    }

    /// Get UDT definition
    pub async fn get_udt(&self, keyspace: &str, name: &str) -> Result<Option<UdtTypeDef>> {
        let registry = self.udt_registry.read().await;
        Ok(registry.get_udt(keyspace, name).cloned())
    }

    /// Get the internal UDT registry (crate-only access for schema manager)
    ///
    /// This method is used by SchemaManager when initialized with a pre-loaded registry
    /// to preserve the UDT definitions loaded during ingestion.
    ///
    /// Gated on `state_machine`: both callers (`storage::sstable::refresh` and the
    /// `cli-helpers`-gated ingestion module) compile out without it, so it would be
    /// dead code under `-D warnings` in the minimal-features build (issue #1972).
    #[cfg(feature = "state_machine")]
    pub(crate) fn get_udt_registry(&self) -> Arc<RwLock<UdtRegistry>> {
        self.udt_registry.clone()
    }

    /// Get ComparatorType for a specific column in a table
    pub async fn get_column_comparator(
        &self,
        keyspace: &str,
        table: &str,
        column: &str,
    ) -> Result<ComparatorType> {
        let schema = self.get_schema(keyspace, table).await?;

        // Find the column
        let column_def = schema
            .columns
            .iter()
            .find(|c| c.name == column)
            .ok_or_else(|| {
                Error::Schema(format!(
                    "Column '{}' not found in table '{}.{}'",
                    column, keyspace, table
                ))
            })?;

        // Parse the column type and create comparator
        let cql_type = CqlType::parse(&column_def.data_type)?;
        ComparatorType::from_cql_type(&cql_type)
    }

    /// Get ComparatorType for all columns in a table
    pub async fn get_table_comparators(
        &self,
        keyspace: &str,
        table: &str,
    ) -> Result<HashMap<String, ComparatorType>> {
        let schema = self.get_schema(keyspace, table).await?;
        let mut comparators = HashMap::new();

        for column in &schema.columns {
            let cql_type = CqlType::parse(&column.data_type)?;
            let comparator = ComparatorType::from_cql_type(&cql_type)?;
            comparators.insert(column.name.clone(), comparator);
        }

        Ok(comparators)
    }

    /// Get ComparatorType for partition key columns (for key comparison)
    pub async fn get_partition_key_comparator(
        &self,
        keyspace: &str,
        table: &str,
    ) -> Result<Vec<ComparatorType>> {
        let schema = self.get_schema(keyspace, table).await?;
        let mut comparators = Vec::new();

        // Get partition keys in order
        let ordered_keys = schema.ordered_partition_keys();
        for key_column in ordered_keys {
            let cql_type = CqlType::parse(&key_column.data_type)?;
            let comparator = ComparatorType::from_cql_type(&cql_type)?;
            comparators.push(comparator);
        }

        Ok(comparators)
    }

    /// Get the complete schema context for parsing operations.
    ///
    /// This is on the read path (per Index.db lookup). For a registered, fresh
    /// entry it does ZERO `TableSchema` deep clones and ZERO `CqlType::parse`
    /// calls: it takes ONE read guard and hands out `Arc` clones of the derived
    /// state precomputed at registration (issue #1709).
    pub async fn get_parsing_context(&self, keyspace: &str, table: &str) -> Result<ParsingContext> {
        let table_id = format!("{}.{}", keyspace, table);

        // Fast path: present, fresh, precomputed. One guard, no await under it,
        // no deep clone, no parse.
        {
            let schemas = self.schemas.read().await;
            if let Some(entry) = schemas.get(&table_id) {
                if !self.is_entry_expired(entry) {
                    match &entry.derived {
                        Some(derived) => return Ok(derived.to_parsing_context()),
                        // Precompute failed (malformed type): reproduce the
                        // original error via on-demand recompute. No await held.
                        None => {
                            return DerivedParsingState::compute(&entry.schema)
                                .map(|d| d.to_parsing_context())
                        }
                    }
                }
            }
        }

        // Cold path: missing or TTL-expired. Trigger the refresh/auto-discovery
        // seam (which reinserts the entry with derived state), then read it
        // back. This is not the post-registration request path.
        self.get_schema(keyspace, table).await?;

        let schemas = self.schemas.read().await;
        match schemas.get(&table_id) {
            Some(entry) => match &entry.derived {
                Some(derived) => Ok(derived.to_parsing_context()),
                None => DerivedParsingState::compute(&entry.schema).map(|d| d.to_parsing_context()),
            },
            None => Err(Error::Schema(format!(
                "Schema not found: {}.{}",
                keyspace, table
            ))),
        }
    }

    /// Get ComparatorType for clustering key columns (for clustering comparison)
    pub async fn get_clustering_key_comparator(
        &self,
        keyspace: &str,
        table: &str,
    ) -> Result<Vec<ComparatorType>> {
        let schema = self.get_schema(keyspace, table).await?;
        let mut comparators = Vec::new();

        // Get clustering keys in order
        let ordered_keys = schema.ordered_clustering_keys();
        for key_column in ordered_keys {
            let cql_type = CqlType::parse(&key_column.data_type)?;
            let comparator = ComparatorType::from_cql_type(&cql_type)?;
            comparators.push(comparator);
        }

        Ok(comparators)
    }

    /// Validate column type compatibility using ComparatorType
    pub async fn validate_column_type_compatibility(
        &self,
        keyspace: &str,
        table: &str,
        column: &str,
        expected_type: &str,
    ) -> Result<bool> {
        let column_comparator = self.get_column_comparator(keyspace, table, column).await?;
        let expected_cql_type = CqlType::parse(expected_type)?;
        let expected_comparator = ComparatorType::from_cql_type(&expected_cql_type)?;

        // Check if comparators are compatible (same type structure)
        Ok(self.comparators_are_compatible(&column_comparator, &expected_comparator))
    }

    /// Check if two ComparatorTypes are compatible
    #[allow(clippy::only_used_in_recursion)]
    fn comparators_are_compatible(&self, left: &ComparatorType, right: &ComparatorType) -> bool {
        match (left, right) {
            // Exact matches
            (ComparatorType::Boolean, ComparatorType::Boolean) => true,
            (ComparatorType::TinyInt, ComparatorType::TinyInt) => true,
            (ComparatorType::SmallInt, ComparatorType::SmallInt) => true,
            (ComparatorType::Int, ComparatorType::Int) => true,
            (ComparatorType::BigInt, ComparatorType::BigInt) => true,
            (ComparatorType::Float32, ComparatorType::Float32) => true,
            (ComparatorType::Float, ComparatorType::Float) => true,
            (ComparatorType::Text, ComparatorType::Text) => true,
            (ComparatorType::Blob, ComparatorType::Blob) => true,
            (ComparatorType::Timestamp, ComparatorType::Timestamp) => true,
            (ComparatorType::Uuid, ComparatorType::Uuid) => true,
            (ComparatorType::Json, ComparatorType::Json) => true,

            // Collection types
            (ComparatorType::List(l_elem), ComparatorType::List(r_elem)) => {
                self.comparators_are_compatible(l_elem, r_elem)
            }
            (ComparatorType::Set(l_elem), ComparatorType::Set(r_elem)) => {
                self.comparators_are_compatible(l_elem, r_elem)
            }
            (ComparatorType::Map(l_key, l_val), ComparatorType::Map(r_key, r_val)) => {
                self.comparators_are_compatible(l_key, r_key)
                    && self.comparators_are_compatible(l_val, r_val)
            }

            // Tuple types
            (ComparatorType::Tuple(l_fields), ComparatorType::Tuple(r_fields)) => {
                l_fields.len() == r_fields.len()
                    && l_fields
                        .iter()
                        .zip(r_fields.iter())
                        .all(|(l, r)| self.comparators_are_compatible(l, r))
            }

            // UDT types
            (
                ComparatorType::Udt {
                    type_name: l_name,
                    keyspace: l_ks,
                    ..
                },
                ComparatorType::Udt {
                    type_name: r_name,
                    keyspace: r_ks,
                    ..
                },
            ) => l_name == r_name && l_ks == r_ks,

            // Frozen types
            (ComparatorType::Frozen(l_inner), ComparatorType::Frozen(r_inner)) => {
                self.comparators_are_compatible(l_inner, r_inner)
            }

            // Custom types
            (ComparatorType::Custom(l_name), ComparatorType::Custom(r_name)) => l_name == r_name,

            // No other combinations are compatible
            _ => false,
        }
    }

    /// Get registry statistics
    pub async fn get_statistics(&self) -> Result<RegistryStatistics> {
        let schemas = self.schemas.read().await;
        let udt_registry = self.udt_registry.read().await;
        let version_history = self.version_history.read().await;

        let mut stats = RegistryStatistics {
            total_schemas: schemas.len(),
            schemas_by_keyspace: HashMap::new(),
            validated_schemas: 0,
            schemas_with_warnings: 0,
            invalid_schemas: 0,
            total_udts: udt_registry.total_udts(),
            total_versions: version_history.values().map(|v| v.len()).sum(),
            auto_discovered_schemas: 0,
            manually_registered_schemas: 0,
            cache_hit_rate: 0.0, // TODO: Implement cache metrics
        };

        // Analyze schema distribution and status
        for entry in schemas.values() {
            let keyspace = &entry.schema.keyspace;
            *stats
                .schemas_by_keyspace
                .entry(keyspace.clone())
                .or_insert(0) += 1;

            match entry.validation_status {
                SchemaValidationStatus::Valid => stats.validated_schemas += 1,
                SchemaValidationStatus::ValidWithWarnings => stats.schemas_with_warnings += 1,
                SchemaValidationStatus::Invalid => stats.invalid_schemas += 1,
                SchemaValidationStatus::NotValidated => {}
            }

            match entry.source {
                SchemaSource::Discovered(_) => stats.auto_discovered_schemas += 1,
                _ => stats.manually_registered_schemas += 1,
            }
        }

        Ok(stats)
    }

    // Private helper methods

    async fn register_discovered_schema(
        &self,
        schema: TableSchema,
        schema_info: Option<SchemaInfo>,
        sstable_files: Vec<PathBuf>,
    ) -> Result<()> {
        let table_id = format!("{}.{}", schema.keyspace, schema.table);
        let source = SchemaSource::Discovered(sstable_files.clone());

        // Precompute derived comparators once (issue #1709); this is the
        // TTL-refresh / auto-discovery seam, so recompute here too.
        let derived = DerivedParsingState::compute(&schema).ok();
        let entry = SchemaEntry {
            schema,
            derived,
            extended_info: schema_info,
            registered_at: SystemTime::now(),
            source,
            validation_status: SchemaValidationStatus::Valid, // Discovery implies validation
            _associated_files: sstable_files,
        };

        let mut schemas = self.schemas.write().await;
        schemas.insert(table_id, entry);

        Ok(())
    }

    fn convert_schema_info_to_table_schema(&self, schema_info: &SchemaInfo) -> Result<TableSchema> {
        let mut columns = Vec::new();
        let mut partition_keys = Vec::new();
        let mut clustering_keys = Vec::new();

        // Convert partition keys
        for (pos, pk) in schema_info.partition_key.iter().enumerate() {
            partition_keys.push(crate::schema::KeyColumn {
                name: pk.name.clone(),
                data_type: pk.data_type.clone(),
                position: pos,
            });
        }

        // Convert clustering keys
        for ck in &schema_info.clustering_keys {
            clustering_keys.push(ck.clone());
        }

        // Convert all columns
        for col in &schema_info.regular_columns {
            columns.push(crate::schema::Column {
                name: col.name.clone(),
                data_type: col.data_type.clone(),
                nullable: col.nullable,
                default: None, // ColumnDefinition doesn't have default_value
                is_static: false,
            });
        }

        // Add static columns
        for col in &schema_info.static_columns {
            columns.push(crate::schema::Column {
                name: col.name.clone(),
                data_type: col.data_type.clone(),
                nullable: col.nullable,
                default: None, // ColumnDefinition doesn't have default_value
                is_static: true,
            });
        }

        Ok(TableSchema {
            keyspace: schema_info.keyspace.clone(),
            table: schema_info.table.clone(),
            partition_keys,
            clustering_keys,
            columns,
            comments: HashMap::new(),
            dropped_columns: HashMap::new(),
        })
    }

    fn is_entry_expired(&self, entry: &SchemaEntry) -> bool {
        if !self.config.enable_caching {
            return false;
        }

        let ttl = std::time::Duration::from_secs(self.config.cache_ttl_seconds);
        entry
            .registered_at
            .elapsed()
            .unwrap_or(std::time::Duration::ZERO)
            > ttl
    }

    async fn refresh_schema(&self, keyspace: &str, table: &str) -> Result<TableSchema> {
        // Implementation for refreshing expired schema
        // For now, just try auto-discovery
        self.auto_discover_schema(keyspace, table).await
    }

    async fn auto_discover_schema(&self, keyspace: &str, table: &str) -> Result<TableSchema> {
        // Try to find SSTable files for this table
        // This is a placeholder - in practice, you'd scan the data directory
        let sstable_files = self.find_sstable_files(keyspace, table).await?;

        if sstable_files.is_empty() {
            return Err(Error::Schema(format!(
                "No SSTables found for {}.{}",
                keyspace, table
            )));
        }

        self.discover_schema(keyspace, table, &sstable_files).await
    }

    async fn find_sstable_files(&self, _keyspace: &str, _table: &str) -> Result<Vec<PathBuf>> {
        // Placeholder implementation
        // In practice, this would scan the data directory structure
        Ok(Vec::new())
    }

    fn matches_query(&self, schema: &TableSchema, query: &SchemaQuery) -> bool {
        // Apply keyspace filter
        if let Some(ref ks) = query.keyspace {
            if &schema.keyspace != ks {
                return false;
            }
        }

        // Apply table pattern filter
        if let Some(ref pattern) = query.table_pattern {
            if !self.matches_pattern(&schema.table, pattern) {
                return false;
            }
        }

        // Other filters would be applied here
        true
    }

    fn matches_pattern(&self, text: &str, pattern: &str) -> bool {
        // Simple wildcard matching (can be enhanced)
        if pattern == "*" {
            return true;
        }

        // For now, just exact match or contains
        text == pattern || text.contains(pattern)
    }

    async fn create_schema_version(&self, table_id: &str, new_schema: &TableSchema) -> Result<()> {
        let mut version_history = self.version_history.write().await;
        let versions = version_history
            .entry(table_id.to_string())
            .or_insert_with(Vec::new);

        let version_number = versions.len() as u32 + 1;
        let changes = if versions.is_empty() {
            vec![SchemaChange {
                change_type: SchemaChangeType::ColumnAdded,
                component: "initial".to_string(),
                description: "Initial schema version".to_string(),
                old_value: None,
                new_value: None,
            }]
        } else {
            // Compare with previous version to detect changes
            self.detect_schema_changes(&versions.last().unwrap().schema, new_schema)
        };

        let new_version = SchemaVersion {
            version: version_number,
            created_at: SystemTime::now(),
            schema: new_schema.clone(),
            changes,
            source: "registry".to_string(),
        };

        versions.push(new_version);

        // Limit version history size
        if versions.len() > self.config.max_versions_per_schema {
            versions.remove(0);
        }

        Ok(())
    }

    fn detect_schema_changes(
        &self,
        old_schema: &TableSchema,
        new_schema: &TableSchema,
    ) -> Vec<SchemaChange> {
        let mut changes = Vec::new();

        // Compare columns
        let old_columns: HashMap<_, _> = old_schema.columns.iter().map(|c| (&c.name, c)).collect();
        let new_columns: HashMap<_, _> = new_schema.columns.iter().map(|c| (&c.name, c)).collect();

        // Find added columns
        for (name, column) in &new_columns {
            if !old_columns.contains_key(name) {
                changes.push(SchemaChange {
                    change_type: SchemaChangeType::ColumnAdded,
                    component: name.to_string(),
                    description: format!(
                        "Column '{}' added with type '{}'",
                        name, column.data_type
                    ),
                    old_value: None,
                    new_value: Some(column.data_type.clone()),
                });
            }
        }

        // Find removed columns
        for name in old_columns.keys() {
            if !new_columns.contains_key(name) {
                changes.push(SchemaChange {
                    change_type: SchemaChangeType::ColumnRemoved,
                    component: name.to_string(),
                    description: format!("Column '{}' removed", name),
                    old_value: None,
                    new_value: None,
                });
            }
        }

        // Find type changes
        for (name, new_column) in &new_columns {
            if let Some(old_column) = old_columns.get(name) {
                if old_column.data_type != new_column.data_type {
                    changes.push(SchemaChange {
                        change_type: SchemaChangeType::ColumnTypeChanged,
                        component: name.to_string(),
                        description: format!("Column '{}' type changed", name),
                        old_value: Some(old_column.data_type.clone()),
                        new_value: Some(new_column.data_type.clone()),
                    });
                }
            }
        }

        changes
    }

    async fn validate_schema_udts(
        &self,
        schema: &TableSchema,
        errors: &mut Vec<ValidationError>,
        warnings: &mut Vec<ValidationWarning>,
    ) {
        let udt_registry = self.udt_registry.read().await;

        for column in &schema.columns {
            // Check if column type references a UDT
            if let Ok(cql_type) = CqlType::parse(&column.data_type) {
                self.validate_cql_type_udts(
                    &cql_type,
                    &schema.keyspace,
                    &udt_registry,
                    errors,
                    warnings,
                );
            }
        }
    }

    #[allow(clippy::only_used_in_recursion)]
    fn validate_cql_type_udts(
        &self,
        cql_type: &CqlType,
        keyspace: &str,
        udt_registry: &UdtRegistry,
        errors: &mut Vec<ValidationError>,
        _warnings: &mut Vec<ValidationWarning>,
    ) {
        match cql_type {
            CqlType::Udt(udt_name, _) => {
                if !udt_registry.contains_udt(keyspace, udt_name)
                    && !udt_registry.contains_udt("system", udt_name)
                {
                    errors.push(ValidationError {
                        code: "UDT_NOT_FOUND".to_string(),
                        message: format!("UDT '{}' not found in keyspace '{}'", udt_name, keyspace),
                        component: Some(udt_name.clone()),
                        severity: ErrorSeverity::High,
                    });
                }
            }
            // `CqlType::parse` represents UDT references as `Custom("udt:<name>")`
            // for mixed-case/underscore/digit names but as a bare `Custom("<name>")`
            // for purely-lowercase names (e.g. `address`). Either way `parse` only
            // yields `Custom` for non-primitive, non-collection type strings — i.e.
            // UDT references — so validate the de-prefixed name (issue #761,
            // roborev job 44).
            CqlType::Custom(name) => {
                let udt_name = name.strip_prefix("udt:").unwrap_or(name);
                let (lookup_keyspace, bare_name) = match udt_name.split_once('.') {
                    Some((ks, n)) => (ks, n),
                    None => (keyspace, udt_name),
                };
                // Skip structural fragments parse couldn't resolve (e.g. an
                // uppercase `SET<TEXT>`); only simple identifiers name a UDT.
                if crate::schema::is_udt_identifier(udt_name)
                    && !udt_registry.contains_udt(lookup_keyspace, bare_name)
                    && !udt_registry.contains_udt("system", bare_name)
                {
                    errors.push(ValidationError {
                        code: "UDT_NOT_FOUND".to_string(),
                        message: format!(
                            "UDT '{}' not found in keyspace '{}'",
                            udt_name, lookup_keyspace
                        ),
                        component: Some(udt_name.to_string()),
                        severity: ErrorSeverity::High,
                    });
                }
            }
            CqlType::List(inner) | CqlType::Set(inner) | CqlType::Frozen(inner) => {
                self.validate_cql_type_udts(inner, keyspace, udt_registry, errors, _warnings);
            }
            CqlType::Map(key_type, value_type) => {
                self.validate_cql_type_udts(key_type, keyspace, udt_registry, errors, _warnings);
                self.validate_cql_type_udts(value_type, keyspace, udt_registry, errors, _warnings);
            }
            CqlType::Tuple(types) => {
                for t in types {
                    self.validate_cql_type_udts(t, keyspace, udt_registry, errors, _warnings);
                }
            }
            _ => {} // Primitive types don't need UDT validation
        }
    }

    async fn validate_column_types(
        &self,
        schema: &TableSchema,
        errors: &mut Vec<ValidationError>,
        _warnings: &mut [ValidationWarning],
    ) {
        for column in &schema.columns {
            if let Err(e) = CqlType::parse(&column.data_type) {
                errors.push(ValidationError {
                    code: "INVALID_COLUMN_TYPE".to_string(),
                    message: format!("Invalid column type '{}': {}", column.data_type, e),
                    component: Some(column.name.clone()),
                    severity: ErrorSeverity::High,
                });
            }
        }
    }

    async fn generate_performance_recommendations(
        &self,
        schema: &TableSchema,
        recommendations: &mut Vec<String>,
    ) {
        // Check for potential performance issues

        // Large partition keys
        if schema.partition_keys.len() > 3 {
            recommendations.push(
                "Consider reducing the number of partition key columns for better performance"
                    .to_string(),
            );
        }

        // Many clustering keys
        if schema.clustering_keys.len() > 5 {
            recommendations
                .push("Large number of clustering keys may impact query performance".to_string());
        }

        // Column count
        if schema.columns.len() > 50 {
            recommendations.push(
                "Consider using UDTs or denormalizing wide tables for better performance"
                    .to_string(),
            );
        }
    }

    fn generate_basic_cql(&self, schema: &TableSchema) -> String {
        let mut cql = format!("CREATE TABLE {}.{} (\n", schema.keyspace, schema.table);

        // Add columns
        for (i, column) in schema.columns.iter().enumerate() {
            if i > 0 {
                cql.push_str(",\n");
            }
            cql.push_str(&format!("  {} {}", column.name, column.data_type));
        }

        // Add primary key
        if !schema.partition_keys.is_empty() {
            cql.push_str(",\n  PRIMARY KEY (");

            if schema.partition_keys.len() == 1 && schema.clustering_keys.is_empty() {
                cql.push_str(&schema.partition_keys[0].name);
            } else {
                // Composite primary key
                cql.push('(');
                for (i, pk) in schema.partition_keys.iter().enumerate() {
                    if i > 0 {
                        cql.push_str(", ");
                    }
                    cql.push_str(&pk.name);
                }
                cql.push(')');

                if !schema.clustering_keys.is_empty() {
                    for ck in &schema.clustering_keys {
                        cql.push_str(", ");
                        cql.push_str(&ck.name);
                    }
                }
            }

            cql.push(')');
        }

        cql.push_str("\n);");
        cql
    }
}

/// Registry statistics
#[derive(Debug, Clone)]
pub struct RegistryStatistics {
    /// Total number of registered schemas
    pub total_schemas: usize,
    /// Schemas grouped by keyspace
    pub schemas_by_keyspace: HashMap<String, usize>,
    /// Number of validated schemas
    pub validated_schemas: usize,
    /// Schemas with validation warnings
    pub schemas_with_warnings: usize,
    /// Invalid schemas
    pub invalid_schemas: usize,
    /// Total UDTs registered
    pub total_udts: usize,
    /// Total schema versions stored
    pub total_versions: usize,
    /// Auto-discovered schemas
    pub auto_discovered_schemas: usize,
    /// Manually registered schemas
    pub manually_registered_schemas: usize,
    /// Cache hit rate
    pub cache_hit_rate: f64,
}

/// Schema-driven parsing context containing all necessary type information.
///
/// Fields are `Arc`-shared with the registry's cached [`SchemaEntry`] derived
/// state (issue #1709), so constructing a context is a set of pointer bumps
/// rather than deep clones + per-column type parsing.
#[derive(Debug, Clone)]
pub struct ParsingContext {
    /// The complete table schema
    pub schema: Arc<TableSchema>,
    /// Comparators for partition key components
    pub partition_comparators: Arc<Vec<ComparatorType>>,
    /// Comparators for clustering key components
    pub clustering_comparators: Arc<Vec<ComparatorType>>,
    /// Comparators for all columns by name
    pub column_comparators: Arc<HashMap<String, ComparatorType>>,
}

impl ParsingContext {
    /// Construct a context from owned parts, sharing each behind an `Arc`.
    ///
    /// The registry hands out contexts from precomputed `Arc`s (issue #1709);
    /// this constructor covers callers that derive the parts ad hoc (e.g.
    /// `schema_aware_reader`) and keeps the `Arc` wrapping an implementation
    /// detail of `ParsingContext`.
    pub fn from_owned(
        schema: TableSchema,
        partition_comparators: Vec<ComparatorType>,
        clustering_comparators: Vec<ComparatorType>,
        column_comparators: HashMap<String, ComparatorType>,
    ) -> Self {
        Self {
            schema: Arc::new(schema),
            partition_comparators: Arc::new(partition_comparators),
            clustering_comparators: Arc::new(clustering_comparators),
            column_comparators: Arc::new(column_comparators),
        }
    }

    /// Get comparator for a specific column
    pub fn get_column_comparator(&self, column_name: &str) -> Option<&ComparatorType> {
        self.column_comparators.get(column_name)
    }

    /// Check if schema-driven parsing is fully configured
    pub fn is_complete(&self) -> bool {
        !self.partition_comparators.is_empty() || !self.schema.partition_keys.is_empty()
    }

    /// Get all key columns (partition + clustering) names in order
    pub fn get_all_key_column_names(&self) -> Vec<String> {
        let mut names = Vec::new();
        names.extend(
            self.schema
                .ordered_partition_keys()
                .iter()
                .map(|k| k.name.clone()),
        );
        names.extend(
            self.schema
                .ordered_clustering_keys()
                .iter()
                .map(|k| k.name.clone()),
        );
        names
    }
}

/// Schema validator for comprehensive validation
#[derive(Debug)]
pub struct SchemaValidator;

impl Default for SchemaValidator {
    fn default() -> Self {
        Self::new()
    }
}

impl SchemaValidator {
    pub fn new() -> Self {
        Self
    }

    pub async fn validate_table_schema(&self, schema: &TableSchema) -> Result<()> {
        schema.validate()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    async fn make_registry(mut reg_config: SchemaRegistryConfig) -> SchemaRegistry {
        reg_config.enable_auto_discovery = false;
        let core_config = Config::default();
        let platform = Arc::new(Platform::new(&core_config).await.expect("platform"));
        SchemaRegistry::new(reg_config, platform, core_config)
            .await
            .expect("registry")
    }

    fn simple_schema(name: &str) -> TableSchema {
        TableSchema {
            keyspace: "test_ks".to_string(),
            table: name.to_string(),
            partition_keys: vec![crate::schema::KeyColumn {
                name: "id".to_string(),
                data_type: "uuid".to_string(),
                position: 0,
            }],
            clustering_keys: vec![],
            columns: vec![crate::schema::Column {
                name: "id".to_string(),
                data_type: "uuid".to_string(),
                nullable: false,
                default: None,
                is_static: false,
            }],
            comments: HashMap::new(),
            dropped_columns: HashMap::new(),
        }
    }

    #[tokio::test]
    async fn test_schema_registry_creation() {
        let registry = make_registry(SchemaRegistryConfig::default()).await;
        let stats = registry.get_statistics().await.unwrap();
        assert_eq!(stats.total_schemas, 0);
    }

    /// Regression (roborev job 44): the registry's `validate_cql_type_udts` must
    /// flag an undefined purely-lowercase UDT (bare `Custom("address")`, no
    /// `udt:` prefix), matching `TableSchema::check_type_udt_references`.
    #[tokio::test]
    async fn test_registry_validate_lowercase_udt_not_found() {
        let registry = make_registry(SchemaRegistryConfig::default()).await;
        let udt_registry = UdtRegistry::new();
        let mut errors = Vec::new();
        let mut warnings = Vec::new();

        let cql_type = CqlType::parse("address").expect("parse");
        registry.validate_cql_type_udts(
            &cql_type,
            "test_ks",
            &udt_registry,
            &mut errors,
            &mut warnings,
        );

        assert!(
            errors
                .iter()
                .any(|e| e.code == "UDT_NOT_FOUND" && e.message.contains("address")),
            "undefined lowercase UDT must be reported, got: {errors:?}"
        );
    }

    /// Regression (roborev job 55): the registry validation path must also catch
    /// an undefined UDT nested inside an UPPERCASE collection, mirroring
    /// `TableSchema::validate_udt_references`.
    #[tokio::test]
    async fn test_registry_validate_uppercase_collection_udt_not_found() {
        let registry = make_registry(SchemaRegistryConfig::default()).await;
        let udt_registry = UdtRegistry::new();

        for ty in ["LIST<MissingType>", "MAP<TEXT, MissingType>"] {
            let mut errors = Vec::new();
            let mut warnings = Vec::new();
            let cql_type = CqlType::parse(ty).expect("parse");
            registry.validate_cql_type_udts(
                &cql_type,
                "test_ks",
                &udt_registry,
                &mut errors,
                &mut warnings,
            );
            assert!(
                errors
                    .iter()
                    .any(|e| e.code == "UDT_NOT_FOUND" && e.message.contains("MissingType")),
                "undefined UDT in '{ty}' must be reported, got: {errors:?}"
            );
        }
    }

    #[test]
    fn test_schema_query_creation() {
        let query = SchemaQuery {
            keyspace: Some("test_ks".to_string()),
            table_pattern: Some("user_*".to_string()),
            source_types: None,
            validated_only: false,
            include_history: false,
        };

        assert_eq!(query.keyspace.as_ref().unwrap(), "test_ks");
        assert_eq!(query.table_pattern.as_ref().unwrap(), "user_*");
    }

    #[tokio::test]
    async fn register_and_retrieve_schema() {
        let registry = make_registry(SchemaRegistryConfig::default()).await;
        let schema = simple_schema("users");

        registry
            .register_schema(schema.clone(), SchemaSource::Manual)
            .await
            .expect("register schema");

        let fetched = registry
            .get_schema("test_ks", "users")
            .await
            .expect("fetch schema");
        assert_eq!(fetched.table, "users");
        assert_eq!(fetched.partition_keys.len(), 1);
    }

    /// Issue #1710: `register_schema` must not hold the `schemas` write guard
    /// across the `create_schema_version(...).await` (that acquires the
    /// `version_history` lock — a latent lock-order hazard). Verified by
    /// structure (version computation is now hoisted before the `schemas`
    /// write in `register_schema`) and behaviorally here: many concurrent
    /// re-registrations of the same table (versioning ON, forcing the version
    /// path every time) must complete without deadlock and leave exactly one
    /// final entry (idempotent, no lost update).
    #[tokio::test]
    async fn test_concurrent_register_schema_idempotent_no_deadlock() {
        let mut cfg = SchemaRegistryConfig::default();
        cfg.enable_versioning = true; // force the create_schema_version path
        let registry = Arc::new(make_registry(cfg).await);

        // Seed once so every concurrent call takes the versioning branch.
        registry
            .register_schema(simple_schema("users"), SchemaSource::Manual)
            .await
            .expect("seed register");

        let mut handles = vec![];
        for _ in 0..16 {
            let r = Arc::clone(&registry);
            handles.push(tokio::spawn(async move {
                r.register_schema(simple_schema("users"), SchemaSource::Manual)
                    .await
                    .expect("concurrent register");
            }));
        }
        for h in handles {
            h.await.expect("task joins (no deadlock)");
        }

        // Exactly one final entry for the table (idempotent upsert).
        let all = registry.list_schemas(None).await.expect("list");
        assert_eq!(
            all.iter().filter(|s| s.table == "users").count(),
            1,
            "concurrent re-registration must leave exactly one entry"
        );
    }

    /// Issue #1710 (roborev follow-up): the existence-check and insert in
    /// `register_schema` must be a SINGLE write-locked critical section.
    /// Here N tasks concurrently register the SAME previously-absent table.
    /// With an atomic check/insert, exactly ONE task observes "absent"
    /// (creates no version) and the other N-1 observe "present" (each records
    /// one version) — so version history holds exactly N-1 entries and there
    /// is exactly one final schema entry. The pre-fix check-then-act shape
    /// (read-lock check, later write-lock insert) let several tasks observe
    /// "absent" at once, dropping version-creation calls (a lost update), so
    /// history would hold FEWER than N-1 entries. A barrier releases all tasks
    /// simultaneously and we repeat to make that concurrent-absent window real.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_first_registration_no_lost_version() {
        const N: usize = 8;
        for _ in 0..40 {
            let mut cfg = SchemaRegistryConfig::default();
            cfg.enable_versioning = true;
            // Large cap so create_schema_version never trims history and the
            // N-1 assertion measures every recorded version.
            cfg.max_versions_per_schema = 10_000;
            let registry = Arc::new(make_registry(cfg).await);
            let barrier = Arc::new(tokio::sync::Barrier::new(N));

            let mut handles = vec![];
            for _ in 0..N {
                let r = Arc::clone(&registry);
                let b = Arc::clone(&barrier);
                handles.push(tokio::spawn(async move {
                    // Release all registrations at the same instant so the
                    // "table absent" window is genuinely contended.
                    b.wait().await;
                    r.register_schema(simple_schema("users"), SchemaSource::Manual)
                        .await
                        .expect("concurrent register");
                }));
            }
            for h in handles {
                h.await.expect("task joins (no deadlock)");
            }

            // Exactly one final entry (idempotent upsert, no lost insert).
            let all = registry.list_schemas(None).await.expect("list");
            assert_eq!(
                all.iter().filter(|s| s.table == "users").count(),
                1,
                "concurrent first registration must leave exactly one entry"
            );

            // Exactly one task saw "absent" (skips versioning); the other N-1
            // each recorded a version. Fewer than N-1 => a lost update.
            let history = registry
                .get_schema_history("test_ks", "users")
                .await
                .expect("history");
            assert_eq!(
                history.len(),
                N - 1,
                "exactly one first-registration + (N-1) versioned updates: no lost update"
            );
        }
    }

    /// A `simple_schema` plus one distinguishing regular column, so different
    /// tasks can register genuinely DIFFERENT schemas for the same table.
    fn schema_with_marker(table: &str, marker: &str) -> TableSchema {
        let mut s = simple_schema(table);
        s.columns.push(crate::schema::Column {
            name: marker.to_string(),
            data_type: "text".to_string(),
            nullable: true,
            default: None,
            is_static: false,
        });
        s
    }

    fn column_names(schema: &TableSchema) -> Vec<String> {
        schema.columns.iter().map(|c| c.name.clone()).collect()
    }

    /// Issue #1710 (roborev Medium): registry and version-history must not
    /// diverge under concurrent updates of the SAME table with DIFFERENT
    /// schemas. Each task registers `users` with a distinct marker column, so
    /// its schema is unique. With the per-table update lock, the whole
    /// {upsert schema -> append version} sequence runs atomically per table, so
    /// the final registry schema ALWAYS equals the newest `version_history`
    /// entry. Without the lock, task A's upsert and task B's history-append can
    /// interleave (registry=A, newest history=B) — divergence this test
    /// detects. A barrier releases all tasks at once and we repeat many
    /// iterations to make the interleaving window real; no deadlock either.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_distinct_schemas_registry_history_consistent() {
        const N: usize = 16;
        for _ in 0..2000 {
            let mut cfg = SchemaRegistryConfig::default();
            cfg.enable_versioning = true;
            cfg.max_versions_per_schema = 10_000; // never trim; keep true newest
            let registry = Arc::new(make_registry(cfg).await);

            // Seed so every concurrent call takes the versioning (update) path.
            registry
                .register_schema(schema_with_marker("users", "seed"), SchemaSource::Manual)
                .await
                .expect("seed register");

            let barrier = Arc::new(tokio::sync::Barrier::new(N));
            let mut handles = vec![];
            for i in 0..N {
                let r = Arc::clone(&registry);
                let b = Arc::clone(&barrier);
                handles.push(tokio::spawn(async move {
                    let marker = format!("c{i}");
                    b.wait().await;
                    r.register_schema(schema_with_marker("users", &marker), SchemaSource::Manual)
                        .await
                        .expect("concurrent register");
                }));
            }
            for h in handles {
                h.await.expect("task joins (no deadlock)");
            }

            // The registry's current schema must equal the newest recorded
            // version — the two cannot have been left describing different
            // updates.
            let current = registry.get_schema("test_ks", "users").await.expect("get");
            let history = registry
                .get_schema_history("test_ks", "users")
                .await
                .expect("history");
            let newest = history.last().expect("at least one version recorded");
            assert_eq!(
                column_names(&current),
                column_names(&newest.schema),
                "registry schema and newest version_history entry must agree \
                 (registry={:?}, newest_version={:?})",
                column_names(&current),
                column_names(&newest.schema),
            );
        }
    }

    #[tokio::test]
    async fn schema_version_history_tracks_changes() {
        let registry = make_registry(SchemaRegistryConfig::default()).await;
        let mut schema = simple_schema("accounts");

        registry
            .register_schema(schema.clone(), SchemaSource::Manual)
            .await
            .expect("register v1");

        schema.columns.push(crate::schema::Column {
            name: "status".to_string(),
            data_type: "text".to_string(),
            nullable: true,
            default: None,
            is_static: false,
        });

        registry
            .register_schema(schema.clone(), SchemaSource::Manual)
            .await
            .expect("register v2");

        let history = registry
            .get_schema_history("test_ks", "accounts")
            .await
            .expect("history");

        assert_eq!(
            history.len(),
            1,
            "Second registration should emit first version"
        );
        assert!(history[0]
            .changes
            .iter()
            .any(|change| matches!(change.change_type, SchemaChangeType::ColumnAdded)));
    }

    /// A schema exercising every derived vector: a 2-component partition key, a
    /// 2-component clustering key, and several regular/collection columns.
    fn multi_column_schema(name: &str) -> TableSchema {
        use crate::schema::{ClusteringColumn, ClusteringOrder, Column, KeyColumn};
        TableSchema {
            keyspace: "test_ks".to_string(),
            table: name.to_string(),
            partition_keys: vec![
                KeyColumn {
                    name: "region".to_string(),
                    data_type: "text".to_string(),
                    position: 0,
                },
                KeyColumn {
                    name: "shard".to_string(),
                    data_type: "int".to_string(),
                    position: 1,
                },
            ],
            clustering_keys: vec![
                ClusteringColumn {
                    name: "ts".to_string(),
                    data_type: "timestamp".to_string(),
                    position: 0,
                    order: ClusteringOrder::Asc,
                },
                ClusteringColumn {
                    name: "seq".to_string(),
                    data_type: "bigint".to_string(),
                    position: 1,
                    order: ClusteringOrder::Desc,
                },
            ],
            columns: vec![
                Column {
                    name: "region".to_string(),
                    data_type: "text".to_string(),
                    nullable: false,
                    default: None,
                    is_static: false,
                },
                Column {
                    name: "shard".to_string(),
                    data_type: "int".to_string(),
                    nullable: false,
                    default: None,
                    is_static: false,
                },
                Column {
                    name: "ts".to_string(),
                    data_type: "timestamp".to_string(),
                    nullable: false,
                    default: None,
                    is_static: false,
                },
                Column {
                    name: "seq".to_string(),
                    data_type: "bigint".to_string(),
                    nullable: false,
                    default: None,
                    is_static: false,
                },
                Column {
                    name: "payload".to_string(),
                    data_type: "text".to_string(),
                    nullable: true,
                    default: None,
                    is_static: false,
                },
                Column {
                    name: "tags".to_string(),
                    data_type: "list<text>".to_string(),
                    nullable: true,
                    default: None,
                    is_static: false,
                },
            ],
            comments: HashMap::new(),
            dropped_columns: HashMap::new(),
        }
    }

    /// Issue #1709 (red on main): after registration, `get_parsing_context` must
    /// perform ZERO `CqlType::parse` calls and ZERO `TableSchema` deep clones on
    /// the request path. Before caching this was `O(columns)` parses and 4 deep
    /// clones per call.
    #[tokio::test]
    async fn get_parsing_context_does_zero_work_after_registration() {
        let registry = make_registry(SchemaRegistryConfig::default()).await;
        registry
            .register_schema(multi_column_schema("metrics"), SchemaSource::Manual)
            .await
            .expect("register");

        // Measure only the request path.
        crate::schema::work_counters::reset();
        let _ctx = registry
            .get_parsing_context("test_ks", "metrics")
            .await
            .expect("context");

        assert_eq!(
            crate::schema::work_counters::parse_calls(),
            0,
            "get_parsing_context must not parse any CQL types after registration"
        );
        assert_eq!(
            crate::schema::work_counters::schema_clones(),
            0,
            "get_parsing_context must not deep-clone the schema after registration"
        );

        // A second call is likewise free.
        crate::schema::work_counters::reset();
        let _ctx2 = registry
            .get_parsing_context("test_ks", "metrics")
            .await
            .expect("context2");
        assert_eq!(crate::schema::work_counters::parse_calls(), 0);
        assert_eq!(crate::schema::work_counters::schema_clones(), 0);
    }

    /// The cached context must be field-by-field IDENTICAL to the pre-cache path
    /// (schema + the three comparator collections derived on demand).
    #[tokio::test]
    async fn cached_parsing_context_matches_on_demand_derivation() {
        for schema in [simple_schema("simple"), multi_column_schema("wide")] {
            let table = schema.table.clone();
            let registry = make_registry(SchemaRegistryConfig::default()).await;
            registry
                .register_schema(schema, SchemaSource::Manual)
                .await
                .expect("register");

            let ctx = registry
                .get_parsing_context("test_ks", &table)
                .await
                .expect("context");

            // Reference: the original per-call derivation helpers.
            let ref_schema = registry
                .get_schema("test_ks", &table)
                .await
                .expect("schema");
            let ref_partition = registry
                .get_partition_key_comparator("test_ks", &table)
                .await
                .expect("partition");
            let ref_clustering = registry
                .get_clustering_key_comparator("test_ks", &table)
                .await
                .expect("clustering");
            let ref_columns = registry
                .get_table_comparators("test_ks", &table)
                .await
                .expect("columns");

            // TableSchema has no PartialEq; compare structurally via serde_json
            // (order-independent for maps), which is a true field-by-field check.
            assert_eq!(
                serde_json::to_value(&*ctx.schema).expect("ctx schema json"),
                serde_json::to_value(&ref_schema).expect("ref schema json"),
                "schema mismatch for {}",
                table
            );
            assert_eq!(
                *ctx.partition_comparators, ref_partition,
                "partition comparators mismatch for {}",
                table
            );
            assert_eq!(
                *ctx.clustering_comparators, ref_clustering,
                "clustering comparators mismatch for {}",
                table
            );
            assert_eq!(
                *ctx.column_comparators, ref_columns,
                "column comparators mismatch for {}",
                table
            );
        }
    }

    /// Two contexts handed out for the same registered schema share the same
    /// underlying `Arc` allocations (pointer bump, not deep copy).
    #[tokio::test]
    async fn parsing_contexts_share_arc_backing() {
        let registry = make_registry(SchemaRegistryConfig::default()).await;
        registry
            .register_schema(multi_column_schema("shared"), SchemaSource::Manual)
            .await
            .expect("register");

        let a = registry
            .get_parsing_context("test_ks", "shared")
            .await
            .expect("a");
        let b = registry
            .get_parsing_context("test_ks", "shared")
            .await
            .expect("b");

        assert!(Arc::ptr_eq(&a.schema, &b.schema));
        assert!(Arc::ptr_eq(
            &a.partition_comparators,
            &b.partition_comparators
        ));
        assert!(Arc::ptr_eq(
            &a.clustering_comparators,
            &b.clustering_comparators
        ));
        assert!(Arc::ptr_eq(&a.column_comparators, &b.column_comparators));
    }

    #[tokio::test]
    async fn expired_cached_schema_invokes_discovery_path() {
        let mut config = SchemaRegistryConfig::default();
        config.cache_ttl_seconds = 0;
        config.enable_auto_discovery = true;
        let registry = make_registry(config).await;

        registry
            .register_schema(simple_schema("events"), SchemaSource::Manual)
            .await
            .expect("register events schema");

        tokio::time::sleep(std::time::Duration::from_millis(10)).await;

        let err = registry
            .get_schema("test_ks", "events")
            .await
            .expect_err("expired schema should attempt discovery");

        assert!(matches!(err, Error::Schema(message) if message.contains("No SSTables found")));
    }

    /// Roborev Medium (issue #1708 follow-up): `find_schema_by_table` — the
    /// single lookup the manager delegates to — must apply the SAME
    /// TTL-expiry/refresh handling `get_schema` uses for the matched entry, and
    /// must NOT serve an expired entry stale. With `cache_ttl_seconds = 0` the
    /// registered entry expires, so the lookup delegates to the refresh seam
    /// (which errors here because no SSTables back a manually-registered schema)
    /// EXACTLY as `get_schema` does. RED on HEAD (read the map directly and
    /// returned `Ok(Some(stale))`).
    #[tokio::test]
    async fn find_schema_by_table_honors_ttl_expiry_like_get_schema() {
        let mut config = SchemaRegistryConfig::default();
        config.cache_ttl_seconds = 0;
        let registry = make_registry(config).await;

        registry
            .register_schema(simple_schema("events"), SchemaSource::Manual)
            .await
            .expect("register events schema");

        tokio::time::sleep(std::time::Duration::from_millis(10)).await;

        // `get_schema` triggers the refresh seam and errors on the expired entry.
        let get_err = registry
            .get_schema("test_ks", "events")
            .await
            .expect_err("expired schema attempts refresh");
        assert!(matches!(&get_err, Error::Schema(m) if m.contains("No SSTables found")));

        // Scoped `keyspace.table` lookup on the EXPIRED entry must behave
        // identically: delegate to refresh (error), never `Ok(Some(stale))`.
        let scoped = registry
            .find_schema_by_table(&Some("test_ks".to_string()), "events")
            .await;
        assert!(
            matches!(&scoped, Err(Error::Schema(m)) if m.contains("No SSTables found")),
            "scoped lookup on an EXPIRED entry must delegate to refresh (not serve stale), got {scoped:?}"
        );

        // The bare-table-name path must honor expiry too.
        let bare = registry.find_schema_by_table(&None, "events").await;
        assert!(
            matches!(&bare, Err(Error::Schema(m)) if m.contains("No SSTables found")),
            "bare lookup on an EXPIRED entry must delegate to refresh, got {bare:?}"
        );
    }

    /// A FRESH matched entry is still returned by value, and an unknown table
    /// returns `Ok(None)` with NO fabrication/auto-discovery (issue #1710
    /// preserved) — the two non-expiry outcomes the refresh path must not alter.
    #[tokio::test]
    async fn find_schema_by_table_returns_fresh_and_none_for_unknown() {
        let registry = make_registry(SchemaRegistryConfig::default()).await;
        registry
            .register_schema(simple_schema("events"), SchemaSource::Manual)
            .await
            .expect("register");

        let found = registry
            .find_schema_by_table(&Some("test_ks".to_string()), "events")
            .await
            .expect("a fresh entry never triggers the refresh error path");
        assert!(
            matches!(found, Some(s) if s.table == "events"),
            "fresh matched entry must be returned by value"
        );

        let missing = registry
            .find_schema_by_table(&None, "never_registered")
            .await
            .expect("unknown table is Ok(None), not an error");
        assert!(
            missing.is_none(),
            "unknown table must be None (no fabrication, no auto-discovery)"
        );
    }
}