[steel thread] Add query plan and eval using DAG model (#202)

* [steel thread] query plan and eval using DAG model

Description of changes
Adds the required changes for enabling query evaluation using a DAG model.

What is our definition of DAG?
We are only interested in DAGs that can be used as execution plans, which leads to the following definition. A DAG is a directed, cycle-free graph G = (V, E) with a denoted root node v0 ∈ V such that all v ∈ V \{v0} are reachable from v0. Note that this is the definition of trees without the condition |E| = |V | − 1. Hence, all trees are DAGs. Reference: https://link.springer.com/article/10.1007/s00450-009-0061-0

In addition
- The steel thread includes only `Scan` and `Project` operators, and passes the test for planning and evaluation of the following query:
```SQL
SELECT a AS b FROM data
```
- The code is _not_ clean, maybe not idiomatic at some instances, and lacks documentation. I'll iterate through the code with a consequent PR when adding new operators.

Algorithm
We're loosely using a push model with DAG as called out in the following: https://link.springer.com/article/10.1007/s00450-009-0061-0 (Section 10.3.5. Pushing)

1. Create a DAG logical plan `LogicalPlan` with each node as a logical operator (for now manual until we integrate with AST -> PLAN).
1. Create an evaluation plan `EvalPlan` with the same structure as logical plan.
1. [Toposort](https://en.wikipedia.org/wiki/Topological_sorting) the `EvalPlan` to ensure all dependencies of a node get evaluated before the node itself.
1. For each operator in `EvalPlan`, eval the node.
1. Store the result in the node.
1. Push the output to each consumer—for now, for final result, we use a `Sink` node that is considered as the node that accumulates output; we need to see if there are better alternatives.

Finally, we're not performing any optimization at this stage (and we won't for extended amount of time).

Author: am357

partiql-eval/Cargo.toml

@@ -22,6 +22,7 @@ edition.workspace = true
 [dependencies]
 partiql-logical = { path = "../partiql-logical" }
 partiql-value = { path = "../partiql-value" }
+petgraph = "0.6.*"
 ordered-float = "3.*"
 itertools = "0.10.*"
 unicase = "2.*"

partiql-eval/benches/bench_eval.rs

@@ -1,3 +1,7 @@
+use std::cell::RefCell;
+use std::rc::Rc;
+use std::time::Duration;
+
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 
 use partiql_eval::env::basic::MapBindings;
@@ -8,9 +12,6 @@ use partiql_eval::eval::{
 use partiql_value::{
     partiql_bag, partiql_list, partiql_tuple, Bag, BindingsName, List, Tuple, Value,
 };
-use std::cell::RefCell;
-use std::rc::Rc;
-use std::time::Duration;
 
 fn data() -> MapBindings<Value> {
     let hr = partiql_tuple![(

partiql-eval/src/env.rs

@@ -1,5 +1,3 @@
-use std::fmt::Debug;
-
 use partiql_value::{BindingsName, Tuple, Value};
 use unicase::UniCase;
 

partiql-eval/src/eval.rs

@@ -1,12 +1,138 @@
-use crate::env::basic::MapBindings;
-use crate::env::Bindings;
-use partiql_value::Value::{Boolean, Missing, Null};
-use partiql_value::{Bag, BindingsName, Tuple, Value};
+use itertools::Itertools;
 use std::cell::RefCell;
 use std::collections::{HashMap, HashSet};
 use std::fmt::Debug;
 use std::rc::Rc;
 
+use petgraph::algo::toposort;
+use petgraph::prelude::StableGraph;
+use petgraph::{Directed, Incoming, Outgoing};
+
+use partiql_value::Value::{Boolean, Missing, Null};
+use partiql_value::{partiql_bag, Bag, BindingsName, Tuple, Value};
+
+use crate::env::basic::MapBindings;
+use crate::env::Bindings;
+
+#[derive(Debug)]
+pub struct EvalPlan(pub StableGraph<Box<dyn DagEvaluable>, (), Directed>);
+
+impl Default for EvalPlan {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl EvalPlan {
+    fn new() -> Self {
+        EvalPlan(StableGraph::<Box<dyn DagEvaluable>, (), Directed>::new())
+    }
+}
+
+pub type EvalResult = Result<Evaluated, EvalErr>;
+
+pub struct Evaluated {
+    pub result: Option<Value>,
+}
+
+pub struct EvalErr {
+    pub errors: Vec<EvaluationError>,
+}
+
+pub enum EvaluationError {
+    InvalidEvaluationPlan(String),
+}
+
+#[derive(Debug)]
+pub struct Scan {
+    pub expr: Box<dyn EvalExpr>,
+    pub as_key: String,
+    pub output: Option<Value>,
+}
+
+impl Scan {
+    pub fn new(expr: Box<dyn EvalExpr>, as_key: &str) -> Self {
+        Scan {
+            expr,
+            as_key: as_key.to_string(),
+            output: None,
+        }
+    }
+}
+
+impl DagEvaluable for Scan {
+    fn evaluate(&mut self, ctx: &dyn EvalContext) -> Option<Value> {
+        let mut value = partiql_bag!();
+        let v = self.expr.evaluate(&Tuple(HashMap::new()), ctx);
+        for t in v.into_iter() {
+            let out = Tuple(HashMap::from([(self.as_key.clone(), t)]));
+            value.push(Value::Tuple(Box::new(out)));
+        }
+        Some(Value::Bag(Box::new(value)))
+    }
+    fn update_input(&mut self, _input: &Value) {
+        todo!("update_input for Scan")
+    }
+}
+
+#[derive(Debug)]
+pub struct Project {
+    pub exprs: HashMap<String, Box<dyn EvalExpr>>,
+    pub input: Option<Value>,
+    pub output: Option<Value>,
+}
+
+impl Project {
+    pub fn new(exprs: HashMap<String, Box<dyn EvalExpr>>) -> Self {
+        Project {
+            exprs,
+            input: None,
+            output: None,
+        }
+    }
+}
+
+impl DagEvaluable for Project {
+    fn evaluate(&mut self, ctx: &dyn EvalContext) -> Option<Value> {
+        let input_value = self
+            .input
+            .as_ref()
+            .expect("Error in retrieving input value")
+            .clone();
+        let mut value = partiql_bag![];
+        for v in input_value.into_iter() {
+            let out = v.coerce_to_tuple();
+
+            let proj: HashMap<String, Value> = self
+                .exprs
+                .iter()
+                .map(|(alias, expr)| (alias.to_string(), expr.evaluate(&out, ctx)))
+                .collect();
+            value.push(Value::Tuple(Box::new(Tuple(proj))));
+        }
+
+        Some(Value::Bag(Box::new(value)))
+    }
+    fn update_input(&mut self, input: &Value) {
+        self.input = Some(input.clone());
+    }
+}
+
+#[derive(Debug)]
+pub struct Sink {
+    pub input: Option<Value>,
+    pub output: Option<Value>,
+}
+
+impl DagEvaluable for Sink {
+    fn evaluate(&mut self, _ctx: &dyn EvalContext) -> Option<Value> {
+        self.input.clone()
+    }
+    fn update_input(&mut self, input: &Value) {
+        self.input = Some(input.clone());
+    }
+}
+
 pub trait EvalContext {
     fn bindings(&self) -> &dyn Bindings<Value>;
 }
@@ -44,10 +170,76 @@ impl Evaluator {
     }
 }
 
+pub struct DagEvaluator {
+    ctx: Box<dyn EvalContext>,
+}
+
+impl DagEvaluator {
+    pub fn new(bindings: MapBindings<Value>) -> Self {
+        let ctx: Box<dyn EvalContext> = Box::new(BasicContext { bindings });
+        DagEvaluator { ctx }
+    }
+
+    pub fn execute_dag(&mut self, plan: EvalPlan) -> Result<Evaluated, EvalErr> {
+        let mut graph = plan.0;
+        // We are only interested in DAGs that can be used as execution plans, which leads to the
+        // following definition.
+        // A DAG is a directed, cycle-free graph G = (V, E) with a denoted root node v0 ∈ V such
+        // that all v ∈ V \{v0} are reachable from v0. Note that this is the definition of trees
+        // without the condition |E| = |V | − 1. Hence, all trees are DAGs.
+        // Reference: https://link.springer.com/article/10.1007/s00450-009-0061-0
+        match graph.externals(Incoming).exactly_one() {
+            Ok(_) => {
+                let sorted_ops = toposort(&graph, None);
+                match sorted_ops {
+                    Ok(ops) => {
+                        let mut result = None;
+                        for idx in ops.into_iter() {
+                            let src = graph
+                                .node_weight_mut(idx)
+                                .expect("Error in retrieving node");
+                            result = src.evaluate(&*self.ctx);
+
+                            let mut ne = graph.neighbors_directed(idx, Outgoing).detach();
+                            while let Some(n) = ne.next_node(&graph) {
+                                let dst =
+                                    graph.node_weight_mut(n).expect("Error in retrieving node");
+                                dst.update_input(
+                                    &result.clone().expect("Error in retrieving source value"),
+                                );
+                            }
+                        }
+                        let evaluated = Evaluated { result };
+                        Ok(evaluated)
+                    }
+                    Err(e) => Err(EvalErr {
+                        errors: vec![EvaluationError::InvalidEvaluationPlan(format!(
+                            "Malformed evaluation plan detected: {:?}",
+                            e
+                        ))],
+                    }),
+                }
+            }
+            Err(e) => Err(EvalErr {
+                errors: vec![EvaluationError::InvalidEvaluationPlan(format!(
+                    "Malformed evaluation plan detected: {:?}",
+                    e
+                ))],
+            }),
+        }
+    }
+}
+
 pub trait Evaluable: Debug {
     fn evaluate(&mut self, ctx: &dyn EvalContext);
 }
 
+// TODO rename to `Evaluable` when moved to DAG model completely
+pub trait DagEvaluable: Debug {
+    fn evaluate(&mut self, ctx: &dyn EvalContext) -> Option<Value>;
+    fn update_input(&mut self, input: &Value);
+}
+
 pub trait TupleSink: Debug {
     fn push_tuple(&mut self, bindings: Tuple, ctx: &dyn EvalContext);
 }
@@ -222,6 +414,13 @@ impl EvalFrom {
     }
 }
 
+impl Evaluable for EvalFrom {
+    fn evaluate(&mut self, ctx: &dyn EvalContext) {
+        let empty = Tuple(HashMap::new());
+        self.push_tuple(empty, ctx);
+    }
+}
+
 impl TupleSink for EvalFrom {
     #[inline]
     fn push_tuple(&mut self, bindings: Tuple, ctx: &dyn EvalContext) {
@@ -239,13 +438,6 @@ impl ValueSink for EvalFrom {
     }
 }
 
-impl Evaluable for EvalFrom {
-    fn evaluate(&mut self, ctx: &dyn EvalContext) {
-        let empty = Tuple(HashMap::new());
-        self.push_tuple(empty, ctx);
-    }
-}
-
 #[derive(Debug)]
 pub struct EvalFromAt {
     expr: Box<dyn EvalExpr>,