Split ColumnOp into one with row indices and one with FieldName &…

… other enabled changes (#1207)

* 1. Split ColumnOp into ColumnOp & FieldOp, former storing ColId
2. Shrink SqlAst to 80 bytes, so it can be passed in registers
3. Store end-result Header in IndexSemiJoin
4. Remove operational use of Header in ColumnOp & build_query
5. Simplify RowRef::{get, project, project_owned}

* Make parts of build_query actually infallible.

1. Make IndexSemiJoin::filter infallible.
2. Make ColumnOp::compare and friends infallible.
3. Make RowRef::{get, project, project_owned} infallible.

* Make RelOps::next itself infallible

* 1. with_select{_cmp}: ensure type safety o query exec cannot panic
2. Document RelValue::{get, read_or_take_column, project_owned}
3. Refactor optimize_select
4. Ensure in optimize_select that conditions are merged with preceding selects

* remove RelOps::{head, row_count}; head is redundant & row_count is useless

* remove Relation trait; it does not carry its weight

* make build_query infallible

* simplify IndexSemiJoin, make it slightly less branchy

* simplify try_index_join

* split IndexSemiJoin into Left & Right parts

* move get_field_pos to test code

* move test version of build_query to test code

crates/bench/benches/subscription.rs

@@ -104,7 +104,7 @@ fn eval(c: &mut Criterion) {
             let query = compile_read_only_query(&raw.db, &tx, sql).unwrap();
             let query: ExecutionSet = query.into();
             let ctx = &ExecutionContext::subscribe(raw.db.address(), SlowQueryConfig::default());
-            b.iter(|| drop(black_box(query.eval(ctx, Protocol::Binary, &raw.db, &tx).unwrap())))
+            b.iter(|| drop(black_box(query.eval(ctx, Protocol::Binary, &raw.db, &tx))))
         });
     };
 
@@ -140,10 +140,7 @@ fn eval(c: &mut Criterion) {
         let query = ExecutionSet::from_iter(query_lhs.into_iter().chain(query_rhs));
         let tx = &tx.into();
 
-        b.iter(|| {
-            let out = query.eval_incr(ctx_incr, &raw.db, tx, &update).unwrap();
-            black_box(out);
-        })
+        b.iter(|| drop(black_box(query.eval_incr(ctx_incr, &raw.db, tx, &update))))
     });
 
     // To profile this benchmark for 30s
@@ -161,10 +158,7 @@ fn eval(c: &mut Criterion) {
         let query: ExecutionSet = query.into();
         let tx = &tx.into();
 
-        b.iter(|| {
-            let out = query.eval_incr(ctx_incr, &raw.db, tx, &update).unwrap();
-            black_box(out);
-        })
+        b.iter(|| drop(black_box(query.eval_incr(ctx_incr, &raw.db, tx, &update))));
     });
 
     // To profile this benchmark for 30s

crates/core/src/db/cursor.rs

@@ -6,12 +6,12 @@ use spacetimedb_sats::AlgebraicValue;
 
 /// Common wrapper for relational iterators that work like cursors.
 pub struct TableCursor<'a> {
-    pub table: DbTable,
+    pub table: &'a DbTable,
     pub iter: Iter<'a>,
 }
 
 impl<'a> TableCursor<'a> {
-    pub fn new(table: DbTable, iter: Iter<'a>) -> Result<Self, DBError> {
+    pub fn new(table: &'a DbTable, iter: Iter<'a>) -> Result<Self, DBError> {
         Ok(Self { table, iter })
     }
 }

crates/core/src/db/datastore/locking_tx_datastore/state_view.rs

@@ -286,10 +286,10 @@ impl<'a> Iterator for Iter<'a> {
                         //
                         // As a result, in MVCC, this branch will need to check if the `row_ref`
                         // also exists in the `tx_state.insert_tables` and ensure it is yielded only once.
-                        if !self
+                        if self
                             .tx_state
-                            .map(|tx_state| tx_state.is_deleted(table_id, row_ref.pointer()))
-                            .unwrap_or(false)
+                            .filter(|tx_state| tx_state.is_deleted(table_id, row_ref.pointer()))
+                            .is_none()
                         {
                             // There either are no state changes for the current tx (`None`),
                             // or there are, but `row_id` specifically has not been changed.

crates/core/src/db/mod.rs

@@ -1,4 +1,3 @@
-pub mod cursor;
 pub mod datastore;
 pub mod db_metrics;
 pub mod relational_db;

crates/core/src/error.rs

@@ -320,6 +320,8 @@ pub enum NodesError {
     SystemName(Box<str>),
     #[error("internal db error: {0}")]
     Internal(#[source] Box<DBError>),
+    #[error(transparent)]
+    BadQuery(#[from] RelationError),
     #[error("invalid index type: {0}")]
     BadIndexType(u8),
 }

crates/core/src/host/instance_env.rs

@@ -19,9 +19,9 @@ use spacetimedb_lib::operator::OpQuery;
 use spacetimedb_lib::ProductValue;
 use spacetimedb_primitives::{ColId, ColListBuilder, TableId};
 use spacetimedb_sats::db::def::{IndexDef, IndexType};
-use spacetimedb_sats::relation::{FieldExpr, FieldName};
+use spacetimedb_sats::relation::FieldName;
 use spacetimedb_sats::Typespace;
-use spacetimedb_vm::expr::{ColumnOp, NoInMemUsed, QueryExpr};
+use spacetimedb_vm::expr::{FieldExpr, FieldOp, NoInMemUsed, QueryExpr};
 
 #[derive(Clone)]
 pub struct InstanceEnv {
@@ -320,27 +320,27 @@ impl InstanceEnv {
     ) -> Result<Vec<Box<[u8]>>, NodesError> {
         use spacetimedb_lib::filter;
 
-        fn filter_to_column_op(table_id: TableId, filter: filter::Expr) -> ColumnOp {
+        fn filter_to_column_op(table_id: TableId, filter: filter::Expr) -> FieldOp {
             match filter {
                 filter::Expr::Cmp(filter::Cmp {
                     op,
                     args: CmpArgs { lhs_field, rhs },
-                }) => ColumnOp::Cmp {
+                }) => FieldOp::Cmp {
                     op: OpQuery::Cmp(op),
-                    lhs: Box::new(ColumnOp::Field(FieldExpr::Name(FieldName::new(
+                    lhs: Box::new(FieldOp::Field(FieldExpr::Name(FieldName::new(
                         table_id,
                         lhs_field.into(),
                     )))),
-                    rhs: Box::new(ColumnOp::Field(match rhs {
+                    rhs: Box::new(FieldOp::Field(match rhs {
                         filter::Rhs::Field(rhs_field) => FieldExpr::Name(FieldName::new(table_id, rhs_field.into())),
                         filter::Rhs::Value(rhs_value) => FieldExpr::Value(rhs_value),
                     })),
                 },
-                filter::Expr::Logic(filter::Logic { lhs, op, rhs }) => ColumnOp::Cmp {
-                    op: OpQuery::Logic(op),
-                    lhs: Box::new(filter_to_column_op(table_id, *lhs)),
-                    rhs: Box::new(filter_to_column_op(table_id, *rhs)),
-                },
+                filter::Expr::Logic(filter::Logic { lhs, op, rhs }) => FieldOp::new(
+                    OpQuery::Logic(op),
+                    filter_to_column_op(table_id, *lhs),
+                    filter_to_column_op(table_id, *rhs),
+                ),
                 filter::Expr::Unary(_) => todo!("unary operations are not yet supported"),
             }
         }
@@ -363,19 +363,19 @@ impl InstanceEnv {
 
         // TODO(Centril): consider caching from `filter: &[u8] -> query: QueryExpr`.
         let query = QueryExpr::new(schema.as_ref())
-            .with_select(filter_to_column_op(table_id, filter))
+            .with_select(filter_to_column_op(table_id, filter))?
             .optimize(&|table_id, table_name| stdb.row_count(table_id, table_name));
 
         // TODO(Centril): Conditionally dump the `query` to a file and compare against integration test.
         // Invent a system where we can make these kinds of "optimization path tests".
 
         let tx: TxMode = tx.into();
         // SQL queries can never reference `MemTable`s, so pass in an empty set.
-        let mut query = build_query(ctx, stdb, &tx, &query, &mut NoInMemUsed)?;
+        let mut query = build_query(ctx, stdb, &tx, &query, &mut NoInMemUsed);
 
         // write all rows and flush at row boundaries.
-        let query_iter = std::iter::from_fn(|| query.next().transpose());
-        let chunks = itertools::process_results(query_iter, |it| ChunkedWriter::collect_iter(it))?;
+        let query_iter = std::iter::from_fn(|| query.next());
+        let chunks = ChunkedWriter::collect_iter(query_iter);
         Ok(chunks)
     }
 }

crates/core/src/sql/ast.rs

@@ -1,14 +1,14 @@
 use crate::config::ReadConfigOption;
 use crate::db::relational_db::{MutTx, RelationalDB, Tx};
 use crate::error::{DBError, PlanError};
-use spacetimedb_data_structures::map::HashMap;
-use spacetimedb_primitives::{ColList, ConstraintKind, Constraints};
+use spacetimedb_data_structures::map::{HashCollectionExt as _, IntMap};
+use spacetimedb_primitives::{ColId, ColList, ConstraintKind, Constraints};
 use spacetimedb_sats::db::def::{ColumnDef, ConstraintDef, TableDef, TableSchema};
 use spacetimedb_sats::db::error::RelationError;
-use spacetimedb_sats::relation::{FieldExpr, FieldName};
+use spacetimedb_sats::relation::{ColExpr, FieldName};
 use spacetimedb_sats::{AlgebraicType, AlgebraicValue};
 use spacetimedb_vm::errors::ErrorVm;
-use spacetimedb_vm::expr::{ColumnOp, DbType, Expr};
+use spacetimedb_vm::expr::{DbType, Expr, FieldExpr, FieldOp};
 use spacetimedb_vm::operator::{OpCmp, OpLogic, OpQuery};
 use spacetimedb_vm::ops::parse::{parse, parse_simple_enum};
 use sqlparser::ast::{
@@ -103,12 +103,12 @@ pub enum Column {
 /// The list of expressions for `SELECT expr1, expr2...` determining what data to extract.
 #[derive(Debug, Clone)]
 pub struct Selection {
-    pub(crate) clause: ColumnOp,
+    pub(crate) clause: FieldOp,
 }
 
 impl Selection {
-    pub fn with_cmp(op: OpQuery, lhs: ColumnOp, rhs: ColumnOp) -> Self {
-        let cmp = ColumnOp::new(op, lhs, rhs);
+    pub fn with_cmp(op: OpQuery, lhs: FieldOp, rhs: FieldOp) -> Self {
+        let cmp = FieldOp::new(op, lhs, rhs);
         Selection { clause: cmp }
     }
 }
@@ -257,25 +257,25 @@ pub fn find_field<'a>(
 pub enum SqlAst {
     Select {
         from: From,
-        project: Vec<Column>,
+        project: Box<[Column]>,
         selection: Option<Selection>,
     },
     Insert {
         table: Arc<TableSchema>,
-        columns: Vec<FieldName>,
-        values: Vec<Vec<FieldExpr>>,
+        columns: Box<[ColId]>,
+        values: Box<[Box<[ColExpr]>]>,
     },
     Update {
         table: Arc<TableSchema>,
-        assignments: HashMap<FieldName, FieldExpr>,
+        assignments: IntMap<ColId, ColExpr>,
         selection: Option<Selection>,
     },
     Delete {
         table: Arc<TableSchema>,
         selection: Option<Selection>,
     },
     CreateTable {
-        table: TableDef,
+        table: Box<TableDef>,
     },
     Drop {
         name: String,
@@ -350,8 +350,8 @@ fn compile_expr_value<'a>(
     tables: impl Clone + Iterator<Item = &'a TableSchema>,
     field: Option<&'a AlgebraicType>,
     of: SqlExpr,
-) -> Result<ColumnOp, PlanError> {
-    Ok(ColumnOp::Field(match of {
+) -> Result<FieldOp, PlanError> {
+    Ok(FieldOp::Field(match of {
         SqlExpr::Identifier(name) => FieldExpr::Name(find_field(tables, &name.value)?.0),
         SqlExpr::CompoundIdentifier(ident) => {
             let col_name = compound_ident(&ident);
@@ -373,7 +373,7 @@ fn compile_expr_value<'a>(
         SqlExpr::BinaryOp { left, op, right } => {
             let (op, lhs, rhs) = compile_bin_op(tables, op, left, right)?;
 
-            return Ok(ColumnOp::new(op, lhs, rhs));
+            return Ok(FieldOp::new(op, lhs, rhs));
         }
         SqlExpr::Nested(x) => {
             return compile_expr_value(tables, field, *x);
@@ -388,7 +388,7 @@ fn compile_expr_value<'a>(
 
 fn compile_expr_field(table: &From, field: Option<&AlgebraicType>, of: SqlExpr) -> Result<FieldExpr, PlanError> {
     match compile_expr_value(table.iter_tables(), field, of)? {
-        ColumnOp::Field(field) => Ok(field),
+        FieldOp::Field(field) => Ok(field),
         x => Err(PlanError::Unsupported {
             feature: format!("Complex expression {x} on insert..."),
         }),
@@ -422,7 +422,7 @@ fn compile_bin_op<'a>(
     op: BinaryOperator,
     lhs: Box<sqlparser::ast::Expr>,
     rhs: Box<sqlparser::ast::Expr>,
-) -> Result<(OpQuery, ColumnOp, ColumnOp), PlanError> {
+) -> Result<(OpQuery, FieldOp, FieldOp), PlanError> {
     let op: OpQuery = match op {
         BinaryOperator::Gt => OpCmp::Gt.into(),
         BinaryOperator::Lt => OpCmp::Lt.into(),
@@ -532,8 +532,8 @@ fn compile_from<T: TableSchemaView>(db: &RelationalDB, tx: &T, from: &[TableWith
                         let tables = base.iter_tables().chain([&*join]);
                         let expr = compile_expr_value(tables, None, x.clone())?;
                         match expr {
-                            ColumnOp::Field(_) => {}
-                            ColumnOp::Cmp { op, lhs, rhs } => {
+                            FieldOp::Field(_) => {}
+                            FieldOp::Cmp { op, lhs, rhs } => {
                                 let op = match op {
                                     OpQuery::Cmp(op) => op,
                                     OpQuery::Logic(op) => {
@@ -543,7 +543,7 @@ fn compile_from<T: TableSchemaView>(db: &RelationalDB, tx: &T, from: &[TableWith
                                     }
                                 };
                                 let (lhs, rhs) = match (*lhs, *rhs) {
-                                    (ColumnOp::Field(FieldExpr::Name(lhs)), ColumnOp::Field(FieldExpr::Name(rhs))) => {
+                                    (FieldOp::Field(FieldExpr::Name(lhs)), FieldOp::Field(FieldExpr::Name(rhs))) => {
                                         (lhs, rhs)
                                     }
                                     (lhs, rhs) => {
@@ -597,7 +597,7 @@ fn compile_select_item(from: &From, select_item: SelectItem) -> Result<Column, P
             sqlparser::ast::Expr::Value(_) => {
                 let value = compile_expr_value(from.iter_tables(), None, expr)?;
                 match value {
-                    ColumnOp::Field(value) => match value {
+                    FieldOp::Field(value) => match value {
                         FieldExpr::Name(_) => Err(PlanError::Unsupported {
                             feature: "Should not be an identifier in Expr::Value".to_string(),
                         }),
@@ -626,11 +626,13 @@ fn compile_select_item(from: &From, select_item: SelectItem) -> Result<Column, P
 /// Compiles the `SELECT ...` clause
 fn compile_select<T: TableSchemaView>(db: &RelationalDB, tx: &T, select: Select) -> Result<SqlAst, PlanError> {
     let from = compile_from(db, tx, &select.from)?;
+
     // SELECT ...
     let mut project = Vec::with_capacity(select.projection.len());
     for select_item in select.projection {
         project.push(compile_select_item(&from, select_item)?);
     }
+    let project = project.into();
 
     let selection = compile_where(&from, select.selection)?;
 
@@ -711,21 +713,21 @@ fn compile_insert<T: TableSchemaView>(
         .map(|x| {
             table
                 .find_field(&format!("{}.{}", &table.root.table_name, x))
-                .map(|(f, _)| f)
+                .map(|(f, _)| f.col)
         })
-        .collect::<Result<Vec<_>, _>>()?;
+        .collect::<Result<Box<[_]>, _>>()?;
 
     let mut values = Vec::with_capacity(data.rows.len());
-
     for x in &data.rows {
         let mut row = Vec::with_capacity(x.len());
         for (pos, v) in x.iter().enumerate() {
             let field_ty = table.root.get_column(pos).map(|col| &col.col_type);
-            row.push(compile_expr_field(&table, field_ty, v.clone())?);
+            row.push(compile_expr_field(&table, field_ty, v.clone())?.strip_table());
         }
-
-        values.push(row);
+        values.push(row.into());
     }
+    let values = values.into();
+
     Ok(SqlAst::Insert {
         table: table.root,
         columns,
@@ -744,19 +746,19 @@ fn compile_update<T: TableSchemaView>(
     let table = From::new(tx.find_table(db, table)?);
     let selection = compile_where(&table, selection)?;
 
-    let mut x = HashMap::with_capacity(assignments.len());
-
+    let mut assigns = IntMap::with_capacity(assignments.len());
     for col in assignments {
         let name: String = col.id.iter().map(|x| x.to_string()).collect();
         let (field_name, field_ty) = table.find_field(&name)?;
+        let col_id = field_name.col;
 
-        let value = compile_expr_field(&table, Some(field_ty), col.value)?;
-        x.insert(field_name, value);
+        let value = compile_expr_field(&table, Some(field_ty), col.value)?.strip_table();
+        assigns.insert(col_id, value);
     }
 
     Ok(SqlAst::Update {
         table: table.root,
-        assignments: x,
+        assignments: assigns,
         selection,
     })
 }
@@ -916,7 +918,7 @@ fn compile_create_table(table: Table, cols: Vec<SqlColumnDef>) -> Result<SqlAst,
         });
     }
 
-    let table = TableDef::new(table.name, columns).with_constraints(constraints);
+    let table = Box::new(TableDef::new(table.name, columns).with_constraints(constraints));
 
     Ok(SqlAst::CreateTable { table })
 }