sql_cli/

non_interactive.rs

use anyhow::{Context, Result};
use comfy_table::presets::*;
use comfy_table::{ContentArrangement, Table};
use std::fs;
use std::io::{self, Write};
use std::path::Path;
use std::time::Instant;
use tracing::{debug, info};

use crate::config::config::Config;
use crate::data::data_view::DataView;
use crate::data::datatable::{DataTable, DataValue};
use crate::data::datatable_loaders::{load_csv_to_datatable, load_json_to_datatable};
use crate::data::temp_table_registry::TempTableRegistry;
use crate::services::query_execution_service::QueryExecutionService;
use crate::sql::parser::ast::{CTEType, TableSource, CTE};
use crate::sql::recursive_parser::Parser;
use crate::sql::script_parser::{ScriptParser, ScriptResult};

/// Check if a query references temporary tables (starting with #)
/// Temporary tables are only valid in script mode
fn check_temp_table_usage(query: &str) -> Result<()> {
    use crate::sql::recursive_parser::Parser;

    let mut parser = Parser::new(query);
    match parser.parse() {
        Ok(stmt) => {
            // Check FROM clause
            if let Some(from_table) = &stmt.from_table {
                if from_table.starts_with('#') {
                    anyhow::bail!(
                        "Temporary table '{}' cannot be used in single-query mode. \
                        Temporary tables are only available in script mode (using -f flag with GO separators).",
                        from_table
                    );
                }
            }

            // Check INTO clause
            if let Some(into_table) = &stmt.into_table {
                anyhow::bail!(
                    "INTO clause for temporary table '{}' is only supported in script mode. \
                    Use -f flag with GO separators to create temporary tables.",
                    into_table.name
                );
            }

            Ok(())
        }
        Err(_) => {
            // If parse fails, let it fail later in the actual execution
            Ok(())
        }
    }
}

/// Extract dependencies from a CTE by analyzing what tables it references
fn extract_cte_dependencies(cte: &CTE) -> Vec<String> {
    // For now, we'll just return the from_table if it exists
    // This could be enhanced to do deeper analysis of the query AST
    let mut deps = Vec::new();

    if let CTEType::Standard(query) = &cte.cte_type {
        if let Some(from_table) = &query.from_table {
            deps.push(from_table.clone());
        }

        // Could also check joins, subqueries, etc.
        for join in &query.joins {
            match &join.table {
                TableSource::Table(table_name) => {
                    deps.push(table_name.clone());
                }
                TableSource::DerivedTable { alias, .. } => {
                    deps.push(alias.clone());
                }
            }
        }
    }

    deps
}

/// Output format for query results
#[derive(Debug, Clone)]
pub enum OutputFormat {
    Csv,
    Json,
    JsonStructured, // Structured JSON with metadata for IDE/plugin integration
    Table,
    Tsv,
}

impl OutputFormat {
    pub fn from_str(s: &str) -> Result<Self> {
        match s.to_lowercase().as_str() {
            "csv" => Ok(OutputFormat::Csv),
            "json" => Ok(OutputFormat::Json),
            "json-structured" => Ok(OutputFormat::JsonStructured),
            "table" => Ok(OutputFormat::Table),
            "tsv" => Ok(OutputFormat::Tsv),
            _ => Err(anyhow::anyhow!(
                "Invalid output format: {}. Use csv, json, json-structured, table, or tsv",
                s
            )),
        }
    }
}

/// Table styling presets for table output format
#[derive(Debug, Clone, Copy)]
pub enum TableStyle {
    /// Current default ASCII style with borders
    Default,
    /// ASCII table with full borders
    AsciiFull,
    /// ASCII table with condensed rows
    AsciiCondensed,
    /// ASCII table with only outer borders
    AsciiBordersOnly,
    /// ASCII table with horizontal lines only
    AsciiHorizontalOnly,
    /// ASCII table with no borders
    AsciiNoBorders,
    /// Markdown-style table
    Markdown,
    /// UTF8 table with box-drawing characters
    Utf8Full,
    /// UTF8 table with condensed rows
    Utf8Condensed,
    /// UTF8 table with only outer borders
    Utf8BordersOnly,
    /// UTF8 table with horizontal lines only
    Utf8HorizontalOnly,
    /// UTF8 table with no borders
    Utf8NoBorders,
    /// No table formatting, just data
    Plain,
}

impl TableStyle {
    pub fn from_str(s: &str) -> Result<Self> {
        match s.to_lowercase().as_str() {
            "default" => Ok(TableStyle::Default),
            "ascii" | "ascii-full" => Ok(TableStyle::AsciiFull),
            "ascii-condensed" => Ok(TableStyle::AsciiCondensed),
            "ascii-borders" | "ascii-borders-only" => Ok(TableStyle::AsciiBordersOnly),
            "ascii-horizontal" | "ascii-horizontal-only" => Ok(TableStyle::AsciiHorizontalOnly),
            "ascii-noborders" | "ascii-no-borders" => Ok(TableStyle::AsciiNoBorders),
            "markdown" | "md" => Ok(TableStyle::Markdown),
            "utf8" | "utf8-full" => Ok(TableStyle::Utf8Full),
            "utf8-condensed" => Ok(TableStyle::Utf8Condensed),
            "utf8-borders" | "utf8-borders-only" => Ok(TableStyle::Utf8BordersOnly),
            "utf8-horizontal" | "utf8-horizontal-only" => Ok(TableStyle::Utf8HorizontalOnly),
            "utf8-noborders" | "utf8-no-borders" => Ok(TableStyle::Utf8NoBorders),
            "plain" | "none" => Ok(TableStyle::Plain),
            _ => Err(anyhow::anyhow!(
                "Invalid table style: {}. Use: default, ascii, ascii-condensed, ascii-borders, ascii-horizontal, ascii-noborders, markdown, utf8, utf8-condensed, utf8-borders, utf8-horizontal, utf8-noborders, plain",
                s
            )),
        }
    }

    pub fn list_styles() -> &'static str {
        "Available table styles:
  default            - Current default ASCII style
  ascii              - ASCII with full borders
  ascii-condensed    - ASCII with condensed rows
  ascii-borders      - ASCII with outer borders only
  ascii-horizontal   - ASCII with horizontal lines only
  ascii-noborders    - ASCII with no borders
  markdown           - GitHub-flavored Markdown table
  utf8               - UTF8 box-drawing characters
  utf8-condensed     - UTF8 with condensed rows
  utf8-borders       - UTF8 with outer borders only
  utf8-horizontal    - UTF8 with horizontal lines only
  utf8-noborders     - UTF8 with no borders
  plain              - No formatting, data only"
    }
}

/// Configuration for non-interactive query execution
pub struct NonInteractiveConfig {
    pub data_file: String,
    pub query: String,
    pub output_format: OutputFormat,
    pub output_file: Option<String>,
    pub case_insensitive: bool,
    pub auto_hide_empty: bool,
    pub limit: Option<usize>,
    pub query_plan: bool,
    pub show_work_units: bool,
    pub execution_plan: bool,
    pub cte_info: bool,
    pub rewrite_analysis: bool,
    pub lift_in_expressions: bool,
    pub script_file: Option<String>, // Path to the script file for relative path resolution
    pub debug_trace: bool,
    pub max_col_width: Option<usize>, // Maximum column width for table output (None = unlimited)
    pub col_sample_rows: usize,       // Number of rows to sample for column width (0 = all rows)
    pub table_style: TableStyle,      // Table styling preset (only affects table output format)
}

/// Execute a query in non-interactive mode
pub fn execute_non_interactive(config: NonInteractiveConfig) -> Result<()> {
    let start_time = Instant::now();

    // Check if query tries to use temporary tables (only valid in script mode)
    check_temp_table_usage(&config.query)?;

    // 1. Load the data file or create DUAL table
    let (data_table, _is_dual) = if config.data_file.is_empty() {
        info!("No data file provided, using DUAL table");
        (crate::data::datatable::DataTable::dual(), true)
    } else {
        info!("Loading data from: {}", config.data_file);
        let table = load_data_file(&config.data_file)?;
        info!(
            "Loaded {} rows with {} columns",
            table.row_count(),
            table.column_count()
        );
        (table, false)
    };
    let _table_name = data_table.name.clone();

    // 2. Create a DataView from the table
    let dataview = DataView::new(std::sync::Arc::new(data_table));

    // 3. Execute the query
    info!("Executing query: {}", config.query);

    // If execution_plan is requested, show detailed execution information
    if config.execution_plan {
        println!("\n=== EXECUTION PLAN ===");
        println!("Query: {}", config.query);
        println!("\nExecution Steps:");
        println!("1. PARSE - Parse SQL query");
        println!("2. LOAD_DATA - Load data from {}", &config.data_file);
        println!(
            "   • Loaded {} rows, {} columns",
            dataview.row_count(),
            dataview.column_count()
        );
    }

    // If show_work_units is requested, analyze and display work units
    if config.show_work_units {
        use crate::query_plan::{ExpressionLifter, QueryAnalyzer};
        use crate::sql::recursive_parser::Parser;

        let mut parser = Parser::new(&config.query);
        match parser.parse() {
            Ok(stmt) => {
                let mut analyzer = QueryAnalyzer::new();
                let mut lifter = ExpressionLifter::new();

                // Check if the query has liftable expressions
                let mut stmt_copy = stmt.clone();
                let lifted = lifter.lift_expressions(&mut stmt_copy);

                // Build the query plan
                match analyzer.analyze(&stmt_copy, config.query.clone()) {
                    Ok(plan) => {
                        println!("\n{}", plan.explain());

                        if !lifted.is_empty() {
                            println!("\nLifted CTEs:");
                            for cte in &lifted {
                                println!("  - {}", cte.name);
                            }
                        }

                        return Ok(());
                    }
                    Err(e) => {
                        eprintln!("Error analyzing query: {}", e);
                        return Err(anyhow::anyhow!("Query analysis failed: {}", e));
                    }
                }
            }
            Err(e) => {
                eprintln!("Error parsing query: {}", e);
                return Err(anyhow::anyhow!("Parse error: {}", e));
            }
        }
    }

    // If query_plan is requested, parse and display the AST
    if config.query_plan {
        use crate::sql::recursive_parser::Parser;
        let mut parser = Parser::new(&config.query);
        match parser.parse() {
            Ok(statement) => {
                println!("\n=== QUERY PLAN (AST) ===");
                println!("{statement:#?}");
                println!("=== END QUERY PLAN ===\n");
            }
            Err(e) => {
                eprintln!("Failed to parse query for plan: {e}");
            }
        }
    }

    // If rewrite analysis is requested, analyze query for optimization opportunities
    if config.rewrite_analysis {
        use crate::sql::query_rewriter::{QueryRewriter, RewriteAnalysis};
        use crate::sql::recursive_parser::Parser;
        use serde_json::json;

        let mut parser = Parser::new(&config.query);
        match parser.parse() {
            Ok(statement) => {
                let mut rewriter = QueryRewriter::new();
                let suggestions = rewriter.analyze(&statement);

                let analysis = RewriteAnalysis::from_suggestions(suggestions);
                println!("{}", serde_json::to_string_pretty(&analysis).unwrap());
                return Ok(());
            }
            Err(e) => {
                let output = json!({
                    "success": false,
                    "error": format!("{}", e),
                    "suggestions": [],
                    "can_auto_rewrite": false,
                });
                println!("{}", serde_json::to_string_pretty(&output).unwrap());
                return Ok(());
            }
        }
    }

    // If CTE info is requested, parse and output CTE information as JSON
    if config.cte_info {
        use crate::sql::recursive_parser::Parser;
        use serde_json::json;

        let mut parser = Parser::new(&config.query);
        match parser.parse() {
            Ok(statement) => {
                let mut cte_info = Vec::new();

                // Extract CTE information
                for (index, cte) in statement.ctes.iter().enumerate() {
                    let cte_json = json!({
                        "index": index,
                        "name": cte.name,
                        "columns": cte.column_list,
                        // We can't easily get line numbers from the AST,
                        // but we can provide the structure
                        "dependencies": extract_cte_dependencies(cte),
                    });
                    cte_info.push(cte_json);
                }

                let output = json!({
                    "success": true,
                    "ctes": cte_info,
                    "total": statement.ctes.len(),
                    "has_final_select": !statement.columns.is_empty() || !statement.select_items.is_empty(),
                });

                println!("{}", serde_json::to_string_pretty(&output).unwrap());
                return Ok(());
            }
            Err(e) => {
                let output = json!({
                    "success": false,
                    "error": format!("{}", e),
                    "ctes": [],
                    "total": 0,
                });
                println!("{}", serde_json::to_string_pretty(&output).unwrap());
                return Ok(());
            }
        }
    }

    let query_start = Instant::now();

    // Load configuration file to get date notation and other settings
    let app_config = Config::load().unwrap_or_else(|e| {
        debug!("Could not load config file: {}. Using defaults.", e);
        Config::default()
    });

    // Initialize global config for function registry
    crate::config::global::init_config(app_config.clone());

    // Use QueryExecutionService with full BehaviorConfig
    let mut behavior_config = app_config.behavior.clone();
    debug!(
        "Using date notation: {}",
        behavior_config.default_date_notation
    );
    // Command line args override config file settings
    if config.case_insensitive {
        behavior_config.case_insensitive_default = true;
    }
    if config.auto_hide_empty {
        behavior_config.hide_empty_columns = true;
    }

    // Parse and potentially rewrite the query
    let exec_start = Instant::now();
    let result = if config.lift_in_expressions {
        use crate::data::query_engine::QueryEngine;
        use crate::query_plan::{CTEHoister, InOperatorLifter};
        use crate::sql::recursive_parser::Parser;

        let mut parser = Parser::new(&config.query);
        match parser.parse() {
            Ok(mut stmt) => {
                // Apply expression lifting for column alias dependencies
                use crate::query_plan::ExpressionLifter;
                let mut expr_lifter = ExpressionLifter::new();
                let lifted = expr_lifter.lift_expressions(&mut stmt);
                if !lifted.is_empty() {
                    info!(
                        "Applied expression lifting - {} CTEs generated",
                        lifted.len()
                    );
                }

                // Apply CTE hoisting to handle nested CTEs
                stmt = CTEHoister::hoist_ctes(stmt);

                // Try to rewrite the query with IN lifting
                let mut lifter = InOperatorLifter::new();
                if lifter.rewrite_query(&mut stmt) {
                    info!("Applied IN expression lifting - query rewritten with CTEs");

                    // Execute the rewritten AST directly
                    let engine = QueryEngine::with_case_insensitive(config.case_insensitive);

                    match engine.execute_statement(dataview.source_arc(), stmt) {
                        Ok(result_view) => {
                            // Create a QueryExecutionResult to match the expected type
                            Ok(
                                crate::services::query_execution_service::QueryExecutionResult {
                                    dataview: result_view,
                                    stats: crate::services::query_execution_service::QueryStats {
                                        row_count: 0, // Will be filled by result_view
                                        column_count: 0,
                                        execution_time: exec_start.elapsed(),
                                        query_engine_time: exec_start.elapsed(),
                                    },
                                    hidden_columns: Vec::new(),
                                    query: config.query.clone(),
                                    execution_plan: None,
                                    debug_trace: None,
                                },
                            )
                        }
                        Err(e) => Err(e),
                    }
                } else {
                    // No lifting needed, execute normally
                    let query_service =
                        QueryExecutionService::with_behavior_config(behavior_config);
                    if config.debug_trace {
                        let debug_ctx = crate::debug_trace::DebugContext::new(
                            crate::debug_trace::DebugLevel::Debug,
                        );
                        query_service.execute_with_debug(
                            &config.query,
                            Some(&dataview),
                            Some(dataview.source()),
                            Some(debug_ctx),
                        )
                    } else {
                        query_service.execute(
                            &config.query,
                            Some(&dataview),
                            Some(dataview.source()),
                        )
                    }
                }
            }
            Err(_) => {
                // Parse failed, execute normally and let it fail with proper error
                let query_service = QueryExecutionService::with_behavior_config(behavior_config);
                if config.debug_trace {
                    let debug_ctx = crate::debug_trace::DebugContext::new(
                        crate::debug_trace::DebugLevel::Debug,
                    );
                    query_service.execute_with_debug(
                        &config.query,
                        Some(&dataview),
                        Some(dataview.source()),
                        Some(debug_ctx),
                    )
                } else {
                    query_service.execute(&config.query, Some(&dataview), Some(dataview.source()))
                }
            }
        }
    } else {
        // Normal execution without lifting
        let query_service = QueryExecutionService::with_behavior_config(behavior_config);

        // Create debug context if debug trace is enabled
        if config.debug_trace {
            let debug_ctx =
                crate::debug_trace::DebugContext::new(crate::debug_trace::DebugLevel::Debug);
            query_service.execute_with_debug(
                &config.query,
                Some(&dataview),
                Some(dataview.source()),
                Some(debug_ctx),
            )
        } else {
            query_service.execute(&config.query, Some(&dataview), Some(dataview.source()))
        }
    }?;
    let exec_time = exec_start.elapsed();

    let query_time = query_start.elapsed();
    info!("Query executed in {:?}", query_time);
    info!(
        "Result: {} rows, {} columns",
        result.dataview.row_count(),
        result.dataview.column_count()
    );

    // Show execution plan details if requested
    if config.execution_plan {
        // Try to get detailed execution plan
        use crate::data::query_engine::QueryEngine;

        let query_engine = QueryEngine::new();

        match query_engine.execute_with_plan(
            std::sync::Arc::new(dataview.source().clone()),
            &config.query,
        ) {
            Ok((_view, plan)) => {
                // Display the detailed execution plan tree
                print!("{}", plan.format_tree());
            }
            Err(e) => {
                // Fall back to simple execution plan display
                eprintln!("Could not generate detailed execution plan: {}", e);
                println!(
                    "3. QUERY_EXECUTION [{:.3}ms]",
                    exec_time.as_secs_f64() * 1000.0
                );

                // Parse query to understand what operations are being performed
                use crate::sql::recursive_parser::Parser;
                let mut parser = Parser::new(&config.query);
                if let Ok(stmt) = parser.parse() {
                    if stmt.where_clause.is_some() {
                        println!("   • WHERE clause filtering applied");
                        println!("   • Rows after filter: {}", result.dataview.row_count());
                    }

                    if let Some(ref order_by) = stmt.order_by {
                        println!("   • ORDER BY: {} column(s)", order_by.len());
                    }

                    if let Some(ref group_by) = stmt.group_by {
                        println!("   • GROUP BY: {} column(s)", group_by.len());
                    }

                    if let Some(limit) = stmt.limit {
                        println!("   • LIMIT: {} rows", limit);
                    }

                    if stmt.distinct {
                        println!("   • DISTINCT applied");
                    }
                }
            }
        }

        println!("\nExecution Statistics:");
        println!(
            "  Preparation:    {:.3}ms",
            (exec_start - start_time).as_secs_f64() * 1000.0
        );
        println!(
            "  Query time:     {:.3}ms",
            exec_time.as_secs_f64() * 1000.0
        );
        println!(
            "  Total time:     {:.3}ms",
            query_time.as_secs_f64() * 1000.0
        );
        println!("  Rows returned:  {}", result.dataview.row_count());
        println!("  Columns:        {}", result.dataview.column_count());
        println!("\n=== END EXECUTION PLAN ===");
        println!();
    }

    // 4. Apply limit if specified
    let final_view = if let Some(limit) = config.limit {
        let limited_table = limit_results(&result.dataview, limit)?;
        DataView::new(std::sync::Arc::new(limited_table))
    } else {
        result.dataview
    };

    // 5. Output debug trace if enabled
    if let Some(ref trace_output) = result.debug_trace {
        eprintln!("{}", trace_output);
    }

    // 6. Output the results
    let exec_time_ms = exec_time.as_secs_f64() * 1000.0;
    let output_result = if let Some(ref path) = config.output_file {
        let mut file = fs::File::create(path)
            .with_context(|| format!("Failed to create output file: {path}"))?;
        output_results(
            &final_view,
            config.output_format,
            &mut file,
            config.max_col_width,
            config.col_sample_rows,
            exec_time_ms,
            config.table_style,
        )?;
        info!("Results written to: {}", path);
        Ok(())
    } else {
        output_results(
            &final_view,
            config.output_format,
            &mut io::stdout(),
            config.max_col_width,
            config.col_sample_rows,
            exec_time_ms,
            config.table_style,
        )?;
        Ok(())
    };

    let total_time = start_time.elapsed();
    debug!("Total execution time: {:?}", total_time);

    // Print stats to stderr so they don't interfere with output
    if config.output_file.is_none() {
        eprintln!(
            "\n# Query completed: {} rows in {:?}",
            final_view.row_count(),
            query_time
        );
    }

    output_result
}

/// Execute a script file with multiple SQL statements separated by GO
pub fn execute_script(config: NonInteractiveConfig) -> Result<()> {
    let _start_time = Instant::now();

    // Parse the script into individual statements with directives
    let parser = ScriptParser::new(&config.query);
    let script_statements = parser.parse_script_statements();

    if script_statements.is_empty() {
        anyhow::bail!("No statements found in script");
    }

    info!("Found {} statements in script", script_statements.len());

    // Determine data file to use (command-line overrides script hint)
    let data_file = if !config.data_file.is_empty() {
        // Command-line argument takes precedence
        config.data_file.clone()
    } else if let Some(hint) = parser.data_file_hint() {
        // Use data file hint from script
        info!("Using data file from script hint: {}", hint);

        // Resolve relative paths relative to script file if provided
        if let Some(script_path) = config.script_file.as_ref() {
            let script_dir = std::path::Path::new(script_path)
                .parent()
                .unwrap_or(std::path::Path::new("."));
            let hint_path = std::path::Path::new(hint);

            if hint_path.is_relative() {
                script_dir.join(hint_path).to_string_lossy().to_string()
            } else {
                hint.to_string()
            }
        } else {
            hint.to_string()
        }
    } else {
        String::new()
    };

    // Load the data file if provided, otherwise use DUAL
    let (data_table, _is_dual) = if data_file.is_empty() {
        // No data file provided, use DUAL table
        info!("No data file provided, using DUAL table");
        (DataTable::dual(), true)
    } else {
        // Check if file exists before trying to load
        if !std::path::Path::new(&data_file).exists() {
            anyhow::bail!(
                "Data file not found: {}\n\
                Please check the path is correct",
                data_file
            );
        }

        info!("Loading data from: {}", data_file);
        let table = load_data_file(&data_file)?;
        info!(
            "Loaded {} rows with {} columns",
            table.row_count(),
            table.column_count()
        );
        (table, false)
    };

    // Track script results
    let mut script_result = ScriptResult::new();
    let mut output = Vec::new();

    // Create Arc<DataTable> once for all statements - avoids expensive cloning
    let arc_data_table = std::sync::Arc::new(data_table);

    // Create temporary table registry for this script execution
    let mut temp_tables = TempTableRegistry::new();

    // Execute each statement
    for (idx, script_stmt) in script_statements.iter().enumerate() {
        let statement_num = idx + 1;
        let stmt_start = Instant::now();

        // Check if this is an EXIT statement
        if script_stmt.is_exit() {
            let exit_code = script_stmt.get_exit_code().unwrap_or(0);
            info!("EXIT statement encountered (code: {})", exit_code);

            // Print message for table format
            if matches!(config.output_format, OutputFormat::Table) {
                if idx > 0 {
                    output.push(String::new());
                }
                output.push(format!("-- Statement {} --", statement_num));
                output.push(format!("Script execution stopped by EXIT {}", exit_code));
            }

            // Mark as success in results
            script_result.add_success(
                statement_num,
                format!("EXIT {}", exit_code),
                0,
                stmt_start.elapsed().as_secs_f64() * 1000.0,
            );

            // Stop execution
            break;
        }

        // Check if this statement should be skipped
        if script_stmt.should_skip() {
            info!(
                "Skipping statement {} due to [SKIP] directive",
                statement_num
            );

            // Print message for table format
            if matches!(config.output_format, OutputFormat::Table) {
                if idx > 0 {
                    output.push(String::new());
                }
                output.push(format!("-- Statement {} [SKIPPED] --", statement_num));
            }

            // Mark as success but with 0 rows
            script_result.add_success(
                statement_num,
                "[SKIPPED]".to_string(),
                0,
                stmt_start.elapsed().as_secs_f64() * 1000.0,
            );

            continue;
        }

        // Get the SQL query
        let statement = match script_stmt.get_query() {
            Some(sql) => sql,
            None => continue, // Should not happen
        };

        // Print separator for table format
        if matches!(config.output_format, OutputFormat::Table) {
            if idx > 0 {
                output.push(String::new()); // Empty line between queries
            }
            output.push(format!("-- Query {} --", statement_num));
        }

        // Step 1: Expand templates in the SQL string FIRST (before parsing)
        // This ensures templates are expanded even when the SQL is re-parsed internally
        use crate::sql::template_expander::TemplateExpander;
        let expander = TemplateExpander::new(&temp_tables);

        let expanded_statement = match expander.parse_templates(statement) {
            Ok(vars) => {
                if vars.is_empty() {
                    statement.to_string()
                } else {
                    match expander.expand(statement, &vars) {
                        Ok(expanded) => {
                            debug!("Expanded templates in SQL: {} vars found", vars.len());
                            for var in &vars {
                                debug!(
                                    "  {} -> expanding from {}",
                                    var.placeholder, var.table_name
                                );
                            }
                            expanded
                        }
                        Err(e) => {
                            script_result.add_failure(
                                statement_num,
                                statement.to_string(),
                                format!("Template expansion error: {}", e),
                                stmt_start.elapsed().as_secs_f64() * 1000.0,
                            );
                            continue; // Skip this statement
                        }
                    }
                }
            }
            Err(e) => {
                script_result.add_failure(
                    statement_num,
                    statement.to_string(),
                    format!("Template parse error: {}", e),
                    stmt_start.elapsed().as_secs_f64() * 1000.0,
                );
                continue; // Skip this statement
            }
        };

        // Use the expanded statement for the rest of the processing
        let statement = expanded_statement.as_str();

        // Step 2: Parse the (possibly expanded) statement to check for INTO clause and temp table references
        let mut parser = Parser::new(statement);
        let parsed_stmt = match parser.parse() {
            Ok(stmt) => stmt,
            Err(e) => {
                // If parsing fails, record error and stop (scripts stop on first error)
                script_result.add_failure(
                    statement_num,
                    statement.to_string(),
                    format!("Parse error: {}", e),
                    stmt_start.elapsed().as_secs_f64() * 1000.0,
                );
                break;
            }
        };

        // Check if FROM clause references a temp table
        let source_table = if let Some(from_table) = &parsed_stmt.from_table {
            if from_table.starts_with('#') {
                // Resolve temp table
                match temp_tables.get(from_table) {
                    Some(temp_table) => temp_table,
                    None => {
                        script_result.add_failure(
                            statement_num,
                            statement.to_string(),
                            format!("Temporary table {} not found", from_table),
                            stmt_start.elapsed().as_secs_f64() * 1000.0,
                        );
                        break;
                    }
                }
            } else {
                // Use the base table
                arc_data_table.clone()
            }
        } else {
            // No FROM clause, use base table
            arc_data_table.clone()
        };

        // Create a fresh DataView for each statement
        let dataview = DataView::new(source_table);

        // If this statement has an INTO clause, we need to remove it before execution
        // because the query executor doesn't understand INTO syntax
        let executable_sql = if parsed_stmt.into_table.is_some() {
            // Remove the INTO clause using AST preprocessing (robust and maintainable)
            use crate::query_plan::IntoClauseRemover;
            use crate::sql::parser::ast_formatter;

            let cleaned_stmt = IntoClauseRemover::remove_into_clause(parsed_stmt.clone());
            ast_formatter::format_select_statement(&cleaned_stmt)
        } else {
            statement.to_string()
        };

        // Execute the statement
        let service = QueryExecutionService::new(config.case_insensitive, config.auto_hide_empty);
        match service.execute_with_temp_tables(
            &executable_sql,
            Some(&dataview),
            None,
            Some(&temp_tables),
        ) {
            Ok(result) => {
                let exec_time = stmt_start.elapsed().as_secs_f64() * 1000.0;
                let final_view = result.dataview;

                // Check if this is an INTO statement - store result in temp table
                if let Some(into_table) = &parsed_stmt.into_table {
                    // Get the source table from the DataView - this contains the query result
                    let result_table = final_view.source_arc();
                    let row_count = result_table.row_count();

                    match temp_tables.insert(into_table.name.clone(), result_table) {
                        Ok(_) => {
                            info!(
                                "Stored {} rows in temporary table {}",
                                row_count, into_table.name
                            );

                            // For INTO statements, output a simple confirmation message
                            let mut statement_output = Vec::new();
                            writeln!(
                                &mut statement_output,
                                "({} rows affected) -> {}",
                                row_count, into_table.name
                            )?;
                            output.extend(
                                String::from_utf8_lossy(&statement_output)
                                    .lines()
                                    .map(String::from),
                            );

                            script_result.add_success(
                                statement_num,
                                statement.to_string(),
                                row_count,
                                exec_time,
                            );
                            continue; // Skip normal output formatting for INTO statements
                        }
                        Err(e) => {
                            script_result.add_failure(
                                statement_num,
                                statement.to_string(),
                                e.to_string(),
                                exec_time,
                            );
                            break;
                        }
                    }
                }

                // Format the output based on the output format (normal query without INTO)
                let mut statement_output = Vec::new();
                match config.output_format {
                    OutputFormat::Csv => {
                        output_csv(&final_view, &mut statement_output, ',')?;
                    }
                    OutputFormat::Json => {
                        output_json(&final_view, &mut statement_output)?;
                    }
                    OutputFormat::JsonStructured => {
                        output_json_structured(&final_view, &mut statement_output, exec_time)?;
                    }
                    OutputFormat::Table => {
                        output_table(
                            &final_view,
                            &mut statement_output,
                            config.max_col_width,
                            config.col_sample_rows,
                            config.table_style,
                        )?;
                        writeln!(
                            &mut statement_output,
                            "Query completed: {} rows in {:.2}ms",
                            final_view.row_count(),
                            exec_time
                        )?;
                    }
                    OutputFormat::Tsv => {
                        output_csv(&final_view, &mut statement_output, '\t')?;
                    }
                }

                // Add to overall output
                output.extend(
                    String::from_utf8_lossy(&statement_output)
                        .lines()
                        .map(String::from),
                );

                script_result.add_success(
                    statement_num,
                    statement.to_string(),
                    final_view.row_count(),
                    exec_time,
                );
            }
            Err(e) => {
                let exec_time = stmt_start.elapsed().as_secs_f64() * 1000.0;
                let error_msg = format!("Query {} failed: {}", statement_num, e);

                if matches!(config.output_format, OutputFormat::Table) {
                    output.push(error_msg.clone());
                }

                script_result.add_failure(
                    statement_num,
                    statement.to_string(),
                    e.to_string(),
                    exec_time,
                );

                // Continue to next statement (don't stop on error)
            }
        }
    }

    // Write output
    if let Some(ref output_file) = config.output_file {
        let mut file = fs::File::create(output_file)?;
        for line in &output {
            writeln!(file, "{}", line)?;
        }
        info!("Results written to: {}", output_file);
    } else {
        for line in &output {
            println!("{}", line);
        }
    }

    // Print summary if in table mode
    if matches!(config.output_format, OutputFormat::Table) {
        println!("\n=== Script Summary ===");
        println!("Total statements: {}", script_result.total_statements);
        println!("Successful: {}", script_result.successful_statements);
        println!("Failed: {}", script_result.failed_statements);
        println!(
            "Total execution time: {:.2}ms",
            script_result.total_execution_time_ms
        );
    }

    if !script_result.all_successful() {
        return Err(anyhow::anyhow!(
            "{} of {} statements failed",
            script_result.failed_statements,
            script_result.total_statements
        ));
    }

    Ok(())
}

/// Load a data file (CSV or JSON) into a `DataTable`
fn load_data_file(path: &str) -> Result<DataTable> {
    let path = Path::new(path);

    if !path.exists() {
        return Err(anyhow::anyhow!("File not found: {}", path.display()));
    }

    // Determine file type by extension
    let extension = path
        .extension()
        .and_then(|ext| ext.to_str())
        .map(str::to_lowercase)
        .unwrap_or_default();

    let table_name = path
        .file_stem()
        .and_then(|stem| stem.to_str())
        .unwrap_or("data")
        .to_string();

    match extension.as_str() {
        "csv" => load_csv_to_datatable(path, &table_name)
            .with_context(|| format!("Failed to load CSV file: {}", path.display())),
        "json" => load_json_to_datatable(path, &table_name)
            .with_context(|| format!("Failed to load JSON file: {}", path.display())),
        _ => Err(anyhow::anyhow!(
            "Unsupported file type: {}. Use .csv or .json",
            extension
        )),
    }
}

/// Limit the number of rows in results
fn limit_results(dataview: &DataView, limit: usize) -> Result<DataTable> {
    let source = dataview.source();
    let mut limited_table = DataTable::new(&source.name);

    // Copy columns
    for col in &source.columns {
        limited_table.add_column(col.clone());
    }

    // Copy limited rows
    let rows_to_copy = dataview.row_count().min(limit);
    for i in 0..rows_to_copy {
        if let Some(row) = dataview.get_row(i) {
            let _ = limited_table.add_row(row.clone());
        }
    }

    Ok(limited_table)
}

/// Output query results in the specified format
fn output_results<W: Write>(
    dataview: &DataView,
    format: OutputFormat,
    writer: &mut W,
    max_col_width: Option<usize>,
    col_sample_rows: usize,
    exec_time_ms: f64,
    table_style: TableStyle,
) -> Result<()> {
    match format {
        OutputFormat::Csv => output_csv(dataview, writer, ','),
        OutputFormat::Tsv => output_csv(dataview, writer, '\t'),
        OutputFormat::Json => output_json(dataview, writer),
        OutputFormat::JsonStructured => output_json_structured(dataview, writer, exec_time_ms),
        OutputFormat::Table => output_table(
            dataview,
            writer,
            max_col_width,
            col_sample_rows,
            table_style,
        ),
    }
}

/// Output results as CSV/TSV
fn output_csv<W: Write>(dataview: &DataView, writer: &mut W, delimiter: char) -> Result<()> {
    // Write headers
    let columns = dataview.column_names();
    for (i, col) in columns.iter().enumerate() {
        if i > 0 {
            write!(writer, "{delimiter}")?;
        }
        write!(writer, "{}", escape_csv_field(col, delimiter))?;
    }
    writeln!(writer)?;

    // Write rows
    for row_idx in 0..dataview.row_count() {
        if let Some(row) = dataview.get_row(row_idx) {
            for (i, value) in row.values.iter().enumerate() {
                if i > 0 {
                    write!(writer, "{delimiter}")?;
                }
                write!(
                    writer,
                    "{}",
                    escape_csv_field(&format_value(value), delimiter)
                )?;
            }
            writeln!(writer)?;
        }
    }

    Ok(())
}

/// Output results as JSON
fn output_json<W: Write>(dataview: &DataView, writer: &mut W) -> Result<()> {
    let columns = dataview.column_names();
    let mut rows = Vec::new();

    for row_idx in 0..dataview.row_count() {
        if let Some(row) = dataview.get_row(row_idx) {
            let mut json_row = serde_json::Map::new();
            for (col_idx, value) in row.values.iter().enumerate() {
                if col_idx < columns.len() {
                    json_row.insert(columns[col_idx].clone(), value_to_json(value));
                }
            }
            rows.push(serde_json::Value::Object(json_row));
        }
    }

    let json = serde_json::to_string_pretty(&rows)?;
    writeln!(writer, "{json}")?;

    Ok(())
}

/// Output results as structured JSON with metadata for IDE/plugin integration
fn output_json_structured<W: Write>(
    dataview: &DataView,
    writer: &mut W,
    exec_time: f64,
) -> Result<()> {
    let column_names = dataview.column_names();
    let data_table = dataview.source();

    // Build column metadata
    let mut columns = Vec::new();
    for (idx, name) in column_names.iter().enumerate() {
        let col_type = data_table
            .columns
            .get(idx)
            .map(|c| format!("{:?}", c.data_type))
            .unwrap_or_else(|| "UNKNOWN".to_string());

        // Calculate max width for this column
        let mut max_width = name.len();
        for row_idx in 0..dataview.row_count() {
            if let Some(row) = dataview.get_row(row_idx) {
                if let Some(value) = row.values.get(idx) {
                    let display_width = match value {
                        DataValue::Null => 4, // "NULL"
                        DataValue::Integer(i) => i.to_string().len(),
                        DataValue::Float(f) => format!("{:.2}", f).len(),
                        DataValue::String(s) => s.len(),
                        DataValue::InternedString(s) => s.len(),
                        DataValue::Boolean(b) => {
                            if *b {
                                4
                            } else {
                                5
                            }
                        } // "true" or "false"
                        DataValue::DateTime(dt) => dt.len(),
                    };
                    max_width = max_width.max(display_width);
                }
            }
        }

        let alignment = match data_table.columns.get(idx).map(|c| &c.data_type) {
            Some(crate::data::datatable::DataType::Integer) => "right",
            Some(crate::data::datatable::DataType::Float) => "right",
            _ => "left",
        };

        let col_meta = serde_json::json!({
            "name": name,
            "type": col_type,
            "max_width": max_width,
            "alignment": alignment
        });
        columns.push(col_meta);
    }

    // Build rows as arrays of strings
    let mut rows = Vec::new();
    for row_idx in 0..dataview.row_count() {
        if let Some(row) = dataview.get_row(row_idx) {
            let row_values: Vec<String> = row
                .values
                .iter()
                .map(|v| match v {
                    DataValue::Null => String::new(),
                    DataValue::Integer(i) => i.to_string(),
                    DataValue::Float(f) => format!("{:.2}", f),
                    DataValue::String(s) => s.clone(),
                    DataValue::InternedString(s) => s.to_string(),
                    DataValue::Boolean(b) => b.to_string(),
                    DataValue::DateTime(dt) => dt.clone(),
                })
                .collect();
            rows.push(serde_json::Value::Array(
                row_values
                    .into_iter()
                    .map(serde_json::Value::String)
                    .collect(),
            ));
        }
    }

    // Build complete structured output
    let output = serde_json::json!({
        "columns": columns,
        "rows": rows,
        "metadata": {
            "total_rows": dataview.row_count(),
            "query_time_ms": exec_time
        }
    });

    let json = serde_json::to_string_pretty(&output)?;
    writeln!(writer, "{json}")?;

    Ok(())
}

/// Output results using the old custom ASCII table format (for Nvim compatibility)
fn output_table_old_style<W: Write>(
    dataview: &DataView,
    writer: &mut W,
    max_col_width: Option<usize>,
) -> Result<()> {
    let columns = dataview.column_names();

    // Calculate column widths
    let mut widths = vec![0; columns.len()];
    for (i, col) in columns.iter().enumerate() {
        widths[i] = col.len();
    }

    // Scan all rows for width calculation
    for row_idx in 0..dataview.row_count() {
        if let Some(row) = dataview.get_row(row_idx) {
            for (i, value) in row.values.iter().enumerate() {
                if i < widths.len() {
                    let value_str = format_value(value);
                    widths[i] = widths[i].max(value_str.len());
                }
            }
        }
    }

    // Apply maximum column width if specified
    if let Some(max_width) = max_col_width {
        for width in &mut widths {
            *width = (*width).min(max_width);
        }
    }

    // Print top border
    write!(writer, "+")?;
    for width in &widths {
        write!(writer, "{}", "-".repeat(*width + 2))?;
        write!(writer, "+")?;
    }
    writeln!(writer)?;

    // Print headers
    write!(writer, "|")?;
    for (i, col) in columns.iter().enumerate() {
        write!(writer, " {:^width$} |", col, width = widths[i])?;
    }
    writeln!(writer)?;

    // Print header separator
    write!(writer, "+")?;
    for width in &widths {
        write!(writer, "{}", "-".repeat(*width + 2))?;
        write!(writer, "+")?;
    }
    writeln!(writer)?;

    // Print data rows (no separators between rows)
    for row_idx in 0..dataview.row_count() {
        if let Some(row) = dataview.get_row(row_idx) {
            write!(writer, "|")?;
            for (i, value) in row.values.iter().enumerate() {
                if i < widths.len() {
                    let value_str = format_value(value);
                    let truncated = if value_str.len() > widths[i] {
                        format!("{}...", &value_str[..widths[i].saturating_sub(3)])
                    } else {
                        value_str
                    };
                    write!(writer, " {:<width$} |", truncated, width = widths[i])?;
                }
            }
            writeln!(writer)?;
        }
    }

    // Print bottom border
    write!(writer, "+")?;
    for width in &widths {
        write!(writer, "{}", "-".repeat(*width + 2))?;
        write!(writer, "+")?;
    }
    writeln!(writer)?;

    Ok(())
}

/// Output results as a table using comfy-table with styling
fn output_table<W: Write>(
    dataview: &DataView,
    writer: &mut W,
    max_col_width: Option<usize>,
    _col_sample_rows: usize, // Not needed with comfy-table
    style: TableStyle,
) -> Result<()> {
    let mut table = Table::new();

    // Apply the selected style preset
    match style {
        TableStyle::Default => {
            // Use custom old-style renderer for Nvim compatibility
            // This matches what the table navigation parser expects
            return output_table_old_style(dataview, writer, max_col_width);
        }
        TableStyle::AsciiFull => {
            table.load_preset(ASCII_FULL);
        }
        TableStyle::AsciiCondensed => {
            table.load_preset(ASCII_FULL_CONDENSED);
        }
        TableStyle::AsciiBordersOnly => {
            table.load_preset(ASCII_BORDERS_ONLY);
        }
        TableStyle::AsciiHorizontalOnly => {
            table.load_preset(ASCII_HORIZONTAL_ONLY);
        }
        TableStyle::AsciiNoBorders => {
            table.load_preset(ASCII_NO_BORDERS);
        }
        TableStyle::Markdown => {
            table.load_preset(ASCII_MARKDOWN);
        }
        TableStyle::Utf8Full => {
            table.load_preset(UTF8_FULL);
        }
        TableStyle::Utf8Condensed => {
            table.load_preset(UTF8_FULL_CONDENSED);
        }
        TableStyle::Utf8BordersOnly => {
            table.load_preset(UTF8_BORDERS_ONLY);
        }
        TableStyle::Utf8HorizontalOnly => {
            table.load_preset(UTF8_HORIZONTAL_ONLY);
        }
        TableStyle::Utf8NoBorders => {
            table.load_preset(UTF8_NO_BORDERS);
        }
        TableStyle::Plain => {
            table.load_preset(NOTHING);
        }
    }

    // Set content arrangement (automatic width adjustment)
    if max_col_width.is_some() {
        table.set_content_arrangement(ContentArrangement::Dynamic);
    }

    // Set column headers
    let columns = dataview.column_names();
    table.set_header(&columns);

    // Add data rows
    for row_idx in 0..dataview.row_count() {
        if let Some(row) = dataview.get_row(row_idx) {
            let row_strings: Vec<String> = row
                .values
                .iter()
                .map(|v| {
                    let s = format_value(v);
                    // Apply max width truncation if specified
                    if let Some(max_width) = max_col_width {
                        if s.len() > max_width {
                            format!("{}...", &s[..max_width.saturating_sub(3)])
                        } else {
                            s
                        }
                    } else {
                        s
                    }
                })
                .collect();
            table.add_row(row_strings);
        }
    }

    // Write the table to the writer
    writeln!(writer, "{}", table)?;

    Ok(())
}

/// Format a `DataValue` for display
fn format_value(value: &DataValue) -> String {
    match value {
        DataValue::Null => String::new(),
        DataValue::Integer(i) => i.to_string(),
        DataValue::Float(f) => f.to_string(),
        DataValue::String(s) => s.clone(),
        DataValue::InternedString(s) => s.to_string(),
        DataValue::Boolean(b) => b.to_string(),
        DataValue::DateTime(dt) => dt.to_string(),
    }
}

/// Convert `DataValue` to JSON
fn value_to_json(value: &DataValue) -> serde_json::Value {
    match value {
        DataValue::Null => serde_json::Value::Null,
        DataValue::Integer(i) => serde_json::Value::Number((*i).into()),
        DataValue::Float(f) => {
            if let Some(n) = serde_json::Number::from_f64(*f) {
                serde_json::Value::Number(n)
            } else {
                serde_json::Value::Null
            }
        }
        DataValue::String(s) => serde_json::Value::String(s.clone()),
        DataValue::InternedString(s) => serde_json::Value::String(s.to_string()),
        DataValue::Boolean(b) => serde_json::Value::Bool(*b),
        DataValue::DateTime(dt) => serde_json::Value::String(dt.to_string()),
    }
}

/// Escape a CSV field if it contains special characters
fn escape_csv_field(field: &str, delimiter: char) -> String {
    if field.contains(delimiter)
        || field.contains('"')
        || field.contains('\n')
        || field.contains('\r')
    {
        format!("\"{}\"", field.replace('"', "\"\""))
    } else {
        field.to_string()
    }
}