engine: combine plan and exec mods (#173)

Author: carllerche

crates/toasty/src/engine.rs

@@ -1,6 +1,3 @@
-mod plan;
-use plan::Plan;
-
 mod eval;
 mod exec;
 mod index;
@@ -48,7 +45,7 @@ impl Engine {
 
         // The plan is called once (single entry record stream) with no arguments
         // (empty record).
-        self.exec_plan(&plan.pipeline, plan.vars).await
+        self.exec_plan(plan).await
     }
 
     /// Returns a new ExprContext

crates/toasty/src/engine/exec.rs

@@ -1,59 +1,82 @@
+mod action;
+pub(crate) use action::Action;
+
 mod delete_by_key;
+pub(crate) use delete_by_key::DeleteByKey;
+
 mod exec_statement;
+pub(crate) use exec_statement::{ExecStatement, ExecStatementOutput};
+
 mod filter;
+pub(crate) use filter::Filter;
+
 mod find_pk_by_index;
+pub(crate) use find_pk_by_index::FindPkByIndex;
+
 mod get_by_key;
+pub(crate) use get_by_key::GetByKey;
+
 mod nested_merge;
+pub(crate) use nested_merge::{MergeQualification, NestedChild, NestedLevel, NestedMerge};
+
+mod output;
+pub(crate) use output::Output;
+
+mod plan;
+pub(crate) use plan::ExecPlan;
+
 mod project;
+pub(crate) use project::Project;
+
 mod query_pk;
+pub(crate) use query_pk::QueryPk;
+
 mod rmw;
+pub(crate) use rmw::ReadModifyWrite;
+
 mod set_var;
+pub(crate) use set_var::SetVar;
+
 mod update_by_key;
+pub(crate) use update_by_key::UpdateByKey;
 
-mod var_store;
-pub(crate) use var_store::VarStore;
+mod var;
+pub(crate) use var::{VarId, VarStore};
 
-use crate::{
-    engine::{
-        plan::{self, Action, VarId},
-        Engine,
-    },
-    Result,
+use crate::{engine::Engine, Result};
+use toasty_core::{
+    driver::Rows,
+    stmt::{self, ValueStream},
 };
-use toasty_core::stmt::ValueStream;
-use toasty_core::{driver::Rows, stmt};
 
 struct Exec<'a> {
     engine: &'a Engine,
     vars: VarStore,
 }
 
 impl Engine {
-    pub(crate) async fn exec_plan(
-        &self,
-        pipeline: &plan::Pipeline,
-        vars: VarStore,
-    ) -> Result<ValueStream> {
-        Exec { engine: self, vars }.exec_pipeline(pipeline).await
-    }
-}
+    pub(crate) async fn exec_plan(&self, plan: ExecPlan) -> Result<ValueStream> {
+        let mut exec = Exec {
+            engine: self,
+            vars: plan.vars,
+        };
 
-impl Exec<'_> {
-    async fn exec_pipeline(&mut self, pipeline: &plan::Pipeline) -> Result<ValueStream> {
-        for step in &pipeline.actions {
-            self.exec_step(step).await?;
+        for step in &plan.actions {
+            exec.exec_step(step).await?;
         }
 
-        Ok(if let Some(returning) = pipeline.returning {
-            match self.vars.load(returning).await? {
+        Ok(if let Some(returning) = plan.returning {
+            match exec.vars.load(returning).await? {
                 Rows::Count(_) => ValueStream::default(),
                 Rows::Values(value_stream) => value_stream,
             }
         } else {
             ValueStream::default()
         })
     }
+}
 
+impl Exec<'_> {
     async fn exec_step(&mut self, action: &Action) -> Result<()> {
         match action {
             Action::DeleteByKey(action) => self.action_delete_by_key(action).await,
@@ -70,7 +93,7 @@ impl Exec<'_> {
         }
     }
 
-    async fn collect_input2(&mut self, input: &[VarId]) -> Result<Vec<stmt::Value>> {
+    async fn collect_input(&mut self, input: &[VarId]) -> Result<Vec<stmt::Value>> {
         let mut ret = Vec::new();
 
         for var_id in input {

crates/toasty/src/engine/plan/action.rs renamed to crates/toasty/src/engine/exec/action.rs

@@ -1,4 +1,4 @@
-use crate::engine::plan::{
+use crate::engine::exec::{
     DeleteByKey, ExecStatement, Filter, FindPkByIndex, GetByKey, NestedMerge, Project, QueryPk,
     ReadModifyWrite, SetVar, UpdateByKey,
 };

crates/toasty/src/engine/exec/delete_by_key.rs

@@ -1,9 +1,31 @@
-use toasty_core::driver::{operation, Rows};
+use toasty_core::{
+    driver::{operation, Rows},
+    schema::db::TableId,
+    stmt,
+};
 
-use super::{plan, Exec, Result};
+use crate::engine::exec::{Action, Output, VarId};
+
+use super::{Exec, Result};
+
+/// Input is the key to delete
+#[derive(Debug)]
+pub(crate) struct DeleteByKey {
+    /// How to access input from the variable table.
+    pub input: VarId,
+
+    /// Where to store the output (impacted row count)
+    pub output: Output,
+
+    /// Which model to get
+    pub table: TableId,
+
+    /// Only delete keys that match the filter
+    pub filter: Option<stmt::Expr>,
+}
 
 impl Exec<'_> {
-    pub(super) async fn action_delete_by_key(&mut self, action: &plan::DeleteByKey) -> Result<()> {
+    pub(super) async fn action_delete_by_key(&mut self, action: &DeleteByKey) -> Result<()> {
         let keys = self
             .vars
             .load(action.input)
@@ -37,3 +59,9 @@ impl Exec<'_> {
         Ok(())
     }
 }
+
+impl From<DeleteByKey> for Action {
+    fn from(src: DeleteByKey) -> Self {
+        Self::DeleteByKey(src)
+    }
+}

crates/toasty/src/engine/exec/exec_statement.rs

@@ -1,14 +1,38 @@
-use super::{plan, Exec, Result};
 use toasty_core::{
     driver::{operation, Rows},
     stmt,
 };
 
+use crate::{
+    engine::exec::{Action, Exec, Output, VarId},
+    Result,
+};
+
+#[derive(Debug)]
+pub(crate) struct ExecStatement {
+    /// Where to get arguments for this action.
+    pub input: Vec<VarId>,
+
+    /// How to handle output
+    pub output: ExecStatementOutput,
+
+    /// The query to execute. This may require input to generate the query.
+    pub stmt: stmt::Statement,
+
+    /// When true, the statement is a conditional update without any returning.
+    pub conditional_update_with_no_returning: bool,
+}
+
+#[derive(Debug)]
+pub(crate) struct ExecStatementOutput {
+    /// Databases always return rows as a vec of values. This specifies the type
+    /// of each value.
+    pub ty: Option<Vec<stmt::Type>>,
+    pub output: Output,
+}
+
 impl Exec<'_> {
-    pub(super) async fn action_exec_statement(
-        &mut self,
-        action: &plan::ExecStatement,
-    ) -> Result<()> {
+    pub(super) async fn action_exec_statement(&mut self, action: &ExecStatement) -> Result<()> {
         let mut stmt = action.stmt.clone();
 
         // Collect input values and substitute into the statement
@@ -110,3 +134,9 @@ impl Exec<'_> {
         Ok(())
     }
 }
+
+impl From<ExecStatement> for Action {
+    fn from(value: ExecStatement) -> Self {
+        Self::ExecStatement(value)
+    }
+}

crates/toasty/src/engine/exec/filter.rs

@@ -1,11 +1,26 @@
 use crate::{
-    engine::{exec::Exec, plan},
+    engine::{
+        eval,
+        exec::{Action, Exec, Output, VarId},
+    },
     Result,
 };
 use toasty_core::driver::Rows;
 
+#[derive(Debug)]
+pub(crate) struct Filter {
+    /// Source of the input
+    pub(crate) input: VarId,
+
+    /// Where to store the output
+    pub(crate) output: Output,
+
+    /// How to project it before storing
+    pub(crate) filter: eval::Func,
+}
+
 impl Exec<'_> {
-    pub(super) async fn action_filter(&mut self, action: &plan::Filter) -> Result<()> {
+    pub(super) async fn action_filter(&mut self, action: &Filter) -> Result<()> {
         // Load the input variable
         let mut input_stream = self.vars.load(action.input).await?.into_values();
 
@@ -30,3 +45,9 @@ impl Exec<'_> {
         Ok(())
     }
 }
+
+impl From<Filter> for Action {
+    fn from(value: Filter) -> Self {
+        Action::Filter(value)
+    }
+}

crates/toasty/src/engine/exec/find_pk_by_index.rs

@@ -1,22 +1,44 @@
 use toasty_core::{
     driver::{operation, Rows},
+    schema::db::{IndexId, TableId},
     stmt,
 };
 
-use super::{plan, Exec, Result};
-use crate::engine::simplify;
+use crate::{
+    engine::{
+        exec::{Action, Exec, Output, VarId},
+        simplify,
+    },
+    Result,
+};
+
+/// Schema: `self` references [index-fields, input-fields] flattened
+#[derive(Debug)]
+pub(crate) struct FindPkByIndex {
+    /// How to access input from the variable table.
+    pub input: Vec<VarId>,
+
+    /// Where to store the output
+    pub output: Output,
+
+    /// Table to query
+    pub table: TableId,
+
+    /// Index to use
+    pub index: IndexId,
+
+    /// Filter to apply to index
+    pub filter: stmt::Expr,
+}
 
 impl Exec<'_> {
-    pub(super) async fn action_find_pk_by_index(
-        &mut self,
-        action: &plan::FindPkByIndex,
-    ) -> Result<()> {
+    pub(super) async fn action_find_pk_by_index(&mut self, action: &FindPkByIndex) -> Result<()> {
         let mut filter = action.filter.clone();
 
         // Collect input values and substitute into the statement
         if !action.input.is_empty() {
             assert!(action.input.len() == 1);
-            let input = self.collect_input2(&action.input).await?;
+            let input = self.collect_input(&action.input).await?;
 
             filter.substitute(&input);
 
@@ -50,3 +72,9 @@ impl Exec<'_> {
         Ok(())
     }
 }
+
+impl From<FindPkByIndex> for Action {
+    fn from(src: FindPkByIndex) -> Self {
+        Self::FindPkByIndex(src)
+    }
+}

crates/toasty/src/engine/exec/get_by_key.rs

@@ -1,9 +1,32 @@
-use super::{plan, Exec, Result};
-use crate::driver::Rows;
-use toasty_core::{driver::operation, stmt::ValueStream};
+use crate::{
+    driver::Rows,
+    engine::exec::{Action, Exec, Output, VarId},
+    Result,
+};
+use toasty_core::{
+    driver::operation,
+    schema::db::{ColumnId, TableId},
+    stmt::ValueStream,
+};
+
+/// Get a model by key
+#[derive(Debug)]
+pub(crate) struct GetByKey {
+    /// Where to get the keys to load
+    pub input: VarId,
+
+    /// Where to store the result
+    pub output: Output,
+
+    /// Table to query
+    pub table: TableId,
+
+    /// Columns to get
+    pub columns: Vec<ColumnId>,
+}
 
 impl Exec<'_> {
-    pub(super) async fn action_get_by_key(&mut self, action: &plan::GetByKey) -> Result<()> {
+    pub(super) async fn action_get_by_key(&mut self, action: &GetByKey) -> Result<()> {
         let keys = self
             .vars
             .load(action.input)
@@ -34,3 +57,9 @@ impl Exec<'_> {
         Ok(())
     }
 }
+
+impl From<GetByKey> for Action {
+    fn from(src: GetByKey) -> Self {
+        Self::GetByKey(src)
+    }
+}