gluesql_core/executor/evaluate/
mod.rs

1mod error;
2mod evaluated;
3mod expr;
4mod function;
5
6use {
7    self::function::BreakCase,
8    super::{context::RowContext, select::select},
9    crate::{
10        ast::{Aggregate, Expr, Function},
11        data::{CustomFunction, Interval, Literal, Row, Value},
12        mock::MockStorage,
13        result::{Error, Result},
14        store::GStore,
15    },
16    async_recursion::async_recursion,
17    chrono::prelude::Utc,
18    futures::{
19        future::{ready, try_join_all},
20        stream::{self, StreamExt, TryStreamExt},
21    },
22    im_rc::HashMap,
23    std::{borrow::Cow, ops::ControlFlow, rc::Rc},
24};
25
26pub use {error::EvaluateError, evaluated::Evaluated};
27
28#[async_recursion(?Send)]
29pub async fn evaluate<'a, 'b: 'a, 'c: 'a, T>(
30    storage: &'a T,
31    context: Option<Rc<RowContext<'b>>>,
32    aggregated: Option<Rc<HashMap<&'c Aggregate, Value>>>,
33    expr: &'a Expr,
34) -> Result<Evaluated<'a>>
35where
36    T: GStore,
37{
38    evaluate_inner(Some(storage), context, aggregated, expr).await
39}
40
41pub async fn evaluate_stateless<'a, 'b: 'a>(
42    context: Option<RowContext<'b>>,
43    expr: &'a Expr,
44) -> Result<Evaluated<'a>> {
45    let context = context.map(Rc::new);
46    let storage: Option<&MockStorage> = None;
47
48    evaluate_inner(storage, context, None, expr).await
49}
50
51#[async_recursion(?Send)]
52async fn evaluate_inner<'a, 'b: 'a, 'c: 'a, T>(
53    storage: Option<&'a T>,
54    context: Option<Rc<RowContext<'b>>>,
55    aggregated: Option<Rc<HashMap<&'c Aggregate, Value>>>,
56    expr: &'a Expr,
57) -> Result<Evaluated<'a>>
58where
59    T: GStore,
60{
61    let eval = |expr| {
62        let context = context.as_ref().map(Rc::clone);
63        let aggregated = aggregated.as_ref().map(Rc::clone);
64
65        evaluate_inner(storage, context, aggregated, expr)
66    };
67
68    match expr {
69        Expr::Literal(ast_literal) => expr::literal(ast_literal),
70        Expr::TypedString { data_type, value } => {
71            expr::typed_string(data_type, Cow::Borrowed(value))
72        }
73        Expr::Identifier(ident) => {
74            let context = context
75                .ok_or_else(|| EvaluateError::ContextRequiredForIdentEvaluation(expr.clone()))?;
76
77            match context.get_value(ident) {
78                Some(value) => Ok(value.clone()),
79                None => Err(EvaluateError::IdentifierNotFound(ident.to_owned()).into()),
80            }
81            .map(Evaluated::Value)
82        }
83        Expr::Nested(expr) => eval(expr).await,
84        Expr::CompoundIdentifier { alias, ident } => {
85            let context = context
86                .ok_or_else(|| EvaluateError::ContextRequiredForIdentEvaluation(expr.clone()))?;
87
88            match context.get_alias_value(alias, ident) {
89                Some(value) => Ok(value.clone()),
90                None => Err(EvaluateError::CompoundIdentifierNotFound {
91                    table_alias: alias.to_owned(),
92                    column_name: ident.to_owned(),
93                }
94                .into()),
95            }
96            .map(Evaluated::Value)
97        }
98        Expr::Subquery(query) => {
99            let storage =
100                storage.ok_or_else(|| EvaluateError::UnsupportedStatelessExpr(expr.clone()))?;
101
102            let evaluations = select(storage, query, context.as_ref().map(Rc::clone))
103                .await?
104                .map(|row| {
105                    let value = match row? {
106                        Row::Vec { columns, values } => {
107                            if columns.len() > 1 {
108                                return Err(EvaluateError::MoreThanOneColumnReturned.into());
109                            }
110                            values
111                        }
112                        Row::Map(_) => {
113                            return Err(EvaluateError::SchemalessProjectionForSubQuery.into());
114                        }
115                    }
116                    .into_iter()
117                    .next();
118
119                    Ok::<_, Error>(value)
120                })
121                .take(2)
122                .try_collect::<Vec<_>>()
123                .await?;
124
125            if evaluations.len() > 1 {
126                return Err(EvaluateError::MoreThanOneRowReturned.into());
127            }
128
129            let value = evaluations
130                .into_iter()
131                .next()
132                .flatten()
133                .unwrap_or(Value::Null);
134
135            Ok(Evaluated::Value(value))
136        }
137        Expr::BinaryOp { op, left, right } => {
138            let left = eval(left).await?;
139            let right = eval(right).await?;
140
141            expr::binary_op(op, left, right)
142        }
143        Expr::UnaryOp { op, expr } => {
144            let v = eval(expr).await?;
145
146            expr::unary_op(op, v)
147        }
148        Expr::Aggregate(aggr) => match aggregated
149            .as_ref()
150            .and_then(|aggregated| aggregated.get(aggr.as_ref()))
151        {
152            Some(value) => Ok(Evaluated::Value(value.clone())),
153            None => Err(EvaluateError::UnreachableEmptyAggregateValue(*aggr.clone()).into()),
154        },
155        Expr::Function(func) => {
156            let context = context.as_ref().map(Rc::clone);
157            let aggregated = aggregated.as_ref().map(Rc::clone);
158
159            evaluate_function(storage, context, aggregated, func).await
160        }
161        Expr::InList {
162            expr,
163            list,
164            negated,
165        } => {
166            let negated = *negated;
167            let target = eval(expr).await?;
168
169            stream::iter(list)
170                .then(eval)
171                .try_filter(|evaluated| ready(evaluated.evaluate_eq(&target)))
172                .try_next()
173                .await
174                .map(|v| v.is_some() ^ negated)
175                .map(Value::Bool)
176                .map(Evaluated::Value)
177        }
178        Expr::InSubquery {
179            expr: target_expr,
180            subquery,
181            negated,
182        } => {
183            let storage =
184                storage.ok_or_else(|| EvaluateError::UnsupportedStatelessExpr(expr.clone()))?;
185            let target = eval(target_expr).await?;
186
187            select(storage, subquery, context)
188                .await?
189                .map(|row| {
190                    let value = match row? {
191                        Row::Vec { values, .. } => values,
192                        Row::Map(_) => {
193                            return Err(EvaluateError::SchemalessProjectionForInSubQuery.into());
194                        }
195                    }
196                    .into_iter()
197                    .next()
198                    .unwrap_or(Value::Null);
199
200                    Ok(Evaluated::Value(value))
201                })
202                .try_filter(|evaluated| ready(evaluated.evaluate_eq(&target)))
203                .try_next()
204                .await
205                .map(|v| v.is_some() ^ negated)
206                .map(Value::Bool)
207                .map(Evaluated::Value)
208        }
209        Expr::Between {
210            expr,
211            negated,
212            low,
213            high,
214        } => {
215            let target = eval(expr).await?;
216            let low = eval(low).await?;
217            let high = eval(high).await?;
218
219            expr::between(target, *negated, low, high)
220        }
221        Expr::Like {
222            expr,
223            negated,
224            pattern,
225        } => {
226            let target = eval(expr).await?;
227            let pattern = eval(pattern).await?;
228            let evaluated = target.like(pattern, true)?;
229
230            Ok(match negated {
231                true => Evaluated::Value(Value::Bool(
232                    evaluated.evaluate_eq(&Evaluated::Literal(Literal::Boolean(false))),
233                )),
234                false => evaluated,
235            })
236        }
237        Expr::ILike {
238            expr,
239            negated,
240            pattern,
241        } => {
242            let target = eval(expr).await?;
243            let pattern = eval(pattern).await?;
244            let evaluated = target.like(pattern, false)?;
245
246            Ok(match negated {
247                true => Evaluated::Value(Value::Bool(
248                    evaluated.evaluate_eq(&Evaluated::Literal(Literal::Boolean(false))),
249                )),
250                false => evaluated,
251            })
252        }
253        Expr::Exists { subquery, negated } => {
254            let storage =
255                storage.ok_or_else(|| EvaluateError::UnsupportedStatelessExpr(expr.clone()))?;
256
257            select(storage, subquery, context)
258                .await?
259                .try_next()
260                .await
261                .map(|v| v.is_some() ^ negated)
262                .map(Value::Bool)
263                .map(Evaluated::Value)
264        }
265        Expr::IsNull(expr) => {
266            let v = eval(expr).await?.is_null();
267
268            Ok(Evaluated::Value(Value::Bool(v)))
269        }
270        Expr::IsNotNull(expr) => {
271            let v = eval(expr).await?.is_null();
272
273            Ok(Evaluated::Value(Value::Bool(!v)))
274        }
275        Expr::Case {
276            operand,
277            when_then,
278            else_result,
279        } => {
280            let operand = match operand {
281                Some(op) => eval(op).await?,
282                None => Evaluated::Value(Value::Bool(true)),
283            };
284
285            for (when, then) in when_then.iter() {
286                let when = eval(when).await?;
287
288                if when.evaluate_eq(&operand) {
289                    return eval(then).await;
290                }
291            }
292
293            match else_result {
294                Some(er) => eval(er).await,
295                None => Ok(Evaluated::Value(Value::Null)),
296            }
297        }
298        Expr::ArrayIndex { obj, indexes } => {
299            let obj = eval(obj).await?;
300            let indexes = try_join_all(indexes.iter().map(eval)).await?;
301            expr::array_index(obj, indexes)
302        }
303        Expr::Array { elem } => try_join_all(elem.iter().map(eval))
304            .await?
305            .into_iter()
306            .map(Value::try_from)
307            .collect::<Result<Vec<_>>>()
308            .map(Value::List)
309            .map(Evaluated::Value),
310        Expr::Interval {
311            expr,
312            leading_field,
313            last_field,
314        } => {
315            let value = eval(expr)
316                .await
317                .and_then(Value::try_from)
318                .map(String::from)?;
319
320            Interval::try_from_str(&value, *leading_field, *last_field)
321                .map(Value::Interval)
322                .map(Evaluated::Value)
323        }
324    }
325}
326
327async fn evaluate_function<'a, 'b: 'a, 'c: 'a, T: GStore>(
328    storage: Option<&'a T>,
329    context: Option<Rc<RowContext<'b>>>,
330    aggregated: Option<Rc<HashMap<&'c Aggregate, Value>>>,
331    func: &'b Function,
332) -> Result<Evaluated<'a>> {
333    use function as f;
334
335    let eval = |expr| {
336        let context = context.as_ref().map(Rc::clone);
337        let aggregated = aggregated.as_ref().map(Rc::clone);
338
339        evaluate_inner(storage, context, aggregated, expr)
340    };
341
342    let name = func.to_string();
343
344    let result = match func {
345        // --- text ---
346        Function::Concat(exprs) => {
347            let exprs = stream::iter(exprs).then(eval).try_collect().await?;
348            f::concat(exprs)
349        }
350        Function::Custom { name, exprs } => {
351            let CustomFunction {
352                func_name,
353                args,
354                body,
355            } = storage
356                .ok_or(EvaluateError::UnsupportedCustomFunction)?
357                .fetch_function(name)
358                .await?
359                .ok_or_else(|| EvaluateError::UnsupportedFunction(name.to_string()))?;
360
361            let min = args.iter().filter(|arg| arg.default.is_none()).count();
362            let max = args.len();
363
364            if !(min..=max).contains(&exprs.len()) {
365                return Err((EvaluateError::FunctionArgsLengthNotWithinRange {
366                    name: func_name.to_owned(),
367                    expected_minimum: min,
368                    expected_maximum: max,
369                    found: exprs.len(),
370                })
371                .into());
372            }
373
374            let exprs = exprs.iter().chain(
375                args.iter()
376                    .skip(exprs.len())
377                    .filter_map(|arg| arg.default.as_ref()),
378            );
379
380            let context = stream::iter(args.iter().zip(exprs))
381                .then(|(arg, expr)| async {
382                    eval(expr)
383                        .await?
384                        .try_into_value(&arg.data_type, true)
385                        .map(|value| (arg.name.to_owned(), value))
386                })
387                .try_collect()
388                .await
389                .map(|values| {
390                    let row = Cow::Owned(Row::Map(values));
391                    let context = RowContext::new(name, row, None);
392                    Some(Rc::new(context))
393                })?;
394
395            return evaluate_inner(storage, context, None, body).await;
396        }
397        Function::ConcatWs { separator, exprs } => {
398            let separator = eval(separator).await?;
399            let exprs = stream::iter(exprs).then(eval).try_collect().await?;
400            f::concat_ws(name, separator, exprs)
401        }
402        Function::IfNull { expr, then } => f::ifnull(eval(expr).await?, eval(then).await?),
403        Function::Lower(expr) => f::lower(name, eval(expr).await?),
404        Function::Initcap(expr) => f::initcap(name, eval(expr).await?),
405        Function::Upper(expr) => f::upper(name, eval(expr).await?),
406        Function::Left { expr, size } | Function::Right { expr, size } => {
407            let expr = eval(expr).await?;
408            let size = eval(size).await?;
409
410            f::left_or_right(name, expr, size)
411        }
412        Function::Replace { expr, old, new } => {
413            let expr = eval(expr).await?;
414            let old = eval(old).await?;
415            let new = eval(new).await?;
416
417            f::replace(name, expr, old, new)
418        }
419        Function::Lpad { expr, size, fill } | Function::Rpad { expr, size, fill } => {
420            let expr = eval(expr).await?;
421            let size = eval(size).await?;
422            let fill = match fill {
423                Some(v) => Some(eval(v).await?),
424                None => None,
425            };
426
427            f::lpad_or_rpad(name, expr, size, fill)
428        }
429        Function::LastDay(expr) => {
430            let expr = eval(expr).await?;
431            f::last_day(name, expr)
432        }
433        Function::Trim {
434            expr,
435            filter_chars,
436            trim_where_field,
437        } => {
438            let expr = eval(expr).await?;
439            let filter_chars = match filter_chars {
440                Some(v) => Some(eval(v).await?),
441                None => None,
442            };
443
444            return expr.trim(name, filter_chars, trim_where_field);
445        }
446        Function::Ltrim { expr, chars } => {
447            let expr = eval(expr).await?;
448            let chars = match chars {
449                Some(v) => Some(eval(v).await?),
450                None => None,
451            };
452
453            return expr.ltrim(name, chars);
454        }
455        Function::Rtrim { expr, chars } => {
456            let expr = eval(expr).await?;
457            let chars = match chars {
458                Some(v) => Some(eval(v).await?),
459                None => None,
460            };
461
462            return expr.rtrim(name, chars);
463        }
464        Function::Reverse(expr) => {
465            let expr = eval(expr).await?;
466
467            f::reverse(name, expr)
468        }
469        Function::Repeat { expr, num } => {
470            let expr = eval(expr).await?;
471            let num = eval(num).await?;
472
473            f::repeat(name, expr, num)
474        }
475        Function::Substr { expr, start, count } => {
476            let expr = eval(expr).await?;
477            let start = eval(start).await?;
478            let count = match count {
479                Some(v) => Some(eval(v).await?),
480                None => None,
481            };
482
483            return expr.substr(name, start, count);
484        }
485        Function::Ascii(expr) => f::ascii(name, eval(expr).await?),
486        Function::Chr(expr) => f::chr(name, eval(expr).await?),
487        Function::Md5(expr) => f::md5(name, eval(expr).await?),
488
489        // --- float ---
490        Function::Abs(expr) => f::abs(name, eval(expr).await?),
491        Function::Sign(expr) => f::sign(name, eval(expr).await?),
492        Function::Sqrt(expr) => f::sqrt(eval(expr).await?),
493        Function::Power { expr, power } => {
494            let expr = eval(expr).await?;
495            let power = eval(power).await?;
496
497            f::power(name, expr, power)
498        }
499        Function::Ceil(expr) => f::ceil(name, eval(expr).await?),
500        Function::Rand(expr) => {
501            let expr = match expr {
502                Some(v) => Some(eval(v).await?),
503                None => None,
504            };
505
506            f::rand(name, expr)
507        }
508        Function::Round(expr) => f::round(name, eval(expr).await?),
509        Function::Floor(expr) => f::floor(name, eval(expr).await?),
510        Function::Radians(expr) => f::radians(name, eval(expr).await?),
511        Function::Degrees(expr) => f::degrees(name, eval(expr).await?),
512        Function::Pi() => return Ok(Evaluated::Value(Value::F64(std::f64::consts::PI))),
513        Function::Exp(expr) => f::exp(name, eval(expr).await?),
514        Function::Log { antilog, base } => {
515            let antilog = eval(antilog).await?;
516            let base = eval(base).await?;
517
518            f::log(name, antilog, base)
519        }
520        Function::Ln(expr) => f::ln(name, eval(expr).await?),
521        Function::Log2(expr) => f::log2(name, eval(expr).await?),
522        Function::Log10(expr) => f::log10(name, eval(expr).await?),
523        Function::Sin(expr) => f::sin(name, eval(expr).await?),
524        Function::Cos(expr) => f::cos(name, eval(expr).await?),
525        Function::Tan(expr) => f::tan(name, eval(expr).await?),
526        Function::Asin(expr) => f::asin(name, eval(expr).await?),
527        Function::Acos(expr) => f::acos(name, eval(expr).await?),
528        Function::Atan(expr) => f::atan(name, eval(expr).await?),
529
530        // --- integer ---
531        Function::Div { dividend, divisor } => {
532            let dividend = eval(dividend).await?;
533            let divisor = eval(divisor).await?;
534
535            f::div(name, dividend, divisor)
536        }
537        Function::Mod { dividend, divisor } => {
538            let dividend = eval(dividend).await?;
539            let divisor = eval(divisor).await?;
540
541            return dividend.modulo(&divisor);
542        }
543        Function::Gcd { left, right } => {
544            let left = eval(left).await?;
545            let right = eval(right).await?;
546
547            f::gcd(name, left, right)
548        }
549        Function::Lcm { left, right } => {
550            let left = eval(left).await?;
551            let right = eval(right).await?;
552
553            f::lcm(name, left, right)
554        }
555
556        // --- spatial ---
557        Function::Point { x, y } => {
558            let x = eval(x).await?;
559            let y = eval(y).await?;
560
561            f::point(name, x, y)
562        }
563        Function::GetX(expr) => f::get_x(name, eval(expr).await?),
564        Function::GetY(expr) => f::get_y(name, eval(expr).await?),
565        Function::CalcDistance {
566            geometry1,
567            geometry2,
568        } => {
569            let geometry1 = eval(geometry1).await?;
570            let geometry2 = eval(geometry2).await?;
571
572            f::calc_distance(name, geometry1, geometry2)
573        }
574
575        // --- etc ---
576        Function::Unwrap { expr, selector } => {
577            let expr = eval(expr).await?;
578            let selector = eval(selector).await?;
579
580            f::unwrap(name, expr, selector)
581        }
582        Function::GenerateUuid() => return Ok(f::generate_uuid()),
583        Function::Greatest(exprs) => {
584            let exprs = stream::iter(exprs).then(eval).try_collect().await?;
585            return f::greatest(name, exprs);
586        }
587        Function::Now() => return Ok(Evaluated::Value(Value::Timestamp(Utc::now().naive_utc()))),
588        Function::Format { expr, format } => {
589            let expr = eval(expr).await?;
590            let format = eval(format).await?;
591
592            f::format(name, expr, format)
593        }
594        Function::ToDate { expr, format } => {
595            let expr = eval(expr).await?;
596            let format = eval(format).await?;
597            f::to_date(name, expr, format)
598        }
599        Function::ToTimestamp { expr, format } => {
600            let expr = eval(expr).await?;
601            let format = eval(format).await?;
602            f::to_timestamp(name, expr, format)
603        }
604        Function::ToTime { expr, format } => {
605            let expr = eval(expr).await?;
606            let format = eval(format).await?;
607            f::to_time(name, expr, format)
608        }
609        Function::Position {
610            from_expr,
611            sub_expr,
612        } => {
613            let from_expr = eval(from_expr).await?;
614            let sub_expr = eval(sub_expr).await?;
615            f::position(from_expr, sub_expr)
616        }
617        Function::FindIdx {
618            from_expr,
619            sub_expr,
620            start,
621        } => {
622            let from_expr = eval(from_expr).await?;
623            let sub_expr = eval(sub_expr).await?;
624            let start = match start {
625                Some(idx) => Some(eval(idx).await?),
626                None => None,
627            };
628            f::find_idx(name, from_expr, sub_expr, start)
629        }
630        Function::Cast { expr, data_type } => return eval(expr).await?.cast(data_type),
631        Function::Extract { field, expr } => {
632            let expr = eval(expr).await?;
633            f::extract(field, expr)
634        }
635        Function::Coalesce(exprs) => {
636            let exprs = stream::iter(exprs).then(eval).try_collect().await?;
637            return f::coalesce(exprs);
638        }
639
640        // --- list ---
641        Function::Append { expr, value } => {
642            let expr = eval(expr).await?;
643            let value = eval(value).await?;
644            f::append(expr, value)
645        }
646        Function::Prepend { expr, value } => {
647            let expr = eval(expr).await?;
648            let value = eval(value).await?;
649            f::prepend(expr, value)
650        }
651        Function::Skip { expr, size } => {
652            let expr = eval(expr).await?;
653            let size = eval(size).await?;
654            f::skip(name, expr, size)
655        }
656        Function::Sort { expr, order } => {
657            let expr = eval(expr).await?;
658            let order = match order {
659                Some(o) => eval(o).await?,
660                None => Evaluated::Value(Value::Str("ASC".to_owned())),
661            };
662            f::sort(expr, order)
663        }
664        Function::Take { expr, size } => {
665            let expr = eval(expr).await?;
666            let size = eval(size).await?;
667            f::take(name, expr, size)
668        }
669        Function::Slice {
670            expr,
671            start,
672            length,
673        } => {
674            let expr = eval(expr).await?;
675            let start = eval(start).await?;
676            let length = eval(length).await?;
677            f::slice(name, expr, start, length)
678        }
679        Function::IsEmpty(expr) => {
680            let expr = eval(expr).await?;
681            f::is_empty(expr)
682        }
683        Function::AddMonth { expr, size } => {
684            let expr = eval(expr).await?;
685            let size = eval(size).await?;
686            f::add_month(name, expr, size)
687        }
688        Function::Length(expr) => f::length(name, eval(expr).await?),
689        Function::Entries(expr) => f::entries(name, eval(expr).await?),
690        Function::Keys(expr) => f::keys(eval(expr).await?),
691        Function::Values(expr) => {
692            let expr = eval(expr).await?;
693            f::values(expr)
694        }
695        Function::Splice {
696            list_data,
697            begin_index,
698            end_index,
699            values,
700        } => {
701            let list_data = eval(list_data).await?;
702            let begin_index = eval(begin_index).await?;
703            let end_index = eval(end_index).await?;
704            let values = match values {
705                Some(v) => Some(eval(v).await?),
706                None => None,
707            };
708            f::splice(name, list_data, begin_index, end_index, values)
709        }
710        Function::Dedup(list) => f::dedup(eval(list).await?),
711    };
712
713    match result {
714        ControlFlow::Continue(v) => Ok(v),
715        ControlFlow::Break(BreakCase::Null) => Ok(Evaluated::Value(Value::Null)),
716        ControlFlow::Break(BreakCase::Err(err)) => Err(err),
717    }
718}
gluesql_core/executor/evaluate/mod.rs

gluesql_core/executor/evaluate/
mod.rs
