diff --git a/crates/bench/src/spacetime_raw.rs b/crates/bench/src/spacetime_raw.rs
index a2ef372cfe5..b33cf47bbd2 100644
--- a/crates/bench/src/spacetime_raw.rs
+++ b/crates/bench/src/spacetime_raw.rs
@@ -3,7 +3,7 @@ use crate::{
     schemas::{table_name, BenchTable, IndexStrategy},
     ResultBench,
 };
-use spacetimedb::db::datastore::traits::{IndexDef, TableDef, TableSchema};
+use spacetimedb::db::datastore::traits::{ColId, IndexDef, TableDef, TableSchema};
 use spacetimedb::db::relational_db::{open_db, RelationalDB};
 use spacetimedb::error::DBError;
 use spacetimedb::sql::execute::run;
@@ -120,8 +120,9 @@ impl BenchDatabase for SpacetimeRaw {
         column_index: u32,
         value: AlgebraicValue,
     ) -> ResultBench<()> {
+        let col: ColId = column_index.into();
         self.db.with_auto_commit(|tx| {
-            for row in self.db.iter_by_col_eq(tx, table.table_id, column_index, value)? {
+            for row in self.db.iter_by_col_eq(tx, table.table_id, col, value)? {
                 black_box(row);
             }
             Ok(())
diff --git a/crates/core/src/db/datastore/locking_tx_datastore/mod.rs b/crates/core/src/db/datastore/locking_tx_datastore/mod.rs
index 485af501891..6f39029b983 100644
--- a/crates/core/src/db/datastore/locking_tx_datastore/mod.rs
+++ b/crates/core/src/db/datastore/locking_tx_datastore/mod.rs
@@ -207,7 +207,7 @@ impl CommittedState {
     pub fn index_seek<'a>(
         &'a self,
         table_id: &TableId,
-        cols: NonEmpty<ColId>,
+        cols: &NonEmpty<ColId>,
         range: &impl RangeBounds<AlgebraicValue>,
     ) -> Option<BTreeIndexRangeIter<'a>> {
         if let Some(table) = self.tables.get(table_id) {
@@ -323,7 +323,7 @@ impl TxState {
     pub fn index_seek<'a>(
         &'a self,
         table_id: &TableId,
-        cols: NonEmpty<ColId>,
+        cols: &NonEmpty<ColId>,
         range: &impl RangeBounds<AlgebraicValue>,
     ) -> Option<BTreeIndexRangeIter<'a>> {
         self.insert_tables.get(table_id)?.index_seek(cols, range)
@@ -582,7 +582,7 @@ impl Inner {
 
     fn drop_table_from_st_tables(&mut self, table_id: TableId) -> super::Result<()> {
         const ST_TABLES_TABLE_ID_COL: ColId = ColId(0);
-        let rows = self.iter_by_col_eq(&ST_TABLES_ID, &ST_TABLES_TABLE_ID_COL, table_id.into())?;
+        let rows = self.iter_by_col_eq(&ST_TABLES_ID, ST_TABLES_TABLE_ID_COL, table_id.into())?;
         let rows = rows.map(|row| row.view().to_owned()).collect::<Vec<_>>();
         if rows.is_empty() {
             return Err(TableError::IdNotFound(table_id.0).into());
@@ -593,7 +593,7 @@ impl Inner {
 
     fn drop_table_from_st_columns(&mut self, table_id: TableId) -> super::Result<()> {
         const ST_COLUMNS_TABLE_ID_COL: ColId = ColId(0);
-        let rows = self.iter_by_col_eq(&ST_COLUMNS_ID, &ST_COLUMNS_TABLE_ID_COL, table_id.into())?;
+        let rows = self.iter_by_col_eq(&ST_COLUMNS_ID, ST_COLUMNS_TABLE_ID_COL, table_id.into())?;
         let rows = rows.map(|row| row.view().to_owned()).collect::<Vec<_>>();
         if rows.is_empty() {
             return Err(TableError::IdNotFound(table_id.0).into());
@@ -619,7 +619,7 @@ impl Inner {
         // If we're out of allocations, then update the sequence row in st_sequences to allocate a fresh batch of sequences.
         const ST_SEQUENCES_SEQUENCE_ID_COL: ColId = ColId(0);
         let old_seq_row = self
-            .iter_by_col_eq(&ST_SEQUENCES_ID, &ST_SEQUENCES_SEQUENCE_ID_COL, seq_id.into())?
+            .iter_by_col_eq(&ST_SEQUENCES_ID, ST_SEQUENCES_SEQUENCE_ID_COL, seq_id.into())?
             .last()
             .unwrap()
             .data;
@@ -684,7 +684,7 @@ impl Inner {
     fn drop_sequence(&mut self, seq_id: SequenceId) -> super::Result<()> {
         const ST_SEQUENCES_SEQUENCE_ID_COL: ColId = ColId(0);
         let old_seq_row = self
-            .iter_by_col_eq(&ST_SEQUENCES_ID, &ST_SEQUENCES_SEQUENCE_ID_COL, seq_id.into())?
+            .iter_by_col_eq(&ST_SEQUENCES_ID, ST_SEQUENCES_SEQUENCE_ID_COL, seq_id.into())?
             .last()
             .unwrap()
             .data;
@@ -698,7 +698,7 @@ impl Inner {
         let seq_name_col: ColId = ColId(1);
         self.iter_by_col_eq(
             &ST_SEQUENCES_ID,
-            &seq_name_col,
+            seq_name_col,
             AlgebraicValue::String(seq_name.to_owned()),
         )
         .map(|mut iter| {
@@ -819,7 +819,7 @@ impl Inner {
         }
 
         // Look up the table_name for the table in question.
-        let table_id_col: ColId = ColId(0);
+        let table_id_col = NonEmpty::new(0);
 
         // TODO(george): As part of the bootstrapping process, we add a bunch of rows
         // and only at very end do we patch things up and create table metadata, indexes,
@@ -829,7 +829,7 @@ impl Inner {
         let value: AlgebraicValue = table_id.into();
         let rows = IterByColRange::Scan(ScanIterByColRange {
             range: value,
-            col_id: table_id_col,
+            cols: table_id_col,
             scan_iter: self.iter(&ST_TABLES_ID)?,
         })
         .collect::<Vec<_>>();
@@ -843,7 +843,7 @@ impl Inner {
         // Look up the columns for the table in question.
         let mut columns = Vec::new();
         const TABLE_ID_COL: ColId = ColId(0);
-        for data_ref in self.iter_by_col_eq(&ST_COLUMNS_ID, &TABLE_ID_COL, table_id.into())? {
+        for data_ref in self.iter_by_col_eq(&ST_COLUMNS_ID, TABLE_ID_COL, table_id.into())? {
             let row = data_ref.view();
 
             let el = StColumnRow::try_from(row)?;
@@ -862,7 +862,7 @@ impl Inner {
         // Look up the indexes for the table in question.
         let mut indexes = Vec::new();
         let table_id_col: ColId = ColId(1);
-        for data_ref in self.iter_by_col_eq(&ST_INDEXES_ID, &table_id_col, table_id.into())? {
+        for data_ref in self.iter_by_col_eq(&ST_INDEXES_ID, table_id_col, table_id.into())? {
             let row = data_ref.view();
 
             let el = StIndexRow::try_from(row)?;
@@ -891,7 +891,7 @@ impl Inner {
         // First drop the tables indexes.
         const ST_INDEXES_TABLE_ID_COL: ColId = ColId(1);
         let rows = self
-            .iter_by_col_eq(&ST_INDEXES_ID, &ST_INDEXES_TABLE_ID_COL, table_id.into())?
+            .iter_by_col_eq(&ST_INDEXES_ID, ST_INDEXES_TABLE_ID_COL, table_id.into())?
             .collect::<Vec<_>>();
         for data_ref in rows {
             let row = data_ref.view();
@@ -902,7 +902,7 @@ impl Inner {
         // Remove the table's sequences from st_sequences.
         const ST_SEQUENCES_TABLE_ID_COL: ColId = ColId(2);
         let rows = self
-            .iter_by_col_eq(&ST_SEQUENCES_ID, &ST_SEQUENCES_TABLE_ID_COL, table_id.into())?
+            .iter_by_col_eq(&ST_SEQUENCES_ID, ST_SEQUENCES_TABLE_ID_COL, table_id.into())?
             .collect::<Vec<_>>();
         for data_ref in rows {
             let row = data_ref.view();
@@ -927,7 +927,7 @@ impl Inner {
         // Update the table's name in st_tables.
         const ST_TABLES_TABLE_ID_COL: ColId = ColId(0);
         let rows = self
-            .iter_by_col_eq(&ST_TABLES_ID, &ST_TABLES_TABLE_ID_COL, table_id.into())?
+            .iter_by_col_eq(&ST_TABLES_ID, ST_TABLES_TABLE_ID_COL, table_id.into())?
             .collect::<Vec<_>>();
         assert!(rows.len() <= 1, "Expected at most one row in st_tables for table_id");
         let row = rows.first().ok_or_else(|| TableError::IdNotFound(table_id.0))?;
@@ -943,7 +943,7 @@ impl Inner {
         let table_name_col: ColId = ColId(1);
         self.iter_by_col_eq(
             &ST_TABLES_ID,
-            &table_name_col,
+            table_name_col,
             AlgebraicValue::String(table_name.to_owned()),
         )
         .map(|mut iter| {
@@ -954,7 +954,7 @@ impl Inner {
 
     fn table_name_from_id(&self, table_id: TableId) -> super::Result<Option<String>> {
         let table_id_col: ColId = ColId(0);
-        self.iter_by_col_eq(&ST_TABLES_ID, &table_id_col, table_id.into())
+        self.iter_by_col_eq(&ST_TABLES_ID, table_id_col, table_id.into())
             .map(|mut iter| {
                 iter.next()
                     .map(|row| row.view().elements[1].as_string().unwrap().to_owned())
@@ -1055,7 +1055,7 @@ impl Inner {
         // Remove the index from st_indexes.
         const ST_INDEXES_INDEX_ID_COL: ColId = ColId(0);
         let old_index_row = self
-            .iter_by_col_eq(&ST_INDEXES_ID, &ST_INDEXES_INDEX_ID_COL, index_id.into())?
+            .iter_by_col_eq(&ST_INDEXES_ID, ST_INDEXES_INDEX_ID_COL, index_id.into())?
             .last()
             .unwrap()
             .data;
@@ -1104,7 +1104,7 @@ impl Inner {
         let index_name_col: ColId = ColId(3);
         self.iter_by_col_eq(
             &ST_INDEXES_ID,
-            &index_name_col,
+            index_name_col,
             AlgebraicValue::String(index_name.to_owned()),
         )
         .map(|mut iter| {
@@ -1195,7 +1195,7 @@ impl Inner {
                     continue;
                 }
                 let st_sequences_table_id_col = ColId(2);
-                for seq_row in self.iter_by_col_eq(&ST_SEQUENCES_ID, &st_sequences_table_id_col, table_id.into())? {
+                for seq_row in self.iter_by_col_eq(&ST_SEQUENCES_ID, st_sequences_table_id_col, table_id.into())? {
                     let seq_row = seq_row.view();
                     let seq_row = StSequenceRow::try_from(seq_row)?;
                     if seq_row.col_id != col.col_id {
@@ -1473,10 +1473,10 @@ impl Inner {
     fn iter_by_col_eq(
         &self,
         table_id: &TableId,
-        col_id: &ColId,
+        cols: impl Into<NonEmpty<ColId>>,
         value: AlgebraicValue,
     ) -> super::Result<IterByColEq<'_>> {
-        self.iter_by_col_range(table_id, col_id, value)
+        self.iter_by_col_range(table_id, cols.into(), value)
     }
 
     /// Returns an iterator,
@@ -1485,7 +1485,7 @@ impl Inner {
     fn iter_by_col_range<'a, R: RangeBounds<AlgebraicValue>>(
         &'a self,
         table_id: &TableId,
-        col_id: &ColId,
+        cols: NonEmpty<ColId>,
         range: R,
     ) -> super::Result<IterByColRange<'a, R>> {
         // We have to index_seek in both the committed state and the current tx state.
@@ -1502,7 +1502,7 @@ impl Inner {
         if let Some(inserted_rows) = self
             .tx_state
             .as_ref()
-            .and_then(|tx_state| tx_state.index_seek(table_id, NonEmpty::new(*col_id), &range))
+            .and_then(|tx_state| tx_state.index_seek(table_id, &cols, &range))
         {
             // The current transaction has modified this table, and the table is indexed.
             let tx_state = self.tx_state.as_ref().unwrap();
@@ -1510,23 +1510,18 @@ impl Inner {
                 table_id: *table_id,
                 tx_state,
                 inserted_rows,
-                committed_rows: self
-                    .committed_state
-                    .index_seek(table_id, NonEmpty::new(*col_id), &range),
+                committed_rows: self.committed_state.index_seek(table_id, &cols, &range),
                 committed_state: &self.committed_state,
             }))
         } else {
             // Either the current transaction has not modified this table, or the table is not
             // indexed.
-            match self
-                .committed_state
-                .index_seek(table_id, NonEmpty::new(*col_id), &range)
-            {
+            match self.committed_state.index_seek(table_id, &cols, &range) {
                 //If we don't have `self.tx_state` yet is likely we are running the bootstrap process
                 Some(committed_rows) => match self.tx_state.as_ref() {
                     None => Ok(IterByColRange::Scan(ScanIterByColRange {
                         range,
-                        col_id: *col_id,
+                        cols: NonEmpty::collect(cols.into_iter().map(|col| col.0)).unwrap(),
                         scan_iter: self.iter(table_id)?,
                     })),
                     Some(tx_state) => Ok(IterByColRange::CommittedIndex(CommittedIndexIter {
@@ -1538,7 +1533,7 @@ impl Inner {
                 },
                 None => Ok(IterByColRange::Scan(ScanIterByColRange {
                     range,
-                    col_id: *col_id,
+                    cols: NonEmpty::collect(cols.into_iter().map(|col| col.0)).unwrap(),
                     scan_iter: self.iter(table_id)?,
                 })),
             }
@@ -1887,7 +1882,7 @@ impl<R: RangeBounds<AlgebraicValue>> Iterator for IterByColRange<'_, R> {
 
 pub struct ScanIterByColRange<'a, R: RangeBounds<AlgebraicValue>> {
     scan_iter: Iter<'a>,
-    col_id: ColId,
+    cols: NonEmpty<u32>,
     range: R,
 }
 
@@ -1898,8 +1893,8 @@ impl<R: RangeBounds<AlgebraicValue>> Iterator for ScanIterByColRange<'_, R> {
     fn next(&mut self) -> Option<Self::Item> {
         for data_ref in &mut self.scan_iter {
             let row = data_ref.view();
-            let value = &row.elements[self.col_id.0 as usize];
-            if self.range.contains(value) {
+            let value = row.project_not_empty(&self.cols).unwrap();
+            if self.range.contains(&value) {
                 return Some(data_ref);
             }
         }
@@ -1920,20 +1915,20 @@ impl TxDatastore for Locking {
         &'a self,
         tx: &'a Self::TxId,
         table_id: TableId,
-        col_id: ColId,
+        cols: NonEmpty<ColId>,
         range: R,
     ) -> super::Result<Self::IterByColRange<'a, R>> {
-        self.iter_by_col_range_mut_tx(tx, table_id, col_id, range)
+        self.iter_by_col_range_mut_tx(tx, table_id, cols, range)
     }
 
     fn iter_by_col_eq_tx<'a>(
         &'a self,
         tx: &'a Self::TxId,
         table_id: TableId,
-        col_id: ColId,
+        cols: NonEmpty<ColId>,
         value: AlgebraicValue,
     ) -> super::Result<Self::IterByColEq<'a>> {
-        self.iter_by_col_eq_mut_tx(tx, table_id, col_id, value)
+        self.iter_by_col_eq_mut_tx(tx, table_id, cols, value)
     }
 
     fn get_tx<'a>(
@@ -2059,20 +2054,20 @@ impl MutTxDatastore for Locking {
         &'a self,
         tx: &'a Self::MutTxId,
         table_id: TableId,
-        col_id: ColId,
+        cols: impl Into<NonEmpty<ColId>>,
         range: R,
     ) -> super::Result<Self::IterByColRange<'a, R>> {
-        tx.lock.iter_by_col_range(&table_id, &col_id, range)
+        tx.lock.iter_by_col_range(&table_id, cols.into(), range)
     }
 
     fn iter_by_col_eq_mut_tx<'a>(
         &'a self,
         tx: &'a Self::MutTxId,
         table_id: TableId,
-        col_id: ColId,
+        cols: impl Into<NonEmpty<ColId>>,
         value: AlgebraicValue,
     ) -> super::Result<Self::IterByColEq<'a>> {
-        tx.lock.iter_by_col_eq(&table_id, &col_id, value)
+        tx.lock.iter_by_col_eq(&table_id, cols, value)
     }
 
     fn get_mut_tx<'a>(
diff --git a/crates/core/src/db/datastore/locking_tx_datastore/table.rs b/crates/core/src/db/datastore/locking_tx_datastore/table.rs
index 5653bd601f3..b93d93e3110 100644
--- a/crates/core/src/db/datastore/locking_tx_datastore/table.rs
+++ b/crates/core/src/db/datastore/locking_tx_datastore/table.rs
@@ -63,9 +63,9 @@ impl Table {
     /// Matching is defined by `Ord for AlgebraicValue`.
     pub(crate) fn index_seek(
         &self,
-        cols: NonEmpty<ColId>,
+        cols: &NonEmpty<ColId>,
         range: &impl RangeBounds<AlgebraicValue>,
     ) -> Option<BTreeIndexRangeIter<'_>> {
-        self.indexes.get(&cols).map(|index| index.seek(range))
+        self.indexes.get(cols).map(|index| index.seek(range))
     }
 }
diff --git a/crates/core/src/db/datastore/traits.rs b/crates/core/src/db/datastore/traits.rs
index 3e21926da48..8ff1b27dc6c 100644
--- a/crates/core/src/db/datastore/traits.rs
+++ b/crates/core/src/db/datastore/traits.rs
@@ -1,5 +1,6 @@
 use crate::db::relational_db::ST_TABLES_ID;
 use core::fmt;
+use derive_more::From;
 use nonempty::NonEmpty;
 use spacetimedb_lib::auth::{StAccess, StTableType};
 use spacetimedb_lib::relation::{DbTable, FieldName, FieldOnly, Header, TableField};
@@ -12,10 +13,17 @@ use std::{ops::RangeBounds, sync::Arc};
 use super::{system_tables::StTableRow, Result};
 
 /// The `id` for [Sequence]
-#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq, Ord, PartialOrd, From)]
 pub struct TableId(pub(crate) u32);
-#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq, Ord, PartialOrd, From)]
 pub struct ColId(pub(crate) u32);
+
+impl From<ColId> for NonEmpty<ColId> {
+    fn from(value: ColId) -> Self {
+        NonEmpty::new(value)
+    }
+}
+
 #[derive(Debug, Copy, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 pub struct IndexId(pub(crate) u32);
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -436,7 +444,7 @@ pub trait TxDatastore: DataRow + Tx {
         &'a self,
         tx: &'a Self::TxId,
         table_id: TableId,
-        col_id: ColId,
+        cols: NonEmpty<ColId>,
         range: R,
     ) -> Result<Self::IterByColRange<'a, R>>;
 
@@ -444,7 +452,7 @@ pub trait TxDatastore: DataRow + Tx {
         &'a self,
         tx: &'a Self::TxId,
         table_id: TableId,
-        col_id: ColId,
+        cols: NonEmpty<ColId>,
         value: AlgebraicValue,
     ) -> Result<Self::IterByColEq<'a>>;
 
@@ -504,14 +512,14 @@ pub trait MutTxDatastore: TxDatastore + MutTx {
         &'a self,
         tx: &'a Self::MutTxId,
         table_id: TableId,
-        col_id: ColId,
+        cols: impl Into<NonEmpty<ColId>>,
         range: R,
     ) -> Result<Self::IterByColRange<'a, R>>;
     fn iter_by_col_eq_mut_tx<'a>(
         &'a self,
         tx: &'a Self::MutTxId,
         table_id: TableId,
-        col_id: ColId,
+        cols: impl Into<NonEmpty<ColId>>,
         value: AlgebraicValue,
     ) -> Result<Self::IterByColEq<'a>>;
     fn get_mut_tx<'a>(
diff --git a/crates/core/src/db/relational_db.rs b/crates/core/src/db/relational_db.rs
index 0ada40ba803..5860e3f1fd8 100644
--- a/crates/core/src/db/relational_db.rs
+++ b/crates/core/src/db/relational_db.rs
@@ -473,16 +473,15 @@ impl RelationalDB {
     /// where the column data identified by `cols` matches `value`.
     ///
     /// Matching is defined by `Ord for AlgebraicValue`.
-    #[tracing::instrument(skip(self, tx, value))]
+    #[tracing::instrument(skip_all)]
     pub fn iter_by_col_eq<'a>(
         &'a self,
         tx: &'a MutTxId,
-        table_id: u32,
-        col_id: u32,
+        table_id: impl Into<TableId>,
+        cols: impl Into<NonEmpty<ColId>>,
         value: AlgebraicValue,
     ) -> Result<IterByColEq<'a>, DBError> {
-        self.inner
-            .iter_by_col_eq_mut_tx(tx, TableId(table_id), ColId(col_id), value)
+        self.inner.iter_by_col_eq_mut_tx(tx, table_id.into(), cols, value)
     }
 
     /// Returns an iterator,
@@ -493,12 +492,11 @@ impl RelationalDB {
     pub fn iter_by_col_range<'a, R: RangeBounds<AlgebraicValue>>(
         &'a self,
         tx: &'a MutTxId,
-        table_id: u32,
-        col_id: u32,
+        table_id: impl Into<TableId>,
+        cols: impl Into<NonEmpty<ColId>>,
         range: R,
     ) -> Result<IterByColRange<'a, R>, DBError> {
-        self.inner
-            .iter_by_col_range_mut_tx(tx, TableId(table_id), ColId(col_id), range)
+        self.inner.iter_by_col_range_mut_tx(tx, table_id.into(), cols, range)
     }
 
     #[tracing::instrument(skip(self, tx, row))]
@@ -642,11 +640,13 @@ mod tests {
     use std::sync::{Arc, Mutex};
 
     use crate::address::Address;
+    use crate::db::datastore::locking_tx_datastore::IterByColEq;
     use crate::db::datastore::system_tables::StIndexRow;
     use crate::db::datastore::system_tables::StSequenceRow;
     use crate::db::datastore::system_tables::StTableRow;
     use crate::db::datastore::system_tables::ST_INDEXES_ID;
     use crate::db::datastore::system_tables::ST_SEQUENCES_ID;
+    use crate::db::datastore::traits::ColId;
     use crate::db::datastore::traits::ColumnDef;
     use crate::db::datastore::traits::IndexDef;
     use crate::db::datastore::traits::TableDef;
@@ -663,6 +663,33 @@ mod tests {
     use spacetimedb_lib::{AlgebraicType, AlgebraicValue, ProductType};
     use spacetimedb_sats::product;
 
+    fn column(name: &str, ty: AlgebraicType) -> ColumnDef {
+        ColumnDef {
+            col_name: name.to_string(),
+            col_type: ty,
+            is_autoinc: false,
+        }
+    }
+
+    fn index(name: &str, cols: &[u32]) -> IndexDef {
+        IndexDef {
+            table_id: 0,
+            cols: NonEmpty::collect(cols.iter().copied()).unwrap(),
+            name: name.to_string(),
+            is_unique: false,
+        }
+    }
+
+    fn table(name: &str, columns: Vec<ColumnDef>, indexes: Vec<IndexDef>) -> TableDef {
+        TableDef {
+            table_name: name.to_string(),
+            columns,
+            indexes,
+            table_type: StTableType::User,
+            table_access: StAccess::Public,
+        }
+    }
+
     #[test]
     fn test() -> ResultTest<()> {
         let (stdb, _tmp_dir) = make_test_db()?;
@@ -816,7 +843,7 @@ mod tests {
         stdb.insert(&mut tx, table_id, product![AlgebraicValue::I32(1)])?;
 
         let mut rows = stdb
-            .iter_by_col_range(&tx, table_id, 0, AlgebraicValue::I32(0)..)?
+            .iter_by_col_range(&tx, table_id, ColId(0), AlgebraicValue::I32(0)..)?
             .map(|r| *r.view().elements[0].as_i32().unwrap())
             .collect::<Vec<i32>>();
         rows.sort();
@@ -842,7 +869,7 @@ mod tests {
 
         let tx = stdb.begin_tx();
         let mut rows = stdb
-            .iter_by_col_range(&tx, table_id, 0, AlgebraicValue::I32(0)..)?
+            .iter_by_col_range(&tx, table_id, ColId(0), AlgebraicValue::I32(0)..)?
             .map(|r| *r.view().elements[0].as_i32().unwrap())
             .collect::<Vec<i32>>();
         rows.sort();
@@ -922,7 +949,7 @@ mod tests {
         stdb.insert(&mut tx, table_id, product![AlgebraicValue::I64(0)])?;
 
         let mut rows = stdb
-            .iter_by_col_range(&tx, table_id, 0, AlgebraicValue::I64(0)..)?
+            .iter_by_col_range(&tx, table_id, ColId(0), AlgebraicValue::I64(0)..)?
             .map(|r| *r.view().elements[0].as_i64().unwrap())
             .collect::<Vec<i64>>();
         rows.sort();
@@ -957,7 +984,7 @@ mod tests {
         stdb.insert(&mut tx, table_id, product![AlgebraicValue::I64(6)])?;
 
         let mut rows = stdb
-            .iter_by_col_range(&tx, table_id, 0, AlgebraicValue::I64(0)..)?
+            .iter_by_col_range(&tx, table_id, ColId(0), AlgebraicValue::I64(0)..)?
             .map(|r| *r.view().elements[0].as_i64().unwrap())
             .collect::<Vec<i64>>();
         rows.sort();
@@ -991,7 +1018,7 @@ mod tests {
         stdb.insert(&mut tx, table_id, product![AlgebraicValue::I64(0)])?;
 
         let mut rows = stdb
-            .iter_by_col_range(&tx, table_id, 0, AlgebraicValue::I64(0)..)?
+            .iter_by_col_range(&tx, table_id, ColId(0), AlgebraicValue::I64(0)..)?
             .map(|r| *r.view().elements[0].as_i64().unwrap())
             .collect::<Vec<i64>>();
         rows.sort();
@@ -1009,7 +1036,7 @@ mod tests {
         stdb.insert(&mut tx, table_id, product![AlgebraicValue::I64(0)])?;
 
         let mut rows = stdb
-            .iter_by_col_range(&tx, table_id, 0, AlgebraicValue::I64(0)..)?
+            .iter_by_col_range(&tx, table_id, ColId(0), AlgebraicValue::I64(0)..)?
             .map(|r| *r.view().elements[0].as_i64().unwrap())
             .collect::<Vec<i64>>();
         rows.sort();
@@ -1051,7 +1078,7 @@ mod tests {
         stdb.insert(&mut tx, table_id, product![AlgebraicValue::I64(1)])?;
 
         let mut rows = stdb
-            .iter_by_col_range(&tx, table_id, 0, AlgebraicValue::I64(0)..)?
+            .iter_by_col_range(&tx, table_id, ColId(0), AlgebraicValue::I64(0)..)?
             .map(|r| *r.view().elements[0].as_i64().unwrap())
             .collect::<Vec<i64>>();
         rows.sort();
@@ -1143,7 +1170,7 @@ mod tests {
         stdb.insert(&mut tx, table_id, product![AlgebraicValue::I64(0)])?;
 
         let mut rows = stdb
-            .iter_by_col_range(&tx, table_id, 0, AlgebraicValue::I64(0)..)?
+            .iter_by_col_range(&tx, table_id, ColId(0), AlgebraicValue::I64(0)..)?
             .map(|r| *r.view().elements[0].as_i64().unwrap())
             .collect::<Vec<i64>>();
         rows.sort();
@@ -1280,6 +1307,59 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_multi_column_index() -> ResultTest<()> {
+        let (stdb, _tmp_dir) = make_test_db()?;
+
+        let columns = vec![
+            column("a", AlgebraicType::U64),
+            column("b", AlgebraicType::U64),
+            column("c", AlgebraicType::U64),
+        ];
+
+        let indexes = vec![index("0", &[0, 1])];
+        let schema = table("t", columns, indexes);
+
+        let mut tx = stdb.begin_tx();
+        let table_id = stdb.create_table(&mut tx, schema)?;
+
+        stdb.insert(
+            &mut tx,
+            table_id,
+            product![AlgebraicValue::U64(0), AlgebraicValue::U64(0), AlgebraicValue::U64(1)],
+        )?;
+        stdb.insert(
+            &mut tx,
+            table_id,
+            product![AlgebraicValue::U64(0), AlgebraicValue::U64(1), AlgebraicValue::U64(2)],
+        )?;
+        stdb.insert(
+            &mut tx,
+            table_id,
+            product![AlgebraicValue::U64(1), AlgebraicValue::U64(2), AlgebraicValue::U64(2)],
+        )?;
+
+        let cols: NonEmpty<ColId> = NonEmpty::collect(vec![ColId(0), ColId(1)]).unwrap();
+        let value: AlgebraicValue = product![AlgebraicValue::U64(0), AlgebraicValue::U64(1)].into();
+
+        let IterByColEq::Index(mut iter) = stdb.iter_by_col_eq(&tx, table_id, cols, value)? else {
+            panic!("expected index iterator");
+        };
+
+        let Some(row) = iter.next() else {
+            panic!("expected non-empty iterator");
+        };
+
+        assert_eq!(
+            row.view(),
+            &product![AlgebraicValue::U64(0), AlgebraicValue::U64(1), AlgebraicValue::U64(2)]
+        );
+
+        // iter should only return a single row, so this count should now be 0.
+        assert_eq!(iter.count(), 0);
+        Ok(())
+    }
+
     // #[test]
     // fn test_rename_column() -> ResultTest<()> {
     //     let (mut stdb, _tmp_dir) = make_test_db()?;
diff --git a/crates/core/src/host/instance_env.rs b/crates/core/src/host/instance_env.rs
index d9e5fb296c8..11046db1ee1 100644
--- a/crates/core/src/host/instance_env.rs
+++ b/crates/core/src/host/instance_env.rs
@@ -7,7 +7,7 @@ use std::sync::Arc;
 use crate::database_instance_context::DatabaseInstanceContext;
 use crate::database_logger::{BacktraceProvider, LogLevel, Record};
 use crate::db::datastore::locking_tx_datastore::MutTxId;
-use crate::db::datastore::traits::{DataRow, IndexDef};
+use crate::db::datastore::traits::{ColId, DataRow, IndexDef};
 use crate::error::{IndexError, NodesError};
 use crate::util::ResultInspectExt;
 
@@ -152,7 +152,7 @@ impl InstanceEnv {
         let eq_value = stdb.decode_column(tx, table_id, col_id, value)?;
 
         // Find all rows in the table where the column data equates to `value`.
-        let seek = stdb.iter_by_col_eq(tx, table_id, col_id, eq_value)?;
+        let seek = stdb.iter_by_col_eq(tx, table_id, ColId(col_id), eq_value)?;
         let seek = seek.map(|x| stdb.data_to_owned(x).into()).collect::<Vec<_>>();
 
         // Delete them and count how many we deleted and error if none.
@@ -313,7 +313,7 @@ impl InstanceEnv {
 
         // Find all rows in the table where the column data matches `value`.
         // Concatenate and return these rows using bsatn encoding.
-        let results = stdb.iter_by_col_eq(tx, table_id, col_id, value)?;
+        let results = stdb.iter_by_col_eq(tx, table_id, ColId(col_id), value)?;
         let mut bytes = Vec::new();
         for result in results {
             bsatn::to_writer(&mut bytes, result.view()).unwrap();
diff --git a/crates/core/src/vm.rs b/crates/core/src/vm.rs
index fb9f9099e16..de907834a9c 100644
--- a/crates/core/src/vm.rs
+++ b/crates/core/src/vm.rs
@@ -1,7 +1,7 @@
 //! The [DbProgram] that execute arbitrary queries & code against the database.
 use crate::db::cursor::{CatalogCursor, IndexCursor, TableCursor};
 use crate::db::datastore::locking_tx_datastore::{IterByColEq, MutTxId};
-use crate::db::datastore::traits::{ColumnDef, IndexDef, TableDef};
+use crate::db::datastore::traits::{ColId, ColumnDef, IndexDef, TableDef};
 use crate::db::relational_db::RelationalDB;
 use itertools::Itertools;
 use nonempty::NonEmpty;
@@ -153,7 +153,7 @@ fn iter_by_col_range<'a>(
     col_id: u32,
     range: impl RangeBounds<AlgebraicValue> + 'a,
 ) -> Result<Box<dyn RelOps + 'a>, ErrorVm> {
-    let iter = db.iter_by_col_range(tx, table.table_id, col_id, range)?;
+    let iter = db.iter_by_col_range(tx, table.table_id, ColId(col_id), range)?;
     Ok(Box::new(IndexCursor::new(table, iter)?) as Box<IterRows<'_>>)
 }
 
@@ -225,7 +225,7 @@ impl<'a, Rhs: RelOps> RelOps for IndexSemiJoin<'a, Rhs> {
                     let table_id = self.index_table;
                     let col_id = self.index_col;
                     let value = value.clone();
-                    let mut index_iter = self.db.iter_by_col_eq(self.tx, table_id, col_id, value)?;
+                    let mut index_iter = self.db.iter_by_col_eq(self.tx, table_id, ColId(col_id), value)?;
                     if let Some(value) = index_iter.next() {
                         self.index_iter = Some(index_iter);
                         return Ok(Some(value.into()));