gluesql / gluesql / 22275856923

pub async fn evaluate<'a, 'b, 'c, T>(

    storage: &'a T,

    context: Option<Arc<RowContext<'b>>>,

    aggregated: Option<Arc<HashMap<&'c Aggregate, Value>>>,

    expr: &'a Expr,

) -> Result<Evaluated<'a>>

where

    'b: 'a,

    'c: 'a,

    T: GStore,

{

    evaluate_inner(Some(storage), context, aggregated, expr).await

}

pub async fn evaluate_stateless<'a, 'b: 'a>(

    context: Option<RowContext<'b>>,

    expr: &'a Expr,

) -> Result<Evaluated<'a>> {

    let context = context.map(Arc::new);

    let storage: Option<&MockStorage> = None;

    evaluate_inner(storage, context, None, expr).await

}

{

    let eval = |expr| {

        let context = context.as_ref().map(Arc::clone);

        let aggregated = aggregated.as_ref().map(Arc::clone);

        evaluate_inner(storage, context, aggregated, expr)

    };

    match expr {

        Expr::Literal(literal) => Ok(expr::literal(literal)),

        Expr::Value(value) => Ok(Evaluated::Value(Cow::Borrowed(value))),

        Expr::TypedString { data_type, value } => expr::typed_string(data_type, value),

        Expr::Identifier(ident) => {

            let context = context.ok_or_else(|| {

                EvaluateError::ContextRequiredForIdentEvaluation(Box::new(expr.clone()))

            })?;

            match context.get_value(ident) {

                Some(value) => Ok(Evaluated::Value(Cow::Owned(value.clone()))),

                None => Err(EvaluateError::IdentifierNotFound(ident.to_owned()).into()),

        Expr::Nested(expr) => eval(expr).await,

        Expr::CompoundIdentifier { alias, ident } => {

            let context = context.ok_or_else(|| {

                EvaluateError::ContextRequiredForIdentEvaluation(Box::new(expr.clone()))

            })?;

            match context.get_alias_value(alias, ident) {

                Some(value) => Ok(Evaluated::Value(Cow::Owned(value.clone()))),

                None => Err(EvaluateError::CompoundIdentifierNotFound {

                    table_alias: alias.to_owned(),

                    column_name: ident.to_owned(),

                }

                .into()),

        Expr::Subquery(query) => {

            let storage = storage

                .ok_or_else(|| EvaluateError::UnsupportedStatelessExpr(Box::new(expr.clone())))?;

            if let SetExpr::Select(select) = &query.body

                && matches!(select.projection, Projection::SchemalessMap)

                return Err(EvaluateError::SchemalessProjectionForSubQuery.into());

            }

            let evaluations = select(storage, query, context.as_ref().map(Arc::clone))

                .await?

                .map(|row| {

                    let values = row?.into_values();

                    if values.len() > 1 {

                        return Err(EvaluateError::MoreThanOneColumnReturned.into());

                    }

                    let value = values.into_iter().next();

                    Ok::<_, Error>(value)

                })

                .take(2)

                .try_collect::<Vec<_>>()

                .await?;

            if evaluations.len() > 1 {

                return Err(EvaluateError::MoreThanOneRowReturned.into());

            }

            let value = evaluations

                .into_iter()

                .next()

                .flatten()

                .unwrap_or(Value::Null);

            Ok(Evaluated::Value(Cow::Owned(value)))

        Expr::BinaryOp { op, left, right } => {

            let left = eval(left).await?;

            let right = eval(right).await?;

            expr::binary_op(op, left, right)

        Expr::UnaryOp { op, expr } => {

            let v = eval(expr).await?;

            expr::unary_op(op, v)

        Expr::Aggregate(aggr) => match aggregated

            .as_ref()

            .and_then(|aggregated| aggregated.get(aggr.as_ref()))

            Some(value) => Ok(Evaluated::Value(Cow::Owned(value.clone()))),

        Expr::Function(func) => {

            let context = context.as_ref().map(Arc::clone);

            let aggregated = aggregated.as_ref().map(Arc::clone);

            evaluate_function(storage, context, aggregated, func).await

            expr,

            list,

            negated,

            let negated = *negated;

            let target = eval(expr).await?;

            if target.is_null() {

                return Ok(target);

            }

            let matched = try_join_all(list.iter().map(eval))

                .await?

                .into_iter()

                .any(|v| v.evaluate_eq(&target).is_true());

            Ok(Evaluated::Value(Cow::Owned(Value::Bool(matched ^ negated))))

            expr: target_expr,

            subquery,

            negated,

            let storage = storage

                .ok_or_else(|| EvaluateError::UnsupportedStatelessExpr(Box::new(expr.clone())))?;

            if let SetExpr::Select(select) = &subquery.body

                && matches!(select.projection, Projection::SchemalessMap)

                return Err(EvaluateError::SchemalessProjectionForInSubQuery.into());

            }

            let target = eval(target_expr).await?;

            select(storage, subquery, context)

                .await?

                .map(|row| {

                    let value = row?.into_values().into_iter().next().unwrap_or(Value::Null);

                    Ok(Evaluated::Value(Cow::Owned(value)))

                })

                .try_filter(|evaluated| ready(evaluated.evaluate_eq(&target).is_true()))

                .try_next()

                .map(|v| v.is_some() ^ negated)

                .map(Value::Bool)

                .map(|v| Evaluated::Value(Cow::Owned(v)))

            expr,

            negated,

            low,

            high,

            let target = eval(expr).await?;

            let low = eval(low).await?;

            let high = eval(high).await?;

            Ok(expr::between(&target, *negated, &low, &high))

            expr,

            negated,

            pattern,

            let target = eval(expr).await?;

            let pattern = eval(pattern).await?;

            let evaluated = target.like(pattern, true)?;

            Ok(match negated {

                    let t =

                        evaluated.evaluate_eq(&Evaluated::Value(Cow::Owned(Value::Bool(false))));

                    Evaluated::Value(Cow::Owned(Value::from(t)))

                false => evaluated,

            expr,

            negated,

            pattern,

            let target = eval(expr).await?;

            let pattern = eval(pattern).await?;

            let evaluated = target.like(pattern, false)?;

            Ok(match negated {

                    let t =

                        evaluated.evaluate_eq(&Evaluated::Value(Cow::Owned(Value::Bool(false))));

                    Evaluated::Value(Cow::Owned(Value::from(t)))

                false => evaluated,

        Expr::Exists { subquery, negated } => {

            let storage = storage

                .ok_or_else(|| EvaluateError::UnsupportedStatelessExpr(Box::new(expr.clone())))?;

            select(storage, subquery, context)

                .await?

                .try_next()

                .map(|v| v.is_some() ^ negated)

                .map(Value::Bool)

                .map(|v| Evaluated::Value(Cow::Owned(v)))

        Expr::IsNull(expr) => {

            let v = eval(expr).await?.is_null();

            Ok(Evaluated::Value(Cow::Owned(Value::Bool(v))))

        Expr::IsNotNull(expr) => {

            let v = eval(expr).await?.is_null();

            Ok(Evaluated::Value(Cow::Owned(Value::Bool(!v))))

            operand,

            when_then,

            else_result,

            let operand = match operand {

                Some(op) => eval(op).await?,

                None => Evaluated::Value(Cow::Owned(Value::Bool(true))),

            for (when, then) in when_then {

                let when = eval(when).await?;

                if when.evaluate_eq(&operand).is_true() {

                    return eval(then).await;

                }

            match else_result {

                Some(er) => eval(er).await,

                None => Ok(Evaluated::Value(Cow::Owned(Value::Null))),

        Expr::ArrayIndex { obj, indexes } => {

            let obj = eval(obj).await?;

            let indexes = try_join_all(indexes.iter().map(eval)).await?;

            expr::array_index(obj, indexes)

        Expr::Array { elem } => try_join_all(elem.iter().map(eval))

            .await?

            .into_iter()

            .map(Value::try_from)

            .collect::<Result<Vec<_>>>()

            .map(Value::List)

            .map(|v| Evaluated::Value(Cow::Owned(v))),

            expr,

            leading_field,

            last_field,

            let value = eval(expr)

                .and_then(Value::try_from)

                .map(String::from)?;

            Interval::try_from_str(&value, *leading_field, *last_field)

                .map(Value::Interval)

                .map(|v| Evaluated::Value(Cow::Owned(v)))

}

async fn evaluate_function<'a, 'b: 'a, 'c: 'a, T: GStore>(

    storage: Option<&'a T>,

    context: Option<Arc<RowContext<'b>>>,

    aggregated: Option<Arc<HashMap<&'c Aggregate, Value>>>,

    func: &'b Function,

) -> Result<Evaluated<'a>> {

    let eval = |expr| {

        let context = context.as_ref().map(Arc::clone);

        let aggregated = aggregated.as_ref().map(Arc::clone);

        evaluate_inner(storage, context, aggregated, expr)

    };

    let name = func.to_string();

    let result = match func {

        Function::Concat(exprs) => {

            let exprs = stream::iter(exprs).then(eval).try_collect().await?;

            f::concat(exprs)

        Function::Custom { name, exprs } => {

                func_name,

                args,

                body,

            } = storage

                .ok_or(EvaluateError::UnsupportedCustomFunction)?

                .fetch_function(name)

                .await?

                .ok_or_else(|| EvaluateError::UnsupportedFunction(name.to_string()))?;

            let min = args.iter().filter(|arg| arg.default.is_none()).count();

            let max = args.len();

            if !(min..=max).contains(&exprs.len()) {

                return Err((EvaluateError::FunctionArgsLengthNotWithinRange {

                    name: func_name.to_owned(),

                    expected_minimum: min,

                    expected_maximum: max,

                    found: exprs.len(),

                })

                .into());

            }

            let exprs = exprs.iter().chain(

                args.iter()

                    .skip(exprs.len())

                    .filter_map(|arg| arg.default.as_ref()),

            let context = stream::iter(args.iter().zip(exprs))

                .then(|(arg, expr)| async {

                    eval(expr)

                        .await?

                        .try_into_value(&arg.data_type, true)

                        .map(|value| (arg.name.clone(), value))

                })

                .try_collect::<Vec<(String, Value)>>()

                .map(|pairs| {

                    let (columns, values): (Vec<_>, Vec<_>) = pairs.into_iter().unzip();

                    let row = Cow::Owned(Row {

                        columns: columns.into(),

                        values,

                    });

                    let context = RowContext::new(name, row, None);

                    Some(Arc::new(context))

                })?;

            return evaluate_inner(storage, context, None, body).await;

        Function::ConcatWs { separator, exprs } => {

            let separator = eval(separator).await?;

            let exprs = stream::iter(exprs).then(eval).try_collect().await?;

            f::concat_ws(&name, separator, exprs)

        Function::IfNull { expr, then } => f::ifnull(eval(expr).await?, eval(then).await?),

        Function::NullIf { expr1, expr2 } => f::nullif(eval(expr1).await?, &eval(expr2).await?),

        Function::Lower(expr) => f::lower(&name, eval(expr).await?),

        Function::Initcap(expr) => f::initcap(&name, eval(expr).await?),

        Function::Upper(expr) => f::upper(&name, eval(expr).await?),

        Function::Left { expr, size } | Function::Right { expr, size } => {

            let expr = eval(expr).await?;

            let size = eval(size).await?;

            f::left_or_right(&name, expr, size)

        Function::Replace { expr, old, new } => {

            let expr = eval(expr).await?;

            let old = eval(old).await?;

            let new = eval(new).await?;

            f::replace(&name, expr, old, new)

        Function::Lpad { expr, size, fill } | Function::Rpad { expr, size, fill } => {

            let expr = eval(expr).await?;

            let size = eval(size).await?;

            let fill = match fill {

                Some(v) => Some(eval(v).await?),

                None => None,

            f::lpad_or_rpad(&name, expr, size, fill)

        Function::LastDay(expr) => {

            let expr = eval(expr).await?;

            f::last_day(&name, expr)

            expr,

            filter_chars,

            trim_where_field,

            let expr = eval(expr).await?;

            let filter_chars = match filter_chars {

                Some(v) => Some(eval(v).await?),

                None => None,

            return expr.trim(name, filter_chars, trim_where_field.as_ref());

        Function::Ltrim { expr, chars } => {

            let expr = eval(expr).await?;

            let chars = match chars {

                Some(v) => Some(eval(v).await?),

                None => None,

            return expr.ltrim(name, chars);

        Function::Rtrim { expr, chars } => {

            let expr = eval(expr).await?;

            let chars = match chars {

                Some(v) => Some(eval(v).await?),

                None => None,

            return expr.rtrim(name, chars);

        Function::Reverse(expr) => {

            let expr = eval(expr).await?;

            f::reverse(&name, expr)

        Function::Repeat { expr, num } => {

            let expr = eval(expr).await?;

            let num = eval(num).await?;

            f::repeat(&name, expr, num)

        Function::Substr { expr, start, count } => {

            let expr = eval(expr).await?;

            let start = eval(start).await?;

            let count = match count {

                Some(v) => Some(eval(v).await?),

                None => None,

            return expr.substr(name, start, count);

        Function::Ascii(expr) => f::ascii(&name, eval(expr).await?),

        Function::Chr(expr) => f::chr(&name, eval(expr).await?),

        Function::Md5(expr) => f::md5(&name, eval(expr).await?),

        Function::Hex(expr) => f::hex(&name, eval(expr).await?),

        Function::Abs(expr) => f::abs(&name, eval(expr).await?),

        Function::Sign(expr) => f::sign(&name, eval(expr).await?),

        Function::Sqrt(expr) => f::sqrt(eval(expr).await?),

        Function::Power { expr, power } => {

            let expr = eval(expr).await?;

            let power = eval(power).await?;

            f::power(&name, expr, power)

        Function::Ceil(expr) => f::ceil(&name, eval(expr).await?),

        Function::Rand(expr) => {

            let expr = match expr {

                Some(v) => Some(eval(v).await?),

                None => None,

            f::rand(&name, expr)

        Function::Round(expr) => f::round(&name, eval(expr).await?),

        Function::Trunc(expr) => f::trunc(&name, eval(expr).await?),

        Function::Floor(expr) => f::floor(&name, eval(expr).await?),

        Function::Radians(expr) => f::radians(&name, eval(expr).await?),

        Function::Degrees(expr) => f::degrees(&name, eval(expr).await?),

            return Ok(Evaluated::Value(Cow::Owned(Value::F64(

                std::f64::consts::PI,

            ))));

        Function::Exp(expr) => f::exp(&name, eval(expr).await?),

        Function::Log { antilog, base } => {

            let antilog = eval(antilog).await?;

            let base = eval(base).await?;

            f::log(&name, antilog, base)

        Function::Ln(expr) => f::ln(&name, eval(expr).await?),

        Function::Log2(expr) => f::log2(&name, eval(expr).await?),

        Function::Log10(expr) => f::log10(&name, eval(expr).await?),

        Function::Sin(expr) => f::sin(&name, eval(expr).await?),

        Function::Cos(expr) => f::cos(&name, eval(expr).await?),

        Function::Tan(expr) => f::tan(&name, eval(expr).await?),

        Function::Asin(expr) => f::asin(&name, eval(expr).await?),

        Function::Acos(expr) => f::acos(&name, eval(expr).await?),

        Function::Atan(expr) => f::atan(&name, eval(expr).await?),

        Function::Div { dividend, divisor } => {

            let dividend = eval(dividend).await?;

            let divisor = eval(divisor).await?;

            f::div(&name, dividend, divisor)

        Function::Mod { dividend, divisor } => {

            let dividend = eval(dividend).await?;

            let divisor = eval(divisor).await?;

            return dividend.modulo(&divisor);

        Function::Gcd { left, right } => {

            let left = eval(left).await?;

            let right = eval(right).await?;

            f::gcd(&name, left, right)

        Function::Lcm { left, right } => {

            let left = eval(left).await?;

            let right = eval(right).await?;

            f::lcm(&name, left, right)

        Function::Point { x, y } => {

            let x = eval(x).await?;

            let y = eval(y).await?;

            f::point(&name, x, y)

        Function::GetX(expr) => f::get_x(&name, eval(expr).await?),

        Function::GetY(expr) => f::get_y(&name, eval(expr).await?),

            geometry1,

            geometry2,

            let geometry1 = eval(geometry1).await?;

            let geometry2 = eval(geometry2).await?;

            f::calc_distance(&name, geometry1, geometry2)

        Function::Unwrap { expr, selector } => {

            let expr = eval(expr).await?;

            let selector = eval(selector).await?;

            f::unwrap(&name, expr, selector)

        Function::GenerateUuid() => return Ok(f::generate_uuid()),

        Function::Greatest(exprs) => {

            let exprs = stream::iter(exprs).then(eval).try_collect().await?;

            return f::greatest(&name, exprs);

            return Ok(Evaluated::Value(Cow::Owned(Value::Timestamp(

                Utc::now().naive_utc(),

            ))));

            return Ok(Evaluated::Value(Cow::Owned(Value::Date(

                Utc::now().date_naive(),

            ))));

            return Ok(Evaluated::Value(Cow::Owned(Value::Time(Utc::now().time()))));

        Function::Format { expr, format } => {

            let expr = eval(expr).await?;

            let format = eval(format).await?;

            f::format(&name, expr, format)

        Function::ToDate { expr, format } => {

            let expr = eval(expr).await?;

            let format = eval(format).await?;

            f::to_date(&name, expr, format)

        Function::ToTimestamp { expr, format } => {

            let expr = eval(expr).await?;

            let format = eval(format).await?;

            f::to_timestamp(&name, expr, format)

        Function::ToTime { expr, format } => {

            let expr = eval(expr).await?;

            let format = eval(format).await?;

            f::to_time(&name, expr, format)

            from_expr,

            sub_expr,

            let from_expr = eval(from_expr).await?;

            let sub_expr = eval(sub_expr).await?;

            f::position(from_expr, sub_expr)

            from_expr,

            sub_expr,

            start,

            let from_expr = eval(from_expr).await?;

            let sub_expr = eval(sub_expr).await?;

            let start = match start {

                Some(idx) => Some(eval(idx).await?),

                None => None,

            f::find_idx(&name, from_expr, sub_expr, start)

        Function::Cast { expr, data_type } => return eval(expr).await?.cast(data_type),

        Function::Extract { field, expr } => {

            let expr = eval(expr).await?;

            f::extract(*field, expr)

        Function::Coalesce(exprs) => {

            let exprs = stream::iter(exprs).then(eval).try_collect().await?;

            return f::coalesce(exprs);

        Function::Append { expr, value } => {

            let expr = eval(expr).await?;

            let value = eval(value).await?;

            f::append(expr, value)

        Function::Prepend { expr, value } => {

            let expr = eval(expr).await?;

            let value = eval(value).await?;

            f::prepend(expr, value)

        Function::Skip { expr, size } => {

            let expr = eval(expr).await?;

            let size = eval(size).await?;

            f::skip(&name, expr, size)

        Function::Sort { expr, order } => {

            let expr = eval(expr).await?;

            let order = match order {

                Some(o) => eval(o).await?,

                None => Evaluated::Value(Cow::Owned(Value::Str("ASC".to_owned()))),

            f::sort(expr, order)

        Function::Take { expr, size } => {

            let expr = eval(expr).await?;

            let size = eval(size).await?;

            f::take(&name, expr, size)

            expr,

            start,

            length,

            let expr = eval(expr).await?;

            let start = eval(start).await?;

            let length = eval(length).await?;

            f::slice(&name, expr, start, length)

        Function::IsEmpty(expr) => {

            let expr = eval(expr).await?;

            f::is_empty(expr)

        Function::AddMonth { expr, size } => {

            let expr = eval(expr).await?;

            let size = eval(size).await?;

            f::add_month(&name, expr, size)

        Function::Length(expr) => f::length(&name, eval(expr).await?),

        Function::Entries(expr) => f::entries(&name, eval(expr).await?),

        Function::Keys(expr) => f::keys(eval(expr).await?),

        Function::Values(expr) => {

            let expr = eval(expr).await?;

            f::values(expr)

            list_data,

            begin_index,

            end_index,

            values,

            let list_data = eval(list_data).await?;

            let begin_index = eval(begin_index).await?;

            let end_index = eval(end_index).await?;

            let values = match values {

                Some(v) => Some(eval(v).await?),

                None => None,

            f::splice(&name, list_data, begin_index, end_index, values)

        Function::Dedup(list) => f::dedup(eval(list).await?),

    match result {

        ControlFlow::Continue(v) => Ok(v),

        ControlFlow::Break(BreakCase::Null) => Ok(Evaluated::Value(Cow::Owned(Value::Null))),

        ControlFlow::Break(BreakCase::Err(err)) => Err(err),

}