query engine integration

Cargo.toml

@@ -19,6 +19,7 @@ members = [
   "crates/paths",
   "crates/physical-plan",
   "crates/primitives",
+  "crates/query",
   "crates/sats",
   "crates/schema",
   "crates/sdk",
@@ -106,6 +107,7 @@ spacetimedb-metrics = { path = "crates/metrics", version = "1.0.0-rc2" }
 spacetimedb-paths = { path = "crates/paths", version = "1.0.0-rc2" }
 spacetimedb-physical-plan = { path = "crates/physical-plan", version = "1.0.0-rc2" }
 spacetimedb-primitives = { path = "crates/primitives", version = "1.0.0-rc2" }
+spacetimedb-query = { path = "crates/query", version = "1.0.0-rc2" }
 spacetimedb-sats = { path = "crates/sats", version = "1.0.0-rc2" }
 spacetimedb-schema = { path = "crates/schema", version = "1.0.0-rc2" }
 spacetimedb-standalone = { path = "crates/standalone", version = "1.0.0-rc2" }

crates/bench/Cargo.toml

@@ -31,9 +31,11 @@ bench = false
 spacetimedb-client-api = { path = "../client-api" }
 spacetimedb-core = { path = "../core", features = ["test"] }
 spacetimedb-data-structures.workspace = true
+spacetimedb-execution = { path = "../execution" }
 spacetimedb-lib = { path = "../lib" }
 spacetimedb-paths.workspace = true
 spacetimedb-primitives = { path = "../primitives" }
+spacetimedb-query = { path = "../query" }
 spacetimedb-sats = { path = "../sats" }
 spacetimedb-schema = { workspace = true, features = ["test"] }
 spacetimedb-standalone = { path = "../standalone" }

crates/bench/benches/subscription.rs

@@ -4,12 +4,14 @@ use spacetimedb::execution_context::Workload;
 use spacetimedb::host::module_host::DatabaseTableUpdate;
 use spacetimedb::identity::AuthCtx;
 use spacetimedb::messages::websocket::BsatnFormat;
+use spacetimedb::sql::ast::SchemaViewer;
 use spacetimedb::subscription::query::compile_read_only_query;
 use spacetimedb::subscription::subscription::ExecutionSet;
 use spacetimedb::{db::relational_db::RelationalDB, messages::websocket::Compression};
 use spacetimedb_bench::database::BenchDatabase as _;
 use spacetimedb_bench::spacetime_raw::SpacetimeRaw;
 use spacetimedb_primitives::{col_list, TableId};
+use spacetimedb_query::SubscribePlan;
 use spacetimedb_sats::{product, AlgebraicType, AlgebraicValue, ProductValue};
 
 fn create_table_location(db: &RelationalDB) -> Result<TableId, DBError> {
@@ -99,6 +101,18 @@ fn eval(c: &mut Criterion) {
     let ins_rhs = insert_op(rhs, "location", new_rhs_row);
     let update = [&ins_lhs, &ins_rhs];
 
+    // A benchmark runner for the new query engine
+    let bench_query = |c: &mut Criterion, name, sql| {
+        c.bench_function(name, |b| {
+            let tx = raw.db.begin_tx(Workload::Subscribe);
+            let auth = AuthCtx::for_testing();
+            let schema_viewer = &SchemaViewer::new(&raw.db, &tx, &auth);
+            let plan = SubscribePlan::compile(sql, schema_viewer).unwrap();
+
+            b.iter(|| drop(black_box(plan.execute_bsatn(&tx))))
+        });
+    };
+
     let bench_eval = |c: &mut Criterion, name, sql| {
         c.bench_function(name, |b| {
             let tx = raw.db.begin_tx(Workload::Update);
@@ -116,22 +130,26 @@ fn eval(c: &mut Criterion) {
         });
     };
 
-    // To profile this benchmark for 30s
-    // samply record -r 10000000 cargo bench --bench=subscription --profile=profiling -- full-scan --exact --profile-time=30
-    // Iterate 1M rows.
-    bench_eval(c, "full-scan", "select * from footprint");
-
-    // To profile this benchmark for 30s
-    // samply record -r 10000000 cargo bench --bench=subscription --profile=profiling -- full-join --exact --profile-time=30
     // Join 1M rows on the left with 12K rows on the right.
-    // Note, this should use an index join so as not to read the entire lhs table.
+    // Note, this should use an index join so as not to read the entire footprint table.
     let name = format!(
         r#"
         select footprint.*
-        from footprint join location on footprint.entity_id = location.entity_id
+        from location join footproint on location.entity_id = footprint.entity_id
         where location.chunk_index = {chunk_index}
         "#
     );
+
+    bench_query(c, "footprint-scan", "select * from footprint");
+    bench_query(c, "footprint-semijoin", &name);
+
+    // To profile this benchmark for 30s
+    // samply record -r 10000000 cargo bench --bench=subscription --profile=profiling -- full-scan --exact --profile-time=30
+    // Iterate 1M rows.
+    bench_eval(c, "full-scan", "select * from footprint");
+
+    // To profile this benchmark for 30s
+    // samply record -r 10000000 cargo bench --bench=subscription --profile=profiling -- full-join --exact --profile-time=30
     bench_eval(c, "full-join", &name);
 
     // To profile this benchmark for 30s

crates/core/Cargo.toml

@@ -29,6 +29,7 @@ spacetimedb-table.workspace = true
 spacetimedb-vm.workspace = true
 spacetimedb-snapshot.workspace = true
 spacetimedb-expr.workspace = true
+spacetimedb-execution.workspace = true
 
 anyhow = { workspace = true, features = ["backtrace"] }
 arrayvec.workspace = true

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

@@ -6,9 +6,12 @@ use super::{
     SharedReadGuard,
 };
 use crate::execution_context::ExecutionContext;
+use spacetimedb_execution::Datastore;
 use spacetimedb_primitives::{ColList, TableId};
 use spacetimedb_sats::AlgebraicValue;
 use spacetimedb_schema::schema::TableSchema;
+use spacetimedb_table::blob_store::BlobStore;
+use spacetimedb_table::table::Table;
 use std::num::NonZeroU64;
 use std::sync::Arc;
 use std::{
@@ -23,6 +26,16 @@ pub struct TxId {
     pub(crate) ctx: ExecutionContext,
 }
 
+impl Datastore for TxId {
+    fn blob_store(&self) -> &dyn BlobStore {
+        &self.committed_state_shared_lock.blob_store
+    }
+
+    fn table(&self, table_id: TableId) -> Option<&Table> {
+        self.committed_state_shared_lock.get_table(table_id)
+    }
+}
+
 impl StateView for TxId {
     fn get_schema(&self, table_id: TableId) -> Option<&Arc<TableSchema>> {
         self.committed_state_shared_lock.get_schema(table_id)

crates/core/src/sql/parser.rs

@@ -3,13 +3,13 @@ use crate::db::relational_db::RelationalDB;
 use crate::sql::ast::SchemaViewer;
 use spacetimedb_expr::check::parse_and_type_sub;
 use spacetimedb_expr::errors::TypingError;
-use spacetimedb_expr::expr::Project;
+use spacetimedb_expr::expr::ProjectName;
 use spacetimedb_lib::db::raw_def::v9::RawRowLevelSecurityDefV9;
 use spacetimedb_lib::identity::AuthCtx;
 use spacetimedb_schema::schema::RowLevelSecuritySchema;
 
 pub struct RowLevelExpr {
-    pub sql: Project,
+    pub sql: ProjectName,
     pub def: RowLevelSecuritySchema,
 }
 

crates/execution/Cargo.toml

@@ -7,7 +7,9 @@ license-file = "LICENSE"
 description = "The SpacetimeDB query engine"
 
 [dependencies]
-spacetimedb-expr.workspace = true
+anyhow.workspace = true
 spacetimedb-lib.workspace = true
+spacetimedb-physical-plan.workspace = true
 spacetimedb-primitives.workspace = true
+spacetimedb-sql-parser.workspace = true
 spacetimedb-table.workspace = true