A pipelined materializing query executor

Fully evaluates each operator before evaluating downstream ones.

This is in contrast to a tuple at a time iterator,
where each row is fully processed by all operators,
before processing the next row.

It is called "materializing" because each operator materializes its output.
In practice however, non-blocking operators will be pipelined,
so as not to perform unnecessary materialization.

crates/execution/src/lib.rs

@@ -5,7 +5,7 @@ use iter::PlanIter;
 use spacetimedb_lib::{
     bsatn::{EncodeError, ToBsatn},
     query::Delta,
-    ser::Serialize,
+    sats::impl_serialize,
     AlgebraicValue, ProductValue,
 };
 use spacetimedb_physical_plan::plan::{ProjectField, ProjectPlan, TupleField};
@@ -17,6 +17,7 @@ use spacetimedb_table::{
 };
 
 pub mod iter;
+pub mod pipelined;
 
 /// The datastore interface required for building an executor
 pub trait Datastore {
@@ -72,12 +73,17 @@ pub trait DeltaStore {
     }
 }
 
-#[derive(Clone, Serialize)]
+#[derive(Clone)]
 pub enum Row<'a> {
     Ptr(RowRef<'a>),
     Ref(&'a ProductValue),
 }
 
+impl_serialize!(['a] Row<'a>, (self, ser) => match self {
+    Self::Ptr(row) => row.serialize(ser),
+    Self::Ref(row) => row.serialize(ser),
+});
+
 impl ToBsatn for Row<'_> {
     fn static_bsatn_size(&self) -> Option<u16> {
         match self {
@@ -104,8 +110,8 @@ impl ToBsatn for Row<'_> {
 impl ProjectField for Row<'_> {
     fn project(&self, field: &TupleField) -> AlgebraicValue {
         match self {
-            Self::Ptr(ptr) => ptr.read_col(field.field_pos).unwrap(),
-            Self::Ref(val) => val.elements.get(field.field_pos).unwrap().clone(),
+            Self::Ptr(ptr) => ptr.project(field),
+            Self::Ref(val) => val.project(field),
         }
     }
 }
@@ -153,6 +159,13 @@ impl<'a> Tuple<'a> {
             }
         }
     }
+
+    fn join(self, with: Self) -> Self {
+        match with {
+            Self::Row(ptr) => self.append(ptr),
+            Self::Join(ptrs) => ptrs.into_iter().fold(self, |tup, ptr| tup.append(ptr)),
+        }
+    }
 }
 
 pub struct DeltaScanIter<'a> {