more type safety for delta plan compilation

clockworklabs · Jan 9, 2025 · cfb0a7b · cfb0a7b
1 parent 8d94b29
commit cfb0a7b
Show file tree

Hide file tree

Showing 10 changed files with 47 additions and 64 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/core/src/subscription/tx.rs b/crates/core/src/subscription/tx.rs
@@ -1,7 +1,7 @@
 use std::ops::Deref;
 
 use spacetimedb_execution::{Datastore, DeltaStore};
-use spacetimedb_lib::ProductValue;
+use spacetimedb_lib::{query::Delta, ProductValue};
 use spacetimedb_primitives::TableId;
 use spacetimedb_table::{blob_store::BlobStore, table::Table};
 
@@ -44,19 +44,19 @@ impl Datastore for DeltaTx<'_> {
 }
 
 impl DeltaStore for DeltaTx<'_> {
-    fn has_inserts(&self, table_id: TableId) -> Option<usize> {
+    fn has_inserts(&self, table_id: TableId) -> Option<Delta> {
         self.data.and_then(|data| {
             data.inserts()
                 .find(|(id, rows)| **id == table_id && !rows.is_empty())
-                .map(|(_, rows)| rows.len())
+                .map(|(_, rows)| Delta::Inserts(rows.len()))
         })
     }
 
-    fn has_deletes(&self, table_id: TableId) -> Option<usize> {
+    fn has_deletes(&self, table_id: TableId) -> Option<Delta> {
         self.data.and_then(|data| {
             data.deletes()
                 .find(|(id, rows)| **id == table_id && !rows.is_empty())
-                .map(|(_, rows)| rows.len())
+                .map(|(_, rows)| Delta::Deletes(rows.len()))
         })
     }
 

diff --git a/crates/execution/src/iter.rs b/crates/execution/src/iter.rs
@@ -1,8 +1,7 @@
 use std::collections::{HashMap, HashSet};
 
 use anyhow::{anyhow, bail, Result};
-use spacetimedb_expr::expr::Delta;
-use spacetimedb_lib::{AlgebraicValue, ProductValue};
+use spacetimedb_lib::{query::Delta, AlgebraicValue, ProductValue};
 use spacetimedb_physical_plan::plan::{
     HashJoin, IxJoin, IxScan, PhysicalExpr, PhysicalPlan, ProjectField, ProjectPlan, Sarg, Semi, TupleField,
 };

diff --git a/crates/execution/src/lib.rs b/crates/execution/src/lib.rs
@@ -4,6 +4,7 @@ use anyhow::{anyhow, Result};
 use iter::PlanIter;
 use spacetimedb_lib::{
     bsatn::{EncodeError, ToBsatn},
+    query::Delta,
     ser::Serialize,
     AlgebraicValue, ProductValue,
 };
@@ -50,8 +51,8 @@ pub trait Datastore {
 }
 
 pub trait DeltaStore {
-    fn has_inserts(&self, table_id: TableId) -> Option<usize>;
-    fn has_deletes(&self, table_id: TableId) -> Option<usize>;
+    fn has_inserts(&self, table_id: TableId) -> Option<Delta>;
+    fn has_deletes(&self, table_id: TableId) -> Option<Delta>;
 
     fn inserts_for_table(&self, table_id: TableId) -> Option<std::slice::Iter<'_, ProductValue>>;
     fn deletes_for_table(&self, table_id: TableId) -> Option<std::slice::Iter<'_, ProductValue>>;

diff --git a/crates/expr/src/expr.rs b/crates/expr/src/expr.rs
@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use spacetimedb_lib::{AlgebraicType, AlgebraicValue};
+use spacetimedb_lib::{query::Delta, AlgebraicType, AlgebraicValue};
 use spacetimedb_primitives::TableId;
 use spacetimedb_schema::schema::TableSchema;
 use spacetimedb_sql_parser::ast::{BinOp, LogOp};
@@ -87,12 +87,6 @@ pub struct Relvar {
     pub delta: Option<Delta>,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum Delta {
-    Inserts(usize),
-    Deletes(usize),
-}
-
 impl RelExpr {
     /// The number of fields this expression returns
     pub fn nfields(&self) -> usize {

diff --git a/crates/lib/src/lib.rs b/crates/lib/src/lib.rs
@@ -11,6 +11,7 @@ pub mod db;
 pub mod error;
 pub mod identity;
 pub mod operator;
+pub mod query;
 pub mod relation;
 pub mod scheduler;
 pub mod version;

diff --git a/crates/lib/src/query.rs b/crates/lib/src/query.rs
@@ -0,0 +1,6 @@
+/// A type used by the query planner for incremental evaluation
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Delta {
+    Inserts(usize),
+    Deletes(usize),
+}
diff --git a/crates/physical-plan/src/plan.rs b/crates/physical-plan/src/plan.rs
@@ -5,8 +5,8 @@ use std::{
 };
 
 use derive_more::From;
-use spacetimedb_expr::{expr::Delta, StatementSource};
-use spacetimedb_lib::AlgebraicValue;
+use spacetimedb_expr::StatementSource;
+use spacetimedb_lib::{query::Delta, AlgebraicValue};
 use spacetimedb_primitives::{ColId, ColSet, IndexId};
 use spacetimedb_schema::schema::{IndexSchema, TableSchema};
 use spacetimedb_sql_parser::ast::{BinOp, LogOp};

diff --git a/crates/query/Cargo.toml b/crates/query/Cargo.toml
@@ -13,6 +13,7 @@ rayon.workspace = true
 spacetimedb-client-api-messages.workspace = true
 spacetimedb-execution.workspace = true
 spacetimedb-expr.workspace = true
+spacetimedb-lib.workspace = true
 spacetimedb-primitives.workspace = true
 spacetimedb-physical-plan.workspace = true
 spacetimedb-sql-parser.workspace = true

diff --git a/crates/query/src/delta.rs b/crates/query/src/delta.rs
@@ -3,10 +3,8 @@ use std::ops::Deref;
 use anyhow::{bail, Result};
 use itertools::Either;
 use spacetimedb_execution::{iter::ProjectIter, Datastore, DeltaStore, Row};
-use spacetimedb_expr::{
-    check::{type_subscription, SchemaView},
-    expr::Delta,
-};
+use spacetimedb_expr::check::{type_subscription, SchemaView};
+use spacetimedb_lib::query::Delta;
 use spacetimedb_physical_plan::{
     compile::compile_project_plan,
     plan::{HashJoin, IxJoin, Label, PhysicalPlan, ProjectPlan},
@@ -205,12 +203,10 @@ impl SelectPlan {
         DeltaPlanEvaluator {
             insert_plans: tx
                 .has_inserts(self.table_id)
-                .map(Delta::Inserts)
                 .map(|delta| delta_plan(&self.plan, self.table_id, delta))
                 .unwrap_or_default(),
             delete_plans: tx
                 .has_deletes(self.table_id)
-                .map(Delta::Deletes)
                 .map(|delta| delta_plan(&self.plan, self.table_id, delta))
                 .unwrap_or_default(),
         }
@@ -331,101 +327,85 @@ impl JoinPlan {
             });
         }
 
-        /// Instantiate an optimized delta plan
-        fn delta_plan_opt(plan: &ProjectPlan, label: Label, delta: Delta) -> ProjectPlan {
+        /// Instantiate and optimize a delta plan
+        fn delta_plan_opt1(plan: &ProjectPlan, label: Label, delta: Delta) -> ProjectPlan {
             let mut plan = plan.clone();
             delta_plan(&mut plan, label, delta);
             plan.optimize()
         }
 
+        /// Instantiate and optimize a delta plan
+        fn delta_plan_opt2(plan: &ProjectPlan, lhs: Label, n: Delta, rhs: Label, m: Delta) -> ProjectPlan {
+            let mut plan = plan.clone();
+            delta_plan(&mut plan, lhs, n);
+            delta_plan(&mut plan, rhs, m);
+            plan.optimize()
+        }
+
         // dr(+)S'
         let dr_ins = tx
             .has_inserts(self.lhs_table)
-            .map(Delta::Inserts)
-            .map(|delta| delta_plan_opt(&self.plan, self.lhs_label, delta))
+            .map(|delta| delta_plan_opt1(&self.plan, self.lhs_label, delta))
             .map(std::iter::once)
             .map(Either::Left)
             .unwrap_or_else(|| Either::Right(std::iter::empty()));
 
         // dr(-)S'
         let dr_del = tx
             .has_deletes(self.lhs_table)
-            .map(Delta::Deletes)
-            .map(|delta| delta_plan_opt(&self.plan, self.lhs_label, delta))
+            .map(|delta| delta_plan_opt1(&self.plan, self.lhs_label, delta))
             .map(std::iter::once)
             .map(Either::Left)
             .unwrap_or_else(|| Either::Right(std::iter::empty()));
 
         // R'ds(+)
         let ds_ins = tx
-            .has_inserts(self.rhs_table)
-            .map(Delta::Inserts)
-            .map(|delta| delta_plan_opt(&self.plan, self.rhs_label, delta))
+            .has_inserts(self.lhs_table)
+            .map(|delta| delta_plan_opt1(&self.plan, self.rhs_label, delta))
             .map(std::iter::once)
             .map(Either::Left)
             .unwrap_or_else(|| Either::Right(std::iter::empty()));
 
         // R'ds(-)
         let ds_del = tx
-            .has_deletes(self.rhs_table)
-            .map(Delta::Deletes)
-            .map(|delta| delta_plan_opt(&self.plan, self.rhs_label, delta))
+            .has_deletes(self.lhs_table)
+            .map(|delta| delta_plan_opt1(&self.plan, self.rhs_label, delta))
             .map(std::iter::once)
             .map(Either::Left)
             .unwrap_or_else(|| Either::Right(std::iter::empty()));
 
         // dr(+)ds(+)
         let dr_ins_ds_ins = tx
             .has_inserts(self.lhs_table)
-            .and_then(|n| tx.has_inserts(self.rhs_table).map(|m| (n, m)))
-            .map(|(n, m)| {
-                let mut plan = self.plan.clone();
-                delta_plan(&mut plan, self.lhs_label, Delta::Inserts(n));
-                delta_plan(&mut plan, self.rhs_label, Delta::Inserts(m));
-                plan.optimize()
-            })
+            .zip(tx.has_inserts(self.rhs_table))
+            .map(|(n, m)| delta_plan_opt2(&self.plan, self.lhs_label, n, self.rhs_label, m))
             .map(std::iter::once)
             .map(Either::Left)
             .unwrap_or_else(|| Either::Right(std::iter::empty()));
 
         // dr(+)ds(-)
         let dr_ins_ds_del = tx
             .has_inserts(self.lhs_table)
-            .and_then(|n| tx.has_deletes(self.rhs_table).map(|m| (n, m)))
-            .map(|(n, m)| {
-                let mut plan = self.plan.clone();
-                delta_plan(&mut plan, self.lhs_label, Delta::Inserts(n));
-                delta_plan(&mut plan, self.rhs_label, Delta::Deletes(m));
-                plan.optimize()
-            })
+            .zip(tx.has_deletes(self.rhs_table))
+            .map(|(n, m)| delta_plan_opt2(&self.plan, self.lhs_label, n, self.rhs_label, m))
             .map(std::iter::once)
             .map(Either::Left)
             .unwrap_or_else(|| Either::Right(std::iter::empty()));
 
         // dr(-)ds(+)
         let dr_del_ds_ins = tx
             .has_deletes(self.lhs_table)
-            .and_then(|n| tx.has_inserts(self.rhs_table).map(|m| (n, m)))
-            .map(|(n, m)| {
-                let mut plan = self.plan.clone();
-                delta_plan(&mut plan, self.lhs_label, Delta::Deletes(n));
-                delta_plan(&mut plan, self.rhs_label, Delta::Inserts(m));
-                plan.optimize()
-            })
+            .zip(tx.has_inserts(self.rhs_table))
+            .map(|(n, m)| delta_plan_opt2(&self.plan, self.lhs_label, n, self.rhs_label, m))
             .map(std::iter::once)
             .map(Either::Left)
             .unwrap_or_else(|| Either::Right(std::iter::empty()));
 
         // dr(-)ds(-)
         let dr_del_ds_del = tx
             .has_deletes(self.lhs_table)
-            .and_then(|n| tx.has_deletes(self.rhs_table).map(|m| (n, m)))
-            .map(|(n, m)| {
-                let mut plan = self.plan.clone();
-                delta_plan(&mut plan, self.lhs_label, Delta::Deletes(n));
-                delta_plan(&mut plan, self.rhs_label, Delta::Deletes(m));
-                plan.optimize()
-            })
+            .zip(tx.has_deletes(self.rhs_table))
+            .map(|(n, m)| delta_plan_opt2(&self.plan, self.lhs_label, n, self.rhs_label, m))
             .map(std::iter::once)
             .map(Either::Left)
             .unwrap_or_else(|| Either::Right(std::iter::empty()));