akar-binder 0.2.3

//! Binder implementation ΓÇö resolves symbols and validates semantics.

#![allow(clippy::collapsible_if, clippy::never_loop)]

mod ddl;

use crate::bound_statement::*;
use akar_catalog::{Catalog, CatalogColumn, CatalogResult, IndexType};
use akar_common::error::BinderError;
use akar_common::types::LogicalTypeID;
use akar_parser::ast::{Clause, Expression, Statement, *};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

/// Resolve SET clause items against the catalog to find column info.
fn resolve_set_items(catalog: &Catalog, items: &[SetItem]) -> Result<Vec<BoundSetItem>, BinderError> {
    let mut result = Vec::new();
    for item in items {
        // Expect property expression like `n.property_name = value`
        match &item.property {
            Expression::PropertyAccess(obj, prop_name) => {
                // Find the variable name by looking at the object
                let _var_name = match obj.as_ref() {
                    Expression::Variable(v) => v.clone(),
                    other => return Err(format!("Unsupported SET target: {:?}", other).into()),
                };
                // Look up the column in the table schema
                // We need to find which table this variable belongs to.
                // Since MERGE is a single-pattern operation, we just use the label.
                let found = catalog.all_entries().find_map(|entry| {
                    entry.columns().iter().find(|c| c.name == *prop_name).map(|_| {
                        let is_node = entry.is_node_table();
                        (entry.name().to_string(), entry.table_id(), is_node)
                    })
                });
                match found {
                    Some((table_name, table_id, is_node)) => {
                        let col_idx = catalog
                            .get_entry_by_name(&table_name)
                            .and_then(|e| e.columns().iter().position(|c| c.name == *prop_name))
                            .unwrap_or(0);
                        result.push(BoundSetItem {
                            property: item.property.clone(),
                            value: item.value.clone(),
                            column_name: prop_name.clone(),
                            column_idx: col_idx,
                            table_name: table_name.to_string(),
                            table_id,
                            is_node,
                        });
                    }
                    None => {
                        return Err(format!("Property '{}' not found in any table", prop_name).into());
                    }
                }
            }
            _ => return Err(format!("Expected property assignment in SET, got: {:?}", item.property).into()),
        }
    }
    Ok(result)
}

/// The binder transforms a parsed AST into a bound statement
/// by resolving symbols against the catalog and validating types.
pub struct Binder {
    catalog: Arc<Mutex<Catalog>>,
}

impl Binder {
    pub fn new(catalog: Arc<Mutex<Catalog>>) -> Self {
        Self { catalog }
    }

    pub fn bind(&self, statement: Statement) -> Result<BoundStatement, BinderError> {
        match statement {
            Statement::Query(query) => self.bind_query(query),
            Statement::CreateNodeTable(t) => self.bind_create_node_table(t),
            Statement::CreateRelTable(t) => self.bind_create_rel_table(t),
            Statement::DropTable(t) => self.bind_drop_table(t),
            Statement::CopyFrom(c) => self.bind_copy_from(c),
            Statement::CopyTo(c) => self.bind_copy_to(c),
            Statement::AlterTable(a) => self.bind_alter_table(a),
            Statement::CreateVectorIndex(v) => self.bind_create_vector_index(v),
            Statement::CreateIndex(v) => self.bind_create_index(v),
            Statement::DropIndex(v) => self.bind_drop_index(v),
            Statement::Union(u) => self.bind_union(u),
            Statement::Merge(m) => self.bind_merge(m),
            Statement::StandaloneCall(c) => self.bind_standalone_call(c),
            Statement::CreateDml(c) => self.bind_create_dml(c, &[]),
            Statement::Explain(e) => self.bind_explain(e),
            Statement::CreateSequence(s) => self.bind_create_sequence(s),
            Statement::DropSequence(s) => self.bind_drop_sequence(s),
            Statement::CreateMacro(m) => self.bind_create_macro(m),
            Statement::ExportDatabase(e) => self.bind_export_database(e),
            Statement::ImportDatabase(i) => self.bind_import_database(i),
            Statement::Analyze(a) => self.bind_analyze(a),
            Statement::CreateFtsIndex(f) => self.bind_create_fts_index(f),
            Statement::Transaction(t) => self.bind_transaction(t),
            Statement::Extension(e) => self.bind_extension(e),
            Statement::AttachDatabase(a) => self.bind_attach_database(a),
            Statement::DetachDatabase(d) => self.bind_detach_database(d),
            Statement::UseDatabase(u) => self.bind_use_database(u),
            Statement::LoadFrom(l) => self.bind_load_from(l),
            Statement::CreateType(t) => self.bind_create_type(t),
            Statement::CommentOnTable(c) => self.bind_comment_on_table(c),
            Statement::CreateGraph(g) => self.bind_create_graph(g),
            Statement::UseGraph(g) => self.bind_use_graph(g),
            Statement::DropGraph(g) => self.bind_drop_graph(g),
        }
    }

    /// Map a string type name to LogicalTypeID.
    pub fn parse_type(type_name: &str) -> Result<LogicalTypeID, BinderError> {
        let upper = type_name.to_uppercase();

        // Handle compound types with no child-type tracking (parse only).
        // An array suffix (`[]` or `[N]`) — any primitive followed by one or
        // more bracket groups — maps to List; the dimension is not tracked at
        // engine level (P80, mirrors the Python translator). Arrays are the
        // only bracket form, so this never collides with MAP/STRUCT/UNION (()
        // delimited) or a scalar primitive (no brackets).
        if upper.contains('[') && upper.ends_with(']') {
            return Ok(LogicalTypeID::List);
        }
        if upper.starts_with("MAP(") {
            return Ok(LogicalTypeID::Map);
        }
        if upper.starts_with("STRUCT(") {
            return Ok(LogicalTypeID::Struct);
        }
        if upper.starts_with("UNION(") {
            return Ok(LogicalTypeID::Union);
        }

        match upper.as_str() {
            "BOOL" | "BOOLEAN" => Ok(LogicalTypeID::Bool),
            "INT64" => Ok(LogicalTypeID::Int64),
            "INT32" => Ok(LogicalTypeID::Int32),
            "INT16" => Ok(LogicalTypeID::Int16),
            "INT8" => Ok(LogicalTypeID::Int8),
            "UINT64" => Ok(LogicalTypeID::UInt64),
            "UINT32" => Ok(LogicalTypeID::UInt32),
            "UINT16" => Ok(LogicalTypeID::UInt16),
            "UINT8" => Ok(LogicalTypeID::UInt8),
            "DOUBLE" => Ok(LogicalTypeID::Double),
            "FLOAT" => Ok(LogicalTypeID::Float),
            "STRING" => Ok(LogicalTypeID::String),
            "BLOB" => Ok(LogicalTypeID::Blob),
            "DATE" => Ok(LogicalTypeID::Date),
            "TIMESTAMP" | "TIMESTAMP_MS" => Ok(LogicalTypeID::Timestamp),
            "TIMESTAMP_SEC" => Ok(LogicalTypeID::TimestampSec),
            "TIMESTAMP_NS" => Ok(LogicalTypeID::TimestampNs),
            "TIMESTAMP_TZ" => Ok(LogicalTypeID::TimestampTz),
            "INTERVAL" => Ok(LogicalTypeID::Interval),
            "SERIAL" => Ok(LogicalTypeID::Serial),
            "UINT128" => Ok(LogicalTypeID::UInt128),
            "JSON" => Ok(LogicalTypeID::Json),
            "TIME" | "DTIME" => Ok(LogicalTypeID::Time),
            _ => Err(format!("Unknown type: {type_name}").into()),
        }
    }

    /// Map a string type name to `LogicalTypeID`, resolving user-defined type
    /// aliases (P84). Follows alias chains (alias-of-alias); falls back to the
    /// plain builtin `parse_type` error when no alias matches.
    pub fn parse_type_resolved(&self, type_name: &str) -> Result<LogicalTypeID, BinderError> {
        if let Ok(t) = Self::parse_type(type_name) {
            return Ok(t);
        }
        let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        let base = catalog
            .resolve_type_alias(type_name)
            .ok_or_else(|| format!("Unknown type: {type_name}"))?;
        Self::parse_type(&base)
    }

    /// Parse compression option.
    pub fn parse_compression(comp: Option<&str>) -> Result<akar_common::enums::CompressionType, BinderError> {
        use akar_common::enums::CompressionType;
        match comp {
            None => Ok(CompressionType::Uncompressed), // Or default based on type
            Some(s) => match s.to_uppercase().as_str() {
                "UNCOMPRESSED" => Ok(CompressionType::Uncompressed),
                "CONSTANT" => Ok(CompressionType::Constant),
                "ONEVALUE" => Ok(CompressionType::OneValue),
                "BOOLEAN" => Ok(CompressionType::Boolean),
                "INTEGER_BITPACKING" => Ok(CompressionType::IntegerBitpacking),
                "STRING_DICTIONARY" => Ok(CompressionType::StringDictionary),
                "FLOAT" => Ok(CompressionType::Float),
                "LIST_DELTA" => Ok(CompressionType::ListDelta),
                _ => Err(format!("Unknown compression type: {s}").into()),
            },
        }
    }

    // ==================== Query Binding ====================

    fn bind_query(&self, query: Query) -> Result<BoundStatement, BinderError> {
        let mut clauses = Vec::new();
        let mut variables: Vec<BoundVariable> = Vec::new();

        for clause in query.clauses {
            let (bound_clause, new_vars) = match clause {
                Clause::Match(m) => {
                    let (bound, vars) = self.bind_match(&m, &variables)?;
                    (BoundClause::BoundMatch(bound), vars)
                }
                Clause::Return(r) => {
                    let bound = self.bind_return(&r, &variables)?;
                    (BoundClause::BoundReturn(bound), Vec::new())
                }
                Clause::With(r) => {
                    let bound = self.bind_return(&r, &variables)?;
                    // P53.18: WITH resets the scope to its projection. Each
                    // projected bare variable (carried through) and each alias
                    // (aggregate/computed) becomes an in-scope variable for the
                    // following clauses — e.g. `WITH m, COUNT(r) AS cnt
                    // WHERE cnt < $n RETURN cnt`. Previously the alias list was
                    // discarded, so the trailing WHERE/RETURN failed with
                    // `Variable 'cnt' not in scope`.
                    let projected: Vec<BoundVariable> = bound
                        .expressions
                        .iter()
                        .filter_map(|be| {
                            if let Expression::Variable(v) = &be.expression {
                                if let Some(existing) = variables.iter().find(|x| &x.name == v) {
                                    let alias = be.alias.clone().unwrap_or_else(|| v.clone());
                                    return Some(BoundVariable {
                                        name: alias,
                                        table_id: existing.table_id,
                                        label: existing.label.clone(),
                                        is_node: existing.is_node,
                                    });
                                }
                            }
                            be.alias.as_ref().map(|alias| BoundVariable {
                                name: alias.clone(),
                                table_id: 0,
                                label: None,
                                is_node: false,
                            })
                        })
                        .collect::<Vec<_>>();
                    (BoundClause::BoundWith(bound), projected)
                }
                Clause::Where(w) => {
                    let bound = self.bind_where(&w, &variables)?;
                    (BoundClause::BoundWhere(bound), Vec::new())
                }
                Clause::Create(c) => {
                    let (bound, vars) = self.bind_match_create(&c, &variables)?;
                    (BoundClause::BoundCreate(bound), vars)
                }
                Clause::Delete(d) => {
                    let bound = self.bind_delete(&d, &variables)?;
                    (BoundClause::BoundDelete(bound), Vec::new())
                }
                Clause::Set(s) => {
                    let bound = self.bind_set(&s, &variables)?;
                    (BoundClause::BoundSet(bound), Vec::new())
                }
                Clause::Unwind(u) => {
                    let bound = self.bind_unwind(&u)?;
                    let new_var = BoundVariable {
                        name: bound.variable.clone(),
                        table_id: 0,
                        label: None,
                        is_node: false,
                    };
                    (BoundClause::BoundUnwind(bound), vec![new_var])
                }
                Clause::Foreach(f) => {
                    let bound = self.bind_foreach(&f, &variables)?;
                    let new_var = BoundVariable {
                        name: bound.variable.clone(),
                        table_id: 0,
                        label: None,
                        is_node: false,
                    };
                    (BoundClause::BoundForeach(bound), vec![new_var])
                }
                Clause::OptionalMatch(m) => {
                    let (bound, vars) = self.bind_optional_match(&m, &variables)?;
                    (BoundClause::BoundOptionalMatch(bound), vars)
                }
                Clause::Merge(m) => {
                    let (bound, vars) = self.bind_merge_clause(&m, &variables)?;
                    (BoundClause::BoundMerge(bound), vars)
                }
            };
            if matches!(bound_clause, BoundClause::BoundWith(_)) {
                // WITH resets the scope to its projection (P53.18): only the
                // projected variables/aliases remain in scope afterwards.
                variables = new_vars;
            } else {
                variables.extend(new_vars);
            }
            clauses.push(bound_clause.clone());

            // Generate implicit WHERE clauses from inline properties for MATCH and CREATE
            if let BoundClause::BoundMatch(bound) = &bound_clause {
                let mut inline_exprs = Vec::new();
                for pattern in &bound.patterns {
                    if let Some(node_var) = &pattern.node_variable {
                        for (key, val_expr) in &pattern.properties {
                            let prop_access = akar_parser::ast::Expression::PropertyAccess(
                                Box::new(akar_parser::ast::Expression::Variable(node_var.clone())),
                                key.clone(),
                            );
                            let equals = akar_parser::ast::Expression::BinaryOp(
                                akar_parser::ast::BinaryOp::Equal,
                                Box::new(prop_access),
                                Box::new(val_expr.clone()),
                            );
                            inline_exprs.push(equals);
                        }
                    }
                    if let Some(edge) = &pattern.edge {
                        if let Some(edge_var) = &edge.variable {
                            for (key, val_expr) in &edge.properties {
                                let prop_access = akar_parser::ast::Expression::PropertyAccess(
                                    Box::new(akar_parser::ast::Expression::Variable(edge_var.clone())),
                                    key.clone(),
                                );
                                let equals = akar_parser::ast::Expression::BinaryOp(
                                    akar_parser::ast::BinaryOp::Equal,
                                    Box::new(prop_access),
                                    Box::new(val_expr.clone()),
                                );
                                inline_exprs.push(equals);
                            }
                        }
                    }
                }

                if !inline_exprs.is_empty() {
                    let combined = inline_exprs
                        .into_iter()
                        .reduce(|acc, e| {
                            akar_parser::ast::Expression::BinaryOp(
                                akar_parser::ast::BinaryOp::And,
                                Box::new(acc),
                                Box::new(e),
                            )
                        })
                        .unwrap();

                    let bound_expr = self.resolve_expression(&combined, &variables)?;
                    clauses.push(BoundClause::BoundWhere(BoundWhereClause { expression: bound_expr }));
                }
            }
        }

        Ok(BoundStatement::BoundQuery(BoundQuery { clauses, variables }))
    }

    // ==================== MATCH Binding ====================

    fn bind_match(
        &self,
        m: &MatchClause,
        existing_vars: &[BoundVariable],
    ) -> Result<(BoundMatchClause, Vec<BoundVariable>), BinderError> {
        let mut patterns = Vec::new();
        let mut new_vars = Vec::new();

        for pattern in &m.patterns {
            let all_vars: Vec<BoundVariable> = existing_vars.iter().cloned().chain(new_vars.iter().cloned()).collect();
            let (bound, nv) = self.bind_pattern(pattern, &all_vars, false)?;
            patterns.push(bound);
            new_vars.extend(nv);
        }

        // Bind optional FTS query
        let fts_query = match m.fts_query.as_ref().map(|fq| -> Result<BoundFtsQuery, String> {
            let catalog = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
            let (table_name, column_name) = catalog
                .get_fts_index(&fq.index_name)
                .map(|(t, c)| (t.to_string(), c.to_string()))
                .unwrap_or_default();
            Ok(BoundFtsQuery {
                index_name: fq.index_name.clone(),
                query_string: fq.query_string.clone(),
                table_name,
                column_name,
            })
        }) {
            Some(r) => Some(r?),
            None => None,
        };

        Ok((
            BoundMatchClause {
                patterns,
                new_variables: new_vars.clone(),
                fts_query,
            },
            new_vars,
        ))
    }

    fn bind_pattern(
        &self,
        pattern: &Pattern,
        existing_vars: &[BoundVariable],
        allow_existing: bool,
    ) -> Result<(BoundPattern, Vec<BoundVariable>), BinderError> {
        let mut new_vars = Vec::new();
        let mut node_table_id = None;
        let mut bound_edge = None;

        // Resolve node
        let (node_var, node_label) = if let Some(ref n) = pattern.node {
            let var = n.variable.clone();
            let label = n.labels.first().cloned();

            // Look up in catalog
            if let Some(ref lbl) = label {
                let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                match catalog.get_entry_by_name(lbl) {
                    Some(entry) if entry.is_node_table() => {
                        node_table_id = Some(entry.table_id());
                    }
                    Some(_entry) => {
                        return Err(format!("'{}' is not a node table", lbl).into());
                    }
                    None => {
                        return Err(format!("Table '{}' not found", lbl).into());
                    }
                }
            }

            // Check for duplicate variable names
            if let Some(ref v) = var {
                if let Some(existing) = existing_vars.iter().find(|bv| bv.name == *v) {
                    // Reusing a variable is allowed when it refers to the same node table
                    // (e.g. `MATCH (a)-[:r1]->(b), (a)-[:r2]->(c)` — `a` is the shared node),
                    // or when the pattern node has no label and names an already-bound
                    // node variable (P53.21: `OPTIONAL MATCH (a)-[existing:Connected]-(b)`
                    // and `MERGE (a)-[r:Connected]->(b)` reuse `a`/`b` from a prior MATCH).
                    let same_node = allow_existing
                        || (existing.is_node
                            && (label.is_none()
                                || (node_table_id.is_some() && existing.table_id == node_table_id.unwrap_or(0))));
                    if same_node {
                        // Reference to already-bound variable (e.g. in CREATE after MATCH,
                        // or the shared node of a multi-pattern MATCH).
                        // Use the existing variable's table_id if we didn't resolve one
                        if node_table_id.is_none() {
                            node_table_id = Some(existing.table_id);
                        }
                        // Don't add to new_vars — it's a reference, not a new binding
                    } else {
                        return Err(format!("Variable '{}' already defined", v).into());
                    }
                } else {
                    new_vars.push(BoundVariable {
                        name: var.clone().unwrap_or_else(|| "_anon_".to_string()),
                        table_id: node_table_id.unwrap_or(0),
                        label: label.clone(),
                        is_node: true,
                    });
                }
            } else {
                new_vars.push(BoundVariable {
                    name: "_anon_".to_string(),
                    table_id: node_table_id.unwrap_or(0),
                    label: label.clone(),
                    is_node: true,
                });
            }

            (var, label)
        } else {
            (None, None)
        };

        // Resolve edge
        if let Some(ref e) = pattern.edge {
            let edge_var = e.variable.clone();
            let edge_label = e.labels.first().cloned();
            let mut rel_table_id = None;

            if let Some(ref lbl) = edge_label {
                let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                match catalog.get_entry_by_name(lbl) {
                    Some(entry) if entry.is_rel_table() => {
                        rel_table_id = Some(entry.table_id());
                    }
                    Some(_) => {
                        return Err(format!("'{}' is not a rel table", lbl).into());
                    }
                    None => {
                        return Err(format!("Rel table '{}' not found", lbl).into());
                    }
                }
            }

            if let Some(ref v) = edge_var {
                if existing_vars.iter().any(|bv| bv.name == *v) || new_vars.iter().any(|bv| bv.name == *v) {
                    return Err(format!("Variable '{}' already defined", v).into());
                }
            }

            new_vars.push(BoundVariable {
                name: edge_var.clone().unwrap_or_else(|| "_anon_edge_".to_string()),
                table_id: rel_table_id.unwrap_or(0),
                label: edge_label.clone(),
                is_node: false,
            });

            bound_edge = Some(BoundEdgePattern {
                variable: e.variable.clone(),
                label: edge_label,
                rel_table_id,
                direction: e.direction.clone(),
                properties: e.properties.clone(),
                lower_bound: e.lower_bound,
                upper_bound: e.upper_bound,
            });
        }

        Ok((
            BoundPattern {
                node_variable: node_var,
                node_label,
                node_table_id,
                properties: pattern.node.as_ref().map(|n| n.properties.clone()).unwrap_or_default(),
                edge: bound_edge,
            },
            new_vars,
        ))
    }

    // ==================== RETURN Binding ====================

    fn bind_return(&self, r: &ReturnClause, variables: &[BoundVariable]) -> Result<BoundReturnClause, BinderError> {
        let mut expressions = Vec::new();
        for item in &r.expressions {
            match &item.expression {
                Expression::Star => {
                    // Expand * to all variables in scope
                    if variables.is_empty() {
                        return Err("RETURN or WITH * is not allowed when there are no variables in scope.".into());
                    }
                    for var in variables {
                        expressions.push(BoundExpression {
                            expression: Expression::Variable(var.name.clone()),
                            resolved_type: if var.is_node {
                                LogicalTypeID::Node
                            } else {
                                LogicalTypeID::Rel
                            },
                            is_constant: false,
                            alias: None,
                        });
                    }
                }
                _ => {
                    let mut resolved = self.resolve_expression(&item.expression, variables)?;
                    resolved.alias = item.alias.clone();
                    expressions.push(resolved);
                }
            }
        }
        // Bind ORDER BY items. Sort keys may reference RETURN/WITH aliases
        // (e.g. `RETURN count(m) AS cnt ORDER BY cnt`); alias shadows any
        // scope variable of the same name (P53.16).
        let alias_types: Vec<(String, LogicalTypeID)> = expressions
            .iter()
            .filter_map(|be| be.alias.clone().map(|a| (a, be.resolved_type)))
            .collect();
        let order_by = r
            .order_by
            .as_ref()
            .map(|items| {
                items
                    .iter()
                    .map(|item| {
                        let resolved = match &item.expression {
                            Expression::Variable(name) => match alias_types.iter().find(|(alias, _)| alias == name) {
                                Some((alias, typ)) => BoundExpression {
                                    expression: Expression::Variable(alias.clone()),
                                    resolved_type: *typ,
                                    is_constant: false,
                                    alias: None,
                                },
                                None => self.resolve_expression(&item.expression, variables)?,
                            },
                            _ => self.resolve_expression(&item.expression, variables)?,
                        };
                        Ok(crate::bound_statement::BoundOrderByItem {
                            expression: resolved,
                            ascending: item.ascending,
                        })
                    })
                    .collect::<Result<Vec<_>, BinderError>>()
            })
            .transpose()?;
        Ok(BoundReturnClause {
            expressions,
            distinct: r.distinct,
            order_by,
            limit: r.limit,
            skip: r.skip,
            limit_param: r.limit_param.clone(),
            skip_param: r.skip_param.clone(),
        })
    }

    // ==================== WHERE Binding ====================

    fn bind_where(&self, w: &WhereClause, variables: &[BoundVariable]) -> Result<BoundWhereClause, BinderError> {
        let resolved = self.resolve_expression(&w.expression, variables)?;
        // WHERE expressions must be boolean
        if resolved.resolved_type != LogicalTypeID::Bool && resolved.resolved_type != LogicalTypeID::Any {
            return Err(format!("WHERE clause must be boolean, got {:?}", resolved.resolved_type).into());
        }
        Ok(BoundWhereClause { expression: resolved })
    }

    // ==================== CREATE (MATCH CREATE) Binding ====================

    fn bind_match_create(
        &self,
        c: &CreateClause,
        existing_vars: &[BoundVariable],
    ) -> Result<(BoundMatchClause, Vec<BoundVariable>), BinderError> {
        // CREATE patterns follow the same structure as MATCH patterns
        let mut patterns = Vec::new();
        let mut new_vars = Vec::new();

        for pattern in &c.patterns {
            let all_vars: Vec<BoundVariable> = existing_vars.iter().cloned().chain(new_vars.iter().cloned()).collect();
            let (bound, nv) = self.bind_pattern(pattern, &all_vars, true)?;
            patterns.push(bound);
            new_vars.extend(nv);
        }

        Ok((
            BoundMatchClause {
                patterns,
                new_variables: new_vars.clone(),
                fts_query: None, // Optional MATCH in Foreach doesn't carry FTS
            },
            new_vars,
        ))
    }

    // ==================== Expression Resolution ====================

    fn resolve_expression(
        &self,
        expr: &Expression,
        variables: &[BoundVariable],
    ) -> Result<BoundExpression, BinderError> {
        match expr {
            Expression::Constant(c) => {
                let typ = match c {
                    Constant::Null => LogicalTypeID::Any,
                    Constant::Bool(_) => LogicalTypeID::Bool,
                    Constant::Integer(_) => LogicalTypeID::Int64,
                    Constant::Float(_) => LogicalTypeID::Double,
                    Constant::String(_) => LogicalTypeID::String,
                };
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: typ,
                    is_constant: true,
                    alias: None,
                })
            }
            Expression::Variable(name) => {
                // Check if variable is in scope
                if let Some(var) = variables.iter().find(|v| v.name == *name) {
                    let typ = if var.is_node {
                        LogicalTypeID::Node
                    } else {
                        LogicalTypeID::Rel
                    };
                    Ok(BoundExpression {
                        expression: expr.clone(),
                        resolved_type: typ,
                        is_constant: false,
                        alias: None,
                    })
                } else if name.to_uppercase() == "COUNT" || name == "*" {
                    // Special handling for COUNT(*)
                    Ok(BoundExpression {
                        expression: expr.clone(),
                        resolved_type: LogicalTypeID::Int64,
                        is_constant: false,
                        alias: None,
                    })
                } else {
                    // Check catalog for table references
                    let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                    if let Some(entry) = catalog.get_entry_by_name(name) {
                        let typ = if entry.is_node_table() {
                            LogicalTypeID::Node
                        } else {
                            LogicalTypeID::Rel
                        };
                        Ok(BoundExpression {
                            expression: expr.clone(),
                            resolved_type: typ,
                            is_constant: false,
                            alias: None,
                        })
                    } else {
                        Err(format!("Variable '{}' not in scope", name).into())
                    }
                }
            }
            Expression::Parameter(_name) => {
                // Parameters are unresolved at bind time; assign Any type.
                // Type checking happens at execute time when values are provided.
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: LogicalTypeID::Any,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::PropertyAccess(obj, prop) => {
                let bound_obj = self.resolve_expression(obj, variables)?;
                // Resolve property type via catalog lookup instead of hardcoded mapping.
                let prop_type = match obj.as_ref() {
                    Expression::Variable(var_name) => {
                        // Find the variable in scope to get its table label
                        if let Some(variable) = variables.iter().find(|v| v.name == *var_name) {
                            if let Some(ref table_label) = variable.label {
                                let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                                match catalog.get_property_type(table_label, prop) {
                                    Some(type_id) => type_id,
                                    None => {
                                        return Err(format!(
                                            "Property '{}' not found on table '{}'",
                                            prop, table_label
                                        )
                                        .into());
                                    }
                                }
                            } else {
                                // Variable has no label (e.g., UNWIND result) — cannot resolve
                                LogicalTypeID::Any
                            }
                        } else {
                            // Variable not in scope — should have failed in resolve_expression
                            bound_obj.resolved_type
                        }
                    }
                    _ => {
                        // Non-variable accessor (e.g., function result) — cannot resolve from catalog
                        LogicalTypeID::Any
                    }
                };
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: prop_type,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::FunctionCall(name, args) => {
                let resolved_args: Result<Vec<BoundExpression>, BinderError> =
                    args.iter().map(|a| self.resolve_expression(a, variables)).collect();
                let _args = resolved_args?;
                // P88: `COUNT_DISTINCT` (from `COUNT(DISTINCT x)`) resolves to
                // the base function's type.
                let upper = name.to_uppercase();
                let base = upper.strip_suffix("_DISTINCT").unwrap_or(&upper);
                let return_type = match base {
                    "COUNT" | "SUM" | "MIN" | "MAX" | "AVG" => LogicalTypeID::Int64,
                    "NEXTVAL" | "CURRVAL" => LogicalTypeID::Int64,
                    "STARTS_WITH" | "ENDS_WITH" | "CONTAINS" => LogicalTypeID::Bool,
                    "TO_UPPER" | "TO_LOWER" | "UPPER" | "LOWER" | "UCASE" | "LCASE" | "TRIM" | "SUBSTRING"
                    | "REPLACE" => LogicalTypeID::String,
                    "ABS" | "CEIL" | "CEILING" | "FLOOR" | "ROUND" | "SQRT" | "LOG" | "EXP" | "SIN" | "COS" | "TAN" => {
                        LogicalTypeID::Double
                    }
                    "DATE" | "TIMESTAMP" => LogicalTypeID::Date,
                    "INT64" | "INT" => LogicalTypeID::Int64,
                    "FLOAT" | "DOUBLE" | "BOOL" | "BOOLEAN" | "STRING" | "BLOB" => LogicalTypeID::String,
                    _ => LogicalTypeID::Any,
                };
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: return_type,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::BinaryOp(op, left, right) => {
                let left = self.resolve_expression(left, variables)?;
                let right = self.resolve_expression(right, variables)?;
                let result_type = match op {
                    BinaryOp::Equal
                    | BinaryOp::NotEqual
                    | BinaryOp::LessThan
                    | BinaryOp::LessThanOrEqual
                    | BinaryOp::GreaterThan
                    | BinaryOp::GreaterThanOrEqual
                    | BinaryOp::And
                    | BinaryOp::Or
                    | BinaryOp::Xor
                    | BinaryOp::In
                    | BinaryOp::NotIn
                    | BinaryOp::StartsWith
                    | BinaryOp::EndsWith
                    | BinaryOp::Contains
                    | BinaryOp::Like => LogicalTypeID::Bool,
                    BinaryOp::Add | BinaryOp::Subtract | BinaryOp::Multiply | BinaryOp::Divide | BinaryOp::Modulo => {
                        // Propagate numeric type
                        if left.resolved_type == LogicalTypeID::Double || right.resolved_type == LogicalTypeID::Double {
                            LogicalTypeID::Double
                        } else {
                            LogicalTypeID::Int64
                        }
                    }
                    BinaryOp::Concat => LogicalTypeID::String,
                };
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: result_type,
                    is_constant: left.is_constant && right.is_constant,
                    alias: None,
                })
            }
            Expression::UnaryOp(op, inner) => {
                let inner = self.resolve_expression(inner, variables)?;
                let result_type = match op {
                    UnaryOp::Not | UnaryOp::IsNull | UnaryOp::IsNotNull => LogicalTypeID::Bool,
                    UnaryOp::Negate => inner.resolved_type,
                };
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: result_type,
                    is_constant: inner.is_constant,
                    alias: None,
                })
            }
            Expression::List(items) => {
                let resolved: Result<Vec<BoundExpression>, BinderError> =
                    items.iter().map(|i| self.resolve_expression(i, variables)).collect();
                resolved?;
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: LogicalTypeID::List,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::Map(entries) => {
                for (_, v) in entries {
                    self.resolve_expression(v, variables)?;
                }
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: LogicalTypeID::Map,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::ExistsSubquery(query) => {
                // Bind the inner query. EXISTS returns Bool.
                // For now, do NOT pass outer variables (uncorrelated subquery).
                let _bound = self.bind_query(*query.clone())?;
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: LogicalTypeID::Bool,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::Case(case_expr) => {
                // Bind subject (if any), all WHEN/THEN expressions, and ELSE.
                // Return type is inferred from the first THEN branch.
                if let Some(subj) = &case_expr.subject {
                    self.resolve_expression(subj, variables)?;
                }
                let mut result_type = LogicalTypeID::Any;
                for alt in &case_expr.alternatives {
                    self.resolve_expression(&alt.when, variables)?;
                    let then_bound = self.resolve_expression(&alt.then, variables)?;
                    if result_type == LogicalTypeID::Any {
                        result_type = then_bound.resolved_type;
                    }
                }
                if let Some(else_e) = &case_expr.else_expr {
                    let else_bound = self.resolve_expression(else_e, variables)?;
                    if result_type == LogicalTypeID::Any {
                        result_type = else_bound.resolved_type;
                    }
                }
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: result_type,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::Star => {
                // Star should be expanded by bind_return before reaching here.
                // If reached, return Any type.
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: LogicalTypeID::Any,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::ListPredicate {
                quantifier: _,
                list,
                var_name,
                predicate,
            } => {
                // Bind both list and predicate expressions
                self.resolve_expression(list, variables)?;

                let mut new_vars = variables.to_vec();
                new_vars.push(crate::bound_statement::BoundVariable {
                    name: var_name.clone(),
                    table_id: 0,
                    label: None,
                    is_node: false,
                });

                self.resolve_expression(predicate, &new_vars)?;
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: LogicalTypeID::Bool,
                    is_constant: false,
                    alias: None,
                })
            }
            Expression::Lambda { var_name: _, body } => {
                // Bind the lambda body — the variable binding is deferred
                // to the evaluator which creates a per-element mini-chunk.
                // For binding purposes, treat the body as unresolved.
                self.resolve_expression(body, variables)?;
                Ok(BoundExpression {
                    expression: expr.clone(),
                    resolved_type: LogicalTypeID::Any,
                    is_constant: false,
                    alias: None,
                })
            }
        }
    }

    // ==================== DDL Binding ====================

    fn bind_create_node_table(&self, t: CreateNodeTable) -> Result<BoundStatement, BinderError> {
        if t.name.is_empty() {
            return Err("Table name cannot be empty".into());
        }

        let mut columns = Vec::new();
        for col in &t.columns {
            let logical_type = self.parse_type_resolved(&col.type_name)?;
            let compression = Self::parse_compression(col.compression.as_deref())?;
            columns.push(CatalogColumn {
                name: col.name.clone(),
                logical_type,
                is_primary_key: col.name == t.primary_key,
                compression,
                default_value: None,
            });
        }

        if columns.is_empty() {
            return Err("Table must have at least one column".into());
        }

        // Verify primary key exists
        if !columns.iter().any(|c| c.is_primary_key) {
            return Err(format!("CREATE NODE TABLE requires a PRIMARY KEY (missing '{}')", t.primary_key).into());
        }

        // Register with catalog
        let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        match catalog.create_node_table(t.name.clone(), columns.clone()) {
            CatalogResult::Created { .. } => {}
            CatalogResult::AlreadyExists if t.if_not_exists => {
                // IF NOT EXISTS: leave any existing table untouched; the
                // physical executor also skips, so this is an idempotent
                // no-op for schema-ensure (P72).
            }
            CatalogResult::AlreadyExists => {
                return Err(format!("Table '{}' already exists", t.name).into());
            }
            _ => return Err("Failed to create table".into()),
        }

        Ok(BoundStatement::BoundCreateNodeTable(BoundCreateNodeTable {
            name: t.name,
            columns,
            primary_key: t.primary_key,
            if_not_exists: t.if_not_exists,
        }))
    }

    fn bind_create_vector_index(&self, v: akar_parser::ast::CreateVectorIndex) -> Result<BoundStatement, BinderError> {
        if v.index_name.is_empty() {
            return Err("Index name cannot be empty".into());
        }
        if v.metric.is_empty() {
            return Err("Metric must be specified (cosine, euclidean, l2, or dot)".into());
        }
        if v.dimensions == 0 {
            return Err("Dimensions must be greater than 0".into());
        }

        // Validate the referenced table and column exist in the catalog.
        // The guard is scoped so it drops before the write lock below —
        // otherwise the second `catalog.lock()` self-deadlocks (P53.x).
        let col_exists = {
            let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
            let entry = catalog
                .get_entry_by_name(&v.table_name)
                .ok_or_else(|| format!("Table '{}' not found", v.table_name))?;
            entry.columns().iter().any(|c| c.name == v.column_name)
        };
        if !col_exists {
            return Err(format!("Column '{}' not found in table '{}'", v.column_name, v.table_name).into());
        }

        // Validate metric value
        match v.metric.to_lowercase().as_str() {
            "cosine" | "euclidean" | "l2" | "dot" => {}
            other => {
                return Err(format!("Unknown metric '{other}'. Supported: cosine, euclidean, l2, dot").into());
            }
        }

        // Register with catalog
        let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        match catalog.create_vector_index(
            v.index_name.clone(),
            v.table_name.clone(),
            v.column_name.clone(),
            v.metric.clone(),
            v.dimensions,
        ) {
            CatalogResult::Created { .. } => {}
            CatalogResult::AlreadyExists => {
                return Err(format!("Vector index '{}' already exists", v.index_name).into());
            }
            CatalogResult::NotFound => {
                return Err(format!("Table '{}' not found", v.table_name).into());
            }
            CatalogResult::Dropped { .. } => {
                return Err("Unexpected: Dropped result from create_vector_index".into());
            }
        }

        Ok(BoundStatement::BoundCreateVectorIndex(BoundCreateVectorIndex {
            index_name: v.index_name,
            table_name: v.table_name,
            column_name: v.column_name,
            metric: v.metric,
            dimensions: v.dimensions,
        }))
    }

    fn bind_create_index(&self, v: akar_parser::ast::CreateIndex) -> Result<BoundStatement, BinderError> {
        if v.index_name.is_empty() {
            return Err("Index name cannot be empty".into());
        }

        // Parse index type
        let index_type = IndexType::from_str(&v.index_type)
            .ok_or_else(|| format!("Unknown index type '{}'. Use ART or HASH", v.index_type))?;

        // Validate table and column exist
        {
            let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
            let entry = catalog
                .get_entry_by_name(&v.table_name)
                .ok_or_else(|| format!("Table '{}' not found", v.table_name))?;

            // Validate column exists and is PK
            let col_exists = entry.columns().iter().any(|c| c.name == v.property);
            if !col_exists {
                return Err(format!("Column '{}' not found in table '{}'", v.property, v.table_name).into());
            }

            let pk_col = entry.columns().iter().find(|c| c.is_primary_key);
            if pk_col.map(|c| c.name.as_str()) != Some(v.property.as_str()) {
                return Err(format!(
                    "Cannot create index on non-PK column '{}'. Only PK columns are supported.",
                    v.property
                )
                .into());
            }
        }

        // Register with catalog (separate lock for mutable access)
        let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        catalog.create_index(&v.table_name, v.index_name.clone(), index_type, &v.property)?;

        Ok(BoundStatement::BoundCreateIndex(BoundCreateIndex {
            index_type,
            index_name: v.index_name,
            table_name: v.table_name,
            column_name: v.property,
        }))
    }

    fn bind_drop_index(&self, v: akar_parser::ast::DropIndex) -> Result<BoundStatement, BinderError> {
        if v.index_name.is_empty() {
            return Err("Index name cannot be empty".into());
        }

        let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        catalog.drop_index(&v.table_name, &v.index_name)?;

        Ok(BoundStatement::BoundDropIndex(BoundDropIndex {
            index_name: v.index_name,
            table_name: v.table_name,
        }))
    }

    fn bind_create_rel_table(&self, t: CreateRelTable) -> Result<BoundStatement, BinderError> {
        if t.name.is_empty() {
            return Err("Table name cannot be empty".into());
        }

        // Validate FROM and TO tables exist
        let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        let src_id = catalog
            .get_table_id(&t.from)
            .ok_or_else(|| format!("Source table '{}' not found", t.from))?;
        let dst_id = catalog
            .get_table_id(&t.to)
            .ok_or_else(|| format!("Destination table '{}' not found", t.to))?;
        drop(catalog);

        let mut columns = Vec::new();
        for col in &t.columns {
            let logical_type = self.parse_type_resolved(&col.type_name)?;
            let compression = Self::parse_compression(col.compression.as_deref())?;
            columns.push(CatalogColumn {
                name: col.name.clone(),
                logical_type,
                is_primary_key: false,
                compression,
                default_value: None,
            });
        }

        // Register with catalog
        let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        match catalog.create_rel_table(t.name.clone(), src_id, dst_id, columns.clone()) {
            CatalogResult::Created { .. } => {}
            CatalogResult::AlreadyExists if t.if_not_exists => {
                // IF NOT EXISTS: leave any existing rel table untouched (P72).
            }
            CatalogResult::AlreadyExists => {
                return Err(format!("Rel table '{}' already exists", t.name).into());
            }
            _ => return Err("Failed to create rel table".into()),
        }

        Ok(BoundStatement::BoundCreateRelTable(BoundCreateRelTable {
            name: t.name,
            from: t.from,
            to: t.to,
            src_table_id: src_id,
            dst_table_id: dst_id,
            columns,
            if_not_exists: t.if_not_exists,
        }))
    }

    fn bind_drop_table(&self, t: DropTable) -> Result<BoundStatement, BinderError> {
        let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        match catalog.drop_table(&t.name) {
            CatalogResult::Dropped { .. } => Ok(BoundStatement::BoundDropTable(BoundDropTable { name: t.name })),
            CatalogResult::NotFound => Err(format!("Table '{}' not found", t.name).into()),
            _ => Err("Failed to drop table".into()),
        }
    }

    fn bind_unwind(&self, u: &akar_parser::ast::UnwindClause) -> Result<BoundUnwindClause, BinderError> {
        // Validate the expression is a list literal, variable reference, or parameter.
        // Parameters are resolved to lists at substitute_params time.
        match &u.expression {
            akar_parser::ast::Expression::List(_) => {}
            akar_parser::ast::Expression::Variable(_) => {}
            akar_parser::ast::Expression::Parameter(_) => {}
            _ => return Err(format!("UNWIND requires a list expression, got: {:?}", u.expression).into()),
        }
        if u.variable.is_empty() {
            return Err("UNWIND requires a variable name".into());
        }
        Ok(BoundUnwindClause {
            expression: u.expression.clone(),
            variable: u.variable.clone(),
        })
    }

    fn bind_foreach(
        &self,
        f: &akar_parser::ast::ForeachClause,
        variables: &[BoundVariable],
    ) -> Result<BoundForeachClause, BinderError> {
        // Validate the expression is a list
        match &f.expression {
            akar_parser::ast::Expression::List(_) | akar_parser::ast::Expression::Variable(_) => {}
            _ => return Err(format!("FOREACH requires a list expression, got: {:?}", f.expression).into()),
        }
        if f.variable.is_empty() {
            return Err("FOREACH requires a variable name".into());
        }
        // Create a new variable scope for the foreach body
        let mut local_vars = variables.to_vec();
        local_vars.push(BoundVariable {
            name: f.variable.clone(),
            table_id: 0,
            label: None,
            is_node: false,
        });

        // Bind sub-statements
        let mut sub_statements = Vec::new();
        for clause in &f.clauses {
            match clause {
                akar_parser::ast::Clause::Create(cc) => {
                    // Bind as DML CREATE (BoundCreateDml), not as a MATCH clause
                    let bound = self.bind_create_dml(cc.clone(), &local_vars)?;
                    sub_statements.push(bound);
                }
                akar_parser::ast::Clause::Set(sc) => {
                    // Manually wrap SET in BoundQuery to preserve variable scope
                    let bound_set = self.bind_set(sc, &local_vars)?;
                    sub_statements.push(BoundStatement::BoundQuery(BoundQuery {
                        clauses: vec![BoundClause::BoundSet(bound_set)],
                        variables: local_vars.clone(),
                    }));
                }
                akar_parser::ast::Clause::Delete(dc) => {
                    // Manually wrap DELETE in BoundQuery to preserve variable scope
                    let bound_delete = self.bind_delete(dc, &local_vars)?;
                    sub_statements.push(BoundStatement::BoundQuery(BoundQuery {
                        clauses: vec![BoundClause::BoundDelete(bound_delete)],
                        variables: local_vars.clone(),
                    }));
                }
                _ => {
                    return Err(format!("Unsupported FOREACH sub-clause: {:?}", clause).into());
                }
            }
        }
        Ok(BoundForeachClause {
            variable: f.variable.clone(),
            expression: f.expression.clone(),
            sub_statements,
        })
    }

    fn bind_optional_match(
        &self,
        m: &akar_parser::ast::OptionalMatchClause,
        existing_vars: &[BoundVariable],
    ) -> Result<(BoundMatchClause, Vec<BoundVariable>), BinderError> {
        let mut patterns = Vec::new();
        let mut new_vars = Vec::new();

        for pattern in &m.patterns {
            let all_vars: Vec<BoundVariable> = existing_vars.iter().cloned().chain(new_vars.iter().cloned()).collect();
            let (bound, nv) = self.bind_pattern(pattern, &all_vars, false)?;
            patterns.push(bound);
            new_vars.extend(nv);
        }

        Ok((
            BoundMatchClause {
                patterns,
                new_variables: new_vars.clone(),
                fts_query: None, // Optional MATCH doesn't carry FTS
            },
            new_vars,
        ))
    }

    fn bind_set(
        &self,
        s: &akar_parser::ast::SetClause,
        variables: &[BoundVariable],
    ) -> Result<BoundSetClause, BinderError> {
        let mut items = Vec::new();
        for item in &s.items {
            // Property must be of form `variable.property`
            match &item.property {
                akar_parser::ast::Expression::PropertyAccess(var_expr, prop_name) => {
                    match var_expr.as_ref() {
                        akar_parser::ast::Expression::Variable(var_name) => {
                            let bound_var = variables
                                .iter()
                                .find(|v| v.name == *var_name)
                                .ok_or_else(|| format!("Variable '{}' not in scope for SET", var_name))?;
                            items.push(BoundSetItem {
                                property: item.property.clone(),
                                value: item.value.clone(),
                                column_name: prop_name.clone(),
                                column_idx: 0, // resolved by catalog lookup
                                table_name: bound_var.label.clone().unwrap_or_default(),
                                table_id: bound_var.table_id,
                                is_node: bound_var.is_node,
                            });
                        }
                        _ => return Err("SET property must be on a variable".into()),
                    }
                }
                _ => return Err("SET requires property access expression (e.g., n.age)".into()),
            }
        }
        Ok(BoundSetClause { items })
    }

    fn bind_union(&self, u: akar_parser::ast::UnionStatement) -> Result<BoundStatement, BinderError> {
        let left = self.bind_query(u.left)?;
        let right = self.bind_query(u.right)?;
        Ok(BoundStatement::BoundUnion(BoundUnion {
            left: Box::new(match left {
                BoundStatement::BoundQuery(q) => q,
                _ => unreachable!(),
            }),
            right: Box::new(match right {
                BoundStatement::BoundQuery(q) => q,
                _ => unreachable!(),
            }),
            all: u.all,
        }))
    }

    fn bind_merge(&self, m: akar_parser::ast::MergeStatement) -> Result<BoundStatement, BinderError> {
        let (bound, _) = self.bind_merge_clause(&m, &[])?;
        Ok(BoundStatement::BoundMerge(bound))
    }

    /// Bind a MERGE pattern (standalone statement or clause in a query chain),
    /// returning the bound merge plus any variables introduced by its patterns.
    ///
    /// `variables` holds the query scope so that label-less node patterns
    /// (e.g. `MERGE (a)-[r:Connected]->(b)` where `a`/`b` were bound by a prior
    /// MATCH) can be resolved to their already-bound table (P53.20).
    fn bind_merge_clause(
        &self,
        m: &akar_parser::ast::MergeStatement,
        variables: &[BoundVariable],
    ) -> Result<(BoundMerge, Vec<BoundVariable>), BinderError> {
        let patterns = self.bind_merge_patterns(&m.patterns, variables)?;
        let primary = patterns
            .iter()
            .find_map(|p| p.node.clone())
            .ok_or("MERGE requires at least one node pattern")?;

        // Resolve ON CREATE SET items
        let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        let on_create = resolve_set_items(&catalog, &m.on_create)?;
        let on_match = resolve_set_items(&catalog, &m.on_match)?;

        let mut new_vars = Vec::new();
        for pat in &patterns {
            if let Some(node) = &pat.node {
                if let Some(name) = &node.variable {
                    // Skip nodes that were resolved as references to existing bindings.
                    let is_reference = variables.iter().any(|v| &v.name == name && v.is_node);
                    if !is_reference {
                        new_vars.push(BoundVariable {
                            name: name.clone(),
                            table_id: node.table_id,
                            label: Some(node.table_name.clone()),
                            is_node: true,
                        });
                    }
                }
            }
            if let Some(edge) = &pat.edge {
                if let Some(name) = &edge.variable {
                    new_vars.push(BoundVariable {
                        name: name.clone(),
                        table_id: edge.table_id,
                        label: Some(edge.table_name.clone()),
                        is_node: false,
                    });
                }
            }
        }

        Ok((
            BoundMerge {
                table_name: primary.table_name,
                table_id: primary.table_id,
                properties: primary.properties,
                patterns,
                on_create,
                on_match,
            },
            new_vars,
        ))
    }

    /// Bind the node/edge elements of a MERGE pattern path, resolving
    /// label-less nodes against the given query scope (P53.20).
    fn bind_merge_patterns(
        &self,
        patterns: &[akar_parser::ast::Pattern],
        variables: &[BoundVariable],
    ) -> Result<Vec<BoundCreatePattern>, BinderError> {
        let mut bound = Vec::with_capacity(patterns.len());
        for (i, pat) in patterns.iter().enumerate() {
            let node = if let Some(ref n) = pat.node {
                match n.labels.first() {
                    Some(label) => {
                        let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                        let entry = catalog
                            .get_entry_by_name(label)
                            .ok_or_else(|| format!("Table '{label}' not found"))?;
                        if !entry.is_node_table() {
                            return Err(format!("'{label}' is not a node table").into());
                        }
                        Some(BoundNodeCreate {
                            variable: n.variable.clone(),
                            table_name: label.clone(),
                            table_id: entry.table_id(),
                            properties: n.properties.clone(),
                        })
                    }
                    None => {
                        // Label-less node: must reference an already-bound node variable.
                        let var_name = n.variable.as_ref().ok_or("MERGE node requires a label or a variable")?;
                        let existing =
                            variables
                                .iter()
                                .find(|v| &v.name == var_name && v.is_node)
                                .ok_or_else(|| {
                                    format!(
                                        "MERGE references unknown variable '{}' (no label to resolve table)",
                                        var_name
                                    )
                                })?;
                        let label = existing.label.clone().unwrap_or_default();
                        let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                        let entry = catalog
                            .get_entry_by_name(&label)
                            .ok_or_else(|| format!("Table '{label}' not found"))?;
                        if !entry.is_node_table() {
                            return Err(format!("'{label}' is not a node table").into());
                        }
                        Some(BoundNodeCreate {
                            variable: n.variable.clone(),
                            table_name: label.clone(),
                            table_id: existing.table_id,
                            properties: n.properties.clone(),
                        })
                    }
                }
            } else {
                None
            };

            let edge = if let Some(ref e) = pat.edge {
                let label = e.labels.first().ok_or("Edge requires a label (rel table name)")?;
                let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                let entry = catalog
                    .get_entry_by_name(label)
                    .ok_or_else(|| format!("Rel table '{label}' not found"))?;
                if !entry.is_rel_table() {
                    return Err(format!("'{label}' is not a rel table").into());
                }
                let cur_var = pat
                    .node
                    .as_ref()
                    .and_then(|n| n.variable.clone())
                    .ok_or("Edge endpoints must be named node variables")?;
                let nxt_var = patterns
                    .get(i + 1)
                    .and_then(|p| p.node.as_ref())
                    .and_then(|n| n.variable.clone())
                    .ok_or("Edge endpoints must be named node variables")?;
                let (src_var, dst_var) = match e.direction {
                    akar_parser::ast::EdgeDirection::RightToLeft => (nxt_var, cur_var),
                    _ => (cur_var, nxt_var),
                };
                Some(BoundEdgeCreate {
                    variable: e.variable.clone(),
                    table_name: label.clone(),
                    table_id: entry.table_id(),
                    src_var,
                    dst_var,
                    properties: e.properties.clone(),
                })
            } else {
                None
            };

            bound.push(BoundCreatePattern { node, edge });
        }
        Ok(bound)
    }

    /// Bind all node/edge elements of a CREATE or MERGE pattern path.
    ///
    /// A path like `(a)-[r:R]->(b)` is flattened by the parser into
    /// `[Pattern{node: a, edge: r}, Pattern{node: b, edge: None}]`. The edge is
    /// attached to the element of its *source* node, so its `src_var` is the
    /// current pattern's node variable and its `dst_var` is the next pattern's
    /// node variable (reversed for `RightToLeft`).
    fn bind_create_patterns(
        &self,
        patterns: &[akar_parser::ast::Pattern],
    ) -> Result<Vec<BoundCreatePattern>, BinderError> {
        let mut bound = Vec::with_capacity(patterns.len());
        for (i, pat) in patterns.iter().enumerate() {
            let node = if let Some(ref n) = pat.node {
                let label = n.labels.first().ok_or("CREATE/MERGE requires a label (table name)")?;
                let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                let entry = catalog
                    .get_entry_by_name(label)
                    .ok_or_else(|| format!("Table '{label}' not found"))?;
                if !entry.is_node_table() {
                    return Err(format!("'{label}' is not a node table").into());
                }
                Some(BoundNodeCreate {
                    variable: n.variable.clone(),
                    table_name: label.clone(),
                    table_id: entry.table_id(),
                    properties: n.properties.clone(),
                })
            } else {
                None
            };

            let edge = if let Some(ref e) = pat.edge {
                let label = e.labels.first().ok_or("Edge requires a label (rel table name)")?;
                let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
                let entry = catalog
                    .get_entry_by_name(label)
                    .ok_or_else(|| format!("Rel table '{label}' not found"))?;
                if !entry.is_rel_table() {
                    return Err(format!("'{label}' is not a rel table").into());
                }
                let cur_var = pat
                    .node
                    .as_ref()
                    .and_then(|n| n.variable.clone())
                    .ok_or("Edge endpoints must be named node variables")?;
                let nxt_var = patterns
                    .get(i + 1)
                    .and_then(|p| p.node.as_ref())
                    .and_then(|n| n.variable.clone())
                    .ok_or("Edge endpoints must be named node variables")?;
                let (src_var, dst_var) = match e.direction {
                    akar_parser::ast::EdgeDirection::RightToLeft => (nxt_var, cur_var),
                    _ => (cur_var, nxt_var),
                };
                Some(BoundEdgeCreate {
                    variable: e.variable.clone(),
                    table_name: label.clone(),
                    table_id: entry.table_id(),
                    src_var,
                    dst_var,
                    properties: e.properties.clone(),
                })
            } else {
                None
            };

            bound.push(BoundCreatePattern { node, edge });
        }
        Ok(bound)
    }

    fn bind_create_dml(
        &self,
        c: akar_parser::ast::CreateClause,
        _variables: &[BoundVariable],
    ) -> Result<BoundStatement, BinderError> {
        let patterns = self.bind_create_patterns(&c.patterns)?;
        if patterns.iter().all(|p| p.node.is_none()) {
            return Err("CREATE DML requires a node pattern".into());
        }

        Ok(BoundStatement::BoundCreateDml(BoundCreateDml { patterns }))
    }

    fn bind_standalone_call(&self, c: akar_parser::ast::StandaloneCall) -> Result<BoundStatement, BinderError> {
        // Note: CALL create_fts_index is superseded by the DDL `CREATE FTS INDEX` statement.
        // CALL is a table function invocation ΓÇö validate the function exists
        // in the function registry. At binding time we just pass through;
        // resolution happens at execution time.
        Ok(BoundStatement::BoundStandaloneCall(BoundStandaloneCall {
            function_name: c.function_name,
            args: c.args,
        }))
    }

    fn bind_explain(&self, e: akar_parser::ast::ExplainStatement) -> Result<BoundStatement, BinderError> {
        // Bind the inner statement recursively
        let inner = self.bind(*e.statement)?;
        Ok(BoundStatement::BoundExplain(BoundExplain {
            inner: Box::new(inner),
            explain_type: e.explain_type,
        }))
    }

    fn bind_create_sequence(&self, s: akar_parser::ast::CreateSequence) -> Result<BoundStatement, BinderError> {
        // Compute defaults matching C++ behavior:
        // - START WITH: 1 for increment > 0, max_value for increment < 0
        // - INCREMENT: 1 (default)
        // - MINVALUE: 1 for increment > 0, i64::MIN for increment < 0
        // - MAXVALUE: i64::MAX for increment > 0, -1 for increment < 0
        // - CYCLE: false (default)
        let increment = s.increment.unwrap_or(1);
        if increment == 0 {
            return Err("INCREMENT must not be zero".into());
        }
        let start_with = s.start_with.unwrap_or(if increment > 0 { 1 } else { -1 });
        let min_value = s.min_value.unwrap_or(if increment > 0 { 1 } else { i64::MIN });
        let max_value = s.max_value.unwrap_or(if increment > 0 { i64::MAX } else { -1 });
        let cycle = s.cycle.unwrap_or(false);

        // Validate min/max/start consistency
        if min_value > max_value {
            return Err(format!(
                "MINVALUE ({}) cannot be greater than MAXVALUE ({})",
                min_value, max_value
            )
            .into());
        }
        if start_with < min_value || start_with > max_value {
            return Err(format!(
                "START WITH ({}) must be between MINVALUE ({}) and MAXVALUE ({})",
                start_with, min_value, max_value
            )
            .into());
        }

        Ok(BoundStatement::BoundCreateSequence(BoundCreateSequence {
            name: s.name,
            if_not_exists: s.if_not_exists,
            or_replace: s.or_replace,
            start_with,
            increment,
            min_value,
            max_value,
            cycle,
        }))
    }

    fn bind_drop_sequence(&self, s: akar_parser::ast::DropSequence) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundDropSequence(BoundDropSequence {
            name: s.name,
            if_exists: s.if_exists,
        }))
    }

    fn bind_create_macro(&self, m: akar_parser::ast::CreateMacro) -> Result<BoundStatement, BinderError> {
        // Convert default args to strings
        let default_args: Vec<(String, String)> = m
            .default_args
            .iter()
            .map(|(name, expr)| (name.clone(), expr_to_debug_string(expr)))
            .collect();
        let expression_str = expr_to_debug_string(&m.expression);
        Ok(BoundStatement::BoundCreateMacro(BoundCreateMacro {
            name: m.name,
            positional_args: m.positional_args,
            default_args,
            expression: expression_str,
        }))
    }

    fn bind_export_database(&self, e: akar_parser::ast::ExportDatabase) -> Result<BoundStatement, BinderError> {
        let file_type = e
            .options
            .get("FORMAT")
            .map(|s| s.to_lowercase())
            .unwrap_or_else(|| "csv".to_string());
        if file_type != "csv" && file_type != "parquet" {
            return Err(format!("Unsupported export format '{file_type}'. Supported: csv, parquet").into());
        }
        let schema_only = e.options.get("SCHEMA_ONLY").map(|s| s == "true").unwrap_or(false);
        Ok(BoundStatement::BoundExportDatabase(BoundExportDatabase {
            file_path: e.file_path,
            file_type,
            schema_only,
            options: e.options,
        }))
    }

    fn bind_import_database(&self, i: akar_parser::ast::ImportDatabase) -> Result<BoundStatement, BinderError> {
        // Validate the import directory (path-traversal hardening) and read the
        // schema/copy/index files through containment-checked helpers.
        let dir = Self::validate_import_database_dir(&i.file_path)?;

        if !dir.join("schema.cypher").exists() {
            return Err(format!("schema.cypher not found in '{}'", i.file_path).into());
        }

        let query = if dir.join("copy.cypher").exists() {
            let schema = Self::read_file_within_dir(&dir, "schema.cypher")?;
            let copy = Self::read_file_within_dir(&dir, "copy.cypher")?;
            format!("{schema}\n{copy}")
        } else {
            Self::read_file_within_dir(&dir, "schema.cypher")?
        };

        let index_query = if dir.join("index.cypher").exists() {
            Self::read_file_within_dir(&dir, "index.cypher")?
        } else {
            String::new()
        };

        Ok(BoundStatement::BoundImportDatabase(BoundImportDatabase {
            file_path: i.file_path,
            query,
            index_query,
        }))
    }

    /// Validate a user-supplied `IMPORT DATABASE` directory path and return its
    /// canonical form.
    ///
    /// Security (path traversal, CWE-22/CWE-59): `file_path` is raw text from
    /// parsed SQL — including queries submitted by remote akar-server clients —
    /// so it may carry `..` components or symlinks that escape the directory
    /// the operator intended. We reject any lexical `..` component outright,
    /// then canonicalize so every later containment check operates on the real
    /// on-disk location instead of the spelled path.
    fn validate_import_database_dir(raw_path: &str) -> Result<std::path::PathBuf, BinderError> {
        if raw_path.is_empty() {
            return Err("IMPORT DATABASE requires a non-empty directory path".into());
        }
        if raw_path.contains('\0') {
            return Err("IMPORT DATABASE path contains a NUL byte".into());
        }
        let path = std::path::Path::new(raw_path);
        for component in path.components() {
            if matches!(component, std::path::Component::ParentDir) {
                return Err(
                    format!("IMPORT DATABASE path '{raw_path}' contains '..' — path traversal is not allowed").into(),
                );
            }
        }
        if !path.exists() {
            return Err(format!("Import directory '{raw_path}' not found").into());
        }
        if !path.is_dir() {
            return Err(format!("'{raw_path}' is not a directory").into());
        }
        std::fs::canonicalize(path).map_err(|e| format!("Cannot resolve import directory '{raw_path}': {e}").into())
    }

    /// Read one fixed-name file from the canonical import directory.
    ///
    /// The file must resolve back inside `dir` after symlink resolution — this
    /// blocks a `schema.cypher -> /etc/passwd` style symlink from escaping the
    /// directory being imported.
    fn read_file_within_dir(dir: &std::path::Path, file_name: &str) -> Result<String, BinderError> {
        let file_path = dir.join(file_name);
        match std::fs::canonicalize(&file_path) {
            Ok(resolved) if resolved.starts_with(dir) => {}
            Ok(resolved) => {
                return Err(format!(
                    "'{file_name}' resolves outside the import directory ({})",
                    resolved.display()
                )
                .into());
            }
            Err(e) => return Err(format!("Cannot read {file_name}: {e}").into()),
        }
        std::fs::read_to_string(&file_path).map_err(|e| format!("Cannot read {file_name}: {e}").into())
    }

    /// Bind ANALYZE statement ΓÇö resolve table names to table IDs.
    fn bind_analyze(&self, a: AnalyzeStatement) -> Result<BoundStatement, BinderError> {
        let cat = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
        let table_ids = if let Some(ref table_name) = a.table_name {
            let id = cat
                .get_table_id(table_name)
                .ok_or_else(|| format!("Table '{table_name}' not found"))?;
            vec![id]
        } else {
            // ANALYZE * ΓÇö collect stats for all node/rel tables
            cat.all_entries()
                .filter(|e| e.is_node_table() || e.is_rel_table())
                .map(|e| e.table_id())
                .collect()
        };
        Ok(BoundStatement::BoundAnalyze(BoundAnalyze {
            table_name: a.table_name,
            table_ids,
        }))
    }

    /// Bind TRANSACTION statement — trivial (no catalog resolution needed).
    fn bind_transaction(&self, t: TransactionStatement) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundTransaction(BoundTransaction { action: t.action }))
    }

    /// Bind EXTENSION statement — trivial (validated at execution time).
    fn bind_extension(&self, e: ExtensionStatement) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundExtension(BoundExtension {
            action: e.action,
            name: e.name,
        }))
    }

    fn bind_attach_database(&self, a: AttachDatabase) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundAttachDatabase(BoundAttachDatabase {
            path: a.path,
            alias: a.alias,
            options: a.options,
        }))
    }

    fn bind_detach_database(&self, d: DetachDatabase) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundDetachDatabase(BoundDetachDatabase {
            alias: d.alias,
        }))
    }

    fn bind_use_database(&self, u: UseDatabase) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundUseDatabase(BoundUseDatabase { alias: u.alias }))
    }

    fn bind_load_from(&self, l: LoadFrom) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundLoadFrom(BoundLoadFrom {
            path: l.path,
            options: l.options,
        }))
    }

    fn bind_create_type(&self, t: CreateType) -> Result<BoundStatement, BinderError> {
        // Validate the type name is a known type (builtin or predefined alias)
        self.parse_type_resolved(&t.type_name)?;
        // Reject re-creating an existing alias
        {
            let catalog = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
            if catalog.get_type_alias(&t.name).is_some() {
                return Err(format!("Type alias '{}' already exists", t.name).into());
            }
        }
        Ok(BoundStatement::BoundCreateType(BoundCreateType {
            name: t.name,
            type_name: t.type_name,
        }))
    }

    fn bind_comment_on_table(&self, c: CommentOnTable) -> Result<BoundStatement, BinderError> {
        // Validate table exists
        {
            let catalog = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
            catalog
                .get_entry_by_name(&c.table_name)
                .ok_or_else(|| format!("Table '{}' not found", c.table_name))?;
        }
        Ok(BoundStatement::BoundCommentOnTable(BoundCommentOnTable {
            table_name: c.table_name,
            comment: c.comment,
        }))
    }

    fn bind_create_graph(&self, g: CreateGraph) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundCreateGraph(BoundCreateGraph {
            name: g.name,
            is_any: g.is_any,
        }))
    }

    fn bind_use_graph(&self, g: UseGraph) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundUseGraph(BoundUseGraph { name: g.name }))
    }

    fn bind_drop_graph(&self, g: DropGraph) -> Result<BoundStatement, BinderError> {
        Ok(BoundStatement::BoundDropGraph(BoundDropGraph { name: g.name }))
    }

    fn bind_create_fts_index(&self, f: CreateFtsIndex) -> Result<BoundStatement, BinderError> {
        // Validate table and column exist
        {
            let catalog = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
            let entry = catalog
                .get_entry_by_name(&f.table_name)
                .ok_or_else(|| format!("Table '{}' not found", f.table_name))?;
            let has_column = entry.columns().iter().any(|c| c.name == f.column_name);
            if !has_column {
                return Err(format!("Column '{}' not found in table '{}'", f.column_name, f.table_name).into());
            }
        }
        let index_name = f.index_name.clone();

        // Register the FTS index → source mapping so `USING FTS INDEX` scans
        // can catch up newly inserted rows and filter soft-deleted ones
        // (P52.39, kept under the P104.2 clean break).
        {
            let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
            catalog
                .register_fts_index(index_name.clone(), f.table_name.clone(), f.column_name.clone())
                .map_err(|e| format!("Failed to register FTS index: {e}"))?;
        }

        Ok(BoundStatement::BoundCreateFtsIndex(BoundCreateFtsIndex {
            index_name: f.index_name,
            table_name: f.table_name,
            column_name: f.column_name,
            tokenizer: f.tokenizer,
            if_not_exists: f.if_not_exists,
        }))
    }

    fn bind_alter_table(&self, a: akar_parser::ast::AlterTable) -> Result<BoundStatement, BinderError> {
        // Validate table exists and extract column info
        let col_names: Vec<String> = {
            let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
            let entry = catalog
                .get_entry_by_name(&a.table_name)
                .ok_or_else(|| format!("Table '{}' not found", a.table_name))?;
            entry.columns().iter().map(|c| c.name.clone()).collect()
        };

        fn has_name(col_names: &[String], name: &str) -> bool {
            col_names.iter().any(|c| c.eq_ignore_ascii_case(name))
        }

        // Validate alter action
        match &a.action {
            akar_parser::ast::AlterAction::AddColumn { name: _, type_name } => {
                self.parse_type_resolved(type_name)?;
            }
            akar_parser::ast::AlterAction::DropColumn { name } => {
                if !has_name(&col_names, name) {
                    return Err(format!("Column '{name}' not found in table '{}'", a.table_name).into());
                }
            }
            akar_parser::ast::AlterAction::RenameColumn { old_name, new_name } => {
                if !has_name(&col_names, old_name) {
                    return Err(format!("Column '{old_name}' not found in table '{}'", a.table_name).into());
                }
                if has_name(&col_names, new_name) {
                    return Err(format!("Column '{new_name}' already exists in table '{}'", a.table_name).into());
                }
            }
            akar_parser::ast::AlterAction::RenameTable { new_name: _ } => {
                // Rename table duplicate check happens at execution time in the catalog
            }
        }

        Ok(BoundStatement::BoundAlterTable(BoundAlterTable {
            table_name: a.table_name,
            action: a.action,
        }))
    }

    fn bind_delete(
        &self,
        d: &akar_parser::ast::DeleteClause,
        variables: &[BoundVariable],
    ) -> Result<BoundDeleteClause, BinderError> {
        if d.expressions.is_empty() {
            return Err("DELETE requires at least one expression".into());
        }

        // Resolve each DELETE variable to its bound table/type. A linear scan per
        // expression is O(N*M) in the worst case (N bound variables, M DELETE
        // expressions). Microbenchmarks (benches/delete_variable_lookup.rs) show the
        // linear scan is faster when the delete list is small — the HashMap build
        // (O(N)) outweighs it — so the name index is only built past that break-even
        // (M >= 8 deletes AND N >= 256 variables in scope).
        let use_index = d.expressions.len() >= 8 && variables.len() >= 256;
        let mut index: Option<HashMap<&str, &BoundVariable>> = None;
        if use_index {
            let mut map = HashMap::with_capacity(variables.len());
            for v in variables {
                // First occurrence wins, matching the linear-scan semantics.
                map.entry(v.name.as_str()).or_insert(v);
            }
            index = Some(map);
        }

        let mut items = Vec::with_capacity(d.expressions.len());
        for expr in &d.expressions {
            match expr {
                akar_parser::ast::Expression::Variable(var_name) => {
                    let var = match &index {
                        Some(map) => map.get(var_name.as_str()).copied(),
                        None => variables.iter().find(|v| v.name == *var_name),
                    }
                    .ok_or_else(|| format!("Variable '{}' not found in scope for DELETE", var_name))?;
                    items.push(BoundDeleteItem {
                        expression: expr.clone(),
                        table_name: var.label.clone().unwrap_or_default(),
                        table_id: var.table_id,
                        primary_key_column: String::new(),
                        is_node: var.is_node,
                    });
                }
                _ => return Err(format!("DELETE only supports variable references, got: {:?}", expr).into()),
            }
        }

        Ok(BoundDeleteClause {
            detach: d.detach,
            items,
        })
    }

    fn bind_copy_from(&self, c: akar_parser::ast::CopyFrom) -> Result<BoundStatement, BinderError> {
        // 1. Look up table in catalog and resolve column schema
        let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
        let entry = catalog
            .get_entry_by_name(&c.table_name)
            .ok_or_else(|| format!("Table '{}' not found", c.table_name))?;
        let table_id = entry.table_id();
        let columns: Vec<akar_catalog::CatalogColumn> = entry.columns().to_vec();
        // Rel tables store their source/destination node IDs in dedicated
        // columns (`src_table_id`/`dst_table_id`), so a rel-table COPY file
        // carries two extra leading columns: [SRC, DST, ...user props].
        let is_rel_table = entry.is_rel_table();
        drop(catalog);

        // 2. Validate file path exists and is accessible
        let path = std::path::Path::new(&c.file_path);
        if !path.exists() {
            return Err(format!("File '{}' not found", c.file_path).into());
        }
        if !path.is_file() {
            return Err(format!("'{}' is not a file", c.file_path).into());
        }

        // 3. If HEADER=true and delimiter is known, peek at first CSV line to
        //    validate column count. If no explicit delimiter option was given,
        //    skip validation (the physical operator handles it with config-aware parsing).
        let header_val = c.options.get("HEADER").or_else(|| c.options.get("header"));
        let delim_val = c.options.get("DELIM").or_else(|| c.options.get("delim"));
        if let Some(hv) = header_val {
            if hv.eq_ignore_ascii_case("true") && delim_val.is_some() {
                let delimiter = delim_val.and_then(|d| d.chars().next()).unwrap_or(',');

                let file = std::fs::File::open(&c.file_path)
                    .map_err(|e| format!("Cannot open file '{}': {}", c.file_path, e))?;
                use std::io::{BufRead, BufReader};
                let mut reader = BufReader::new(file);
                let mut first_line = String::new();
                reader
                    .read_line(&mut first_line)
                    .map_err(|e| format!("Cannot read file '{}': {}", c.file_path, e))?;

                let trimmed = first_line.trim();
                if trimmed.is_empty() {
                    return Err(format!("File '{}' is empty, cannot validate header", c.file_path).into());
                }

                let csv_col_count = trimmed.split(delimiter).count();
                // For rel tables the file has [SRC, DST] plus user properties.
                let expected_col_count = if is_rel_table { columns.len() + 2 } else { columns.len() };
                if csv_col_count != expected_col_count {
                    return Err(format!(
                        "Column count mismatch: CSV header has {csv_col_count} columns \
                         but table '{}' has {expected_col_count} columns",
                        c.table_name,
                    )
                    .into());
                }
            }
        }

        Ok(BoundStatement::BoundCopyFrom(BoundCopyFrom {
            table_name: c.table_name,
            table_id,
            file_path: c.file_path,
            options: c.options,
            columns,
        }))
    }

    /// Bind COPY TO ΓÇö export query results to a file.
    fn bind_copy_to(&self, c: akar_parser::ast::CopyTo) -> Result<BoundStatement, BinderError> {
        // Bind the inner query
        let bound_query = match self.bind(Statement::Query(c.query))? {
            BoundStatement::BoundQuery(q) => q,
            _ => return Err("COPY TO inner statement must be a query".into()),
        };

        Ok(BoundStatement::BoundCopyTo(BoundCopyTo {
            file_path: c.file_path,
            format: c.format,
            header: c.header,
            query: bound_query,
        }))
    }
}

/// Convert an Expression AST to a debug string for storage.
/// Used by macro definition storage.
fn expr_to_debug_string(expr: &akar_parser::ast::Expression) -> String {
    format!("{:?}", expr)
}

#[cfg(test)]
mod tests {
    use super::*;
    use akar_parser::ast::{Constant, DeleteClause, Expression};

    fn binder() -> Binder {
        Binder::new(Arc::new(Mutex::new(Catalog::new())))
    }

    fn var(name: &str, table_id: u64, label: Option<&str>, is_node: bool) -> BoundVariable {
        BoundVariable {
            name: name.to_string(),
            table_id,
            label: label.map(|s| s.to_string()),
            is_node,
        }
    }

    fn delete_clause(expressions: Vec<Expression>) -> DeleteClause {
        DeleteClause {
            detach: true,
            expressions,
        }
    }

    #[test]
    fn delete_resolves_bound_variable_fields() {
        let b = binder();
        let variables = vec![var("a", 7, Some("Person"), true), var("r", 3, Some("Knows"), false)];
        let d = delete_clause(vec![Expression::Variable("a".into()), Expression::Variable("r".into())]);
        let BoundDeleteClause { detach, items } = b.bind_delete(&d, &variables).unwrap();
        {
            assert!(detach);
            assert_eq!(items.len(), 2);
            assert_eq!(items[0].table_name, "Person");
            assert_eq!(items[0].table_id, 7);
            assert!(items[0].is_node);
            assert_eq!(items[1].table_name, "Knows");
            assert_eq!(items[1].table_id, 3);
            assert!(!items[1].is_node);
        }
    }

    #[test]
    fn delete_missing_variable_reports_original_error() {
        let b = binder();
        let variables = vec![var("a", 1, Some("Person"), true)];
        let d = delete_clause(vec![Expression::Variable("ghost".into())]);
        let err = b.bind_delete(&d, &variables).unwrap_err();
        assert!(
            err.to_string()
                .contains("Variable 'ghost' not found in scope for DELETE"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn delete_rejects_non_variable_expression() {
        let b = binder();
        let variables = vec![var("a", 1, Some("Person"), true)];
        let d = delete_clause(vec![Expression::Constant(Constant::Integer(42))]);
        let err = b.bind_delete(&d, &variables).unwrap_err();
        assert!(
            err.to_string().contains("DELETE only supports variable references"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn delete_requires_at_least_one_expression() {
        let b = binder();
        let d = delete_clause(vec![]);
        let err = b.bind_delete(&d, &[]).unwrap_err();
        assert!(
            err.to_string().contains("at least one expression"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn delete_first_occurrence_wins_in_linear_path() {
        let b = binder();
        // N < 100 and M < 8 => linear scan path. A duplicate name must resolve
        // to the first occurrence, exactly like the pre-optimization code.
        let variables = vec![var("dup", 1, Some("First"), true), var("dup", 2, Some("Second"), false)];
        let d = delete_clause(vec![Expression::Variable("dup".into())]);
        let bound = b.bind_delete(&d, &variables).unwrap();
        assert_eq!(bound.items[0].table_id, 1, "first occurrence must win");
        assert_eq!(bound.items[0].table_name, "First");
    }

    #[test]
    fn delete_first_occurrence_wins_in_index_path() {
        let b = binder();
        // N >= 256 and M >= 8 => HashMap index path. The index must still
        // prefer the first occurrence (entry().or_insert()), matching the
        // linear-scan semantics.
        let mut variables: Vec<BoundVariable> = (0..300).map(|i| var(&format!("v{i}"), i as u64, None, true)).collect();
        variables[0].name = "dup".into();
        variables[0].table_id = 999;
        variables[150].name = "dup".into();
        variables[150].table_id = 555;
        let d = delete_clause(vec![Expression::Variable("dup".into()); 8]);
        let bound = b.bind_delete(&d, &variables).unwrap();
        assert_eq!(bound.items.len(), 8);
        for item in &bound.items {
            assert_eq!(item.table_id, 999, "first occurrence must win in the index path");
        }
    }

    // ==================== IMPORT DATABASE path-traversal hardening ====================

    fn unique_temp_dir(tag: &str) -> std::path::PathBuf {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let dir = std::env::temp_dir().join(format!("akar_binder_sec_{}_{nanos}_{tag}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    #[test]
    fn import_validator_rejects_parent_dir_components() {
        for evil in ["..", "../etc", "safe/../../../etc", "a/../b", "x/.."] {
            let err = Binder::validate_import_database_dir(evil).unwrap_err();
            assert!(
                err.to_string().contains("path traversal"),
                "'{evil}' must be rejected as traversal, got: {err}"
            );
        }
    }

    #[cfg(windows)]
    #[test]
    fn import_validator_rejects_backslash_traversal_on_windows() {
        for evil in ["..\\etc", "safe\\..\\..\\etc"] {
            let err = Binder::validate_import_database_dir(evil).unwrap_err();
            assert!(
                err.to_string().contains("path traversal"),
                "'{evil}' must be rejected as traversal, got: {err}"
            );
        }
    }

    #[test]
    fn import_validator_rejects_empty_and_nul_paths() {
        assert!(Binder::validate_import_database_dir("").is_err());
        assert!(Binder::validate_import_database_dir("dir\0x").is_err());
    }

    #[test]
    fn import_bind_rejects_traversal_before_filesystem_access() {
        let b = binder();
        let stmt = akar_parser::parse("IMPORT DATABASE 'x/../../y'").unwrap();
        let err = b.bind(stmt).unwrap_err();
        assert!(
            err.to_string().contains("path traversal"),
            "bind must reject traversal, got: {err}"
        );
    }

    #[test]
    fn import_read_helper_accepts_files_inside_directory() {
        let dir = unique_temp_dir("ok");
        std::fs::write(
            dir.join("schema.cypher"),
            "CREATE NODE TABLE T(id INT64, PRIMARY KEY(id));",
        )
        .unwrap();
        let canonical = std::fs::canonicalize(&dir).unwrap();
        let content = Binder::read_file_within_dir(&canonical, "schema.cypher").unwrap();
        assert!(content.contains("CREATE NODE TABLE T"));
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn import_read_helper_errors_for_missing_file() {
        let dir = unique_temp_dir("missing");
        let canonical = std::fs::canonicalize(&dir).unwrap();
        assert!(Binder::read_file_within_dir(&canonical, "schema.cypher").is_err());
        std::fs::remove_dir_all(&dir).ok();
    }

    #[cfg(unix)]
    #[test]
    fn import_read_helper_blocks_symlink_escape() {
        let dir = unique_temp_dir("sym");
        let outside = unique_temp_dir("outside");
        let secret = outside.join("secret.txt");
        std::fs::write(&secret, "top secret").unwrap();
        std::os::unix::fs::symlink(&secret, dir.join("schema.cypher")).unwrap();

        let canonical = std::fs::canonicalize(&dir).unwrap();
        let err = Binder::read_file_within_dir(&canonical, "schema.cypher").unwrap_err();
        assert!(
            err.to_string().contains("outside the import directory"),
            "symlink escape must be blocked, got: {err}"
        );
        std::fs::remove_dir_all(&dir).ok();
        std::fs::remove_dir_all(&outside).ok();
    }

    #[test]
    fn parse_type_array_with_dims_maps_to_list() {
        // P80: `FLOAT[384]` (array with a capacity) must map to List, matching
        // the existing empty-bracket `FLOAT[]` form and the Python translator.
        assert_eq!(Binder::parse_type("FLOAT[384]").unwrap(), LogicalTypeID::List);
        assert_eq!(Binder::parse_type("FLOAT[]").unwrap(), LogicalTypeID::List);
        assert_eq!(Binder::parse_type("FLOAT[384][]").unwrap(), LogicalTypeID::List);
        // Scalars without brackets must still parse to their primitive type.
        assert_eq!(Binder::parse_type("FLOAT").unwrap(), LogicalTypeID::Float);
        assert_eq!(Binder::parse_type("INT64").unwrap(), LogicalTypeID::Int64);
    }
}