(feat) Optimizer & Memo Full Logical -> Logical (#101)

Implements the in-memory memo for logical -> logical, as well as the
optimizer task graph, and the support of `*` stored expressions in the
compiler.

All features are tested accordingly, except the optimizer itself which
will be postponed in a future PR.

---------

Co-authored-by: Connor Tsui <[email protected]>

Author: AlSchlo

README.md

@@ -2,4 +2,4 @@
 
 # optd
 
-Query Optimizer Service
+Query Optimizer Service

optd-cli/examples/higher_order.opt

@@ -1,7 +1,7 @@
 data Logical =
     | Operation = 
         | Binary = 
-            | Arithmetic = 
+            | Arith = 
                 | Add(left: Logical, right: Logical)
                 | Mul(left: Logical, right: Logical)
                 | Div(left: Logical, right: Logical)

optd-cli/examples/logical_rules.opt

@@ -138,9 +138,10 @@ fn build_calculator_expr() =
     const3 = Const(3),
     const4 = Const(4),
     addition = Add(const2, const3),
-    multiplication = Mult(addition, const4)
+    multiplication = Mult(addition, const4),
+    pow = Pow(multiplication, Const(10)),
   in
-    multiplication.evaluate()
+    pow.evaluate()
 
 [run]
 fn run_mult_commute() =

optd-cli/examples/tutorial.opt

@@ -227,7 +227,7 @@ fn (catalog: Catalog) get_table_schema(name: String): Schema
 
 fn (logical: Logical*) properties(): LogicalProperties
 
-fn (costed: Physical$) statistics(): CostedProperties
+fn (costed: Physical$) statistics(): Statistics
 
 // -------------------------
 // 9. Transformation Rules
@@ -239,11 +239,11 @@ fn (costed: Physical$) statistics(): CostedProperties
 // Helper function for scalar rewrites.
 fn (expr: Scalar) remap(bindings: {I64 : I64}): Scalar = 
     match expr
-        | ColumnRef(idx) -> 
-            if bindings(idx) != none then // TODO: Fix none bug... Should not be a type! Rather Option<Nothing>.
-                ColumnRef(0) // TODO: Add ! once we have the `!` syntax.
-            else
-                ColumnRef(idx)
+        | ColumnRef(idx) -> ColumnRef(0)
+            // if bindings(idx) != none then // TODO: Fix EQ inference: just require EqHash & adapt engine.
+            //    ColumnRef(0) // TODO: Add ! once we have the `!` syntax.
+            // else
+               // ColumnRef(idx)
         | IntLiteral(value) -> IntLiteral(value)
         | StringLiteral(value) -> StringLiteral(value)
         | BoolLiteral(value) -> BoolLiteral(value)
@@ -277,12 +277,12 @@ fn (expr: Logical*) join_commute(): Logical? = match expr
             right_indices = 0..right_len,
             left_indices = 0..left_len,
 
-            remapping = (left_indices.map(i -> (i, i + right_len)) ++ 
-                right_indices.map(i -> (i + left_len, i))).to_map(),
+            remapping = (left_indices.map((i: I64) -> (i, i + right_len)) ++ 
+                right_indices.map((i: I64) -> (i + left_len, i))).to_map(),
         in
             Project(
                 Join(right, left, Inner, predicate.remap(remapping)),
-                (right_indices ++ left_indices).map(i -> ColumnRef(i))
+                (right_indices ++ left_indices).map((i: I64) -> ColumnRef(i))
             )
     \ _ -> none
 
@@ -306,8 +306,8 @@ fn (plan: Logical*) join_associativity(): Logical? = match plan
             b_indices = 0..b_len,
             c_indices = 0..c_len,
 
-            pred_bc_remapping = (b_indices.map(i -> (a_len + i, i)) ++
-                c_indices.map(i -> (a_len + b_len + i, b_len + i))).to_map(),
+            pred_bc_remapping = (b_indices.map((i: I64) -> (a_len + i, i)) ++
+                c_indices.map((i: I64) -> (a_len + b_len + i, b_len + i))).to_map(),
 
             remapped_pred_bc = pred_bc.remap(pred_bc_remapping),
 
@@ -316,8 +316,8 @@ fn (plan: Logical*) join_associativity(): Logical? = match plan
             b_indices_after = 0..b_len,
             c_indices_after = 0..c_len,
 
-            pred_ab_remapping = (b_indices_after.map(i -> (a_len + i, a_len + i)) ++
-                c_indices_after.map(i -> (a_len + b_len + i, a_len + b_len + i))).to_map(),
+            pred_ab_remapping = (b_indices_after.map((i: I64) -> (a_len + i, a_len + i)) ++
+                c_indices_after.map((i: I64) -> (a_len + b_len + i, a_len + b_len + i))).to_map(),
 
             remapped_pred_ab = pred_ab.remap(pred_ab_remapping)
         in
@@ -353,10 +353,11 @@ fn (expr: Logical*) impl_filter_enforce(props: PhysicalProperties?): Physical? =
             let 
                 result = PhysFilter(child.optimize(none), predicate)
             in
-                if props != none then
-                    PhysSort(result, props#order_by)
-                else
-                    result
+                result
+                // if props != none then
+                    // PhysSort(result, props#order_by) // TODO: same ! problem
+                // else
+                //    result
         \ _ -> none
 
 [implementation]
@@ -365,14 +366,14 @@ fn (expr: Logical*) impl_filter_passthrough(props: PhysicalProperties?): Physica
         | Filter(child, predicate) -> PhysFilter(child.optimize(props), predicate)
         \ _ -> none
 
-[implementation]
-fn (expr: Logical*) impl_sort(props: PhysicalProperties?): Physical? = match expr
-    | Sort(child, order_by) -> 
-        if props == none then // TODO: The incoming `?` syntax will make this cleaner
-            PhysSort(child.optimize(none), order_by)
-        else
-            none
-    \ _ -> none
+// [implementation]
+// fn (expr: Logical*) impl_sort(props: PhysicalProperties?): Physical? = match expr
+//     | Sort(child, order_by) -> 
+//         if props == none then // TODO: The incoming `?` syntax will make this cleaner. Also == problem for now.
+//             PhysSort(child.optimize(none), order_by)
+//         else
+//             none
+//     \ _ -> none
 
 // -------------------------
 // 11. Other Required Functions
@@ -381,14 +382,14 @@ fn (expr: Logical*) impl_sort(props: PhysicalProperties?): Physical? = match exp
 // Cost function for physical operators - used by the cost-based optimizer to evaluate 
 // different physical implementation alternatives.
 
-fn (expr: Physical*) cost(): F64 = 0
+fn (expr: Physical*) cost(): F64 = 0.0
 
 // Derive logical properties from a logical plan - propagates schema information and other
 // logical properties (like cardinality estimates, uniqueness, or functional dependencies)
 // through the logical plan.
 
-fn (log: Logical*) derive(): LogicalProperties = match log
-    | Get(table_name) -> catalog.get_table_schema(table_name)
+fn (log: Logical*) derive(): LogicalProperties? = match log
+    | Get(table_name) -> LogicalProperties(Catalog.get_table_schema(table_name))
     | Filter(child, _) -> child.properties()
     | Join(left, right, join_type, _) ->
         let 

optd-cli/src/main.rs

@@ -40,9 +40,10 @@ use optd::dsl::analyzer::hir::{CoreData, HIR, Udf, Value};
 use optd::dsl::compile::{Config, compile_hir};
 use optd::dsl::engine::{Continuation, Engine, EngineResponse};
 use optd::dsl::utils::errors::{CompileError, Diagnose};
+use optd::dsl::utils::retriever::{MockRetriever, Retriever};
 use std::collections::HashMap;
 use std::sync::Arc;
-use tokio::runtime::Runtime;
+use tokio::runtime::Builder;
 use tokio::task::JoinSet;
 
 #[derive(Parser)]
@@ -66,7 +67,11 @@ enum Commands {
 }
 
 /// A unimplemented user-defined function.
-pub fn unimplemented_udf(_args: &[Value], _catalog: &dyn Catalog) -> Value {
+pub fn unimplemented_udf(
+    _args: &[Value],
+    _catalog: &dyn Catalog,
+    _retriever: &dyn Retriever,
+) -> Value {
     println!("This user-defined function is unimplemented!");
     Value::new(CoreData::<Value>::None)
 }
@@ -122,7 +127,13 @@ fn run_all_functions(hir: &HIR) -> Result<(), Vec<CompileError>> {
     println!("Found {} functions to run", functions.len());
 
     // Create a multi-threaded runtime for parallel execution.
-    let runtime = Runtime::new().unwrap();
+    // TODO: We increase the stack size by x64 to avoid stack overflow
+    // given the lack of tail recursion in the engine (yet...)
+    let runtime = Builder::new_multi_thread()
+        .thread_stack_size(128 * 1024 * 1024)
+        .enable_all()
+        .build()
+        .unwrap();
     let function_results = runtime.block_on(run_functions_in_parallel(hir, functions));
 
     // Process and display function results.
@@ -139,10 +150,11 @@ fn run_all_functions(hir: &HIR) -> Result<(), Vec<CompileError>> {
 
 async fn run_functions_in_parallel(hir: &HIR, functions: Vec<String>) -> Vec<FunctionResult> {
     let catalog = Arc::new(memory_catalog());
+    let retriever = Arc::new(MockRetriever::new());
     let mut set = JoinSet::new();
 
     for function_name in functions {
-        let engine = Engine::new(hir.context.clone(), catalog.clone());
+        let engine = Engine::new(hir.context.clone(), catalog.clone(), retriever.clone());
         let name = function_name.clone();
 
         set.spawn(async move {
@@ -151,7 +163,7 @@ async fn run_functions_in_parallel(hir: &HIR, functions: Vec<String>) -> Vec<Fun
                 Arc::new(|value| Box::pin(async move { value }));
 
             // Launch the function with an empty vector of arguments.
-            let result = engine.launch_rule(&name, vec![], result_handler).await;
+            let result = engine.launch(&name, vec![], result_handler).await;
             FunctionResult { name, result }
         });
     }

optd/src/dsl/analyzer/from_ast/converter.rs

@@ -182,6 +182,7 @@ mod converter_tests {
     use crate::dsl::analyzer::hir::{CoreData, FunKind};
     use crate::dsl::analyzer::type_checks::registry::{Generic, TypeKind};
     use crate::dsl::parser::ast::{self, Adt, Function, Item, Module, Type as AstType};
+    use crate::dsl::utils::retriever::Retriever;
     use crate::dsl::utils::span::{Span, Spanned};
 
     // Helper functions to create test items
@@ -381,7 +382,11 @@ mod converter_tests {
         let ext_func = create_simple_function("external_function", false);
         let module = create_module_with_functions(vec![ext_func]);
 
-        pub fn external_function(_args: &[Value], _catalog: &dyn Catalog) -> Value {
+        pub fn external_function(
+            _args: &[Value],
+            _catalog: &dyn Catalog,
+            _retriever: &dyn Retriever,
+        ) -> Value {
             println!("Hello from UDF!");
             Value::new(CoreData::<Value>::None)
         }

optd/src/dsl/analyzer/hir/mod.rs

@@ -17,6 +17,7 @@
 
 use super::type_checks::registry::Type;
 use crate::catalog::Catalog;
+use crate::dsl::utils::retriever::Retriever;
 use crate::dsl::utils::span::Span;
 use context::Context;
 use map::Map;
@@ -76,12 +77,17 @@ pub struct Udf {
     /// The function pointer to the user-defined function.
     ///
     /// Note that [`Value`]s passed to and returned from this UDF do not have associated metadata.
-    pub func: fn(&[Value], &dyn Catalog) -> Value,
+    pub func: fn(&[Value], &dyn Catalog, &dyn Retriever) -> Value,
 }
 
 impl Udf {
-    pub fn call(&self, values: &[Value], catalog: &dyn Catalog) -> Value {
-        (self.func)(values, catalog)
+    pub fn call(
+        &self,
+        values: &[Value],
+        catalog: &dyn Catalog,
+        retriever: &dyn Retriever,
+    ) -> Value {
+        (self.func)(values, catalog, retriever)
     }
 }
 

optd/src/dsl/analyzer/into_hir/converter.rs

@@ -234,7 +234,12 @@ fn convert_field_access(
 ) -> ExprKind {
     use ExprKind::*;
 
-    let expr_type = registry.resolve_type(&expr.metadata.ty);
+    let mut expr_type = registry.resolve_type(&expr.metadata.ty);
+
+    // Unwrap stored type if necessary.
+    while let TypeKind::Stored(inner_type) = *expr_type.value {
+        expr_type = inner_type;
+    }
 
     match &*expr_type.value {
         TypeKind::Tuple(_) => {
@@ -262,7 +267,10 @@ fn convert_field_access(
             )
         }
 
-        _ => panic!("Field access on non-struct, non-tuple type: error in type inference"),
+        _ => panic!(
+            "Field access on non-struct, non-tuple type: {:?}",
+            expr_type
+        ),
     }
 }
 

optd/src/dsl/analyzer/type_checks/glb.rs

@@ -221,13 +221,13 @@ impl TypeRegistry {
         let result = glb_kind.into();
 
         // Verify post-condition in debug mode only
-        debug_assert!(
+        assert!(
             self.is_subtype(&result, type1),
             "GLB post-condition failed: {:?} is not a subtype of {:?}",
             result,
             type1
         );
-        debug_assert!(
+        assert!(
             self.is_subtype(&result, type2),
             "GLB post-condition failed: {:?} is not a subtype of {:?}",
             result,

optd/src/dsl/analyzer/type_checks/lub.rs

@@ -200,13 +200,13 @@ impl TypeRegistry {
         let result = lub_kind.into();
 
         // Verify post-condition in debug mode only
-        debug_assert!(
+        assert!(
             self.is_subtype(type1, &result),
             "LUB post-condition failed: {:?} is not a subtype of {:?}",
             type1,
             result
         );
-        debug_assert!(
+        assert!(
             self.is_subtype(type2, &result),
             "LUB post-condition failed: {:?} is not a subtype of {:?}",
             type2,