icydb-core 0.114.1

use crate::db::sql::lowering::{SqlLoweringError, aggregate::lower_aggregate_call};
use crate::{
    db::{
        query::{
            builder::NumericProjectionExpr,
            plan::expr::{BinaryOp, CaseWhenArm, Expr, FieldId, Function, UnaryOp},
        },
        sql::parser::{SqlExpr, SqlExprBinaryOp, SqlExprUnaryOp, SqlScalarFunction},
    },
    value::Value,
};

///
/// SqlExprPhase
///
/// Lowering-time SQL expression phase boundary.
/// Clause owners pass this to the shared SQL-expression lowering seam so
/// aggregate admission stays explicit instead of leaking through wrappers.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db::sql::lowering) enum SqlExprPhase {
    Scalar,
    PreAggregate,
    PostAggregate,
}

// Lower one SQL expression tree into the canonical planner expression tree
// while enforcing the aggregate-admission rule for the owning clause phase.
pub(in crate::db::sql::lowering) fn lower_sql_expr(
    expr: &SqlExpr,
    phase: SqlExprPhase,
) -> Result<Expr, SqlLoweringError> {
    match expr {
        SqlExpr::Field(field) => Ok(Expr::Field(FieldId::new(field.clone()))),
        SqlExpr::Aggregate(aggregate) => {
            if !phase_allows_aggregate(phase) {
                return Err(phase_aggregate_error(phase));
            }

            Ok(Expr::Aggregate(lower_aggregate_call(aggregate.clone())?))
        }
        SqlExpr::Literal(literal) => Ok(Expr::Literal(literal.clone())),
        SqlExpr::Param { index } => Err(SqlLoweringError::unsupported_parameter_placement(
            Some(*index),
            "unbound SQL parameter reached expression lowering without prepare-time binding",
        )),
        SqlExpr::Membership {
            expr,
            values,
            negated,
        } => lower_sql_membership_expr(expr.as_ref(), values.as_slice(), *negated, phase),
        SqlExpr::NullTest { expr, negated } => Ok(Expr::FunctionCall {
            function: if *negated {
                Function::IsNotNull
            } else {
                Function::IsNull
            },
            args: vec![lower_sql_expr(expr.as_ref(), phase)?],
        }),
        SqlExpr::FunctionCall { function, args } => lower_sql_function_call(*function, args, phase),
        SqlExpr::Unary { op, expr } => Ok(Expr::Unary {
            op: lower_sql_unary_op(*op),
            expr: Box::new(lower_sql_expr(expr.as_ref(), phase)?),
        }),
        SqlExpr::Binary { op, left, right } => {
            lower_sql_binary_expr(*op, left.as_ref(), right.as_ref(), phase)
        }
        SqlExpr::Case { arms, else_expr } => Ok(Expr::Case {
            when_then_arms: arms
                .iter()
                .map(|arm| {
                    Ok(CaseWhenArm::new(
                        lower_sql_expr(&arm.condition, phase)?,
                        lower_sql_expr(&arm.result, phase)?,
                    ))
                })
                .collect::<Result<Vec<_>, SqlLoweringError>>()?,
            else_expr: Box::new(match else_expr.as_ref() {
                Some(else_expr) => lower_sql_expr(else_expr.as_ref(), phase)?,
                None => Expr::Literal(Value::Null),
            }),
        }),
    }
}

// Lower one parser-owned membership surface onto the existing boolean compare
// expression family so later WHERE compilation can still reuse the shipped
// normalized predicate path.
fn lower_sql_membership_expr(
    expr: &SqlExpr,
    values: &[Value],
    negated: bool,
    phase: SqlExprPhase,
) -> Result<Expr, SqlLoweringError> {
    let Some((first, rest)) = values.split_first() else {
        unreachable!("parsed membership expression must keep at least one literal");
    };

    let compare_op = if negated {
        SqlExprBinaryOp::Ne
    } else {
        SqlExprBinaryOp::Eq
    };
    let join_op = if negated {
        SqlExprBinaryOp::And
    } else {
        SqlExprBinaryOp::Or
    };

    let mut lowered =
        lower_sql_binary_expr(compare_op, expr, &SqlExpr::Literal(first.clone()), phase)?;
    for value in rest {
        lowered = Expr::Binary {
            op: lower_sql_binary_op(join_op),
            left: Box::new(lowered),
            right: Box::new(lower_sql_binary_expr(
                compare_op,
                expr,
                &SqlExpr::Literal(value.clone()),
                phase,
            )?),
        };
    }

    Ok(lowered)
}

fn lower_sql_binary_expr(
    op: SqlExprBinaryOp,
    left: &SqlExpr,
    right: &SqlExpr,
    phase: SqlExprPhase,
) -> Result<Expr, SqlLoweringError> {
    if let (SqlExpr::Field(field), SqlExpr::Literal(literal)) = (left, right)
        && let Some(expr) = lower_field_literal_numeric_expr(op, field.as_str(), literal)?
    {
        return Ok(expr);
    }

    Ok(Expr::Binary {
        op: lower_sql_binary_op(op),
        left: Box::new(lower_sql_expr(left, phase)?),
        right: Box::new(lower_sql_expr(right, phase)?),
    })
}

fn lower_field_literal_numeric_expr(
    op: SqlExprBinaryOp,
    field: &str,
    literal: &Value,
) -> Result<Option<Expr>, SqlLoweringError> {
    let builder = match op {
        SqlExprBinaryOp::Add => Some(NumericProjectionExpr::add_value(
            field.to_string(),
            literal.clone(),
        )),
        SqlExprBinaryOp::Sub => Some(NumericProjectionExpr::sub_value(
            field.to_string(),
            literal.clone(),
        )),
        SqlExprBinaryOp::Mul => Some(NumericProjectionExpr::mul_value(
            field.to_string(),
            literal.clone(),
        )),
        SqlExprBinaryOp::Div => Some(NumericProjectionExpr::div_value(
            field.to_string(),
            literal.clone(),
        )),
        SqlExprBinaryOp::Or
        | SqlExprBinaryOp::And
        | SqlExprBinaryOp::Eq
        | SqlExprBinaryOp::Ne
        | SqlExprBinaryOp::Lt
        | SqlExprBinaryOp::Lte
        | SqlExprBinaryOp::Gt
        | SqlExprBinaryOp::Gte => None,
    };

    builder
        .transpose()
        .map(|projection| projection.map(|projection| projection.expr().clone()))
        .map_err(SqlLoweringError::from)
}

const fn phase_allows_aggregate(phase: SqlExprPhase) -> bool {
    matches!(phase, SqlExprPhase::PostAggregate)
}

fn phase_aggregate_error(phase: SqlExprPhase) -> SqlLoweringError {
    match phase {
        SqlExprPhase::Scalar => SqlLoweringError::unsupported_select_projection(),
        SqlExprPhase::PreAggregate => SqlLoweringError::unsupported_aggregate_input_expressions(),
        SqlExprPhase::PostAggregate => {
            unreachable!("post-aggregate lowering allows aggregate leaves")
        }
    }
}

const fn lower_sql_unary_op(op: SqlExprUnaryOp) -> UnaryOp {
    match op {
        SqlExprUnaryOp::Not => UnaryOp::Not,
    }
}

pub(in crate::db::sql::lowering) const fn lower_sql_binary_op(op: SqlExprBinaryOp) -> BinaryOp {
    match op {
        SqlExprBinaryOp::Or => BinaryOp::Or,
        SqlExprBinaryOp::And => BinaryOp::And,
        SqlExprBinaryOp::Eq => BinaryOp::Eq,
        SqlExprBinaryOp::Ne => BinaryOp::Ne,
        SqlExprBinaryOp::Lt => BinaryOp::Lt,
        SqlExprBinaryOp::Lte => BinaryOp::Lte,
        SqlExprBinaryOp::Gt => BinaryOp::Gt,
        SqlExprBinaryOp::Gte => BinaryOp::Gte,
        SqlExprBinaryOp::Add => BinaryOp::Add,
        SqlExprBinaryOp::Sub => BinaryOp::Sub,
        SqlExprBinaryOp::Mul => BinaryOp::Mul,
        SqlExprBinaryOp::Div => BinaryOp::Div,
    }
}

// Report whether one parsed SQL expression shape is admitted as a boolean
// result surface by SQL lowering before unbound parameters are substituted.
// This stays parser/lowering-owned because it classifies SQL expression
// structure, not planner-owned semantic type results.
pub(in crate::db::sql::lowering) const fn sql_expr_is_boolean_shape(expr: &SqlExpr) -> bool {
    match expr {
        SqlExpr::Membership { .. }
        | SqlExpr::NullTest { .. }
        | SqlExpr::Unary { .. }
        | SqlExpr::FunctionCall { .. }
        | SqlExpr::Case { .. } => true,
        SqlExpr::Binary { op, .. } => matches!(
            op,
            SqlExprBinaryOp::Or
                | SqlExprBinaryOp::And
                | SqlExprBinaryOp::Eq
                | SqlExprBinaryOp::Ne
                | SqlExprBinaryOp::Lt
                | SqlExprBinaryOp::Lte
                | SqlExprBinaryOp::Gt
                | SqlExprBinaryOp::Gte
        ),
        SqlExpr::Field(_) | SqlExpr::Literal(_) | SqlExpr::Param { .. } | SqlExpr::Aggregate(_) => {
            false
        }
    }
}

///
/// PreparedSqlPredicateTemplateShape
///
/// Parser-owned shape classification for the bounded symbolic scalar prepared
/// predicate lane. Session binding consumes this to pair parsed SQL predicate
/// structure with planner-owned compare metadata without re-deriving the SQL
/// tree shape locally.
///

pub(in crate::db) enum PreparedSqlPredicateTemplateShape<'a> {
    And {
        left: &'a SqlExpr,
        right: &'a SqlExpr,
    },
    Or {
        left: &'a SqlExpr,
        right: &'a SqlExpr,
    },
    Not {
        expr: &'a SqlExpr,
    },
    CompareWithParamRhs {
        slot_index: usize,
    },
}

// Classify one parsed SQL predicate subtree for the bounded symbolic scalar
// prepared lane. This stays lowering-owned because it is only interpreting
// parser structure, not planner predicate meaning.
pub(in crate::db) fn sql_expr_prepared_predicate_template_shape(
    expr: &SqlExpr,
) -> Option<PreparedSqlPredicateTemplateShape<'_>> {
    match expr {
        SqlExpr::Binary {
            op: SqlExprBinaryOp::And,
            left,
            right,
        } => Some(PreparedSqlPredicateTemplateShape::And { left, right }),
        SqlExpr::Binary {
            op: SqlExprBinaryOp::Or,
            left,
            right,
        } => Some(PreparedSqlPredicateTemplateShape::Or { left, right }),
        SqlExpr::Unary { expr, .. } => Some(PreparedSqlPredicateTemplateShape::Not { expr }),
        SqlExpr::Binary {
            op:
                SqlExprBinaryOp::Eq
                | SqlExprBinaryOp::Ne
                | SqlExprBinaryOp::Lt
                | SqlExprBinaryOp::Lte
                | SqlExprBinaryOp::Gt
                | SqlExprBinaryOp::Gte,
            left,
            right,
        } => match (&**left, &**right) {
            (SqlExpr::Field(_), SqlExpr::Param { index }) => {
                Some(PreparedSqlPredicateTemplateShape::CompareWithParamRhs { slot_index: *index })
            }
            _ => None,
        },
        _ => None,
    }
}

///
/// PreparedSqlTemplateExprScope
///
/// Scoped template-admission classifier for prepared SQL expression lanes.
/// Lowering owns this parser-shape decision so prepared statement assembly can
/// consume one shared expression classifier instead of keeping a local ladder
/// inside the prepared lowering owner.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db::sql::lowering) enum PreparedSqlTemplateExprScope {
    Filter,
    Having,
}

// Return whether one parsed SQL expression uses parameter slots inside one
// general expression-owned shape that must stay off template lanes for the
// selected scope.
pub(in crate::db) fn sql_expr_uses_general_template_expr_parameters(
    scope: PreparedSqlTemplateExprScope,
    expr: &SqlExpr,
) -> bool {
    match expr {
        SqlExpr::Field(_) | SqlExpr::Literal(_) | SqlExpr::Param { .. } | SqlExpr::Aggregate(_) => {
            false
        }
        SqlExpr::Membership { expr, .. } => sql_expr_contains_param(expr.as_ref()),
        SqlExpr::NullTest { expr, .. } => !sql_null_test_target_is_template_shape(scope, expr),
        SqlExpr::Unary { expr, .. } => sql_expr_uses_general_template_expr_parameters(scope, expr),
        SqlExpr::Binary { op, left, right } => match op {
            SqlExprBinaryOp::And | SqlExprBinaryOp::Or => {
                sql_expr_uses_general_template_expr_parameters(scope, left)
                    || sql_expr_uses_general_template_expr_parameters(scope, right)
            }
            SqlExprBinaryOp::Eq
            | SqlExprBinaryOp::Ne
            | SqlExprBinaryOp::Lt
            | SqlExprBinaryOp::Lte
            | SqlExprBinaryOp::Gt
            | SqlExprBinaryOp::Gte => {
                !(sql_compare_target_is_template_shape(scope, left)
                    && sql_compare_target_is_template_shape(scope, right))
                    && (sql_expr_contains_param(left) || sql_expr_contains_param(right))
            }
            SqlExprBinaryOp::Add
            | SqlExprBinaryOp::Sub
            | SqlExprBinaryOp::Mul
            | SqlExprBinaryOp::Div => {
                sql_expr_contains_param(left) || sql_expr_contains_param(right)
            }
        },
        SqlExpr::FunctionCall { function, args } => {
            !sql_function_is_template_shape(scope, *function, args.as_slice())
                && args.iter().any(sql_expr_contains_param)
        }
        SqlExpr::Case { arms, else_expr } => {
            arms.iter().any(|arm| {
                sql_expr_contains_param(&arm.condition) || sql_expr_contains_param(&arm.result)
            }) || else_expr
                .as_ref()
                .is_some_and(|expr| sql_expr_contains_param(expr))
        }
    }
}

// Return whether one parsed SQL expression is a direct compare/text-predicate
// operand shape that prepared predicate/access templates are still allowed to
// own.
fn sql_compare_target_is_template_shape(
    scope: PreparedSqlTemplateExprScope,
    expr: &SqlExpr,
) -> bool {
    match scope {
        PreparedSqlTemplateExprScope::Filter => match expr {
            SqlExpr::Field(_) | SqlExpr::Literal(_) | SqlExpr::Param { .. } => true,
            SqlExpr::FunctionCall { function, args }
                if matches!(
                    function,
                    SqlScalarFunction::Lower | SqlScalarFunction::Upper
                ) && matches!(args.as_slice(), [SqlExpr::Field(_)]) =>
            {
                true
            }
            _ => false,
        },
        PreparedSqlTemplateExprScope::Having => matches!(
            expr,
            SqlExpr::Field(_) | SqlExpr::Literal(_) | SqlExpr::Param { .. } | SqlExpr::Aggregate(_)
        ),
    }
}

// Return whether one parsed SQL null-test target still matches the predicate
// lane that prepared templates are allowed to rebuild.
fn sql_null_test_target_is_template_shape(
    scope: PreparedSqlTemplateExprScope,
    expr: &SqlExpr,
) -> bool {
    match scope {
        PreparedSqlTemplateExprScope::Filter => matches!(expr, SqlExpr::Field(_)),
        PreparedSqlTemplateExprScope::Having => !sql_expr_contains_param(expr),
    }
}

// Return whether one parsed SQL function call is one direct text predicate
// shape still owned by prepared predicate templates.
fn sql_function_is_template_shape(
    scope: PreparedSqlTemplateExprScope,
    function: SqlScalarFunction,
    args: &[SqlExpr],
) -> bool {
    match scope {
        PreparedSqlTemplateExprScope::Filter => {
            matches!(
                function,
                SqlScalarFunction::StartsWith
                    | SqlScalarFunction::EndsWith
                    | SqlScalarFunction::Contains
            ) && matches!(
                args,
                [left, right]
                    if sql_compare_target_is_template_shape(
                        PreparedSqlTemplateExprScope::Filter,
                        left
                    ) && sql_compare_target_is_template_shape(
                        PreparedSqlTemplateExprScope::Filter,
                        right
                    )
            )
        }
        PreparedSqlTemplateExprScope::Having => false,
    }
}

// Return whether one parsed SQL boolean expression is compound enough that
// grouped symbolic template ownership must keep failing closed instead of
// claiming a combined access-plus-residual route. This remains lowering-owned
// because it classifies parser structure, not session binding policy.
pub(in crate::db) const fn sql_expr_is_compound_boolean_shape(expr: &SqlExpr) -> bool {
    matches!(
        expr,
        SqlExpr::Binary {
            op: SqlExprBinaryOp::And | SqlExprBinaryOp::Or,
            ..
        } | SqlExpr::Unary { .. }
    )
}

// Walk one parsed SQL expression and report whether it references any runtime
// parameter slot. Lowering owns this parser-tree traversal so prepared
// lowering and session binding do not each keep their own local walker.
pub(in crate::db) fn sql_expr_contains_param(expr: &SqlExpr) -> bool {
    sql_expr_first_param_index(expr).is_some()
}

// Walk one parsed SQL expression tree and report whether it already contains
// any literal equal to one of the supplied values. Lowering owns this parser
// traversal so session binding can reuse one shared SQL-expression walk for
// template-sentinel collision checks instead of keeping its own copy.
pub(in crate::db) fn sql_expr_contains_any_literal(expr: &SqlExpr, values: &[Value]) -> bool {
    match expr {
        SqlExpr::Field(_) | SqlExpr::Param { .. } => false,
        SqlExpr::Aggregate(aggregate) => {
            aggregate
                .input
                .as_ref()
                .is_some_and(|expr| sql_expr_contains_any_literal(expr, values))
                || aggregate
                    .filter_expr
                    .as_ref()
                    .is_some_and(|expr| sql_expr_contains_any_literal(expr, values))
        }
        SqlExpr::Literal(value) => values.contains(value),
        SqlExpr::Membership {
            expr,
            values: members,
            ..
        } => {
            sql_expr_contains_any_literal(expr, values)
                || members.iter().any(|value| values.contains(value))
        }
        SqlExpr::NullTest { expr, .. } | SqlExpr::Unary { expr, .. } => {
            sql_expr_contains_any_literal(expr, values)
        }
        SqlExpr::FunctionCall { args, .. } => args
            .iter()
            .any(|arg| sql_expr_contains_any_literal(arg, values)),
        SqlExpr::Binary { left, right, .. } => {
            sql_expr_contains_any_literal(left, values)
                || sql_expr_contains_any_literal(right, values)
        }
        SqlExpr::Case { arms, else_expr } => {
            arms.iter().any(|arm| {
                sql_expr_contains_any_literal(&arm.condition, values)
                    || sql_expr_contains_any_literal(&arm.result, values)
            }) || else_expr
                .as_ref()
                .is_some_and(|expr| sql_expr_contains_any_literal(expr, values))
        }
    }
}

// Return the first runtime parameter slot referenced by one parsed SQL
// expression. This stays lowering-owned because it classifies parser-owned
// SQL structure rather than prepared-lane semantics.
pub(in crate::db) fn sql_expr_first_param_index(expr: &SqlExpr) -> Option<usize> {
    match expr {
        SqlExpr::Param { index } => Some(*index),
        SqlExpr::Field(_) | SqlExpr::Literal(_) => None,
        SqlExpr::Aggregate(aggregate) => aggregate
            .input
            .as_deref()
            .and_then(sql_expr_first_param_index)
            .or_else(|| {
                aggregate
                    .filter_expr
                    .as_deref()
                    .and_then(sql_expr_first_param_index)
            }),
        SqlExpr::Membership { expr, .. }
        | SqlExpr::NullTest { expr, .. }
        | SqlExpr::Unary { expr, .. } => sql_expr_first_param_index(expr),
        SqlExpr::FunctionCall { args, .. } => sql_expr_first_param_index_from_args(args.as_slice()),
        SqlExpr::Binary { left, right, .. } => {
            sql_expr_first_param_index(left).or_else(|| sql_expr_first_param_index(right))
        }
        SqlExpr::Case { arms, else_expr } => arms
            .iter()
            .find_map(|arm| {
                sql_expr_first_param_index(&arm.condition)
                    .or_else(|| sql_expr_first_param_index(&arm.result))
            })
            .or_else(|| else_expr.as_deref().and_then(sql_expr_first_param_index)),
    }
}

// Return the first runtime parameter slot referenced by one parsed SQL
// argument list.
pub(in crate::db) fn sql_expr_first_param_index_from_args(args: &[SqlExpr]) -> Option<usize> {
    args.iter().find_map(sql_expr_first_param_index)
}

// Return the first runtime parameter slot referenced across one parsed SQL
// left/right pair.
pub(in crate::db) fn sql_expr_first_param_index_from_pair(
    left: &SqlExpr,
    right: &SqlExpr,
) -> Option<usize> {
    sql_expr_first_param_index(left).or_else(|| sql_expr_first_param_index(right))
}

pub(in crate::db::sql::lowering) const fn lower_sql_scalar_function(
    function: SqlScalarFunction,
) -> Function {
    match function {
        SqlScalarFunction::Trim => Function::Trim,
        SqlScalarFunction::Ltrim => Function::Ltrim,
        SqlScalarFunction::Rtrim => Function::Rtrim,
        SqlScalarFunction::Round => Function::Round,
        SqlScalarFunction::Coalesce => Function::Coalesce,
        SqlScalarFunction::NullIf => Function::NullIf,
        SqlScalarFunction::Abs => Function::Abs,
        SqlScalarFunction::Ceil => Function::Ceil,
        SqlScalarFunction::Ceiling => Function::Ceiling,
        SqlScalarFunction::Floor => Function::Floor,
        SqlScalarFunction::Lower => Function::Lower,
        SqlScalarFunction::Upper => Function::Upper,
        SqlScalarFunction::Length => Function::Length,
        SqlScalarFunction::Left => Function::Left,
        SqlScalarFunction::Right => Function::Right,
        SqlScalarFunction::StartsWith => Function::StartsWith,
        SqlScalarFunction::EndsWith => Function::EndsWith,
        SqlScalarFunction::Contains => Function::Contains,
        SqlScalarFunction::Position => Function::Position,
        SqlScalarFunction::Replace => Function::Replace,
        SqlScalarFunction::Substring => Function::Substring,
    }
}

fn lower_sql_function_call(
    function: SqlScalarFunction,
    args: &[SqlExpr],
    phase: SqlExprPhase,
) -> Result<Expr, SqlLoweringError> {
    if matches!(function, SqlScalarFunction::Round) {
        return lower_sql_round_function_call(args, phase);
    }

    Ok(Expr::FunctionCall {
        function: lower_sql_scalar_function(function),
        args: args
            .iter()
            .map(|arg| lower_sql_expr(arg, phase))
            .collect::<Result<Vec<_>, SqlLoweringError>>()?,
    })
}

fn lower_sql_round_function_call(
    args: &[SqlExpr],
    phase: SqlExprPhase,
) -> Result<Expr, SqlLoweringError> {
    if !(1..=2).contains(&args.len()) {
        return Err(crate::db::QueryError::unsupported_query(format!(
            "ROUND(...) expects 1 or 2 args, found {}",
            args.len()
        ))
        .into());
    }

    let input = lower_sql_expr(&args[0], phase)?;
    let scale = match args.get(1) {
        Some(SqlExpr::Literal(scale)) => Expr::Literal(Value::Uint(u64::from(
            validate_round_projection_scale(scale.clone())?,
        ))),
        Some(other) => lower_sql_expr(other, phase)?,
        None => Expr::Literal(Value::Uint(0)),
    };

    Ok(Expr::FunctionCall {
        function: Function::Round,
        args: vec![input, scale],
    })
}

fn validate_round_projection_scale(scale: Value) -> Result<u32, SqlLoweringError> {
    match scale {
        Value::Int(value) => u32::try_from(value).map_err(|_| {
            crate::db::QueryError::unsupported_query(format!(
                "ROUND(...) requires non-negative integer scale, found {value}",
            ))
            .into()
        }),
        Value::Uint(value) => u32::try_from(value).map_err(|_| {
            crate::db::QueryError::unsupported_query(format!(
                "ROUND(...) scale exceeds supported integer range, found {value}",
            ))
            .into()
        }),
        other => Err(crate::db::QueryError::unsupported_query(format!(
            "ROUND(...) requires integer scale, found {other:?}",
        ))
        .into()),
    }
}