heliosdb-nano 3.49.0

//! DDL (Data Definition Language) operations
//!
//! This module handles CREATE/DROP INDEX and other DDL operations.

#![allow(elided_lifetimes_in_paths)]

use super::scan::ScanOperator;
use super::{Executor, PhysicalOperator};
use crate::sql::LogicalPlan;
use crate::{Error, Result};
use rocksdb::{IteratorMode, ReadOptions};
use std::sync::Arc;

fn empty_ddl_result(executor: &Executor) -> Box<dyn PhysicalOperator> {
    Box::new(
        ScanOperator::new(
            "".to_string(),
            Arc::new(crate::Schema { columns: vec![] }),
            None,
            vec![],
            vec![],
        )
        .with_timeout(executor.timeout_ctx()),
    )
}

fn create_art_secondary_index(
    storage: &crate::storage::StorageEngine,
    name: &str,
    table_name: &str,
    column_name: &str,
    if_not_exists: bool,
) -> Result<Option<usize>> {
    if storage.is_branch_active() {
        return Err(Error::query_execution(
            "CREATE INDEX for ART secondary indexes must run on the main branch",
        ));
    }

    let art_manager = storage.art_indexes();
    if art_manager.index_exists(name) {
        if if_not_exists {
            return Ok(None);
        }
        return Err(Error::query_execution(format!("ART index '{}' already exists", name)));
    }

    let catalog = storage.catalog();
    let schema = catalog.get_table_schema(table_name)?;
    if !schema.columns.iter().any(|c| c.name == column_name) {
        return Err(Error::query_execution(format!(
            "Column '{}' not found in table '{}'",
            column_name, table_name
        )));
    }

    let columns = vec![column_name.to_string()];
    art_manager
        .create_manual_index(name, table_name, &columns)
        .map_err(|e| Error::query_execution(format!("Failed to create ART index: {}", e)))?;

    let tuples = storage.scan_table_with_schema(table_name, &schema)?;
    match art_manager.backfill_manual_index(name, &schema, &tuples) {
        Ok(backfilled) => Ok(Some(backfilled)),
        Err(e) => {
            let _ = art_manager.drop_index(name);
            Err(Error::query_execution(format!("Failed to backfill ART index: {}", e)))
        }
    }
}

fn vector_distance_metric(options: &[crate::sql::logical_plan::IndexOption]) -> Result<crate::vector::DistanceMetric> {
    use crate::sql::logical_plan::IndexOption;
    use crate::vector::DistanceMetric;

    let mut metric = DistanceMetric::L2;
    for option in options {
        if let IndexOption::DistanceMetric(name) = option {
            metric = match name.as_str() {
                "l2" | "euclidean" => DistanceMetric::L2,
                "cosine" => DistanceMetric::Cosine,
                "ip" | "inner_product" => DistanceMetric::InnerProduct,
                other => {
                    return Err(Error::query_execution(format!(
                        "Unsupported vector index metric '{}'",
                        other
                    )))
                }
            };
        }
    }
    Ok(metric)
}

fn collect_existing_vectors(
    storage: &crate::storage::StorageEngine,
    schema: &crate::Schema,
    table_name: &str,
    column_name: &str,
    dimension: usize,
) -> Result<Vec<(u64, crate::vector::Vector)>> {
    let col_idx = schema
        .get_column_index(column_name)
        .ok_or_else(|| Error::query_execution(format!("Column '{}' not found in schema", column_name)))?;
    let tuples = storage.scan_table_with_schema_columns(table_name, schema, &[col_idx])?;
    let mut vectors = Vec::with_capacity(tuples.len());

    for tuple in tuples {
        match tuple.values.get(col_idx) {
            Some(crate::Value::Vector(vec)) => {
                if vec.len() != dimension {
                    return Err(Error::query_execution(format!(
                        "Vector dimension mismatch while backfilling '{}.{}': expected {}, got {}",
                        table_name,
                        column_name,
                        dimension,
                        vec.len()
                    )));
                }
                let row_id = tuple.row_id.ok_or_else(|| {
                    Error::query_execution(format!(
                        "Cannot backfill vector index on '{}.{}' from tuple without row_id",
                        table_name, column_name
                    ))
                })?;
                vectors.push((row_id, vec.clone()));
            }
            Some(crate::Value::Null) | None => {}
            Some(other) => {
                return Err(Error::query_execution(format!(
                    "Cannot backfill vector index on '{}.{}' from non-vector value {:?}",
                    table_name, column_name, other
                )))
            }
        }
    }

    Ok(vectors)
}

fn backfill_vector_index(
    vector_indexes: &crate::storage::VectorIndexManager,
    index_name: &str,
    vectors: &[(u64, crate::vector::Vector)],
) -> Result<usize> {
    // Parallel bulk build: hands the whole batch to the index, which uses
    // Rayon (`parallel_insert`) to construct the HNSW graph across all cores.
    // This replaces the old single-threaded per-row `insert_vector` loop and
    // is the dominant cost of `CREATE INDEX ... USING hnsw` on a populated
    // table (e.g. ~1120s → far less for 678k vectors). Results are identical
    // to sequential insertion; only build wall-time changes.
    if let Err(e) = vector_indexes.insert_vectors_batch(index_name, vectors) {
        let _ = vector_indexes.drop_index(index_name);
        return Err(Error::query_execution(format!(
            "Failed to backfill HNSW index '{}': {}",
            index_name, e
        )));
    }
    Ok(vectors.len())
}

fn persist_index_definition(
    storage: &crate::storage::StorageEngine,
    name: &str,
    table_name: &str,
    column_name: &str,
    index_type: Option<&str>,
    options: &[crate::sql::logical_plan::IndexOption],
) -> Result<()> {
    let definition = crate::storage::PersistedIndexDefinition {
        table_name: table_name.to_string(),
        column_name: column_name.to_string(),
        index_type: index_type.map(str::to_string),
        options: options.to_vec(),
    };
    storage.catalog().save_index_definition(name, &definition)
}

fn encoded_index_options(options: &[crate::sql::logical_plan::IndexOption]) -> Vec<u8> {
    bincode::serialize(options).unwrap_or_default()
}

/// Handle CREATE INDEX logical plan node
pub(super) fn handle_create_index(executor: &Executor, plan: &LogicalPlan) -> Result<Box<dyn PhysicalOperator>> {
    if let LogicalPlan::CreateIndex {
        name,
        table_name,
        column_name,
        index_type,
        if_not_exists,
        options,
    } = plan
    {
        if let Some(storage) = executor.storage() {
            if matches!(index_type.as_deref(), None | Some("art" | "btree" | "hash")) {
                if let Some(backfilled) =
                    create_art_secondary_index(storage, name, table_name, column_name, *if_not_exists)?
                {
                    tracing::info!(
                        "Created ART index '{}' on table '{}' column '{}' with {} existing rows",
                        name,
                        table_name,
                        column_name,
                        backfilled
                    );
                    persist_index_definition(storage, name, table_name, column_name, Some("art"), options)?;

                    let encoded_options = encoded_index_options(options);
                    if let Err(e) =
                        storage.log_create_index(name, table_name, column_name, Some("art"), &encoded_options)
                    {
                        tracing::warn!("Failed to log CREATE INDEX to WAL: {}", e);
                    }
                }
            } else if let Some(idx_type) = index_type {
                if idx_type == "gin" || idx_type == "gist" {
                    // Postgres FTS/GIN/GiST index.
                    //
                    // Accepted for syntactic compatibility (Django, Rails,
                    // and hand-written migrations emit CREATE INDEX ...
                    // USING gin) but does NOT yet build a real inverted
                    // index — the @@ operator walks the table row by row
                    // using the in-evaluator BM25 scorer. On realistic
                    // text volumes this is fine; at scale, consider the
                    // native search::bm25 API until a persistent GIN
                    // backend lands.
                    //
                    // See docs/compatibility/fts.md for the full list of
                    // behaviours we do and do not implement.
                    tracing::info!(
                        "Accepted CREATE INDEX {} USING {} ON {} ({}) — \
                         DDL-only (no backing index yet)",
                        name,
                        idx_type,
                        table_name,
                        column_name
                    );
                    persist_index_definition(storage, name, table_name, column_name, Some(idx_type.as_str()), options)?;
                    let encoded_options = encoded_index_options(options);
                    if let Err(e) = storage.log_create_index(
                        name,
                        table_name,
                        column_name,
                        Some(idx_type.as_str()),
                        &encoded_options,
                    ) {
                        tracing::warn!("Failed to log CREATE INDEX to WAL: {}", e);
                    }
                } else if idx_type == "hnsw" {
                    // Check if index already exists
                    let vector_indexes = storage.vector_indexes();
                    if vector_indexes.index_exists(name) {
                        if *if_not_exists {
                            // IF NOT EXISTS specified, return silently
                            return Ok(empty_ddl_result(executor));
                        } else {
                            // Error: index already exists
                            return Err(Error::query_execution(format!("Index '{}' already exists", name)));
                        }
                    }

                    let catalog = storage.catalog();
                    let schema = catalog.get_table_schema(table_name)?;

                    // Find the column to index
                    let column = schema.get_column(column_name).ok_or_else(|| {
                        Error::query_execution(format!("Column '{}' not found in table '{}'", column_name, table_name))
                    })?;

                    // Extract vector dimension from Vector(n) type
                    let dimension = match column.data_type {
                        crate::DataType::Vector(dim) => dim,
                        _ => {
                            return Err(Error::query_execution(format!(
                                "Column '{}' is not a vector type, cannot create HNSW index",
                                column_name
                            )))
                        }
                    };
                    let distance_metric = vector_distance_metric(options)?;
                    let existing_vectors =
                        collect_existing_vectors(storage, &schema, table_name, column_name, dimension)?;

                    // Parse quantization options
                    use crate::sql::logical_plan::{IndexOption, QuantizationType};

                    let mut quantization_type = QuantizationType::None;
                    let mut pq_subquantizers: Option<usize> = None;
                    let mut pq_centroids: Option<usize> = None;
                    let mut persistent = false;
                    #[cfg(feature = "vector-persist")]
                    let mut rerank_precision: Option<crate::vector::persistent::VectorPrecision> = None;
                    #[cfg(not(feature = "vector-persist"))]
                    let rerank_precision: Option<()> = None;

                    for option in options {
                        match option {
                            IndexOption::Quantization(qt) => quantization_type = *qt,
                            IndexOption::PqSubquantizers(n) => pq_subquantizers = Some(*n),
                            IndexOption::PqCentroids(n) => pq_centroids = Some(*n),
                            IndexOption::Persistent(enabled) => persistent = *enabled,
                            IndexOption::RerankPrecision(precision) => {
                                #[cfg(feature = "vector-persist")]
                                {
                                    rerank_precision = Some(match precision.as_str() {
                                        "f32" => crate::vector::persistent::VectorPrecision::F32,
                                        "f16" => crate::vector::persistent::VectorPrecision::F16,
                                        "i8" => crate::vector::persistent::VectorPrecision::I8,
                                        other => {
                                            return Err(Error::query_execution(format!(
                                                "Unsupported rerank_precision '{}'",
                                                other
                                            )))
                                        }
                                    });
                                }
                                #[cfg(not(feature = "vector-persist"))]
                                {
                                    let _ = precision;
                                }
                            }
                            _ => {} // Ignore other options for now
                        }
                    }

                    if persistent {
                        let training_vectors: Vec<crate::vector::Vector> =
                            existing_vectors.iter().map(|(_, vector)| vector.clone()).collect();

                        let pq_config = if quantization_type == QuantizationType::Product {
                            let mut cfg = crate::vector::ProductQuantizerConfig::default_for_dimension(dimension)
                                .map_err(|e| Error::query_execution(format!("Invalid PQ config: {}", e)))?;
                            if let Some(n) = pq_subquantizers {
                                cfg.num_subquantizers = n;
                            }
                            if let Some(n) = pq_centroids {
                                cfg.num_centroids = n;
                            }
                            cfg.validate()
                                .map_err(|e| Error::query_execution(format!("Invalid PQ config: {}", e)))?;
                            Some(cfg)
                        } else {
                            None
                        };

                        vector_indexes.create_persistent_index(
                            name.clone(),
                            table_name.clone(),
                            column_name.clone(),
                            dimension,
                            distance_metric,
                            pq_config,
                            rerank_precision,
                            &training_vectors,
                            storage.db(),
                        )?;
                        let backfilled = backfill_vector_index(vector_indexes, name, &existing_vectors)?;
                        tracing::info!(
                            "Created persistent HNSW index '{}' on table '{}' column '{}' with {} existing vectors",
                            name,
                            table_name,
                            column_name,
                            backfilled
                        );
                        persist_index_definition(
                            storage,
                            name,
                            table_name,
                            column_name,
                            Some("persistent_hnsw"),
                            options,
                        )?;

                        let encoded_options = encoded_index_options(options);
                        if let Err(e) = storage.log_create_index(
                            name,
                            table_name,
                            column_name,
                            Some("persistent_hnsw"),
                            &encoded_options,
                        ) {
                            tracing::warn!("Failed to log CREATE INDEX to WAL: {}", e);
                        }
                        return Ok(empty_ddl_result(executor));
                    }

                    // Check if we should create a quantized index
                    match quantization_type {
                        QuantizationType::Product => {
                            // Create quantized index
                            use crate::vector::ProductQuantizerConfig;

                            // Build PQ config
                            let mut pq_config = ProductQuantizerConfig::default_for_dimension(dimension)
                                .map_err(|e| Error::query_execution(format!("Invalid PQ config: {}", e)))?;

                            if let Some(n) = pq_subquantizers {
                                pq_config.num_subquantizers = n;
                            }
                            if let Some(n) = pq_centroids {
                                pq_config.num_centroids = n;
                            }

                            // Validate config
                            pq_config
                                .validate()
                                .map_err(|e| Error::query_execution(format!("Invalid PQ config: {}", e)))?;

                            let training_vectors: Vec<crate::vector::Vector> =
                                existing_vectors.iter().map(|(_, vector)| vector.clone()).collect();

                            vector_indexes.create_quantized_index(
                                name.clone(),
                                table_name.clone(),
                                column_name.clone(),
                                dimension,
                                distance_metric,
                                pq_config,
                                &training_vectors,
                            )?;
                            let backfilled = backfill_vector_index(vector_indexes, name, &existing_vectors)?;
                            tracing::info!(
                                "Created quantized HNSW index '{}' on table '{}' column '{}' with {} existing vectors",
                                name,
                                table_name,
                                column_name,
                                backfilled
                            );
                            persist_index_definition(storage, name, table_name, column_name, Some("hnsw_pq"), options)?;

                            // Log to WAL for replication
                            let encoded_options = encoded_index_options(options);
                            if let Err(e) = storage.log_create_index(
                                name,
                                table_name,
                                column_name,
                                Some("hnsw_pq"),
                                &encoded_options,
                            ) {
                                tracing::warn!("Failed to log CREATE INDEX to WAL: {}", e);
                            }
                        }
                        _ => {
                            // Create standard non-quantized index
                            vector_indexes.create_index(
                                name.clone(),
                                table_name.clone(),
                                column_name.clone(),
                                dimension,
                                distance_metric,
                            )?;
                            let backfilled = backfill_vector_index(vector_indexes, name, &existing_vectors)?;
                            tracing::info!(
                                "Created HNSW index '{}' on table '{}' column '{}' with {} existing vectors",
                                name,
                                table_name,
                                column_name,
                                backfilled
                            );
                            persist_index_definition(
                                storage,
                                name,
                                table_name,
                                column_name,
                                index_type.as_deref(),
                                options,
                            )?;

                            // Log to WAL for replication
                            let encoded_options = encoded_index_options(options);
                            if let Err(e) = storage.log_create_index(
                                name,
                                table_name,
                                column_name,
                                index_type.as_deref(),
                                &encoded_options,
                            ) {
                                tracing::warn!("Failed to log CREATE INDEX to WAL: {}", e);
                            }
                        }
                    }
                }
            }
        }

        // Return empty result set for DDL
        Ok(empty_ddl_result(executor))
    } else {
        Err(Error::query_execution("Expected CreateIndex plan node"))
    }
}

/// Handle DROP TABLE logical plan node
pub(super) fn handle_drop_table(
    executor: &Executor,
    table_name: &str,
    if_exists: bool,
) -> Result<Box<dyn PhysicalOperator>> {
    if let Some(storage) = executor.storage() {
        let catalog = storage.catalog();

        // Check if table exists
        match catalog.get_table_schema(table_name) {
            Ok(_) => {
                // Table exists - drop it
                catalog.drop_table(table_name)?;
                // KanttBan #23 (v3.31.1 phase 2): clean up the
                // identity side-table record. Best-effort; missing
                // record is fine.
                let _ = catalog.drop_identity_columns(table_name);
            }
            Err(_) => {
                // Table doesn't exist
                if !if_exists {
                    return Err(Error::query_execution(format!("Table '{}' does not exist", table_name)));
                }
                // If IF EXISTS, silently succeed
            }
        }

        // Return empty result set for DDL
        Ok(Box::new(
            ScanOperator::new(
                "".to_string(),
                Arc::new(crate::Schema { columns: vec![] }),
                None,
                vec![],
                vec![],
            )
            .with_timeout(executor.timeout_ctx()),
        ))
    } else {
        Err(Error::query_execution("No storage engine available"))
    }
}

/// Handle TRUNCATE logical plan node
pub(super) fn handle_truncate(executor: &Executor, table_name: &str) -> Result<Box<dyn PhysicalOperator>> {
    if let Some(storage) = executor.storage() {
        let catalog = storage.catalog();

        // Check if table exists
        if catalog.get_table_schema(table_name).is_err() {
            return Err(Error::query_execution(format!("Table '{}' does not exist", table_name)));
        }

        // Delete all rows from the table
        let prefix = format!("data:{}:", table_name);
        let prefix_bytes = prefix.as_bytes();
        let mut keys_to_delete = Vec::new();

        // Collect all keys for this table
        // Use total_order_seek to bypass prefix bloom filter for full table scans
        let mut read_opts = ReadOptions::default();
        read_opts.set_total_order_seek(true);
        let iter = storage.db.iterator_opt(IteratorMode::Start, read_opts);
        for item in iter {
            let (key, _) = item.map_err(|e| Error::storage(format!("Iterator error: {}", e)))?;

            if !key.starts_with(prefix_bytes) {
                if let (Some(&k), Some(&p)) = (key.first(), prefix_bytes.first()) {
                    if k > p {
                        break;
                    }
                }
                continue;
            }

            keys_to_delete.push(key.to_vec());
        }

        // Delete all collected keys
        for key in &keys_to_delete {
            storage.delete(key)?;
        }

        // R3.3: purge columnar sidecars (col:/colz:/colzm:/colp:/colpm:) with
        // the rows — stale batches or live-row presence must not resurrect
        // truncated rows. No-op prefix seeks for row-only tables.
        crate::storage::ColumnarStore::purge_table_sidecars(&storage.db, table_name)?;

        // Clear ART index entries for this table so that stale PK/UNIQUE
        // values do not block re-insertion of the same values.
        // Skip clearing if branches exist or time-travel snapshots are
        // retained, because branch data and snapshots may still
        // reference the indexed values.
        // Check for user-created branches (exclude the auto-created "main" branch).
        // Branch data uses separate key prefixes and does not share the ART index,
        // but as a safety measure we skip clearing when user branches exist.
        let has_user_branches = storage
            .list_branches()
            .map(|b| b.iter().any(|br| br.name != "main"))
            .unwrap_or(false);
        if !has_user_branches {
            storage.art_indexes().clear_table_indexes(table_name);
        }

        // Log to WAL for replication
        if let Err(e) = storage.log_truncate(table_name) {
            tracing::warn!("Failed to log TRUNCATE to WAL: {}", e);
        }

        // Return empty result set for DDL
        Ok(Box::new(
            ScanOperator::new(
                "".to_string(),
                Arc::new(crate::Schema { columns: vec![] }),
                None,
                vec![],
                vec![],
            )
            .with_timeout(executor.timeout_ctx()),
        ))
    } else {
        Err(Error::query_execution("No storage engine available"))
    }
}

// =============================================================================
// HA Operations (ha-tier1 feature)
// =============================================================================

/// Handle SWITCHOVER to target node
/// Example: SELECT helios_switchover('node-uuid')
#[cfg(feature = "ha-tier1")]
pub(super) fn handle_switchover(_executor: &Executor, target_node: &str) -> Result<Box<dyn PhysicalOperator>> {
    use crate::replication::ha_state::ha_state;
    use crate::replication::topology_manager;
    use uuid::Uuid;

    // Resolve target node (can be alias or UUID)
    let target_uuid = topology_manager()
        .resolve_node_id(target_node)
        .or_else(|| {
            // Fallback: try parsing as UUID directly if not in topology
            Uuid::parse_str(target_node).ok()
        })
        .ok_or_else(|| {
            Error::query_execution(format!(
                "Target node '{}' not found. Specify a valid node alias or UUID.",
                target_node
            ))
        })?;

    // Get HA state registry
    let ha_registry = ha_state();

    // Check if this node is primary
    if ha_registry.get_role() != crate::replication::ha_state::HARole::Primary {
        return Err(Error::query_execution(
            "Switchover can only be initiated from the primary node",
        ));
    }

    // Check if target standby exists and is healthy
    let standbys = ha_registry.get_standbys();
    let target_standby = standbys.iter().find(|s| s.node_id == target_uuid);

    if target_standby.is_none() {
        return Err(Error::query_execution(format!(
            "Target standby '{}' ({}) not found or not connected",
            target_node, target_uuid
        )));
    }

    // Get the display name for user feedback
    let display_name = topology_manager()
        .get_node(target_uuid)
        .map(|n| n.display_name())
        .unwrap_or_else(|| target_node.to_string());

    // For now, return a message indicating switchover would be initiated
    // Full implementation requires async coordination with SwitchoverCoordinator
    let msg = format!(
        "Switchover to node {} ({}) initiated. This is a placeholder - full async switchover requires runtime integration.",
        display_name, target_uuid
    );

    Ok(Box::new(super::StatusMessageOperator::new(msg)))
}

/// Handle SWITCHOVER CHECK to validate preconditions
/// Example: SELECT helios_switchover_check('node-uuid') or SELECT helios_switchover_check('alias')
#[cfg(feature = "ha-tier1")]
pub(super) fn handle_switchover_check(_executor: &Executor, target_node: &str) -> Result<Box<dyn PhysicalOperator>> {
    use crate::replication::ha_state::ha_state;
    use crate::replication::topology_manager;
    use crate::{Column, DataType, Schema, Tuple, Value};
    use uuid::Uuid;

    // Resolve target node (can be alias or UUID)
    let target_uuid = topology_manager()
        .resolve_node_id(target_node)
        .or_else(|| {
            // Fallback: try parsing as UUID directly if not in topology
            Uuid::parse_str(target_node).ok()
        })
        .ok_or_else(|| {
            Error::query_execution(format!(
                "Target node '{}' not found. Specify a valid node alias or UUID.",
                target_node
            ))
        })?;

    // Get HA state registry
    let ha_registry = ha_state();

    // Build check result
    let mut can_proceed = true;
    let mut target_healthy = false;
    let mut target_lsn: u64 = 0;
    let primary_lsn = ha_registry.get_lsn();
    let mut warnings = Vec::new();
    let mut blockers = Vec::new();

    // Check if this node is primary
    if ha_registry.get_role() != crate::replication::ha_state::HARole::Primary {
        can_proceed = false;
        blockers.push("This node is not the primary".to_string());
    }

    // Check target standby
    let standbys = ha_registry.get_standbys();
    if let Some(standby) = standbys.iter().find(|s| s.node_id == target_uuid) {
        target_healthy = true;
        target_lsn = standby.apply_lsn;

        let lag = primary_lsn.saturating_sub(target_lsn);
        if lag > 0 {
            warnings.push(format!("Target standby is {} LSN behind", lag));
        }
    } else {
        can_proceed = false;
        blockers.push(format!("Target node {} ({}) not found", target_node, target_uuid));
    }

    let lag_bytes = primary_lsn.saturating_sub(target_lsn) as i64;

    // Create result tuple
    let schema = Arc::new(Schema {
        columns: vec![
            Column::new("can_proceed", DataType::Boolean),
            Column::new("target_healthy", DataType::Boolean),
            Column::new("target_lsn", DataType::Int8),
            Column::new("primary_lsn", DataType::Int8),
            Column::new("lag_bytes", DataType::Int8),
            Column::new("warnings", DataType::Text),
            Column::new("blockers", DataType::Text),
        ],
    });

    let tuple = Tuple::new(vec![
        Value::Boolean(can_proceed),
        Value::Boolean(target_healthy),
        Value::Int8(target_lsn as i64),
        Value::Int8(primary_lsn as i64),
        Value::Int8(lag_bytes),
        Value::String(warnings.join("; ")),
        Value::String(blockers.join("; ")),
    ]);

    Ok(Box::new(SingleTupleOperator::new(tuple, schema)))
}

/// Handle CLUSTER STATUS query
/// Example: SELECT * FROM helios_cluster_status()
#[cfg(feature = "ha-tier1")]
pub(super) fn handle_cluster_status(_executor: &Executor) -> Result<Box<dyn PhysicalOperator>> {
    use crate::replication::ha_state::{ha_state, HARole};
    use crate::{Column, DataType, Schema, Tuple, Value};

    let ha_registry = ha_state();

    let schema = Arc::new(Schema {
        columns: vec![
            Column::new("node_id", DataType::Text),
            Column::new("role", DataType::Text),
            Column::new("address", DataType::Text),
            Column::new("is_healthy", DataType::Boolean),
            Column::new("lsn", DataType::Int8),
            Column::new("lag_ms", DataType::Int8),
            Column::new("priority", DataType::Int4),
        ],
    });

    let mut tuples = Vec::new();

    // Add primary info if available
    if let Some(config) = ha_registry.get_config() {
        let role_str = match ha_registry.get_role() {
            HARole::Primary => "primary",
            HARole::Standby => "standby",
            HARole::Standalone => "standalone",
            HARole::Observer => "observer",
        };

        tuples.push(Tuple::new(vec![
            Value::String(config.node_id.to_string()),
            Value::String(role_str.to_string()),
            Value::String(config.listen_addr.clone()),
            Value::Boolean(true), // Local node is always "healthy" from its perspective
            Value::Int8(ha_registry.get_lsn() as i64),
            Value::Int8(0),   // No lag for self
            Value::Int4(100), // Default priority - config doesn't store priority yet
        ]));
    }

    // Add standby info
    for standby in ha_registry.get_standbys() {
        tuples.push(Tuple::new(vec![
            Value::String(standby.node_id.to_string()),
            Value::String("standby".to_string()),
            Value::String(standby.address.clone()),
            Value::Boolean(true), // Connected standbys are healthy
            Value::Int8(standby.apply_lsn as i64),
            Value::Int8(standby.lag_ms as i64),
            Value::Int4(0), // Priority not stored in StandbyInfo yet
        ]));
    }

    Ok(Box::new(MultiTupleOperator::new(tuples, schema)))
}

/// Single tuple operator for returning one result row
#[cfg(feature = "ha-tier1")]
struct SingleTupleOperator {
    tuple: Option<crate::Tuple>,
    schema: Arc<crate::Schema>,
}

#[cfg(feature = "ha-tier1")]
impl SingleTupleOperator {
    fn new(tuple: crate::Tuple, schema: Arc<crate::Schema>) -> Self {
        Self {
            tuple: Some(tuple),
            schema,
        }
    }
}

#[cfg(feature = "ha-tier1")]
impl super::PhysicalOperator for SingleTupleOperator {
    fn next(&mut self) -> Result<Option<crate::Tuple>> {
        Ok(self.tuple.take())
    }

    fn schema(&self) -> Arc<crate::Schema> {
        self.schema.clone()
    }
}

/// Multi tuple operator for returning multiple result rows
#[cfg(feature = "ha-tier1")]
struct MultiTupleOperator {
    tuples: std::collections::VecDeque<crate::Tuple>,
    schema: Arc<crate::Schema>,
}

#[cfg(feature = "ha-tier1")]
impl MultiTupleOperator {
    fn new(tuples: Vec<crate::Tuple>, schema: Arc<crate::Schema>) -> Self {
        Self {
            tuples: tuples.into_iter().collect(),
            schema,
        }
    }
}

#[cfg(feature = "ha-tier1")]
impl super::PhysicalOperator for MultiTupleOperator {
    fn next(&mut self) -> Result<Option<crate::Tuple>> {
        Ok(self.tuples.pop_front())
    }

    fn schema(&self) -> Arc<crate::Schema> {
        self.schema.clone()
    }
}

/// Handle SET NODE ALIAS command
#[cfg(feature = "ha-tier1")]
pub(super) fn handle_set_node_alias(
    _executor: &Executor,
    node_id: &str,
    alias: &Option<String>,
) -> Result<Box<dyn PhysicalOperator>> {
    use crate::replication::topology_manager;
    use crate::{Column, DataType, Schema, Tuple, Value};
    use uuid::Uuid;

    let topology = topology_manager();

    // Resolve the node_id (could be existing alias or UUID)
    let target_uuid = topology
        .resolve_node_id(node_id)
        .or_else(|| Uuid::parse_str(node_id).ok())
        .ok_or_else(|| {
            Error::query_execution(format!(
                "Node '{}' not found in cluster topology. Use SHOW TOPOLOGY to see available nodes.",
                node_id
            ))
        })?;

    // Set or clear the alias
    let result_msg = if let Some(ref new_alias) = alias {
        // Validate alias format (no spaces, not a valid UUID pattern)
        if new_alias.contains(' ') {
            return Err(Error::query_execution("Alias cannot contain spaces"));
        }
        if Uuid::parse_str(new_alias).is_ok() {
            return Err(Error::query_execution("Alias cannot be a valid UUID format"));
        }
        if new_alias.is_empty() {
            return Err(Error::query_execution("Alias cannot be empty"));
        }

        if !topology.set_alias(target_uuid, Some(new_alias.clone())) {
            return Err(Error::query_execution(format!(
                "Failed to set alias: '{}' is already in use by another node",
                new_alias
            )));
        }

        format!("Node alias '{}' set for node '{}'", new_alias, target_uuid)
    } else {
        // Clearing alias - this should always succeed if the node exists
        topology.set_alias(target_uuid, None);
        format!("Node alias removed for node '{}'", target_uuid)
    };

    let schema = Arc::new(Schema {
        columns: vec![Column::new("result", DataType::Text)],
    });

    let tuple = Tuple::new(vec![Value::String(result_msg)]);

    Ok(Box::new(SingleTupleOperator::new(tuple, schema)))
}

/// Handle SHOW TOPOLOGY command - displays detailed cluster topology
#[cfg(feature = "ha-tier1")]
pub(super) fn handle_show_topology(_executor: &Executor) -> Result<Box<dyn PhysicalOperator>> {
    use crate::replication::ha_state::{ha_state, HARole};
    use crate::replication::topology_manager;
    use crate::{Column, DataType, Schema, Tuple, Value};

    let ha_registry = ha_state();
    let topology = topology_manager();

    let schema = Arc::new(Schema {
        columns: vec![
            Column::new("node_id", DataType::Text),
            Column::new("alias", DataType::Text),
            Column::new("role", DataType::Text),
            Column::new("client_addr", DataType::Text),
            Column::new("replication_addr", DataType::Text),
            Column::new("healthy", DataType::Boolean),
            Column::new("health_msg", DataType::Text),
            Column::new("last_seen_secs", DataType::Int8),
            Column::new("lsn", DataType::Int8),
            Column::new("lag_ms", DataType::Int8),
            Column::new("priority", DataType::Int4),
            Column::new("weight", DataType::Int4),
        ],
    });

    let mut tuples = Vec::new();

    // Helper to get alias for a node
    let get_alias = |node_id: uuid::Uuid| -> Value {
        topology
            .get_node(node_id)
            .and_then(|n| n.alias.clone())
            .map(Value::String)
            .unwrap_or(Value::Null)
    };

    // Helper to get node info from topology
    let get_topology_info = |node_id: uuid::Uuid| -> (u32, u32, Option<String>) {
        topology
            .get_node(node_id)
            .map(|n| (n.priority, n.weight, n.health_message.clone()))
            .unwrap_or((100, 100, None))
    };

    // Add local node info
    if let Some(config) = ha_registry.get_config() {
        let role_str = match ha_registry.get_role() {
            HARole::Primary => "Primary",
            HARole::Standby => "Standby",
            HARole::Standalone => "Standalone",
            HARole::Observer => "Observer",
        };

        let alias = get_alias(config.node_id);
        let (priority, weight, health_msg) = get_topology_info(config.node_id);

        tuples.push(Tuple::new(vec![
            Value::String(config.node_id.to_string()),
            alias,
            Value::String(role_str.to_string()),
            Value::String(config.listen_addr.clone()),
            Value::String(format!("{}:{}", config.listen_addr, config.replication_port)),
            Value::Boolean(true), // Local node is always "healthy" from its perspective
            Value::String(health_msg.unwrap_or_else(|| "OK".to_string())),
            Value::Int8(0), // last seen
            Value::Int8(ha_registry.get_lsn() as i64),
            Value::Int8(0), // No lag for self
            Value::Int4(priority as i32),
            Value::Int4(weight as i32),
        ]));
    }

    // Add standby info from HA registry, enriched with topology data
    for standby in ha_registry.get_standbys() {
        let alias = get_alias(standby.node_id);
        let (priority, weight, health_msg) = get_topology_info(standby.node_id);

        tuples.push(Tuple::new(vec![
            Value::String(standby.node_id.to_string()),
            alias,
            Value::String("Standby".to_string()),
            Value::String(standby.address.clone()),
            Value::String(standby.address.clone()), // replication addr same as client for now
            Value::Boolean(true),                   // Connected standbys are healthy
            Value::String(health_msg.unwrap_or_else(|| "Connected".to_string())),
            Value::Int8(0), // last seen
            Value::Int8(standby.apply_lsn as i64),
            Value::Int8(standby.lag_ms as i64),
            Value::Int4(priority as i32),
            Value::Int4(weight as i32),
        ]));
    }

    Ok(Box::new(MultiTupleOperator::new(tuples, schema)))
}