Reuse TxState, stashing in CommittedState btw txes

crates/bench/benches/subscription.rs

@@ -1,10 +1,12 @@
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
+use spacetimedb::client::consume_each_list::ConsumeEachBuffer;
 use spacetimedb::error::DBError;
 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_queryset;
+use spacetimedb::subscription::row_list_builder_pool::BsatnRowListBuilderPool;
 use spacetimedb::subscription::subscription::ExecutionSet;
 use spacetimedb::subscription::tx::DeltaTx;
 use spacetimedb::subscription::{collect_table_update, TableUpdateType};
@@ -119,6 +121,8 @@ fn eval(c: &mut Criterion) {
     let ins_rhs = insert_op(rhs, "location", new_rhs_row);
     let update = [&ins_lhs, &ins_rhs];
 
+    let bsatn_rlb_pool = black_box(BsatnRowListBuilderPool::new());
+
     // A benchmark runner for the new query engine
     let bench_query = |c: &mut Criterion, name, sql| {
         c.bench_function(name, |b| {
@@ -134,13 +138,17 @@ fn eval(c: &mut Criterion) {
             let tx = DeltaTx::from(&tx);
 
             b.iter(|| {
-                drop(black_box(collect_table_update::<_, BsatnFormat>(
+                let updates = black_box(collect_table_update::<BsatnFormat>(
                     &plans,
                     table_id,
                     table_name.clone(),
                     &tx,
                     TableUpdateType::Subscribe,
-                )))
+                    &bsatn_rlb_pool,
+                ));
+                if let Ok((updates, _)) = updates {
+                    updates.consume_each_list(&mut |buffer| bsatn_rlb_pool.try_put(buffer));
+                }
             })
         });
     };
@@ -152,12 +160,9 @@ fn eval(c: &mut Criterion) {
             let query: ExecutionSet = query.into();
 
             b.iter(|| {
-                drop(black_box(query.eval::<BsatnFormat>(
-                    &raw.db,
-                    &tx,
-                    None,
-                    Compression::None,
-                )))
+                let updates =
+                    black_box(query.eval::<BsatnFormat>(&raw.db, &tx, &bsatn_rlb_pool, None, Compression::None));
+                updates.consume_each_list(&mut |buffer| bsatn_rlb_pool.try_put(buffer));
             })
         });
     };

crates/bindings-macro/src/table.rs

@@ -466,11 +466,13 @@ impl ValidatedIndex<'_> {
         };
         let vis = superize_vis(vis);
 
+        let num_cols = cols.len();
         let mut decl = quote! {
             #typeck_direct_index
 
             #vis struct #index_ident;
             impl spacetimedb::table::Index for #index_ident {
+                const NUM_COLS_INDEXED: usize = #num_cols;
                 fn index_id() -> spacetimedb::table::IndexId {
                     static INDEX_ID: std::sync::OnceLock<spacetimedb::table::IndexId> = std::sync::OnceLock::new();
                     *INDEX_ID.get_or_init(|| {

crates/bindings-sys/src/lib.rs

@@ -588,7 +588,6 @@ pub mod raw {
         ///
         /// - `out_ptr` is NULL or `out` is not in bounds of WASM memory.
         pub fn identity(out_ptr: *mut u8);
-
     }
 
     // See comment on previous `extern "C"` block re: ABI version.
@@ -775,6 +774,86 @@ pub mod raw {
         ) -> u16;
     }
 
+    #[link(wasm_import_module = "spacetime_10.4")]
+    extern "C" {
+        /// Finds all rows in the index identified by `index_id`,
+        /// according to `point = point_ptr[..point_len]` in WASM memory.
+        ///
+        /// The index itself has a schema/type.
+        /// Matching defined by first BSATN-decoding `point` to that `AlgebraicType`
+        /// and then comparing the decoded `point` to the keys in the index
+        /// using `Ord for AlgebraicValue`.
+        /// to the keys in the index.
+        /// The `point` is BSATN-decoded to that `AlgebraicType`.
+        /// A match happens when `Ordering::Equal` is returned from `fn cmp`.
+        /// This occurs exactly when the row's BSATN-encoding
+        /// is equal to the encoding of the `AlgebraicValue`.
+        ///
+        /// This ABI is not limited to single column indices.
+        /// Multi-column indices can be queried by providing
+        /// a BSATN-encoded `ProductValue`
+        /// that is typed at the `ProductType` of the index.
+        ///
+        /// The relevant table for the index is found implicitly via the `index_id`,
+        /// which is unique for the module.
+        ///
+        /// On success, the iterator handle is written to the `out` pointer.
+        /// This handle can be advanced by [`row_iter_bsatn_advance`].
+        ///
+        /// # Traps
+        ///
+        /// Traps if:
+        /// - `point_ptr` is NULL or `point` is not in bounds of WASM memory.
+        /// - `out` is NULL or `out[..size_of::<RowIter>()]` is not in bounds of WASM memory.
+        ///
+        /// # Errors
+        ///
+        /// Returns an error:
+        ///
+        /// - `NOT_IN_TRANSACTION`, when called outside of a transaction.
+        /// - `NO_SUCH_INDEX`, when `index_id` is not a known ID of an index.
+        /// - `WRONG_INDEX_ALGO` if the index is not a range-scan compatible index.
+        /// - `BSATN_DECODE_ERROR`, when `point` cannot be decoded to an `AlgebraicValue`
+        ///   typed at the index's key type (`AlgebraicType`).
+        pub fn datastore_index_scan_point_bsatn(
+            index_id: IndexId,
+            point_ptr: *const u8, // AlgebraicValue
+            point_len: usize,
+            out: *mut RowIter,
+        ) -> u16;
+
+        /// Deletes all rows found in the index identified by `index_id`,
+        /// according to `point = point_ptr[..point_len]` in WASM memory.
+        ///
+        /// This syscall will delete all the rows found by
+        /// [`datastore_index_scan_point_bsatn`] with the same arguments passed.
+        /// See `datastore_index_scan_point_bsatn` for details.
+        ///
+        /// The number of rows deleted is written to the WASM pointer `out`.
+        ///
+        /// # Traps
+        ///
+        /// Traps if:
+        /// - `point_ptr` is NULL or `point` is not in bounds of WASM memory.
+        /// - `out` is NULL or `out[..size_of::<u32>()]` is not in bounds of WASM memory.
+        ///
+        /// # Errors
+        ///
+        /// Returns an error:
+        ///
+        /// - `NOT_IN_TRANSACTION`, when called outside of a transaction.
+        /// - `NO_SUCH_INDEX`, when `index_id` is not a known ID of an index.
+        /// - `WRONG_INDEX_ALGO` if the index is not a range-compatible index.
+        /// - `BSATN_DECODE_ERROR`, when `point` cannot be decoded to an `AlgebraicValue`
+        ///   typed at the index's key type (`AlgebraicType`).
+        pub fn datastore_delete_by_index_scan_point_bsatn(
+            index_id: IndexId,
+            point_ptr: *const u8, // AlgebraicValue
+            point_len: usize,
+            out: *mut u32,
+        ) -> u16;
+    }
+
     /// What strategy does the database index use?
     ///
     /// See also: <https://www.postgresql.org/docs/current/sql-createindex.html>
@@ -1095,6 +1174,44 @@ pub fn datastore_table_scan_bsatn(table_id: TableId) -> Result<RowIter> {
     Ok(RowIter { raw })
 }
 
+/// Finds all rows in the index identified by `index_id`,
+/// according to the `point.
+///
+/// The index itself has a schema/type.
+/// Matching defined by first BSATN-decoding `point` to that `AlgebraicType`
+/// and then comparing the decoded `point` to the keys in the index
+/// using `Ord for AlgebraicValue`.
+/// to the keys in the index.
+/// The `point` is BSATN-decoded to that `AlgebraicType`.
+/// A match happens when `Ordering::Equal` is returned from `fn cmp`.
+/// This occurs exactly when the row's BSATN-encoding
+/// is equal to the encoding of the `AlgebraicValue`.
+///
+/// This ABI is not limited to single column indices.
+/// Multi-column indices can be queried by providing
+/// a BSATN-encoded `ProductValue`
+/// that is typed at the `ProductType` of the index.
+///
+/// The relevant table for the index is found implicitly via the `index_id`,
+/// which is unique for the module.
+///
+/// On success, the iterator handle is written to the `out` pointer.
+/// This handle can be advanced by [`RowIter::read`].
+///
+/// # Errors
+///
+/// Returns an error:
+///
+/// - `NOT_IN_TRANSACTION`, when called outside of a transaction.
+/// - `NO_SUCH_INDEX`, when `index_id` is not a known ID of an index.
+/// - `WRONG_INDEX_ALGO` if the index is not a range-compatible index.
+/// - `BSATN_DECODE_ERROR`, when `point` cannot be decoded to an `AlgebraicValue`
+///   typed at the index's key type (`AlgebraicType`).
+pub fn datastore_index_scan_point_bsatn(index_id: IndexId, point: &[u8]) -> Result<RowIter> {
+    let raw = unsafe { call(|out| raw::datastore_index_scan_point_bsatn(index_id, point.as_ptr(), point.len(), out))? };
+    Ok(RowIter { raw })
+}
+
 /// Finds all rows in the index identified by `index_id`,
 /// according to the `prefix`, `rstart`, and `rend`.
 ///
@@ -1169,6 +1286,28 @@ pub fn datastore_index_scan_range_bsatn(
     Ok(RowIter { raw })
 }
 
+/// Deletes all rows found in the index identified by `index_id`,
+/// according to the `point.
+///
+/// This syscall will delete all the rows found by
+/// [`datastore_index_scan_point_bsatn`] with the same arguments passed.
+/// See `datastore_index_scan_point_bsatn` for details.
+///
+/// The number of rows deleted is returned on success.
+///
+/// # Errors
+///
+/// Returns an error:
+///
+/// - `NOT_IN_TRANSACTION`, when called outside of a transaction.
+/// - `NO_SUCH_INDEX`, when `index_id` is not a known ID of an index.
+/// - `WRONG_INDEX_ALGO` if the index is not a range-compatible index.
+/// - `BSATN_DECODE_ERROR`, when `point` cannot be decoded to an `AlgebraicValue`
+///   typed at the index's key type (`AlgebraicType`).
+pub fn datastore_delete_by_index_scan_point_bsatn(index_id: IndexId, point: &[u8]) -> Result<u32> {
+    unsafe { call(|out| raw::datastore_delete_by_index_scan_point_bsatn(index_id, point.as_ptr(), point.len(), out)) }
+}
+
 /// Deletes all rows found in the index identified by `index_id`,
 /// according to the `prefix`, `rstart`, and `rend`.
 ///