hamelin_sql 0.7.3

use anyhow::bail;

use hamelin_lib::func::def::ParameterBinding;
use hamelin_lib::sql::expression::apply::{
    BinaryOperatorApply, FunctionCallApply, Lambda, UnaryOperatorApply,
};
use hamelin_lib::sql::expression::literal::{
    BooleanLiteral, IntegerLiteral, NullLiteral, StringLiteral,
};
use hamelin_lib::sql::expression::operator::Operator;
use hamelin_lib::sql::expression::IndexLookup;
use hamelin_lib::sql::expression::{Leaf, SQLExpression};
use hamelin_lib::translation::ExpressionTranslation;
use hamelin_lib::types::TIMESTAMP;

use crate::range_builder::RangeBuilder;

/// Default translation for a function call. It just calls a SQL function with the same name.
pub fn direct_function_translation(
    func: &str,
    bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let all_args = bindings.into_iter().map(|t| t.sql).collect();
    let fca: SQLExpression = FunctionCallApply::with_positional(func, all_args).into();

    Ok(fca)
}

pub fn function_translation_respecting_ignore_nulls(
    func: &str,
    mut bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let ignore_nulls = bindings.take_by_name("ignore_nulls")?;
    let all_args = bindings.into_iter().map(|t| t.sql).collect();
    let mut fca = FunctionCallApply::with_positional(func, all_args);

    match ignore_nulls.sql {
        SQLExpression::Leaf(Leaf::BooleanLiteral(b)) => {
            if b.value {
                fca = fca.with_ignore_nulls();
            }
        }
        _ => bail!("ignore_nulls must be boolean literal"),
    }

    Ok(fca.into())
}

/// Default translation for a unary operator.
pub fn direct_unary_operator_translation(
    func: &str,
    mut bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let sql_op: Operator = hamelin_lib::operator::Operator::of(func)?.try_into()?;
    Ok(UnaryOperatorApply::new(sql_op, bindings.take()?.sql).into())
}

/// Default translation for a binary operator.
pub fn direct_binary_operator_translation(
    func: &str,
    mut bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let left = bindings.take()?.sql;
    let right = bindings.take()?.sql;
    let sql_op: Operator = hamelin_lib::operator::Operator::of(func)?.try_into()?;
    Ok(BinaryOperatorApply::new(sql_op, left, right).into())
}

/// Default translation for a binary operator that calls SQL's concatenation operator.
pub fn to_sql_concat(
    _: &str,
    mut bindings: ParameterBinding<ExpressionTranslation>,
) -> anyhow::Result<SQLExpression> {
    let p0 = bindings.take()?;
    let p1 = bindings.take()?;
    Ok(BinaryOperatorApply::new(Operator::Concat, p0.sql, p1.sql).into())
}

/// Converts an interval expression to a timestamp range from now() + interval to now().
pub fn interval_to_range(interval: SQLExpression) -> anyhow::Result<RangeBuilder> {
    Ok(RangeBuilder::default()
        .with_begin(interval_to_timestamp(interval), TIMESTAMP.to_sql()?.into())
        .with_end(
            FunctionCallApply::with_no_arguments("now").into(),
            TIMESTAMP.to_sql()?.into(),
        ))
}

/// Converts a timestamp expression to a timestamp range from timestamp to now().
pub fn timestamp_to_range(timestamp: SQLExpression) -> anyhow::Result<RangeBuilder> {
    Ok(RangeBuilder::default()
        .with_begin(timestamp, TIMESTAMP.to_sql()?.into())
        .with_end(
            FunctionCallApply::with_no_arguments("now").into(),
            TIMESTAMP.to_sql()?.into(),
        ))
}

/// Converts an interval expression to a timestamp by calculating now() + interval.
pub fn interval_to_timestamp(interval: SQLExpression) -> SQLExpression {
    BinaryOperatorApply::new(
        Operator::Plus,
        FunctionCallApply::with_no_arguments("now").into(),
        interval,
    )
    .into()
}

/// Converts a timestamp range from an interval range by converting each interval to a timestamp by
/// adding it to now().
pub fn interval_range_to_timestamp_range(
    range_expr: SQLExpression,
) -> anyhow::Result<RangeBuilder> {
    let ts_sql = TIMESTAMP.to_sql()?;
    if let Some(mut rb) = RangeBuilder::from_literal(range_expr.clone()) {
        if !matches!(&rb.begin, SQLExpression::Leaf(Leaf::NullLiteral(_))) {
            rb = rb
                .clone()
                .with_begin(interval_to_timestamp(rb.begin.clone()), ts_sql.clone());
        }
        if !matches!(&rb.end, SQLExpression::Leaf(Leaf::NullLiteral(_))) {
            rb = rb
                .clone()
                .with_end(interval_to_timestamp(rb.end.clone()), ts_sql);
        }
        Ok(rb)
    } else {
        Ok(RangeBuilder::default()
            .with_begin(
                interval_to_timestamp(range_expr.clone().index(IntegerLiteral::new("1").into())),
                ts_sql.clone().into(),
            )
            .with_end(
                interval_to_timestamp(range_expr.clone().index(IntegerLiteral::new("2").into())),
                ts_sql.into(),
            ))
    }
}

/// A helper that implements WITHIN (range) and also (exp) IN (range) where range is a literal
pub fn within_range(left: SQLExpression, range: RangeBuilder) -> SQLExpression {
    BinaryOperatorApply::new(
        Operator::And,
        if let SQLExpression::Leaf(Leaf::NullLiteral(_)) = range.begin {
            BooleanLiteral::new(true).into()
        } else {
            BinaryOperatorApply::new(Operator::Gte, left.clone(), range.begin).into()
        },
        if let SQLExpression::Leaf(Leaf::NullLiteral(_)) = range.end {
            BooleanLiteral::new(true).into()
        } else {
            BinaryOperatorApply::new(Operator::Lte, left, range.end).into()
        },
    )
    .into()
}

/// A helper that implements WITHIN (range) and also (exp) IN (range) where range is an expression
pub fn within_range_expr(left: SQLExpression, right: SQLExpression) -> SQLExpression {
    BinaryOperatorApply::new(
        Operator::And,
        FunctionCallApply::with_two(
            "coalesce",
            BinaryOperatorApply::new(
                Operator::Gte,
                left.clone(),
                IndexLookup::new(right.clone(), IntegerLiteral::new("1").into()).into(),
            )
            .into(),
            BooleanLiteral::new(true).into(),
        )
        .into(),
        FunctionCallApply::with_two(
            "coalesce",
            BinaryOperatorApply::new(
                Operator::Lt,
                left,
                IndexLookup::new(right.clone(), IntegerLiteral::new("2").into()).into(),
            )
            .into(),
            BooleanLiteral::new(true).into(),
        )
        .into(),
    )
    .into()
}

/// Converts nested transform calls to a different function.
///
/// For example, if the original expression is `transform(a, x -> transform(b, y -> f(x, y)))`
/// and `toFunction` is `g`, this function will return `g(a, x -> g(b, y -> f(x, y)))`.
///
/// # Arguments
/// - `original` - the original expression
/// - `to_function` - the function to convert the nested transform calls to
///
/// # Returns
/// The converted expression
pub fn convert_nested_transforms(original: SQLExpression, to_function: &str) -> SQLExpression {
    if let SQLExpression::FunctionCallApply(fca) = &original {
        if fca.function_name == "transform" {
            if let (Some(first_arg), Some(SQLExpression::Lambda(lmbda))) =
                (fca.arguments.first(), fca.arguments.get(1))
            {
                return FunctionCallApply::with_two(
                    to_function,
                    first_arg.clone(),
                    Lambda::new(
                        lmbda.arguments.clone(),
                        convert_nested_transforms(lmbda.body.as_ref().clone(), to_function),
                    )
                    .into(),
                )
                .into();
            }
        }
    }

    original
}

pub fn wrap_timestamp(expr: SQLExpression) -> SQLExpression {
    FunctionCallApply::with_one(
        "try",
        FunctionCallApply::with_one(
            "from_iso8601_timestamp",
            FunctionCallApply::with_three(
                "replace",
                expr,
                StringLiteral::new(" ").into(),
                StringLiteral::new("T").into(),
            )
            .into(),
        )
        .into(),
    )
    .into()
}

/// Get the length (cardinality) of an array expression
pub fn array_length(array_expr: SQLExpression) -> SQLExpression {
    FunctionCallApply::with_one("cardinality", array_expr).into()
}

/// Get the length of a string expression
pub fn string_length(string_expr: SQLExpression) -> SQLExpression {
    FunctionCallApply::with_one("length", string_expr).into()
}

/// Convert a negative index to positive based on the provided length expression
/// This maintains Hamelin's zero-based indexing semantics
fn negative_index_to_positive(
    index_expr: SQLExpression,
    length_expr: SQLExpression,
) -> SQLExpression {
    FunctionCallApply::with_three(
        "if",
        BinaryOperatorApply::new(
            Operator::Lt,
            index_expr.clone(),
            IntegerLiteral::new("0").into(),
        )
        .into(),
        BinaryOperatorApply::new(Operator::Plus, length_expr, index_expr.clone()).into(),
        index_expr,
    )
    .into()
}

pub fn array_negative_index_to_positive(
    index_expr: SQLExpression,
    array_expr: SQLExpression,
) -> SQLExpression {
    negative_index_to_positive(index_expr.clone(), array_length(array_expr))
}

pub fn string_negative_index_to_positive(
    index_expr: SQLExpression,
    string_expr: SQLExpression,
) -> SQLExpression {
    negative_index_to_positive(index_expr.clone(), string_length(string_expr))
}

/// Convert a zero-based Hamelin index to one-based SQL index, handling negative indices.
/// After conversion, out-of-bounds negative indices produce values <= 0. Since Trino's
/// `element_at` throws on index 0 and treats negative indices as "from end", we guard
/// against both by returning NULL for any converted index < 1.
pub fn hamelin_array_index_to_sql_with_negatives(
    index_expr: SQLExpression,
    array_expr: SQLExpression,
) -> SQLExpression {
    let positive_index = array_negative_index_to_positive(index_expr.clone(), array_expr);

    // Convert to SQL space: always add 1 to convert from zero-based to one-based indexing
    // This applies to both originally positive and originally negative indices after conversion
    let one_based: SQLExpression = BinaryOperatorApply::new(
        Operator::Plus,
        positive_index,
        IntegerLiteral::new("1").into(),
    )
    .into();

    // Guard: if the 1-based index is < 1, the negative index was out of bounds.
    // Return NULL instead of 0 or a negative value (which would cause errors or
    // wrong results with element_at).
    FunctionCallApply::with_three(
        "if",
        BinaryOperatorApply::new(
            Operator::Lt,
            one_based.clone(),
            IntegerLiteral::new("1").into(),
        )
        .into(),
        NullLiteral::default().into(),
        one_based,
    )
    .into()
}

/// Convert a zero-based Hamelin string index to one-based SQL index, handling negative indices
/// For string operations like substr where negative indices need special handling
pub fn hamelin_string_index_to_sql_with_negatives(
    index_expr: SQLExpression,
    string_expr: SQLExpression,
) -> SQLExpression {
    let positive_index = string_negative_index_to_positive(index_expr.clone(), string_expr);

    // Convert to SQL space: always add 1 to convert from zero-based to one-based indexing
    // This applies to both originally positive and originally negative indices after conversion
    BinaryOperatorApply::new(
        Operator::Plus,
        positive_index,
        IntegerLiteral::new("1").into(),
    )
    .into()
}