//! SQL executor: DDL and DML operations.

mod aggregate;
pub(crate) mod ann_persist;
mod ann_topk;
pub(crate) mod compile;
mod correlated;
mod cte;
mod ddl;
mod dml;
mod explain;
pub(crate) mod helpers;
mod join;
pub(crate) mod matviews;
mod scan;
mod select;
pub(crate) mod triggers;
mod view;
mod window;
pub(crate) mod write;
pub use ann_persist::AnnSegmentInfo;
pub use ann_topk::AnnIndexSource;
pub(crate) use ann_topk::{ann_appends_safe, ann_dml_gen_key};
pub(crate) use ann_topk::{ann_cache_status, persist_ann_index};
pub use compile::{compile, CompiledPlan};
use cte::*;
use ddl::*;
pub use dml::exec_insert_in_txn;
use dml::*;
use explain::*;
use join::*;
use scan::*;
use select::*;
use view::*;
use window::*;
use write::*;

use citadel::Database;
use citadel_txn::read_txn::ReadTxn;
use rustc_hash::FxHashMap;

use crate::error::{Result, SqlError};
use crate::parser::*;
use crate::schema::SchemaManager;
use crate::types::*;

type CteContext = FxHashMap<String, QueryResult>;
type ScanTableFn<'a> = &'a mut dyn FnMut(&str) -> Result<(TableSchema, Vec<Vec<Value>>)>;

pub fn execute(
    db: &Database,
    schema: &mut SchemaManager,
    stmt: &Statement,
    params: &[Value],
) -> Result<ExecutionResult> {
    match stmt {
        Statement::CreateTable(ct) => exec_create_table(db, schema, ct),
        Statement::DropTable(dt) => exec_drop_table(db, schema, dt),
        Statement::CreateIndex(ci) => exec_create_index(db, schema, ci),
        Statement::DropIndex(di) => exec_drop_index(db, schema, di),
        Statement::CreateView(cv) => exec_create_view(db, schema, cv),
        Statement::DropView(dv) => exec_drop_view(db, schema, dv),
        Statement::AlterTable(at) => exec_alter_table(db, schema, at),
        Statement::Insert(ins) => exec_insert(db, schema, ins, params),
        Statement::Select(sq) => exec_select_query(db, schema, sq),
        Statement::Update(upd) => exec_update(db, schema, upd),
        Statement::Delete(del) => exec_delete(db, schema, del),
        Statement::Truncate(t) => exec_truncate(db, schema, t),
        Statement::Explain(inner) => explain(schema, inner),
        Statement::CreateTrigger(ct) => triggers::exec_create_trigger(db, schema, ct),
        Statement::DropTrigger(dt) => triggers::exec_drop_trigger(db, schema, dt),
        Statement::CreateMaterializedView(mv) => matviews::exec_create_matview(db, schema, mv),
        Statement::RefreshMaterializedView(rmv) => matviews::exec_refresh_matview(db, schema, rmv),
        Statement::DropMaterializedView(dmv) => matviews::exec_drop_matview(db, schema, dmv),
        Statement::Begin { .. }
        | Statement::Commit
        | Statement::Rollback
        | Statement::Savepoint(_)
        | Statement::ReleaseSavepoint(_)
        | Statement::RollbackTo(_)
        | Statement::SetTimezone(_) => Err(SqlError::Unsupported(
            "transaction / session control handled by Connection".into(),
        )),
    }
}

pub fn execute_with_read(
    rtx: &mut citadel_txn::read_txn::ReadTxn<'_>,
    schema: &SchemaManager,
    stmt: &Statement,
    _params: &[Value],
) -> Result<ExecutionResult> {
    match stmt {
        Statement::Select(sq) => cte::exec_select_query_with_read(rtx, schema, sq),
        Statement::Explain(inner) => explain(schema, inner),
        Statement::CreateTable(_)
        | Statement::DropTable(_)
        | Statement::CreateIndex(_)
        | Statement::DropIndex(_)
        | Statement::CreateView(_)
        | Statement::DropView(_)
        | Statement::AlterTable(_)
        | Statement::Insert(_)
        | Statement::Update(_)
        | Statement::Delete(_)
        | Statement::Truncate(_)
        | Statement::CreateTrigger(_)
        | Statement::DropTrigger(_)
        | Statement::CreateMaterializedView(_)
        | Statement::RefreshMaterializedView(_)
        | Statement::DropMaterializedView(_) => Err(SqlError::Unsupported(
            "cannot execute mutating statement inside a read-only transaction".into(),
        )),
        Statement::Begin { .. }
        | Statement::Commit
        | Statement::Rollback
        | Statement::Savepoint(_)
        | Statement::ReleaseSavepoint(_)
        | Statement::RollbackTo(_)
        | Statement::SetTimezone(_) => Err(SqlError::Unsupported(
            "transaction / session control handled by Connection".into(),
        )),
    }
}

/// Execute a parsed SQL statement within an existing write transaction.
pub fn execute_in_txn(
    wtx: &mut citadel_txn::write_txn::WriteTxn<'_>,
    schema: &mut SchemaManager,
    stmt: &Statement,
    params: &[Value],
) -> Result<ExecutionResult> {
    match stmt {
        Statement::CreateTable(ct) => exec_create_table_in_txn(wtx, schema, ct),
        Statement::DropTable(dt) => exec_drop_table_in_txn(wtx, schema, dt),
        Statement::CreateIndex(ci) => exec_create_index_in_txn(wtx, schema, ci),
        Statement::DropIndex(di) => exec_drop_index_in_txn(wtx, schema, di),
        Statement::CreateView(cv) => exec_create_view_in_txn(wtx, schema, cv),
        Statement::DropView(dv) => exec_drop_view_in_txn(wtx, schema, dv),
        Statement::AlterTable(at) => exec_alter_table_in_txn(wtx, schema, at),
        Statement::Insert(ins) => exec_insert_in_txn(wtx, schema, ins, params),
        Statement::Select(sq) => exec_select_query_in_txn(wtx, schema, sq),
        Statement::Update(upd) => exec_update_in_txn(wtx, schema, upd),
        Statement::Delete(del) => exec_delete_in_txn(wtx, schema, del),
        Statement::Truncate(t) => exec_truncate_in_txn(wtx, schema, t),
        Statement::Explain(inner) => explain(schema, inner),
        Statement::CreateTrigger(ct) => triggers::exec_create_trigger_in_txn(wtx, schema, ct),
        Statement::DropTrigger(dt) => triggers::exec_drop_trigger_in_txn(wtx, schema, dt),
        Statement::CreateMaterializedView(mv) => {
            matviews::exec_create_matview_in_txn(wtx, schema, mv)
        }
        Statement::RefreshMaterializedView(rmv) => {
            matviews::exec_refresh_matview_in_txn(wtx, schema, rmv)
        }
        Statement::DropMaterializedView(dmv) => {
            matviews::exec_drop_matview_in_txn(wtx, schema, dmv)
        }
        Statement::Begin { .. }
        | Statement::Commit
        | Statement::Rollback
        | Statement::Savepoint(_)
        | Statement::ReleaseSavepoint(_)
        | Statement::RollbackTo(_)
        | Statement::SetTimezone(_) => {
            Err(SqlError::Unsupported("nested transaction control".into()))
        }
    }
}

pub(super) fn scan_table_with_read(
    rtx: &mut ReadTxn<'_>,
    schema: &SchemaManager,
    name: &str,
) -> Result<(TableSchema, Vec<Vec<Value>>)> {
    let table_schema = schema
        .get(name)
        .ok_or_else(|| SqlError::TableNotFound(name.to_string()))?;
    let (rows, _) = collect_rows_with_read(rtx, table_schema, &None, None)?;
    Ok((table_schema.clone(), rows))
}

pub(super) fn scan_table_with_read_or_view(
    rtx: &mut ReadTxn<'_>,
    schema: &SchemaManager,
    name: &str,
) -> Result<(TableSchema, Vec<Vec<Value>>)> {
    if let Some(ts) = schema.get(name) {
        let (rows, _) = collect_rows_with_read(rtx, ts, &None, None)?;
        return Ok((ts.clone(), rows));
    }
    if let Some(vd) = schema.get_view(name) {
        let qr = exec_view_with_read(rtx, schema, vd)?;
        let vs = build_view_schema(name, &qr);
        return Ok((vs, qr.rows));
    }
    if let Some(vt) = schema.get_virtual(name) {
        let qr = vt.scan(schema)?;
        let vs = build_view_schema(name, &qr);
        return Ok((vs, qr.rows));
    }
    Err(SqlError::TableNotFound(name.to_string()))
}

pub(super) fn scan_table_write(
    wtx: &mut citadel_txn::write_txn::WriteTxn<'_>,
    schema: &SchemaManager,
    name: &str,
) -> Result<(TableSchema, Vec<Vec<Value>>)> {
    let table_schema = schema
        .get(name)
        .ok_or_else(|| SqlError::TableNotFound(name.to_string()))?;
    let (rows, _) = collect_rows_write(wtx, table_schema, &None, None)?;
    Ok((table_schema.clone(), rows))
}

pub(super) fn scan_table_write_or_view(
    wtx: &mut citadel_txn::write_txn::WriteTxn<'_>,
    schema: &SchemaManager,
    name: &str,
) -> Result<(TableSchema, Vec<Vec<Value>>)> {
    if let Some(ts) = schema.get(name) {
        let (rows, _) = collect_rows_write(wtx, ts, &None, None)?;
        return Ok((ts.clone(), rows));
    }
    if let Some(vd) = schema.get_view(name) {
        let qr = exec_view_write(wtx, schema, vd)?;
        let vs = build_view_schema(name, &qr);
        return Ok((vs, qr.rows));
    }
    Err(SqlError::TableNotFound(name.to_string()))
}

pub(super) fn resolve_table_or_cte(
    name: &str,
    ctes: &CteContext,
    scan_table: ScanTableFn<'_>,
) -> Result<(TableSchema, Vec<Vec<Value>>)> {
    let lower = name.to_ascii_lowercase();
    if let Some(cte_result) = ctes.get(&lower) {
        let schema = build_cte_schema(&lower, cte_result);
        Ok((schema, cte_result.rows.clone()))
    } else {
        scan_table(&lower)
    }
}

pub(super) fn exec_select_join_with_ctes(
    stmt: &SelectStmt,
    ctes: &CteContext,
    scan_table: ScanTableFn<'_>,
) -> Result<ExecutionResult> {
    let (from_schema, from_rows) = resolve_table_or_cte(&stmt.from, ctes, scan_table)?;
    let from_alias = table_alias_or_name(&stmt.from, &stmt.from_alias);

    let mut tables: Vec<(String, TableSchema)> = vec![(from_alias.clone(), from_schema)];
    let mut join_rows: Vec<Vec<Vec<Value>>> = Vec::new();

    for join in &stmt.joins {
        let jname = &join.table.name;
        let (js, jrows) = resolve_table_or_cte(jname, ctes, scan_table)?;
        let jalias = table_alias_or_name(jname, &join.table.alias);
        tables.push((jalias, js));
        join_rows.push(jrows);
    }

    let mut outer_rows = from_rows;
    let mut cur_tables: Vec<(String, &TableSchema)> = vec![(from_alias.clone(), &tables[0].1)];

    for (ji, join) in stmt.joins.iter().enumerate() {
        let inner_schema = &tables[ji + 1].1;
        let inner_alias = &tables[ji + 1].0;
        let inner_rows = &mut join_rows[ji];

        let mut preview_tables = cur_tables.clone();
        preview_tables.push((inner_alias.clone(), inner_schema));
        let combined_cols = build_joined_columns(&preview_tables);

        let outer_col_count = if outer_rows.is_empty() {
            cur_tables.iter().map(|(_, s)| s.columns.len()).sum()
        } else {
            outer_rows[0].len()
        };
        let inner_col_count = inner_schema.columns.len();

        let (equi_pairs, is_pure_equi) =
            compute_equi_join_meta(join, &combined_cols, outer_col_count);
        outer_rows = exec_join_step(
            outer_rows,
            inner_rows,
            join,
            &combined_cols,
            outer_col_count,
            inner_col_count,
            None,
            None,
            &equi_pairs,
            is_pure_equi,
        );
        cur_tables.push((inner_alias.clone(), inner_schema));
    }

    let joined_cols = build_joined_columns(&cur_tables);
    process_select(&joined_cols, outer_rows, stmt, false)
}
