sql_rs/query/

parser.rs

use crate::{SqlRsError, Result, Value};

#[derive(Debug, Clone, PartialEq)]
pub enum AggregateFunction {
    Count,
    Sum,
    Avg,
    Min,
    Max,
}

#[derive(Debug, Clone)]
pub enum SelectColumn {
    Star,
    Column(String),
    Aggregate {
        function: AggregateFunction,
        column: String,
    },
}

#[derive(Debug, Clone)]
pub enum Statement {
    CreateTable {
        name: String,
        columns: Vec<(String, String)>,
    },
    Insert {
        table: String,
        columns: Vec<String>,
        values: Vec<Value>,
    },
    Select {
        table: String,
        columns: Vec<SelectColumn>,
        joins: Vec<Join>,
        where_clause: Option<WhereClause>,
        group_by: Option<Vec<String>>,
        order_by: Option<OrderBy>,
        limit: Option<usize>,
        offset: Option<usize>,
    },
    Update {
        table: String,
        sets: Vec<(String, Value)>,
        where_clause: Option<WhereClause>,
    },
    Delete {
        table: String,
        where_clause: Option<WhereClause>,
    },
    DropTable {
        name: String,
    },
    CreateIndex {
        index_name: String,
        table: String,
        column: String,
    },
    DropIndex {
        index_name: String,
    },
}

#[derive(Debug, Clone)]
pub enum WhereExpression {
    Condition {
        column: String,
        operator: String,
        value: Value,
    },
    And {
        left: Box<WhereExpression>,
        right: Box<WhereExpression>,
    },
    Or {
        left: Box<WhereExpression>,
        right: Box<WhereExpression>,
    },
}

// For backward compatibility, WhereClause is now a type alias
pub type WhereClause = WhereExpression;

#[derive(Debug, Clone)]
pub struct OrderBy {
    pub column: String,
    pub ascending: bool,
}

#[derive(Debug, Clone, PartialEq)]
pub enum JoinType {
    Inner,
    Left,
    Right,
}

#[derive(Debug, Clone)]
pub struct Join {
    pub join_type: JoinType,
    pub table: String,
    pub on_left: String,
    pub on_right: String,
}

pub fn parse_sql(sql: &str) -> Result<Statement> {
    let sql = sql.trim();
    let upper = sql.to_uppercase();
    
    // Use split_whitespace to handle multiple spaces between keywords
    let words: Vec<&str> = upper.split_whitespace().collect();
    
    if words.len() >= 2 && words[0] == "CREATE" && words[1] == "TABLE" {
        parse_create_table(sql)
    } else if words.len() >= 2 && words[0] == "CREATE" && words[1] == "INDEX" {
        parse_create_index(sql)
    } else if words.len() >= 2 && words[0] == "INSERT" && words[1] == "INTO" {
        parse_insert(sql)
    } else if words.len() >= 1 && words[0] == "SELECT" {
        parse_select(sql)
    } else if words.len() >= 1 && words[0] == "UPDATE" {
        parse_update(sql)
    } else if words.len() >= 2 && words[0] == "DELETE" && words[1] == "FROM" {
        parse_delete(sql)
    } else if words.len() >= 2 && words[0] == "DROP" && words[1] == "TABLE" {
        parse_drop_table(sql)
    } else if words.len() >= 2 && words[0] == "DROP" && words[1] == "INDEX" {
        parse_drop_index(sql)
    } else {
        Err(SqlRsError::Query(format!(
            "Unsupported SQL statement: {}",
            sql
        )))
    }
}

fn parse_create_table(sql: &str) -> Result<Statement> {
    let parts: Vec<&str> = sql.split_whitespace().collect();

    if parts.len() < 4 {
        return Err(SqlRsError::Query(
            "Invalid CREATE TABLE syntax".to_string(),
        ));
    }

    let name = parts[2].to_string();

    let columns_start = sql
        .find('(')
        .ok_or_else(|| SqlRsError::Query("Missing column definitions".to_string()))?;
    let columns_end = sql
        .rfind(')')
        .ok_or_else(|| SqlRsError::Query("Missing closing parenthesis".to_string()))?;

    let columns_str = &sql[columns_start + 1..columns_end];
    let column_defs: Vec<&str> = columns_str.split(',').map(|s| s.trim()).collect();

    let mut columns = Vec::new();
    for def in column_defs {
        let parts: Vec<&str> = def.split_whitespace().collect();
        if parts.len() >= 2 {
            columns.push((parts[0].to_string(), parts[1].to_uppercase()));
        }
    }

    if columns.is_empty() {
        return Err(SqlRsError::Query("No valid column definitions found".to_string()));
    }

    Ok(Statement::CreateTable { name, columns })
}

fn parse_insert(sql: &str) -> Result<Statement> {
    let upper = sql.to_uppercase();

    let table_start = upper
        .find("INTO")
        .ok_or_else(|| SqlRsError::Query("Missing INTO".to_string()))?
        + 4;
    let values_start = upper
        .find("VALUES")
        .ok_or_else(|| SqlRsError::Query("Missing VALUES".to_string()))?;

    let table_part = sql[table_start..values_start].trim();

    let (table, columns) = if let Some(paren_start) = table_part.find('(') {
        let table = table_part[..paren_start].trim().to_string();
        let paren_end = table_part
            .rfind(')')
            .ok_or_else(|| SqlRsError::Query("Missing closing parenthesis".to_string()))?;
        let cols_str = &table_part[paren_start + 1..paren_end];
        let columns: Vec<String> = cols_str.split(',').map(|s| s.trim().to_string()).collect();
        (table, columns)
    } else {
        (table_part.to_string(), Vec::new())
    };

    let values_part = &sql[values_start + 6..].trim();
    let values_start = values_part
        .find('(')
        .ok_or_else(|| SqlRsError::Query("Missing values parenthesis".to_string()))?;
    let values_end = values_part
        .rfind(')')
        .ok_or_else(|| SqlRsError::Query("Missing closing parenthesis".to_string()))?;

    let values_str = &values_part[values_start + 1..values_end];
    let value_strs: Vec<&str> = values_str.split(',').map(|s| s.trim()).collect();

    let mut values = Vec::new();
    for val_str in value_strs {
        values.push(parse_value(val_str)?);
    }

    Ok(Statement::Insert {
        table,
        columns,
        values,
    })
}

fn parse_select(sql: &str) -> Result<Statement> {
    let upper = sql.to_uppercase();

    let from_pos = upper
        .find("FROM")
        .ok_or_else(|| SqlRsError::Query("Missing FROM".to_string()))?;

    let cols_str = sql[6..from_pos].trim();
    let columns = parse_select_columns(cols_str)?;

    let after_from = &sql[from_pos + 4..];
    let upper_after_from = after_from.to_uppercase();

    // Find positions of JOIN types, WHERE, GROUP BY, ORDER BY, LIMIT, OFFSET
    let inner_join_pos = upper_after_from.find("INNER JOIN");
    let left_join_pos = upper_after_from.find("LEFT JOIN");
    let right_join_pos = upper_after_from.find("RIGHT JOIN");
    let join_pos = upper_after_from.find("JOIN");
    let where_pos = upper_after_from.find("WHERE");
    let group_pos = upper_after_from.find("GROUP BY");
    let order_pos = upper_after_from.find("ORDER BY");
    let limit_pos = upper_after_from.find("LIMIT");
    let offset_pos = upper_after_from.find("OFFSET");

    // Find the first JOIN position (if any)
    let first_join_pos = [inner_join_pos, left_join_pos, right_join_pos, join_pos]
        .iter()
        .filter_map(|&p| p)
        .min();

    // Determine the end of the table name
    let table_end = [first_join_pos, where_pos, group_pos, order_pos, limit_pos, offset_pos]
        .iter()
        .filter_map(|&p| p)
        .min()
        .unwrap_or(after_from.len());

    let table = after_from[..table_end].trim().to_string();

    // Parse JOIN clauses
    let joins = if first_join_pos.is_some() {
        parse_joins(after_from, first_join_pos.unwrap())?
    } else {
        Vec::new()
    };

    // Parse WHERE clause
    let where_clause = if let Some(pos) = where_pos {
        let where_end = [group_pos, order_pos, limit_pos, offset_pos]
            .iter()
            .filter_map(|&p| p)
            .filter(|&p| p > pos)
            .min()
            .unwrap_or(after_from.len());
        let where_str = after_from[pos + 5..where_end].trim();
        Some(parse_where(where_str)?)
    } else {
        None
    };

    // Parse GROUP BY clause
    let group_by = if let Some(pos) = group_pos {
        let group_end = [order_pos, limit_pos, offset_pos]
            .iter()
            .filter_map(|&p| p)
            .filter(|&p| p > pos)
            .min()
            .unwrap_or(after_from.len());
        let group_str = after_from[pos + 8..group_end].trim();
        let columns: Vec<String> = group_str.split(',').map(|s| s.trim().to_string()).collect();
        Some(columns)
    } else {
        None
    };

    // Parse ORDER BY clause
    let order_by = if let Some(pos) = order_pos {
        let order_end = [limit_pos, offset_pos]
            .iter()
            .filter_map(|&p| p)
            .filter(|&p| p > pos)
            .min()
            .unwrap_or(after_from.len());
        let order_str = after_from[pos + 8..order_end].trim();
        Some(parse_order_by(order_str)?)
    } else {
        None
    };

    // Parse LIMIT clause
    let limit = if let Some(pos) = limit_pos {
        let limit_end = offset_pos
            .filter(|&p| p > pos)
            .unwrap_or(after_from.len());
        let limit_str = after_from[pos + 5..limit_end].trim();
        Some(limit_str.parse::<usize>().map_err(|_| {
            SqlRsError::Query("Invalid LIMIT value".to_string())
        })?)
    } else {
        None
    };

    // Parse OFFSET clause
    let offset = if let Some(pos) = offset_pos {
        let offset_str = after_from[pos + 6..].trim();
        Some(offset_str.parse::<usize>().map_err(|_| {
            SqlRsError::Query("Invalid OFFSET value".to_string())
        })?)
    } else {
        None
    };

    Ok(Statement::Select {
        table,
        columns,
        joins,
        where_clause,
        group_by,
        order_by,
        limit,
        offset,
    })
}

fn parse_joins(after_from: &str, start_pos: usize) -> Result<Vec<Join>> {
    let upper = after_from.to_uppercase();
    let mut joins = Vec::new();
    let mut current_pos = start_pos;

    // Find WHERE, GROUP BY, ORDER BY, LIMIT, OFFSET to know where JOINs end
    let where_pos = upper.find("WHERE");
    let group_pos = upper.find("GROUP BY");
    let order_pos = upper.find("ORDER BY");
    let limit_pos = upper.find("LIMIT");
    let offset_pos = upper.find("OFFSET");
    
    let joins_end = [where_pos, group_pos, order_pos, limit_pos, offset_pos]
        .iter()
        .filter_map(|&p| p)
        .min()
        .unwrap_or(after_from.len());

    while current_pos < joins_end {
        let remaining = &after_from[current_pos..joins_end];
        let remaining_upper = remaining.to_uppercase();

        // Determine join type
        let (join_type, keyword_len) = if remaining_upper.starts_with("INNER JOIN") {
            (JoinType::Inner, 10)
        } else if remaining_upper.starts_with("LEFT JOIN") {
            (JoinType::Left, 9)
        } else if remaining_upper.starts_with("RIGHT JOIN") {
            (JoinType::Right, 10)
        } else if remaining_upper.starts_with("JOIN") {
            (JoinType::Inner, 4) // Default JOIN is INNER JOIN
        } else {
            break;
        };

        // Skip the JOIN keyword
        let after_join = &remaining[keyword_len..].trim_start();
        let after_join_upper = after_join.to_uppercase();

        // Find ON keyword
        let on_pos = after_join_upper.find(" ON ")
            .ok_or_else(|| SqlRsError::Query("Missing ON clause in JOIN".to_string()))?;

        let table_name = after_join[..on_pos].trim().to_string();

        // Parse ON condition
        let after_on = &after_join[on_pos + 4..];
        
        // Find the end of ON clause (next JOIN or end of joins section)
        let next_inner_join = after_on.to_uppercase().find("INNER JOIN");
        let next_left_join = after_on.to_uppercase().find("LEFT JOIN");
        let next_right_join = after_on.to_uppercase().find("RIGHT JOIN");
        let next_join = after_on.to_uppercase().find(" JOIN");
        
        let on_end = [next_inner_join, next_left_join, next_right_join, next_join]
            .iter()
            .filter_map(|&p| p)
            .min()
            .unwrap_or(after_on.len());

        let on_clause = after_on[..on_end].trim();

        // Parse ON condition (e.g., "table1.id = table2.id")
        let parts: Vec<&str> = on_clause.split('=').map(|s| s.trim()).collect();
        if parts.len() != 2 {
            return Err(SqlRsError::Query("Invalid ON clause format".to_string()));
        }

        let on_left = parts[0].to_string();
        let on_right = parts[1].to_string();

        joins.push(Join {
            join_type,
            table: table_name,
            on_left,
            on_right,
        });

        // Move to next potential JOIN
        // We need to move past the current JOIN keyword, table name, ON keyword, and ON clause
        let consumed = keyword_len + on_pos + 4 + on_end;
        current_pos = current_pos + consumed;
        
        // Skip any whitespace
        while current_pos < joins_end && after_from.as_bytes().get(current_pos).map_or(false, |&b| b.is_ascii_whitespace()) {
            current_pos += 1;
        }
    }

    Ok(joins)
}

fn parse_select_columns(cols_str: &str) -> Result<Vec<SelectColumn>> {
    let cols_str = cols_str.trim();
    
    if cols_str.is_empty() {
        return Err(SqlRsError::Query("Empty column list in SELECT".to_string()));
    }
    
    if cols_str == "*" {
        return Ok(vec![SelectColumn::Star]);
    }

    let mut columns = Vec::new();
    for col_str in cols_str.split(',') {
        let col_str = col_str.trim();
        let col_upper = col_str.to_uppercase();

        // Check for aggregate functions
        if col_upper.starts_with("COUNT(") {
            let inner = &col_str[6..col_str.len().saturating_sub(1)].trim();
            columns.push(SelectColumn::Aggregate {
                function: AggregateFunction::Count,
                column: inner.to_string(),
            });
        } else if col_upper.starts_with("SUM(") {
            let inner = &col_str[4..col_str.len().saturating_sub(1)].trim();
            columns.push(SelectColumn::Aggregate {
                function: AggregateFunction::Sum,
                column: inner.to_string(),
            });
        } else if col_upper.starts_with("AVG(") {
            let inner = &col_str[4..col_str.len().saturating_sub(1)].trim();
            columns.push(SelectColumn::Aggregate {
                function: AggregateFunction::Avg,
                column: inner.to_string(),
            });
        } else if col_upper.starts_with("MIN(") {
            let inner = &col_str[4..col_str.len().saturating_sub(1)].trim();
            columns.push(SelectColumn::Aggregate {
                function: AggregateFunction::Min,
                column: inner.to_string(),
            });
        } else if col_upper.starts_with("MAX(") {
            let inner = &col_str[4..col_str.len().saturating_sub(1)].trim();
            columns.push(SelectColumn::Aggregate {
                function: AggregateFunction::Max,
                column: inner.to_string(),
            });
        } else {
            // Regular column
            columns.push(SelectColumn::Column(col_str.to_string()));
        }
    }

    Ok(columns)
}

fn parse_update(sql: &str) -> Result<Statement> {
    let upper = sql.to_uppercase();

    // Find UPDATE and table name
    let update_pos = 0;
    let set_pos = upper
        .find("SET")
        .ok_or_else(|| SqlRsError::Query("Missing SET in UPDATE".to_string()))?;

    let table = sql[update_pos + 6..set_pos].trim().to_string();

    // Find WHERE clause if exists
    let after_set = &sql[set_pos + 3..].trim();

    let (set_where_str, where_clause) =
        if let Some(where_pos) = after_set.to_uppercase().find("WHERE") {
            let set_str = &after_set[..where_pos];
            let where_str = &after_set[where_pos + 5..];
            (set_str.to_string(), Some(parse_where(where_str.trim())?))
        } else {
            (after_set.to_string(), None)
        };

    // Parse SET clauses (e.g., "col1 = val1, col2 = val2")
    let mut sets = Vec::new();
    for set_clause in set_where_str.split(',') {
        let set_clause = set_clause.trim();
        if let Some(eq_pos) = set_clause.find('=') {
            let column = set_clause[..eq_pos].trim().to_string();
            let value_str = set_clause[eq_pos + 1..].trim();
            let value = parse_value(value_str)?;
            sets.push((column, value));
        } else {
            return Err(SqlRsError::Query(format!(
                "Invalid SET clause: {}",
                set_clause
            )));
        }
    }

    if sets.is_empty() {
        return Err(SqlRsError::Query("No SET clauses provided".to_string()));
    }

    Ok(Statement::Update {
        table,
        sets,
        where_clause,
    })
}

fn parse_where(where_str: &str) -> Result<WhereExpression> {
    // Parse OR (lowest precedence)
    parse_or_expression(where_str)
}

fn parse_or_expression(where_str: &str) -> Result<WhereExpression> {
    let upper = where_str.to_uppercase();

    // Find all OR positions that are not within parentheses
    let or_positions = find_logical_operator_positions(&upper, "OR");

    // If we found OR, split and parse recursively
    if !or_positions.is_empty() {
        // Use the first OR position (left-to-right associativity)
        let or_pos = or_positions[0];
        let left_str = &where_str[..or_pos];
        let right_str = &where_str[or_pos + 3..]; // Skip "OR"

        return Ok(WhereExpression::Or {
            left: Box::new(parse_and_expression(left_str.trim())?),
            right: Box::new(parse_or_expression(right_str.trim())?),
        });
    }

    // No OR found, try AND
    parse_and_expression(where_str)
}

fn parse_and_expression(where_str: &str) -> Result<WhereExpression> {
    let upper = where_str.to_uppercase();

    // Find all AND positions that are not within parentheses
    let and_positions = find_logical_operator_positions(&upper, "AND");

    // If we found AND, split and parse recursively
    if !and_positions.is_empty() {
        // Use the first AND position (left-to-right associativity)
        let and_pos = and_positions[0];
        let left_str = &where_str[..and_pos];
        let right_str = &where_str[and_pos + 4..]; // Skip "AND "

        return Ok(WhereExpression::And {
            left: Box::new(parse_and_expression(left_str.trim())?),
            right: Box::new(parse_and_expression(right_str.trim())?),
        });
    }

    // No AND found, parse as a single condition
    parse_condition(where_str)
}

fn find_logical_operator_positions(upper: &str, operator: &str) -> Vec<usize> {
    let mut positions = Vec::new();
    let mut depth = 0;
    let bytes = upper.as_bytes();
    let op_bytes = operator.as_bytes();
    let op_len = op_bytes.len();

    for i in 0..bytes.len() {
        match bytes[i] {
            b'(' => depth += 1,
            b')' => depth -= 1,
            _ if depth == 0 => {
                // Check if we can find the operator at this position
                if i + op_len <= bytes.len() && &bytes[i..i + op_len] == op_bytes {
                    // Check if it's surrounded by spaces (or at boundaries with proper spacing)
                    let before_ok = i == 0 || bytes[i - 1].is_ascii_whitespace();
                    let after_ok = i + op_len >= bytes.len() || bytes[i + op_len].is_ascii_whitespace();

                    if before_ok && after_ok {
                        positions.push(i);
                    }
                }
            }
            _ => {}
        }
    }

    positions
}

fn parse_condition(where_str: &str) -> Result<WhereExpression> {
    let trimmed = where_str.trim();

    // Handle parentheses for grouping
    if trimmed.starts_with('(') {
        // Find matching closing parenthesis
        let mut depth = 0;
        let mut close_pos = 0;
        let mut chars = trimmed.chars().enumerate();

        while let Some((pos, c)) = chars.next() {
            match c {
                '(' => depth += 1,
                ')' => {
                    depth -= 1;
                    if depth == 0 {
                        close_pos = pos;
                        break;
                    }
                }
                _ => {}
            }
        }

        if close_pos > 0 {
            let inner = &trimmed[1..close_pos];
            let remaining = &trimmed[close_pos + 1..].trim();

            // If there's remaining text after the parentheses, parse it as an AND/OR
            if remaining.is_empty() {
                return parse_or_expression(inner);
            } else {
                // Combine the parenthesized expression with the remaining part
                // This handles cases like "(a = 1) OR (b = 2)"
                let left_expr = parse_or_expression(inner)?;

                // Try to parse the remaining part as AND/OR
                let remaining_upper = remaining.to_uppercase();
                if remaining_upper.starts_with("AND ") {
                    let right_str = remaining[4..].trim();
                    let right_expr = parse_or_expression(right_str)?;
                    return Ok(WhereExpression::And {
                        left: Box::new(left_expr),
                        right: Box::new(right_expr),
                    });
                } else if remaining_upper.starts_with("OR ") {
                    let right_str = remaining[3..].trim();
                    let right_expr = parse_or_expression(right_str)?;
                    return Ok(WhereExpression::Or {
                        left: Box::new(left_expr),
                        right: Box::new(right_expr),
                    });
                }
            }
        }
    }

    // Parse as a simple condition
    let operators = [">=", "<=", "!=", "=", ">", "<"];

    for op in &operators {
        if let Some(pos) = trimmed.find(op) {
            let column = trimmed[..pos].trim().to_string();
            let value_str = trimmed[pos + op.len()..].trim();
            let value = parse_value(value_str)?;

            return Ok(WhereExpression::Condition {
                column,
                operator: op.to_string(),
                value,
            });
        }
    }

    Err(SqlRsError::Query("Invalid WHERE clause".to_string()))
}

fn parse_delete(sql: &str) -> Result<Statement> {
    let upper = sql.to_uppercase();

    let from_pos = upper
        .find("FROM")
        .ok_or_else(|| SqlRsError::Query("Missing FROM in DELETE".to_string()))?;

    let after_from = &sql[from_pos + 4..].trim();

    let (table, where_clause) = if let Some(where_pos) = after_from.to_uppercase().find("WHERE") {
        let table = after_from[..where_pos].trim().to_string();
        let where_str = after_from[where_pos + 5..].trim();
        (table, Some(parse_where(where_str)?))
    } else {
        (after_from.to_string(), None)
    };

    if table.is_empty() {
        return Err(SqlRsError::Query("Missing table name in DELETE".to_string()));
    }

    Ok(Statement::Delete {
        table,
        where_clause,
    })
}

fn parse_drop_table(sql: &str) -> Result<Statement> {
    let parts: Vec<&str> = sql.split_whitespace().collect();

    if parts.len() < 3 {
        return Err(SqlRsError::Query(
            "Invalid DROP TABLE syntax".to_string(),
        ));
    }

    let name = parts[2].to_string();

    Ok(Statement::DropTable { name })
}

fn parse_order_by(order_str: &str) -> Result<OrderBy> {
    let upper = order_str.to_uppercase();
    let ascending = !upper.ends_with("DESC");
    
    let column = if upper.ends_with("ASC") {
        order_str[..order_str.len() - 3].trim().to_string()
    } else if upper.ends_with("DESC") {
        order_str[..order_str.len() - 4].trim().to_string()
    } else {
        order_str.trim().to_string()
    };
    
    Ok(OrderBy { column, ascending })
}

fn parse_value(s: &str) -> Result<Value> {
    let s = s.trim();

    if s.eq_ignore_ascii_case("NULL") {
        return Ok(Value::Null);
    }

    if s.eq_ignore_ascii_case("TRUE") {
        return Ok(Value::Boolean(true));
    }

    if s.eq_ignore_ascii_case("FALSE") {
        return Ok(Value::Boolean(false));
    }

    if (s.starts_with('\'') && s.ends_with('\'')) || (s.starts_with('"') && s.ends_with('"')) {
        return Ok(Value::Text(s[1..s.len() - 1].to_string()));
    }

    if let Ok(i) = s.parse::<i64>() {
        return Ok(Value::Integer(i));
    }

    if let Ok(f) = s.parse::<f64>() {
        return Ok(Value::Float(f));
    }

    Ok(Value::Text(s.to_string()))
}

fn parse_create_index(sql: &str) -> Result<Statement> {
    // CREATE INDEX index_name ON table_name (column_name)
    let upper = sql.to_uppercase();
    
    let on_pos = upper
        .find(" ON ")
        .ok_or_else(|| SqlRsError::Query("Missing ON in CREATE INDEX".to_string()))?;
    
    let index_part = &sql[12..on_pos].trim(); // Skip "CREATE INDEX"
    let index_name = index_part.to_string();
    
    if index_name.is_empty() {
        return Err(SqlRsError::Query("Missing index name in CREATE INDEX".to_string()));
    }
    
    let after_on = &sql[on_pos + 4..].trim();
    
    // Find opening parenthesis
    let paren_start = after_on
        .find('(')
        .ok_or_else(|| SqlRsError::Query("Missing column specification in CREATE INDEX".to_string()))?;
    
    let table = after_on[..paren_start].trim().to_string();
    
    if table.is_empty() {
        return Err(SqlRsError::Query("Missing table name in CREATE INDEX".to_string()));
    }
    
    let paren_end = after_on
        .rfind(')')
        .ok_or_else(|| SqlRsError::Query("Missing closing parenthesis in CREATE INDEX".to_string()))?;
    
    let column = after_on[paren_start + 1..paren_end].trim().to_string();
    
    if column.is_empty() {
        return Err(SqlRsError::Query("Missing column name in CREATE INDEX".to_string()));
    }
    
    Ok(Statement::CreateIndex {
        index_name,
        table,
        column,
    })
}

fn parse_drop_index(sql: &str) -> Result<Statement> {
    // DROP INDEX index_name
    let parts: Vec<&str> = sql.split_whitespace().collect();
    
    if parts.len() < 3 {
        return Err(SqlRsError::Query("Invalid DROP INDEX syntax".to_string()));
    }
    
    let index_name = parts[2].to_string();
    
    if index_name.is_empty() {
        return Err(SqlRsError::Query("Missing index name in DROP INDEX".to_string()));
    }
    
    Ok(Statement::DropIndex { index_name })
}