Merge branch 'master' into shub/hack-idc

Author: Shubham8287

crates/bindings-typescript/src/sdk/db_connection_impl.ts

@@ -158,6 +158,10 @@ export class DbConnectionImpl<RemoteModule extends UntypedRemoteModule>
   #reducerCallInfo = new Map<number, { name: string; args: object }>();
   #procedureCallbacks = new Map<number, ProcedureCallback>();
   #rowDeserializers: Record<string, Deserializer<any>>;
+  #rowIdMetadata: Record<
+    string,
+    { primaryKeyColName?: string; primaryKeyColType?: AlgebraicType }
+  >;
   #reducerArgsSerializers: Record<
     string,
     { serialize: Serializer<any>; deserialize: Deserializer<any> }
@@ -205,6 +209,7 @@ export class DbConnectionImpl<RemoteModule extends UntypedRemoteModule>
     this.#emitter = emitter;
 
     this.#rowDeserializers = Object.create(null);
+    this.#rowIdMetadata = Object.create(null);
     this.#sourceNameToTableDef = Object.create(null);
     for (const table of Object.values(remoteModule.tables)) {
       this.#rowDeserializers[table.sourceName] = ProductType.makeDeserializer(
@@ -213,6 +218,15 @@ export class DbConnectionImpl<RemoteModule extends UntypedRemoteModule>
       this.#sourceNameToTableDef[table.sourceName] = table as Values<
         RemoteModule['tables']
       >;
+      const primaryKeyColumn = Object.entries(table.columns).find(
+        ([, column]) => column.columnMetadata.isPrimaryKey
+      );
+      this.#rowIdMetadata[table.sourceName] = primaryKeyColumn
+        ? {
+            primaryKeyColName: primaryKeyColumn[0],
+            primaryKeyColType: primaryKeyColumn[1].typeBuilder.algebraicType,
+          }
+        : {};
     }
 
     this.#reducerArgsSerializers = Object.create(null);
@@ -302,8 +316,6 @@ export class DbConnectionImpl<RemoteModule extends UntypedRemoteModule>
   #makeReducers(def: RemoteModule): ReducersView<RemoteModule> {
     const out: Record<string, unknown> = {};
 
-    const writer = new BinaryWriter(1024);
-
     for (const reducer of def.reducers) {
       const reducerName = reducer.name;
       const key = reducer.accessorName;
@@ -312,6 +324,7 @@ export class DbConnectionImpl<RemoteModule extends UntypedRemoteModule>
         this.#reducerArgsSerializers[reducerName];
 
       (out as any)[key] = (params: InferTypeOfRow<typeof reducer.params>) => {
+        const writer = this.#reducerArgsEncoder;
         writer.clear();
         serializeArgs(writer, params);
         const argsBuffer = writer.getBuffer();
@@ -428,20 +441,13 @@ export class DbConnectionImpl<RemoteModule extends UntypedRemoteModule>
     const rows: Operation[] = [];
 
     const deserializeRow = this.#rowDeserializers[tableName];
-    const table = this.#sourceNameToTableDef[tableName];
-    // TODO: performance
-    const columnsArray = Object.entries(table.columns);
-    const primaryKeyColumnEntry = columnsArray.find(
-      col => col[1].columnMetadata.isPrimaryKey
-    );
+    const { primaryKeyColName, primaryKeyColType } =
+      this.#rowIdMetadata[tableName];
     let previousOffset = 0;
     while (reader.remaining > 0) {
       const row = deserializeRow(reader);
       let rowId: ComparablePrimitive | undefined = undefined;
-      if (primaryKeyColumnEntry !== undefined) {
-        const primaryKeyColName = primaryKeyColumnEntry[0];
-        const primaryKeyColType =
-          primaryKeyColumnEntry[1].typeBuilder.algebraicType;
+      if (primaryKeyColName !== undefined && primaryKeyColType !== undefined) {
         rowId = AlgebraicType.intoMapKey(
           primaryKeyColType,
           row[primaryKeyColName]
@@ -548,6 +554,7 @@ export class DbConnectionImpl<RemoteModule extends UntypedRemoteModule>
     }
   }
 
+  #reducerArgsEncoder = new BinaryWriter(1024);
   #clientMessageEncoder = new BinaryWriter(1024);
   #sendMessage(message: ClientMessage): void {
     const writer = this.#clientMessageEncoder;
@@ -906,7 +913,8 @@ export class DbConnectionImpl<RemoteModule extends UntypedRemoteModule>
     _paramsType: ProductType,
     params: object
   ): Promise<void> {
-    const writer = new BinaryWriter(1024);
+    const writer = this.#reducerArgsEncoder;
+    writer.clear();
     this.#reducerArgsSerializers[reducerName].serialize(writer, params);
     const argsBuffer = writer.getBuffer();
     return this.callReducer(reducerName, argsBuffer, params);

crates/bindings-typescript/src/sdk/table_cache.ts

@@ -66,6 +66,7 @@ export class TableCacheImpl<
   TableName extends TableNamesOf<RemoteModule>,
 > implements ClientTableCoreImplementable<RemoteModule, TableName>
 {
+  private readonly hasPrimaryKey: boolean;
   private rows: Map<
     ComparablePrimitive,
     [RowType<TableDefForTableName<RemoteModule, TableName>>, number]
@@ -83,6 +84,9 @@ export class TableCacheImpl<
     this.tableDef = tableDef;
     this.rows = new Map();
     this.emitter = new EventEmitter();
+    this.hasPrimaryKey = Object.values(this.tableDef.columns).some(
+      col => col.columnMetadata.isPrimaryKey === true
+    );
     // Build index views from the resolved runtime index metadata.
     //
     // We intentionally use `resolvedIndexes` rather than `indexes`:
@@ -281,11 +285,7 @@ export class TableCacheImpl<
       return pendingCallbacks;
     }
 
-    // TODO: performance
-    const hasPrimaryKey = Object.values(this.tableDef.columns).some(
-      col => col.columnMetadata.isPrimaryKey === true
-    );
-    if (hasPrimaryKey) {
+    if (this.hasPrimaryKey) {
       const insertMap = new Map<
         ComparablePrimitive,
         [

crates/bindings-typescript/src/server/runtime.ts

@@ -667,9 +667,37 @@ function makeTableView(
       index = base as UniqueIndex<any, any>;
     } else if (serializeSinglePoint) {
       // numColumns == 1
+
+      const serializeSingleRange = !isHashIndex
+        ? (buffer: ResizableBuffer, range: Range<any>): IndexScanArgs => {
+            BINARY_WRITER.reset(buffer);
+            const writer = BINARY_WRITER;
+            const writeBound = (bound: Bound<any>) => {
+              const tags = { included: 0, excluded: 1, unbounded: 2 };
+              writer.writeU8(tags[bound.tag]);
+              if (bound.tag !== 'unbounded')
+                serializeSingleElement!(writer, bound.value);
+            };
+            writeBound(range.from);
+            const rstartLen = writer.offset;
+            writeBound(range.to);
+            const rendLen = writer.offset - rstartLen;
+            return [0, 0, rstartLen, rendLen];
+          }
+        : null;
+
       const rawIndex = {
         filter: (range: any): IteratorObject<RowType<any>> => {
           const buf = LEAF_BUF;
+          if (serializeSingleRange && range instanceof Range) {
+            const args = serializeSingleRange(buf, range);
+            const iter_id = sys.datastore_index_scan_range_bsatn(
+              index_id,
+              buf.buffer,
+              ...args
+            );
+            return tableIterator(iter_id, deserializeRow);
+          }
           const point_len = serializeSinglePoint(buf, range);
           const iter_id = sys.datastore_index_scan_point_bsatn(
             index_id,
@@ -680,6 +708,14 @@ function makeTableView(
         },
         delete: (range: any): u32 => {
           const buf = LEAF_BUF;
+          if (serializeSingleRange && range instanceof Range) {
+            const args = serializeSingleRange(buf, range);
+            return sys.datastore_delete_by_index_scan_range_bsatn(
+              index_id,
+              buf.buffer,
+              ...args
+            );
+          }
           const point_len = serializeSinglePoint(buf, range);
           return sys.datastore_delete_by_index_scan_point_bsatn(
             index_id,

crates/core/src/config.rs

@@ -56,11 +56,12 @@ impl MetadataFile {
         path.write(self.to_string())
     }
 
-    fn check_compatibility(previous: &Self, current: &Self) -> anyhow::Result<()> {
+    fn check_compatibility(previous: &Self, current: &Self, metafile: &Path) -> anyhow::Result<()> {
         anyhow::ensure!(
             previous.edition == current.edition,
-            "metadata.toml indicates that this database is from a different \
+            "metadata.toml at {} indicates that this database is from a different \
             edition of SpacetimeDB (running {:?}, but this database is {:?})",
+            metafile.display(),
             current.edition,
             previous.edition,
         );
@@ -110,8 +111,8 @@ impl MetadataFile {
     /// `self` is the metadata file read from a database, and current is
     /// the default metadata file that the active database version would
     /// right to a new database.
-    pub fn check_compatibility_and_update(mut self, current: Self) -> anyhow::Result<Self> {
-        Self::check_compatibility(&self, &current)?;
+    pub fn check_compatibility_and_update(mut self, current: Self, metafile: &Path) -> anyhow::Result<Self> {
+        Self::check_compatibility(&self, &current, metafile)?;
         // bump the version in the file only if it's being run in a newer database.
         self.version = std::cmp::max(self.version, current.version);
         Ok(self)
@@ -344,67 +345,67 @@ mod tests {
     fn check_metadata_compatibility_checking() {
         assert_eq!(
             mkmeta(1, 0, 0)
-                .check_compatibility_and_update(mkmeta(1, 0, 1))
+                .check_compatibility_and_update(mkmeta(1, 0, 1), Path::new("metadata.toml"))
                 .unwrap()
                 .version,
             mkver(1, 0, 1)
         );
         assert_eq!(
             mkmeta(1, 0, 1)
-                .check_compatibility_and_update(mkmeta(1, 0, 0))
+                .check_compatibility_and_update(mkmeta(1, 0, 0), Path::new("metadata.toml"))
                 .unwrap()
                 .version,
             mkver(1, 0, 1)
         );
 
         mkmeta(1, 1, 0)
-            .check_compatibility_and_update(mkmeta(1, 0, 5))
+            .check_compatibility_and_update(mkmeta(1, 0, 5), Path::new("metadata.toml"))
             .unwrap_err();
         mkmeta(2, 0, 0)
-            .check_compatibility_and_update(mkmeta(1, 3, 5))
+            .check_compatibility_and_update(mkmeta(1, 3, 5), Path::new("metadata.toml"))
             .unwrap_err();
         assert_eq!(
             mkmeta(1, 12, 0)
-                .check_compatibility_and_update(mkmeta(2, 0, 0))
+                .check_compatibility_and_update(mkmeta(2, 0, 0), Path::new("metadata.toml"))
                 .unwrap()
                 .version,
             mkver(2, 0, 0)
         );
         mkmeta(2, 0, 0)
-            .check_compatibility_and_update(mkmeta(3, 0, 0))
+            .check_compatibility_and_update(mkmeta(3, 0, 0), Path::new("metadata.toml"))
             .unwrap_err();
     }
 
     #[test]
     fn check_metadata_compatibility_prerelease() {
         mkmeta(1, 9, 0)
-            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"))
+            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"), Path::new("metadata.toml"))
             .unwrap();
 
         mkmeta_pre(2, 0, 0, "rc1")
-            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"))
+            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"), Path::new("metadata.toml"))
             .unwrap();
 
         mkmeta_pre(2, 0, 0, "rc1")
-            .check_compatibility_and_update(mkmeta(2, 0, 1))
+            .check_compatibility_and_update(mkmeta(2, 0, 1), Path::new("metadata.toml"))
             .unwrap();
 
         mkmeta_pre(2, 0, 0, "rc1")
-            .check_compatibility_and_update(mkmeta(2, 0, 0))
+            .check_compatibility_and_update(mkmeta(2, 0, 0), Path::new("metadata.toml"))
             .unwrap();
 
         // Now check some failures..
 
         mkmeta_pre(2, 0, 0, "rc1")
-            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc2"))
+            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc2"), Path::new("metadata.toml"))
             .unwrap_err();
 
         mkmeta_pre(2, 0, 0, "rc2")
-            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"))
+            .check_compatibility_and_update(mkmeta_pre(2, 0, 0, "rc1"), Path::new("metadata.toml"))
             .unwrap_err();
 
         mkmeta(2, 0, 0)
-            .check_compatibility_and_update(mkmeta_pre(2, 1, 0, "rc1"))
+            .check_compatibility_and_update(mkmeta_pre(2, 1, 0, "rc1"), Path::new("metadata.toml"))
             .unwrap_err();
     }
 

crates/core/src/db/relational_db.rs

@@ -17,7 +17,7 @@ use spacetimedb_datastore::locking_tx_datastore::datastore::TxMetrics;
 use spacetimedb_datastore::locking_tx_datastore::state_view::{
     IterByColEqMutTx, IterByColRangeMutTx, IterMutTx, StateView,
 };
-use spacetimedb_datastore::locking_tx_datastore::{MutTxId, TxId};
+use spacetimedb_datastore::locking_tx_datastore::{IndexScanPointOrRange, MutTxId, TxId};
 use spacetimedb_datastore::system_tables::{
     system_tables, StModuleRow, ST_CLIENT_ID, ST_CONNECTION_CREDENTIALS_ID, ST_VIEW_SUB_ID,
 };
@@ -55,10 +55,11 @@ use spacetimedb_snapshot::{ReconstructedSnapshot, SnapshotError, SnapshotReposit
 use spacetimedb_table::indexes::RowPointer;
 use spacetimedb_table::page_pool::PagePool;
 use spacetimedb_table::table::{RowRef, TableScanIter};
+use spacetimedb_table::table_index::IndexKey;
 use std::borrow::Cow;
 use std::io;
 use std::num::NonZeroUsize;
-use std::ops::{Bound, RangeBounds};
+use std::ops::RangeBounds;
 use std::sync::Arc;
 use tokio::sync::watch;
 
@@ -1394,32 +1395,24 @@ impl RelationalDB {
         Ok(self.inner.iter_by_col_range_tx(tx, table_id.into(), cols, range)?)
     }
 
-    pub fn index_scan_range<'a>(
+    pub fn index_scan_range<'de, 'a>(
         &'a self,
         tx: &'a MutTx,
         index_id: IndexId,
-        prefix: &[u8],
+        prefix: &'de [u8],
         prefix_elems: ColId,
-        rstart: &[u8],
-        rend: &[u8],
-    ) -> Result<
-        (
-            TableId,
-            Bound<AlgebraicValue>,
-            Bound<AlgebraicValue>,
-            impl Iterator<Item = RowRef<'a>> + use<'a>,
-        ),
-        DBError,
-    > {
+        rstart: &'de [u8],
+        rend: &'de [u8],
+    ) -> Result<(TableId, IndexScanPointOrRange<'de, 'a>), DBError> {
         Ok(tx.index_scan_range(index_id, prefix, prefix_elems, rstart, rend)?)
     }
 
-    pub fn index_scan_point<'a>(
+    pub fn index_scan_point<'a, 'p>(
         &'a self,
         tx: &'a MutTx,
         index_id: IndexId,
-        point: &[u8],
-    ) -> Result<(TableId, AlgebraicValue, impl Iterator<Item = RowRef<'a>> + use<'a>), DBError> {
+        point: &'p [u8],
+    ) -> Result<(TableId, IndexKey<'p>, impl Iterator<Item = RowRef<'a>> + use<'a>), DBError> {
         Ok(tx.index_scan_point(index_id, point)?)
     }
 

crates/core/src/host/instance_env.rs

@@ -18,7 +18,7 @@ use spacetimedb_client_api_messages::energy::EnergyQuanta;
 use spacetimedb_datastore::db_metrics::DB_METRICS;
 use spacetimedb_datastore::execution_context::Workload;
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
-use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, MutTxId};
+use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, IndexScanPointOrRange, MutTxId};
 use spacetimedb_datastore::traits::IsolationLevel;
 use spacetimedb_lib::{http as st_http, ConnectionId, Identity, Timestamp};
 use spacetimedb_primitives::{ColId, ColList, IndexId, TableId};
@@ -539,9 +539,12 @@ impl InstanceEnv {
         let tx = &mut *self.get_tx()?;
 
         // Find all rows in the table to delete.
-        let (table_id, _, _, iter) = stdb.index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
+        let (table_id, iter) = stdb.index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
         // Re. `SmallVec`, `delete_by_field` only cares about 1 element, so optimize for that.
-        let rows_to_delete = iter.map(|row_ref| row_ref.pointer()).collect::<SmallVec<[_; 1]>>();
+        let rows_to_delete = match iter {
+            IndexScanPointOrRange::Point(_, iter) => iter.map(|row_ref| row_ref.pointer()).collect(),
+            IndexScanPointOrRange::Range(iter) => iter.map(|row_ref| row_ref.pointer()).collect(),
+        };
 
         Ok(Self::datastore_delete_by_index_scan(stdb, tx, table_id, rows_to_delete))
     }
@@ -703,19 +706,22 @@ impl InstanceEnv {
         let tx = &mut *self.get_tx()?;
 
         // Open index iterator
-        let (table_id, lower, upper, iter) =
+        let (table_id, iter) =
             self.relational_db()
                 .index_scan_range(tx, index_id, prefix, prefix_elems, rstart, rend)?;
 
         // Scan the index and serialize rows to BSATN.
-        let (chunks, rows_scanned, bytes_scanned) = ChunkedWriter::collect_iter(pool, iter);
+        let (point, (chunks, rows_scanned, bytes_scanned)) = match iter {
+            IndexScanPointOrRange::Point(point, iter) => (Some(point), ChunkedWriter::collect_iter(pool, iter)),
+            IndexScanPointOrRange::Range(iter) => (None, ChunkedWriter::collect_iter(pool, iter)),
+        };
 
         // Record the number of rows and the number of bytes scanned by the iterator.
         tx.metrics.index_seeks += 1;
         tx.metrics.bytes_scanned += bytes_scanned;
         tx.metrics.rows_scanned += rows_scanned;
 
-        tx.record_index_scan_range(&self.func_type, table_id, index_id, lower, upper);
+        tx.record_index_scan_range(&self.func_type, table_id, index_id, point);
 
         Ok(chunks)
     }