flowcode-core 0.4.3-alpha

// Verb Mappers for DuckDB Backend
// All verb mappers receive &[TypedValue] where:
// - 0 = first argument (often left table or primary entity)
// - 1 = second argument (often right table or secondary entity)
// - 2.. = additional arguments (e.g., raw predicate tokens for conditions).

// DuckDB Verb Mapping for FlowCode v0.4-alpha
// This file outlines the mapping of FlowCode operations to SQL queries for DuckDB.

use crate::types::TypedValue;
use crate::duckdb_session::DuckSession;
use crate::error::FCError;
use std::collections::HashSet;

/// Maps a FlowCode verb to a DuckDB SQL query string.
pub fn map_verb_to_sql(verb: &str, args: &[TypedValue], session: &DuckSession) -> Result<String, FCError> {
    match verb.to_lowercase().as_str() {
        "select" => {
            // If no arguments, return a basic SELECT query
            if args.is_empty() {
                Ok("SELECT * FROM employees".to_string())
            } else {
                // Use the first argument as the table name
                let table = args[0].to_string();
                // If more args are provided, treat them as column names
                let columns = if args.len() > 1 {
                    args[1..].iter().map(|arg| {
                        match &arg.value {
                            crate::ast::ArgValue::String(s) => s.clone(),
                            _ => "*".to_string(),
                        }
                    }).collect::<Vec<String>>().join(", ")
                } else {
                    "*".to_string()
                };
                Ok(format!("SELECT {} FROM {}", columns, table))
            }
        },
        "filter" => {
            if args.len() < 2 {
                Err(FCError::InvalidArgument("FILTER command requires a table name and a condition".to_string()))
            } else {
                let table = args[0].to_string();
                let condition = args[1..].iter().map(|arg| arg.to_string()).collect::<Vec<String>>().join(" ");
                Ok(format!("SELECT * FROM {} WHERE {}", table, condition))
            }
        },
        "group" => {
            if args.len() < 2 {
                Err(FCError::InvalidArgument("GROUP command requires a table name and at least one column to group by".to_string()))
            } else {
                let table = args[0].to_string();
                let group_by_columns = args[1..].iter().map(|arg| arg.to_string()).collect::<Vec<String>>().join(", ");
                Ok(format!("SELECT {} FROM {} GROUP BY {}", group_by_columns, table, group_by_columns))
            }
        },
        "join" => {
            map_join(args, session)
        },
        "sort" => {
            map_sort(args)
        },
        "load" => {
            if args.is_empty() {
                Err(FCError::InvalidArgument("LOAD command requires a file path or data source".to_string()))
            } else {
                let source = args[0].to_string();
                let table_name = if args.len() > 1 { args[1].to_string() } else { "loaded_data".to_string() };
                Ok(format!("CREATE TABLE {} AS SELECT * FROM read_csv_auto('{}')", table_name, source))
            }
        },
        "combine" => {
            // COMBINE is not supported in DuckDB, return an error
            Err(FCError::BackendError("COMBINE command is not supported in DuckDB backend. Use numeric operations in SELECT statements instead.".to_string()))
        },
        "predict" => {
            // PREDICT is not supported in DuckDB, return an error
            Err(FCError::BackendError("PREDICT command is not supported in DuckDB backend. Use SELECT to view data instead.".to_string()))
        },
        "show" => {
            // SHOW is not supported in DuckDB, return an error
            Err(FCError::BackendError("SHOW command is not supported in DuckDB backend. Use SELECT to view data instead.".to_string()))
        },
        "sql" => {
            map_sql(args)
        },
        _ => Err(FCError::BackendError(format!("Unsupported verb for DuckDB mapping: {}", verb))),
    }
}

/// Maps the 'select' verb to a SQL SELECT statement.
#[allow(dead_code)]
fn map_select(_args: &[TypedValue]) -> Result<String, FCError> {
    Err(FCError::InvalidArgument("Select verb requires at least one argument (table or columns)".to_string()))
}

/// Maps the 'filter' verb to a SQL WHERE clause.
#[allow(dead_code)]
fn map_filter(args: &[TypedValue]) -> Result<String, FCError> {
    if args.len() < 2 {
        Err(FCError::InvalidArgument("Filter verb requires a table and at least one condition".to_string()))
    } else {
        let table = args[0].to_string();
        let condition = args[1..].iter().map(|arg| arg.to_string()).collect::<Vec<String>>().join(" ");
        Ok(format!("SELECT * FROM {} WHERE {}", table, condition))
    }
}

/// Maps the 'group' verb to a SQL GROUP BY clause.
#[allow(dead_code)]
fn map_group(_args: &[TypedValue]) -> Result<String, FCError> {
    Err(FCError::InvalidArgument("Group verb requires a table and at least one grouping column".to_string()))
}

/// Maps the 'join' verb to a SQL JOIN clause.
/// This function constructs a SQL query for joining two tables based on provided arguments.
/// - It uses LEFT JOIN as the join type.
/// - When no specific columns are provided after a 'column' keyword, it defaults to SELECT *.
/// - For user-specified columns, it prefixes each column with the appropriate table name (left or right) based on uniqueness.
///   If a column exists in both tables or is unknown, it defaults to the left table prefix.
/// - Table column information is fetched dynamically via session.get_table_columns, with fallback to common column sets if retrieval fails.
fn map_join(args: &[TypedValue], session: &DuckSession) -> Result<String, FCError> {
    if args.len() < 3 {
        return Err(FCError::InvalidArgument("join needs: join <left> <right> \"predicate\" [column <cols…>]".to_string()));
    }

    let left = args[0].to_string();
    let right = args[1].to_string();

    // Find optional 'column' / 'columns' keyword
    let keyword_pos = args.iter().position(|t| {
        let s = t.to_string().to_ascii_lowercase();
        s == "column" || s == "columns"
    });

    let (pred_tokens, col_tokens) = match keyword_pos {
        Some(pos) => (&args[2..pos], &args[pos + 1..]),
        None => (&args[2..], &[] as &[TypedValue]),
    };

    let on_clause = pred_tokens
        .iter()
        .map(|t| t.to_string())
        .collect::<Vec<_>>()
        .join(" ");

    let select_clause = if col_tokens.is_empty() {
        "*".to_string()
    } else {
        // Attempt to fetch column sets for each table. Fall back to common heuristics if query fails.
        let left_cols: HashSet<String> = session
            .get_table_columns(&left)
            .unwrap_or_else(|_| vec!["id", "name", "department"].into_iter().map(|s| s.to_string()).collect())
            .into_iter()
            .collect();
        let right_cols: HashSet<String> = session
            .get_table_columns(&right)
            .unwrap_or_else(|_| vec!["location", "manager", "name"].into_iter().map(|s| s.to_string()).collect())
            .into_iter()
            .collect();

        col_tokens
            .iter()
            .map(|token| {
                let col = token.to_string();
                // Prefix with the table where the column uniquely exists; default to left if in both/unknown
                if left_cols.contains(&col) && !right_cols.contains(&col) {
                    format!("{}.{}", left, col)
                } else if right_cols.contains(&col) && !left_cols.contains(&col) {
                    format!("{}.{}", right, col)
                } else {
                    // Ambiguous or unknown: prefix with left
                    format!("{}.{}", left, col)
                }
            })
            .collect::<Vec<_>>()
            .join(", ")
    };

    Ok(format!(
        "SELECT {} FROM {} LEFT JOIN {} ON {}",
        select_clause, left, right, on_clause
    ))
}

/// Maps the 'sort' verb to a SQL ORDER BY clause.
#[allow(dead_code)]
fn map_sort(args: &[TypedValue]) -> Result<String, FCError> {
    if args.len() < 2 {
        return Err(FCError::InvalidArgument("SORT command requires a table name and at least one column to sort by".to_string()));
    }

    let table = args[0].to_string();

    // Build ORDER BY parts, supporting optional ASC/DESC per column.
    // Pattern: <col1> [asc|desc] <col2> [asc|desc] ...
    let mut parts: Vec<String> = Vec::new();
    let mut idx = 1;
    while idx < args.len() {
        let col = args[idx].to_string();
        if col.eq_ignore_ascii_case("asc") || col.eq_ignore_ascii_case("desc") {
            return Err(FCError::InvalidArgument("Column name expected before ASC/DESC".to_string()));
        }

        // Default direction ASC
        let mut direction = "ASC".to_string();
        if idx + 1 < args.len() {
            let next_token = args[idx + 1].to_string();
            if next_token.eq_ignore_ascii_case("asc") || next_token.eq_ignore_ascii_case("desc") {
                direction = next_token.to_uppercase();
                idx += 1; // Consume direction token
            }
        }

        parts.push(format!("{} {}", col, direction));
        idx += 1; // Move to next token (column)
    }

    let order_clause = parts.join(", ");
    Ok(format!("SELECT * FROM {} ORDER BY {}", table, order_clause))
}

/// Maps the 'sql' verb to a direct SQL query.
fn map_sql(args: &[TypedValue]) -> Result<String, FCError> {
    if args.is_empty() {
        return Err(FCError::InvalidArgument(
            "sql command requires a query string as the first argument".to_string()
        ));
    }

    let query = match &args[0].value {
        crate::ast::ArgValue::String(s) => s.clone(),
        _ => return Err(FCError::InvalidArgument(
            "sql command requires a string argument for the query".to_string()
        )),
    };
    // Return the raw SQL query as a string to be executed by DuckSession
    Ok(query)
}