part: allocate watch channels lazily

any_table.go

@@ -22,8 +22,13 @@ func (t AnyTable) NumObjects(txn ReadTxn) int {
 }
 
 func (t AnyTable) All(txn ReadTxn) iter.Seq2[any, Revision] {
-	all, _ := t.AllWatch(txn)
-	return all
+	indexTxn := txn.mustIndexReadTxn(t.Meta, PrimaryIndexPos)
+	all := indexTxn.allNoWatch()
+	return func(yield func(any, Revision) bool) {
+		all.All(func(_ []byte, iobj object) bool {
+			return yield(iobj.data, iobj.revision)
+		})
+	}
 }
 
 func (t AnyTable) AllWatch(txn ReadTxn) (iter.Seq2[any, Revision], <-chan struct{}) {
@@ -42,7 +47,7 @@ func (t AnyTable) UnmarshalYAML(data []byte) (any, error) {
 
 func (t AnyTable) Insert(txn WriteTxn, obj any) (old any, hadOld bool, err error) {
 	var iobj object
-	iobj, hadOld, _, err = txn.unwrap().insert(t.Meta, Revision(0), obj)
+	iobj, hadOld, _, err = txn.unwrap().insert(t.Meta, Revision(0), obj, false)
 	if hadOld {
 		old = iobj.data
 	}
@@ -63,7 +68,7 @@ func (t AnyTable) Get(txn ReadTxn, index string, key string) (any, Revision, boo
 	if err != nil {
 		return nil, 0, false, err
 	}
-	obj, _, found := itxn.get(rawKey)
+	obj, found := itxn.getNoWatch(rawKey)
 	if found {
 		return obj.data, obj.revision, found, nil
 	}
@@ -75,7 +80,7 @@ func (t AnyTable) Prefix(txn ReadTxn, index string, key string) (iter.Seq2[any,
 	if err != nil {
 		return nil, err
 	}
-	iter, _ := itxn.prefix(rawKey)
+	iter := itxn.prefixNoWatch(rawKey)
 	return objSeq[any](iter), nil
 }
 
@@ -84,7 +89,7 @@ func (t AnyTable) LowerBound(txn ReadTxn, index string, key string) (iter.Seq2[a
 	if err != nil {
 		return nil, err
 	}
-	iter, _ := itxn.lowerBound(rawKey)
+	iter := itxn.lowerBoundNoWatch(rawKey)
 	return objSeq[any](iter), nil
 }
 
@@ -93,7 +98,7 @@ func (t AnyTable) List(txn ReadTxn, index string, key string) (iter.Seq2[any, Re
 	if err != nil {
 		return nil, err
 	}
-	iter, _ := itxn.list(rawKey)
+	iter := itxn.listNoWatch(rawKey)
 	return objSeq[any](iter), nil
 }
 

deletetracker.go

@@ -37,7 +37,7 @@ func (dt *deleteTracker[Obj]) getRevision() uint64 {
 // called!
 func (dt *deleteTracker[Obj]) deleted(txn ReadTxn, minRevision Revision) *iterator[Obj] {
 	indexEntry := txn.root()[dt.table.tablePos()].indexes[GraveyardRevisionIndexPos]
-	objs, _ := indexEntry.lowerBoundNext(index.Uint64(minRevision))
+	objs := indexEntry.lowerBoundNextNoWatch(index.Uint64(minRevision))
 	return &iterator[Obj]{objs}
 }
 

graveyard.go

@@ -63,7 +63,7 @@ func graveyardWorker(db *DB, ctx context.Context, gcRateLimitInterval time.Durat
 			// to the low watermark.
 			indexTree := rtxn.mustIndexReadTxn(table.meta, GraveyardRevisionIndexPos)
 
-			iter, _ := indexTree.all()
+			iter := indexTree.allNoWatch()
 			for key, obj := range iter.All {
 				if obj.revision > lowWatermark {
 					break

http.go

@@ -131,9 +131,9 @@ type QueryResponse struct {
 func runQuery(reader tableIndexReader, lowerbound bool, queryKey index.Key, onObject func(object) error) {
 	var iter tableIndexIterator
 	if lowerbound {
-		iter, _ = reader.lowerBound(queryKey)
+		iter = reader.lowerBoundNoWatch(queryKey)
 	} else {
-		iter, _ = reader.list(queryKey)
+		iter = reader.listNoWatch(queryKey)
 	}
 	for _, obj := range iter.All {
 		if err := onObject(obj); err != nil {

iterator.go

@@ -172,7 +172,7 @@ func (it *changeIterator[Obj]) refresh(txn ReadTxn) {
 		panic(fmt.Sprintf("Table[%T].Changes().Next() called with the target table locked. This is not supported.", obj))
 	}
 	indexEntry := tableEntry.indexes[RevisionIndexPos]
-	updated, _ := indexEntry.lowerBoundNext(index.Uint64(it.revision + 1))
+	updated := indexEntry.lowerBoundNextNoWatch(index.Uint64(it.revision + 1))
 	updateIter := &iterator[Obj]{updated}
 	deleteIter := it.dt.deleted(txn, it.deleteRevision+1)
 	it.iter = newDualIterator(deleteIter, updateIter)

lpm_index.go

@@ -203,6 +203,10 @@ func (l lpmIndex) all() (tableIndexIterator, <-chan struct{}) {
 	return newLPMIterator(l.lpm.All()), l.watch
 }
 
+func (l lpmIndex) allNoWatch() tableIndexIterator {
+	return newLPMIterator(l.lpm.All())
+}
+
 // get implements tableIndex.
 func (l lpmIndex) get(ikey index.Key) (object, <-chan struct{}, bool) {
 	entry, found := l.lpm.Lookup(ikey)
@@ -215,6 +219,14 @@ func (l lpmIndex) get(ikey index.Key) (object, <-chan struct{}, bool) {
 	return object{}, l.watch, false
 }
 
+func (l lpmIndex) getNoWatch(ikey index.Key) (object, bool) {
+	entry, found := l.lpm.Lookup(ikey)
+	if !found {
+		return object{}, false
+	}
+	return entry.first()
+}
+
 // len implements tableIndex.
 func (l lpmIndex) len() int {
 	return l.size
@@ -229,16 +241,32 @@ func (l lpmIndex) list(key index.Key) (tableIndexIterator, <-chan struct{}) {
 	return &entry, l.watch
 }
 
+func (l lpmIndex) listNoWatch(key index.Key) tableIndexIterator {
+	entry, found := l.lpm.Lookup(key)
+	if !found || entry.len() == 0 {
+		return emptyTableIndexIterator
+	}
+	return &entry
+}
+
 // lowerBound implements tableIndex.
 func (l lpmIndex) lowerBound(key index.Key) (tableIndexIterator, <-chan struct{}) {
 	return newLPMIterator(l.lpm.LowerBound(key)), l.watch
 }
 
+func (l lpmIndex) lowerBoundNoWatch(key index.Key) tableIndexIterator {
+	return newLPMIterator(l.lpm.LowerBound(key))
+}
+
 // lowerBoundNext implements tableIndexTxn.
 func (l lpmIndex) lowerBoundNext(key index.Key) (func() ([]byte, object, bool), <-chan struct{}) {
 	return newLPMNextFunc(l.lpm.LowerBound(key)), l.watch
 }
 
+func (l lpmIndex) lowerBoundNextNoWatch(key index.Key) func() ([]byte, object, bool) {
+	return newLPMNextFunc(l.lpm.LowerBound(key))
+}
+
 // objectToKey implements tableIndex.
 func (l lpmIndex) objectToKey(obj object) index.Key {
 	return l.objectToKeys(obj).First()
@@ -249,6 +277,10 @@ func (l lpmIndex) prefix(key index.Key) (tableIndexIterator, <-chan struct{}) {
 	return newLPMIterator(l.lpm.Prefix(key)), l.watch
 }
 
+func (l lpmIndex) prefixNoWatch(key index.Key) tableIndexIterator {
+	return newLPMIterator(l.lpm.Prefix(key))
+}
+
 // rootWatch implements tableIndex.
 func (l lpmIndex) rootWatch() <-chan struct{} {
 	return l.watch
@@ -287,6 +319,10 @@ func (l *lpmIndexTxn) all() (tableIndexIterator, <-chan struct{}) {
 	return newLPMIterator(l.tx.All()), l.index.watch
 }
 
+func (l *lpmIndexTxn) allNoWatch() tableIndexIterator {
+	return newLPMIterator(l.tx.All())
+}
+
 // commit implements tableIndexTxn.
 func (l *lpmIndexTxn) commit() (tableIndex, tableIndexTxnNotify) {
 	lpm := l.tx.Commit()
@@ -311,11 +347,19 @@ func (l *lpmIndexTxn) insert(key index.Key, obj object) (old object, hadOld bool
 	panic("LPM index cannot be the primary index")
 }
 
+func (l *lpmIndexTxn) insertNoWatch(key index.Key, obj object) (old object, hadOld bool) {
+	panic("LPM index cannot be the primary index")
+}
+
 // modify implements tableIndexTxn.
 func (l *lpmIndexTxn) modify(key index.Key, obj object, mod func(old, new object) object) (old object, newObj object, hadOld bool, watch <-chan struct{}) {
 	panic("LPM index cannot be the primary index")
 }
 
+func (l *lpmIndexTxn) modifyNoWatch(key index.Key, obj object, mod func(old, new object) object) (old object, newObj object, hadOld bool) {
+	panic("LPM index cannot be the primary index")
+}
+
 // get implements tableIndexTxn.
 func (l *lpmIndexTxn) get(key index.Key) (object, <-chan struct{}, bool) {
 	entry, found := l.tx.Lookup(key)
@@ -328,6 +372,14 @@ func (l *lpmIndexTxn) get(key index.Key) (object, <-chan struct{}, bool) {
 	return object{}, l.index.watch, false
 }
 
+func (l *lpmIndexTxn) getNoWatch(key index.Key) (object, bool) {
+	entry, found := l.tx.Lookup(key)
+	if !found {
+		return object{}, false
+	}
+	return entry.first()
+}
+
 // len implements tableIndexTxn.
 func (l *lpmIndexTxn) len() int {
 	return l.size
@@ -342,16 +394,32 @@ func (l *lpmIndexTxn) list(key index.Key) (tableIndexIterator, <-chan struct{})
 	return &entry, l.index.watch
 }
 
+func (l *lpmIndexTxn) listNoWatch(key index.Key) tableIndexIterator {
+	entry, found := l.tx.Lookup(key)
+	if !found || entry.len() == 0 {
+		return emptyTableIndexIterator
+	}
+	return &entry
+}
+
 // lowerBound implements tableIndexTxn.
 func (l *lpmIndexTxn) lowerBound(key index.Key) (tableIndexIterator, <-chan struct{}) {
 	return newLPMIterator(l.tx.LowerBound(key)), l.index.watch
 }
 
+func (l *lpmIndexTxn) lowerBoundNoWatch(key index.Key) tableIndexIterator {
+	return newLPMIterator(l.tx.LowerBound(key))
+}
+
 // lowerBoundNext implements tableIndexTxn.
 func (l *lpmIndexTxn) lowerBoundNext(key index.Key) (func() ([]byte, object, bool), <-chan struct{}) {
 	return newLPMNextFunc(l.tx.LowerBound(key)), l.index.watch
 }
 
+func (l *lpmIndexTxn) lowerBoundNextNoWatch(key index.Key) func() ([]byte, object, bool) {
+	return newLPMNextFunc(l.tx.LowerBound(key))
+}
+
 // notify implements tableIndexTxn.
 func (l *lpmIndexTxn) notify() {
 	if l.index.watch != nil {
@@ -370,6 +438,10 @@ func (l *lpmIndexTxn) prefix(key index.Key) (tableIndexIterator, <-chan struct{}
 	return newLPMIterator(l.tx.Prefix(key)), l.index.watch
 }
 
+func (l *lpmIndexTxn) prefixNoWatch(key index.Key) tableIndexIterator {
+	return newLPMIterator(l.tx.Prefix(key))
+}
+
 // reindex implements tableIndexTxn.
 func (l *lpmIndexTxn) reindex(primaryKey index.Key, old object, new object) {
 	var newKeys index.KeySet

part/iterator.go

@@ -138,7 +138,7 @@ func newIterator[T any](start *header[T]) Iterator[T] {
 	return Iterator[T]{start: start}
 }
 
-func prefixSearch[T any](root *header[T], rootWatch <-chan struct{}, prefix []byte) (Iterator[T], <-chan struct{}) {
+func prefixSearch[T any](root *header[T], rootWatch *watchState, prefix []byte) (Iterator[T], *watchState) {
 	if root == nil {
 		return newIterator[T](nil), rootWatch
 	}

part/map.go

@@ -75,7 +75,7 @@ func (m Map[K, V]) Get(key K) (value V, found bool) {
 	if !m.hasTree {
 		return
 	}
-	kv, _, found := m.tree.Get(m.keyToBytes(key))
+	kv, found := m.tree.Get(m.keyToBytes(key))
 	return kv.Value, found
 }
 
@@ -178,7 +178,7 @@ func (m Map[K, V]) Prefix(prefix K) iter.Seq2[K, V] {
 	if !m.hasTree {
 		return toSeq2[K, V](Iterator[mapKVPair[K, V]]{})
 	}
-	iter, _ := m.tree.Prefix(m.keyToBytes(prefix))
+	iter := m.tree.Prefix(m.keyToBytes(prefix))
 	return toSeq2(iter)
 }
 
@@ -442,14 +442,14 @@ func (txn MapTxn[K, V]) Delete(key K) bool {
 
 // Get a value from the map by its key.
 func (txn MapTxn[K, V]) Get(key K) (value V, found bool) {
-	kv, _, found := txn.txn.Get(txn.bytesFromKeyFunc(key))
+	kv, found := txn.txn.Get(txn.bytesFromKeyFunc(key))
 	return kv.Value, found
 }
 
 // Prefix iterates in order over all keys that start with
 // the given prefix.
 func (txn MapTxn[K, V]) Prefix(prefix K) iter.Seq2[K, V] {
-	iter, _ := txn.txn.Prefix(txn.bytesFromKeyFunc(prefix))
+	iter := txn.txn.Prefix(txn.bytesFromKeyFunc(prefix))
 	return toSeq2(iter)
 }