Add error handling for ranged seeks on non-range compat indices (IndexCannotSeekRange).

Centril · Centril · commit 540a61cfef9f · 2026-01-09T14:55:30.000+01:00
Use point scans in query execution (`IxScanEq`).
Refactor table index code with macro `same_for_all_types`.
diff --git a/crates/datastore/src/error.rs b/crates/datastore/src/error.rs
@@ -129,6 +129,8 @@ pub enum IndexError {
     NotUnique(IndexId),
     #[error("Key {1:?} was not found in index {0:?}")]
     KeyNotFound(IndexId, AlgebraicValue),
+    #[error("IndexId {0:?} does not support seeking for a range")]
+    IndexCannotSeekRange(IndexId),
 }
 
 #[derive(Error, Debug, PartialEq, Eq)]
diff --git a/crates/datastore/src/locking_tx_datastore/committed_state.rs b/crates/datastore/src/locking_tx_datastore/committed_state.rs
@@ -50,6 +50,7 @@ use spacetimedb_table::{
     indexes::{RowPointer, SquashedOffset},
     page_pool::PagePool,
     table::{IndexScanPointIter, IndexScanRangeIter, InsertError, RowRef, Table, TableAndIndex, TableScanIter},
+    table_index::IndexSeekRangeResult,
 };
 use std::collections::BTreeMap;
 use std::sync::Arc;
@@ -213,8 +214,8 @@ impl StateView for CommittedState {
         range: R,
     ) -> Result<Self::IterByColRange<'_, R>> {
         match self.index_seek_range(table_id, &cols, &range) {
-            Some(iter) => Ok(ScanOrIndex::Index(iter)),
-            None => Ok(ScanOrIndex::Scan(ApplyFilter::new(
+            Some(Ok(iter)) => Ok(ScanOrIndex::Index(iter)),
+            None | Some(Err(_)) => Ok(ScanOrIndex::Scan(ApplyFilter::new(
                 RangeOnColumn { cols, range },
                 self.iter(table_id)?,
             ))),
@@ -948,7 +949,7 @@ impl CommittedState {
         table_id: TableId,
         cols: &ColList,
         range: &impl RangeBounds<AlgebraicValue>,
-    ) -> Option<IndexScanRangeIter<'a>> {
+    ) -> Option<IndexSeekRangeResult<IndexScanRangeIter<'a>>> {
         self.tables
             .get(&table_id)?
             .get_index_by_cols_with_table(&self.blob_store, cols)
diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs
@@ -67,7 +67,7 @@ use spacetimedb_table::{
         BlobNumBytes, DuplicateError, IndexScanPointIter, IndexScanRangeIter, InsertError, RowRef, Table,
         TableAndIndex, UniqueConstraintViolation,
     },
-    table_index::TableIndex,
+    table_index::{IndexCannotSeekRange, IndexSeekRangeResult, TableIndex},
 };
 use std::{
     sync::Arc,
@@ -429,7 +429,8 @@ impl Datastore for MutTxId {
             .get_table_and_index(index_id)
             .ok_or_else(|| IndexError::NotFound(index_id))?;
 
-        Ok(self.index_scan_range_inner(table_id, tx_index, commit_index, range))
+        self.index_scan_range_inner(table_id, tx_index, commit_index, range)
+            .map_err(|IndexCannotSeekRange| IndexError::IndexCannotSeekRange(index_id).into())
     }
 
     fn index_scan_point<'a>(
@@ -1373,27 +1374,36 @@ impl MutTxId {
         let bounds =
             Self::range_scan_decode_bounds(index_ty, prefix, prefix_elems, rstart, rend).map_err(IndexError::Decode)?;
 
-        let iter = self.index_scan_range_inner(table_id, tx_index, commit_index, &bounds);
+        let iter = self
+            .index_scan_range_inner(table_id, tx_index, commit_index, &bounds)
+            .map_err(|IndexCannotSeekRange| IndexError::IndexCannotSeekRange(index_id))?;
 
         let (lower, upper) = bounds;
         Ok((table_id, lower, upper, iter))
     }
 
     /// See [`MutTxId::index_scan_range`].
+    #[inline(always)]
     fn index_scan_range_inner<'a>(
         &'a self,
         table_id: TableId,
         tx_index: Option<TableAndIndex<'a>>,
         commit_index: TableAndIndex<'a>,
         bounds: &impl RangeBounds<AlgebraicValue>,
-    ) -> IndexScanRanged<'a> {
+    ) -> IndexSeekRangeResult<IndexScanRanged<'a>> {
         // Get an index seek iterator for the tx and committed state.
-        let tx_iter = tx_index.map(|i| i.seek_range(bounds));
+        let tx_iter = tx_index.map(|i| i.seek_range(bounds)).transpose();
         let commit_iter = commit_index.seek_range(bounds);
 
+        // Deal with not having ranged indices.
+        let (tx_iter, commit_iter) = match (tx_iter, commit_iter) {
+            (Ok(t), Ok(c)) => (t, c),
+            (Err(e), _) | (_, Err(e)) => return Err(e),
+        };
+
         // Combine it all.
         let dt = self.tx_state.get_delete_table(table_id);
-        ScanMutTx::combine(dt, tx_iter, commit_iter)
+        Ok(ScanMutTx::combine(dt, tx_iter, commit_iter))
     }
 
     /// Translate `index_id` to the table id, and commit/tx indices.
@@ -3176,8 +3186,10 @@ fn iter_by_col_range<'a, R: RangeBounds<AlgebraicValue>>(
     // If there's an index, use that.
     // It's sufficient to check that the committed state has an index
     // as index schema changes are applied immediately.
-    if let Some(commit_iter) = committed_state.index_seek_range(table_id, &cols, &range) {
-        let tx_iter = tx_state.index_seek_range_by_cols(table_id, &cols, &range);
+    if let Some(Ok(commit_iter)) = committed_state.index_seek_range(table_id, &cols, &range) {
+        let tx_iter = tx_state
+            .index_seek_range_by_cols(table_id, &cols, &range)
+            .map(|r| r.expect("got a commit index so we should have a compatible tx index"));
         let delete_table = tx_state.get_delete_table(table_id);
         let iter = ScanMutTx::combine(delete_table, tx_iter, commit_iter);
         Ok(ScanOrIndex::Index(iter))
diff --git a/crates/datastore/src/locking_tx_datastore/tx.rs b/crates/datastore/src/locking_tx_datastore/tx.rs
@@ -4,14 +4,17 @@ use super::{
     state_view::{IterByColRangeTx, StateView},
     IterByColEqTx, SharedReadGuard,
 };
-use crate::execution_context::ExecutionContext;
+use crate::{error::IndexError, execution_context::ExecutionContext};
 use spacetimedb_durability::TxOffset;
 use spacetimedb_execution::Datastore;
 use spacetimedb_lib::metrics::ExecutionMetrics;
 use spacetimedb_primitives::{ColList, IndexId, TableId};
 use spacetimedb_sats::AlgebraicValue;
 use spacetimedb_schema::schema::TableSchema;
-use spacetimedb_table::table::{IndexScanPointIter, IndexScanRangeIter, TableAndIndex, TableScanIter};
+use spacetimedb_table::{
+    table::{IndexScanPointIter, IndexScanRangeIter, TableAndIndex, TableScanIter},
+    table_index::IndexCannotSeekRange,
+};
 use std::sync::Arc;
 use std::{future, num::NonZeroU64};
 use std::{
@@ -64,7 +67,8 @@ impl Datastore for TxId {
         index_id: IndexId,
         range: &impl RangeBounds<AlgebraicValue>,
     ) -> anyhow::Result<Self::RangeIndexIter<'a>> {
-        self.with_index(table_id, index_id, |i| i.seek_range(range))
+        self.with_index(table_id, index_id, |i| i.seek_range(range))?
+            .map_err(|IndexCannotSeekRange| IndexError::IndexCannotSeekRange(index_id).into())
     }
 
     fn index_scan_point<'a>(
diff --git a/crates/datastore/src/locking_tx_datastore/tx_state.rs b/crates/datastore/src/locking_tx_datastore/tx_state.rs
@@ -11,7 +11,7 @@ use spacetimedb_table::{
     pointer_map::PointerMap,
     static_assert_size,
     table::{IndexScanPointIter, IndexScanRangeIter, RowRef, Table, TableAndIndex},
-    table_index::TableIndex,
+    table_index::{IndexSeekRangeResult, TableIndex},
 };
 use std::collections::{btree_map, BTreeMap};
 use thin_vec::ThinVec;
@@ -142,7 +142,7 @@ impl TxState {
         table_id: TableId,
         cols: &ColList,
         range: &impl RangeBounds<AlgebraicValue>,
-    ) -> Option<IndexScanRangeIter<'a>> {
+    ) -> Option<IndexSeekRangeResult<IndexScanRangeIter<'a>>> {
         self.insert_tables
             .get(&table_id)?
             .get_index_by_cols_with_table(&self.blob_store, cols)
diff --git a/crates/execution/src/pipelined.rs b/crates/execution/src/pipelined.rs
@@ -284,7 +284,8 @@ impl PipelinedProject {
 #[derive(Debug)]
 pub enum PipelinedExecutor {
     TableScan(PipelinedScan),
-    IxScan(PipelinedIxScan),
+    IxScanEq(PipelinedIxScanEq),
+    IxScanRange(PipelinedIxScanRange),
     IxJoin(PipelinedIxJoin),
     IxDeltaScan(PipelinedIxDeltaScan),
     IxDeltaJoin(PipelinedIxDeltaJoin),
@@ -302,7 +303,22 @@ impl From<PhysicalPlan> for PipelinedExecutor {
                 limit,
                 delta,
             }),
-            PhysicalPlan::IxScan(scan @ IxScan { delta: None, .. }, _) => Self::IxScan(scan.into()),
+            PhysicalPlan::IxScan(
+                scan @ IxScan {
+                    delta: None,
+                    arg: Sarg::Eq(..),
+                    ..
+                },
+                _,
+            ) => Self::IxScanEq(scan.into()),
+            PhysicalPlan::IxScan(
+                scan @ IxScan {
+                    delta: None,
+                    arg: Sarg::Range(..),
+                    ..
+                },
+                _,
+            ) => Self::IxScanRange(scan.into()),
             PhysicalPlan::IxScan(scan, _) => Self::IxDeltaScan(scan.into()),
             PhysicalPlan::IxJoin(
                 IxJoin {
@@ -391,15 +407,16 @@ impl PipelinedExecutor {
                 lhs.visit(f);
                 rhs.visit(f);
             }
-            Self::TableScan(..) | Self::IxScan(..) | Self::IxDeltaScan(..) => {}
+            Self::TableScan(..) | Self::IxScanEq(..) | Self::IxScanRange(..) | Self::IxDeltaScan(..) => {}
         }
     }
 
     /// Does this operation contain an empty delta scan?
     pub fn is_empty(&self, tx: &impl DeltaStore) -> bool {
         match self {
             Self::TableScan(scan) => scan.is_empty(tx),
-            Self::IxScan(scan) => scan.is_empty(tx),
+            Self::IxScanEq(scan) => scan.is_empty(tx),
+            Self::IxScanRange(scan) => scan.is_empty(tx),
             Self::IxDeltaScan(scan) => scan.is_empty(tx),
             Self::IxJoin(join) => join.is_empty(tx),
             Self::IxDeltaJoin(join) => join.is_empty(tx),
@@ -418,7 +435,8 @@ impl PipelinedExecutor {
     ) -> Result<()> {
         match self {
             Self::TableScan(scan) => scan.execute(tx, metrics, f),
-            Self::IxScan(scan) => scan.execute(tx, metrics, f),
+            Self::IxScanEq(scan) => scan.execute(tx, metrics, f),
+            Self::IxScanRange(scan) => scan.execute(tx, metrics, f),
             Self::IxDeltaScan(scan) => scan.execute(tx, metrics, f),
             Self::IxJoin(join) => join.execute(tx, metrics, f),
             Self::IxDeltaJoin(join) => join.execute(tx, metrics, f),
@@ -506,7 +524,7 @@ impl PipelinedScan {
 
 /// An index scan executor for a delta table.
 ///
-/// TODO: There is much overlap between this executor and [PipelinedIxScan].
+/// TODO: There is much overlap between this executor and [PipelinedIxScanRange].
 /// But merging them requires merging the [Datastore] and [DeltaStore] traits,
 /// since the index scan interface is right now split between both.
 #[derive(Debug)]
@@ -659,9 +677,9 @@ impl PipelinedIxDeltaScan {
     }
 }
 
-/// A pipelined executor for scanning an index
+/// A pipelined executor for range scanning an index
 #[derive(Debug)]
-pub struct PipelinedIxScan {
+pub struct PipelinedIxScanRange {
     /// The table id
     pub table_id: TableId,
     /// The index id
@@ -675,7 +693,7 @@ pub struct PipelinedIxScan {
     pub upper: Bound<AlgebraicValue>,
 }
 
-impl From<IxScan> for PipelinedIxScan {
+impl From<IxScan> for PipelinedIxScanRange {
     fn from(scan: IxScan) -> Self {
         match scan {
             IxScan {
@@ -712,7 +730,7 @@ impl From<IxScan> for PipelinedIxScan {
     }
 }
 
-impl PipelinedIxScan {
+impl PipelinedIxScanRange {
     /// We don't know statically if an index scan will return rows
     pub fn is_empty(&self, _: &impl DeltaStore) -> bool {
         false
@@ -794,6 +812,89 @@ impl PipelinedIxScan {
     }
 }
 
+/// A pipelined executor for equality scanning an index
+#[derive(Debug)]
+pub struct PipelinedIxScanEq {
+    /// The table id
+    pub table_id: TableId,
+    /// The index id
+    pub index_id: IndexId,
+    pub limit: Option<u64>,
+    /// The point to scan the index for.
+    pub point: AlgebraicValue,
+}
+
+impl From<IxScan> for PipelinedIxScanEq {
+    fn from(scan: IxScan) -> Self {
+        match scan {
+            IxScan {
+                schema,
+                limit,
+                delta: _,
+                index_id,
+                prefix,
+                arg: Sarg::Eq(_, v),
+            } => {
+                let point = if prefix.is_empty() {
+                    let mut elems = Vec::with_capacity(prefix.len() + 1);
+                    elems.extend(prefix.into_iter().map(|(_, v)| v));
+                    elems.push(v);
+                    AlgebraicValue::product(elems)
+                } else {
+                    v
+                };
+
+                Self {
+                    table_id: schema.table_id,
+                    index_id,
+                    limit,
+                    point,
+                }
+            }
+            IxScan {
+                arg: Sarg::Range(..), ..
+            } => unreachable!(),
+        }
+    }
+}
+
+impl PipelinedIxScanEq {
+    /// We don't know statically if an index scan will return rows
+    pub fn is_empty(&self, _: &impl DeltaStore) -> bool {
+        false
+    }
+
+    pub fn execute<'a, Tx: Datastore + DeltaStore>(
+        &self,
+        tx: &'a Tx,
+        metrics: &mut ExecutionMetrics,
+        f: &mut dyn FnMut(Tuple<'a>) -> Result<()>,
+    ) -> Result<()> {
+        // Scan without a row limit.
+        let scan = || tx.index_scan_point(self.table_id, self.index_id, &self.point);
+        // Scan with an optional row limit.
+        let scan_opt_limit = |limit| match limit {
+            None => scan().map(Either::Left),
+            Some(n) => scan().map(|iter| iter.take(n)).map(Either::Right),
+        };
+        let mut n = 0;
+        let mut f = |t| {
+            n += 1;
+            f(t)
+        };
+        for ptr in scan_opt_limit(self.limit.map(|n| n as usize))?
+            .map(Row::Ptr)
+            .map(Tuple::Row)
+        {
+            f(ptr)?;
+        }
+
+        metrics.index_seeks += 1;
+        metrics.rows_scanned += n;
+        Ok(())
+    }
+}
+
 /// A pipelined index join executor
 #[derive(Debug)]
 pub struct PipelinedIxJoin {
diff --git a/crates/physical-plan/src/plan.rs b/crates/physical-plan/src/plan.rs
@@ -1185,6 +1185,17 @@ pub struct IxScan {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Sarg {
     Eq(ColId, AlgebraicValue),
+    /// NOTE(centril): We currently never construct this variant.
+    /// We do have non-ranged hash indices.
+    /// This means that when we get around to using this variant,
+    /// we must change the rewrite rules such that we do not emit
+    /// [`IxScan`] on a hash index.
+    ///
+    /// Moreover, an equality scan (the variant above)
+    /// `(a0, b0)` on an index `(a, b, c)` is actually a ranged scan.
+    /// We also currently do not emit such `IxScan`s where the number
+    /// of equalities provided are fewer than the number of columns in the index.
+    /// When we do, we must also account for hash indices in the rewrite rules.
     Range(ColId, Bound<AlgebraicValue>, Bound<AlgebraicValue>),
 }
 
diff --git a/crates/subscription/src/lib.rs b/crates/subscription/src/lib.rs
@@ -1,8 +1,8 @@
 use anyhow::{bail, Result};
 use spacetimedb_execution::{
     pipelined::{
-        PipelinedExecutor, PipelinedIxDeltaJoin, PipelinedIxDeltaScan, PipelinedIxJoin, PipelinedIxScan,
-        PipelinedProject,
+        PipelinedExecutor, PipelinedIxDeltaJoin, PipelinedIxDeltaScan, PipelinedIxJoin, PipelinedIxScanEq,
+        PipelinedIxScanRange, PipelinedProject,
     },
     Datastore, DeltaStore, Row,
 };
@@ -36,7 +36,8 @@ impl Fragments {
         let mut index_ids = HashSet::new();
         for plan in self.insert_plans.iter().chain(self.delete_plans.iter()) {
             plan.visit(&mut |plan| match plan {
-                PipelinedExecutor::IxScan(PipelinedIxScan { table_id, index_id, .. })
+                PipelinedExecutor::IxScanEq(PipelinedIxScanEq { table_id, index_id, .. })
+                | PipelinedExecutor::IxScanRange(PipelinedIxScanRange { table_id, index_id, .. })
                 | PipelinedExecutor::IxDeltaScan(PipelinedIxDeltaScan { table_id, index_id, .. })
                 | PipelinedExecutor::IxJoin(PipelinedIxJoin {
                     rhs_table: table_id,
diff --git a/crates/table/benches/page_manager.rs b/crates/table/benches/page_manager.rs
@@ -917,7 +917,7 @@ fn index_seek(c: &mut Criterion) {
                     let mut elapsed = WallTime.zero();
                     for _ in 0..num_iters {
                         let (row, none) = time(&mut elapsed, || {
-                            let mut iter = index.seek_range(&col_to_seek);
+                            let mut iter = index.seek_range(&col_to_seek).unwrap();
                             (iter.next(), iter.next())
                         });
                         assert!(
diff --git a/crates/table/src/table.rs b/crates/table/src/table.rs
diff --git a/crates/table/src/table_index/index.rs b/crates/table/src/table_index/index.rs
diff --git a/crates/table/src/table_index/mod.rs b/crates/table/src/table_index/mod.rs

Original file line number	Diff line number	Diff line change
`@@ -129,6 +129,8 @@ pub enum IndexError {`
`129`	`129`	`NotUnique(IndexId),`
`130`	`130`	`#[error("Key {1:?} was not found in index {0:?}")]`
`131`	`131`	`KeyNotFound(IndexId, AlgebraicValue),`
	`132`	`+ #[error("IndexId {0:?} does not support seeking for a range")]`
	`133`	`+ IndexCannotSeekRange(IndexId),`
`132`	`134`	`}`
`133`	`135`
`134`	`136`	`#[derive(Error, Debug, PartialEq, Eq)]`