perf: maintain warm indexes and counts across exclusive commits (#22)

README.md

@@ -546,6 +546,10 @@ Rollback, compare failure, DDL, restart, and ambiguous transport errors never
 leave speculative ownership entries. NULL-bearing keys remain exempt; composite
 keys use the constraint encoder, not the SELECT candidate encoder. Expression
 UNIQUE constraints retain scan validation until evaluator determinism is proven.
+Exclusive-gate commits also maintain warm column-based SELECT indexes and exact
+counts incrementally. Shared-gate optimistic commits still invalidate those read
+caches to avoid publishing them out of durable CAS order. Expression read indexes
+are rebuilt after such commits instead of assuming evaluator determinism.
 
 **PRAGMA:**
 ```sql

pkg/executor/cache_commit_test.go

@@ -0,0 +1,66 @@
+package executor
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+	"github.com/danfragoso/pizzasql-next/pkg/testkv"
+)
+
+func TestExclusiveCommitsMaintainWarmIndexesAndCounts(t *testing.T) {
+	kv := testkv.New(t)
+	pool := kv.Pool(8)
+	defer pool.Close()
+	schema := storage.NewSchemaManager(pool, "cache_commit")
+	table := storage.NewTableManager(pool, schema, "cache_commit")
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (pk TEXT PRIMARY KEY, v INTEGER, u INTEGER UNIQUE)")
+	execMust(t, e, "CREATE INDEX idx_v ON t(v)")
+	execMust(t, e, "INSERT INTO t VALUES ('a',1,10),('b',2,20)")
+	execMust(t, e, "SELECT pk FROM t WHERE v=1")
+	execMust(t, e, "SELECT pk FROM t WHERE u=10")
+	execMust(t, e, "SELECT COUNT(*) FROM t")
+	check := func(sql string, want [][]interface{}) {
+		t.Helper()
+		before := kv.RequestCount(10)
+		got := execMust(t, e, sql).Rows
+		if len(got) != len(want) || (len(got) > 0 && !reflect.DeepEqual(got, want)) {
+			t.Fatalf("%s: got %v want %v", sql, got, want)
+		}
+		if pages := kv.RequestCount(10) - before; pages != 0 {
+			t.Fatalf("%s rebuilt a warm cache (%d pages)", sql, pages)
+		}
+	}
+	execMust(t, e, "INSERT INTO t VALUES ('c',3,30)")
+	check("SELECT pk FROM t WHERE v=3", [][]interface{}{{"c"}})
+	check("SELECT COUNT(*) FROM t", [][]interface{}{{int64(3)}})
+	execMust(t, e, "UPDATE t SET v=4,u=40 WHERE pk='c'")
+	check("SELECT pk FROM t WHERE v=3", nil)
+	check("SELECT pk FROM t WHERE v=4", [][]interface{}{{"c"}})
+	check("SELECT pk FROM t WHERE u=30", nil)
+	check("SELECT pk FROM t WHERE u=40", [][]interface{}{{"c"}})
+	execMust(t, e, "UPDATE t SET pk='z' WHERE pk='c'")
+	check("SELECT pk FROM t WHERE v=4", [][]interface{}{{"z"}})
+	check("SELECT pk FROM t WHERE u=40", [][]interface{}{{"z"}})
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "DELETE FROM t WHERE pk='a'")
+	execMust(t, e, "INSERT INTO t VALUES ('a',9,90)")
+	execMust(t, e, "COMMIT")
+	check("SELECT pk FROM t WHERE v=1", nil)
+	check("SELECT pk FROM t WHERE v=9", [][]interface{}{{"a"}})
+	check("SELECT COUNT(*) FROM t", [][]interface{}{{int64(3)}})
+	execMust(t, e, "UPDATE t SET u=110-u WHERE pk IN ('a','b')")
+	check("SELECT pk FROM t WHERE u=20", [][]interface{}{{"a"}})
+	check("SELECT pk FROM t WHERE u=90", [][]interface{}{{"b"}})
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "INSERT INTO t VALUES ('c',3,30)")
+	execMust(t, e, "ROLLBACK")
+	check("SELECT pk FROM t WHERE v=3", nil)
+	execMust(t, e, "DELETE FROM t WHERE pk='b'")
+	check("SELECT pk FROM t WHERE u=90", nil)
+	check("SELECT COUNT(*) FROM t", [][]interface{}{{int64(2)}})
+	execMust(t, e, "DELETE FROM t")
+	check("SELECT pk FROM t WHERE v=4", nil)
+	check("SELECT COUNT(*) FROM t", [][]interface{}{{int64(0)}})
+}

pkg/storage/cache_commit.go

@@ -0,0 +1,151 @@
+package storage
+
+import "strings"
+
+type cachedRowChange struct {
+	pk     string
+	rowid  int64
+	absent bool
+	values map[string]string
+}
+
+type tableCachePlan struct {
+	table      string
+	rows       []cachedRowChange
+	indexes    []string
+	drop       []string
+	countDelta int
+}
+
+// Prepare under an exclusive table gate, before durable publication. Shared-gate
+// optimistic commits still invalidate: otherwise same-key commits could publish
+// derived state in the opposite order from their durable CAS operations.
+func (m *TableManager) prepareTableCache(table string, entries map[string]*overlayEntry, reads map[string]uint64) (*tableCachePlan, error) {
+	indexes, err := m.schema.ListTableIndexes(table)
+	if err != nil {
+		return nil, err
+	}
+	plan := &tableCachePlan{table: strings.ToLower(table)}
+	var warm []*Index
+	m.cacheMu.RLock()
+	for _, idx := range indexes {
+		name := strings.ToLower(idx.Name)
+		if _, ok := m.indexCache[name]; !ok {
+			continue
+		}
+		if !cacheableColumnIndex(idx) {
+			plan.drop = append(plan.drop, name)
+			continue
+		}
+		warm = append(warm, idx)
+		plan.indexes = append(plan.indexes, name)
+	}
+	m.cacheMu.RUnlock()
+	for key, entry := range entries {
+		if entry.absent {
+			if reads[key] != 0 {
+				plan.countDelta--
+			}
+		} else if reads[key] == 0 {
+			plan.countDelta++
+		}
+		if len(warm) == 0 {
+			continue
+		}
+		row := cachedRowChange{pk: strings.TrimPrefix(key, m.dataPrefix(table)), absent: entry.absent, values: make(map[string]string, len(warm))}
+		if !entry.absent {
+			row.rowid, _ = rowIDFromRow(entry.row)
+			for _, idx := range warm {
+				value, err := m.buildIndexValue(entry.row, idx.Columns)
+				if err != nil {
+					return nil, err
+				}
+				row.values[strings.ToLower(idx.Name)] = formatIndexValue(value)
+			}
+		}
+		plan.rows = append(plan.rows, row)
+	}
+	return plan, nil
+}
+
+func (m *TableManager) applyTableCache(plan *tableCachePlan) {
+	m.cacheMu.Lock()
+	defer m.cacheMu.Unlock()
+	for _, name := range plan.drop {
+		delete(m.indexCache, name)
+		delete(m.indexRowValues, name)
+		delete(m.indexTable, name)
+	}
+	keys := m.rowKeyCache[plan.table]
+	reverse := m.keyRowCache[plan.table]
+	// Remove every previous owner before adding any replacement (PK moves, rowid
+	// replacement and swaps must not depend on Go map iteration order).
+	for _, row := range plan.rows {
+		id, ok := reverse[row.pk]
+		if !ok {
+			continue
+		}
+		for _, name := range plan.indexes {
+			if value, ok := m.indexRowValues[name][id]; ok {
+				m.removeIndexEntryLocked(name, value, id)
+			}
+		}
+		delete(keys, id)
+		delete(reverse, row.pk)
+	}
+	for _, row := range plan.rows {
+		if row.absent {
+			continue
+		}
+		keys[row.rowid] = row.pk
+		reverse[row.pk] = row.rowid
+		for name, value := range row.values {
+			m.indexCache[name][value] = append(m.indexCache[name][value], row.rowid)
+			m.indexRowValues[name][row.rowid] = value
+		}
+	}
+	if m.countsInit[plan.table] {
+		m.counts[plan.table] += plan.countDelta
+	}
+}
+
+// Call with cacheMu held when publishing a complete durable index snapshot.
+func (m *TableManager) installIndexCacheLocked(name, table string, values map[string][]int64, keys map[int64]string) {
+	m.indexCache[name] = values
+	m.indexTable[name] = table
+	m.rowKeyCache[table] = keys
+	reverse := make(map[string]int64, len(keys))
+	for id, key := range keys {
+		reverse[key] = id
+	}
+	m.keyRowCache[table] = reverse
+	indexed := make(map[int64]string, len(keys))
+	for value, ids := range values {
+		for _, id := range ids {
+			indexed[id] = value
+		}
+	}
+	m.indexRowValues[name] = indexed
+}
+
+func (m *TableManager) removeIndexEntryLocked(name, value string, id int64) {
+	values := m.indexCache[name]
+	if values == nil {
+		return
+	}
+	ids := values[value]
+	remaining := ids[:0]
+	for _, candidate := range ids {
+		if candidate != id {
+			remaining = append(remaining, candidate)
+		}
+	}
+	if len(remaining) == 0 {
+		delete(values, value)
+	} else {
+		values[value] = remaining
+	}
+	if indexed, ok := m.indexRowValues[name][id]; ok && indexed == value {
+		delete(m.indexRowValues[name], id)
+	}
+}

pkg/storage/table.go

@@ -31,11 +31,13 @@ type TableManager struct {
 	schema   *SchemaManager
 	database string
 
-	cacheMu     sync.RWMutex
-	uniqueCache map[string]*uniqueIndexCache
-	indexCache  map[string]map[string][]int64 // index name → indexed value → rowids
-	indexTable  map[string]string             // index name → table name
-	rowKeyCache map[string]map[int64]string   // table name → rowid → primary key
+	cacheMu        sync.RWMutex
+	uniqueCache    map[string]*uniqueIndexCache
+	indexRowValues map[string]map[int64]string   // index name → rowid → candidate key
+	keyRowCache    map[string]map[string]int64   // table name → primary key → rowid
+	indexCache     map[string]map[string][]int64 // index name → indexed value → rowids
+	indexTable     map[string]string             // index name → table name
+	rowKeyCache    map[string]map[int64]string   // table name → rowid → primary key
 	// disabledIndexes prevents a concurrent lookup from rebuilding an index
 	// after DROP has cleared it but before the schema entry is removed.
 	disabledIndexes map[string]bool
@@ -92,6 +94,8 @@ func NewTableManager(pool *KVPool, schema *SchemaManager, database string) *Tabl
 		schema:               schema,
 		database:             database,
 		uniqueCache:          make(map[string]*uniqueIndexCache),
+		indexRowValues:       make(map[string]map[int64]string),
+		keyRowCache:          make(map[string]map[string]int64),
 		indexCache:           make(map[string]map[string][]int64),
 		indexTable:           make(map[string]string),
 		rowKeyCache:          make(map[string]map[int64]string),
@@ -206,12 +210,14 @@ func (m *TableManager) invalidateCache(table string) {
 	m.cacheMu.Lock()
 	key := strings.ToLower(table)
 	delete(m.rowKeyCache, key)
+	delete(m.keyRowCache, key)
 	delete(m.counts, key)
 	delete(m.countsInit, key)
 	delete(m.countGeneration, key)
 	for indexName, tableName := range m.indexTable {
 		if tableName == key {
 			delete(m.indexCache, indexName)
+			delete(m.indexRowValues, indexName)
 			delete(m.indexTable, indexName)
 		}
 	}
@@ -607,7 +613,7 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 	}
 
 	m.applyUniquePlan(uniquePlan, nil)
-	if err := m.updateIndexesForRow(table, nr, true); err != nil {
+	if err := m.updateIndexesForRow(table, nr, true, key); err != nil {
 		return 0, err
 	}
 	// Publish derived index state before its generations. A reader that races
@@ -856,7 +862,7 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 		m.applyUniquePlan(uniquePlan, accepted)
 	}
 	for i := 0; i < numOK; i++ {
-		if err := m.updateIndexesForRow(table, encoded[i], true); err != nil {
+		if err := m.updateIndexesForRow(table, encoded[i], true, string(ops[i].Key)); err != nil {
 			return numOK, 0, err
 		}
 	}
@@ -881,7 +887,7 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 // updateIndexesForRow adds or removes entries from already-built in-memory
 // indexes. Index entries are rebuildable from durable row data, so this method
 // intentionally does not write idx:* keys to KV.
-func (m *TableManager) updateIndexesForRow(table string, row Row, add bool) error {
+func (m *TableManager) updateIndexesForRow(table string, row Row, add bool, actualKey string) error {
 	indexes, err := m.schema.ListTableIndexes(table)
 	if err != nil || len(indexes) == 0 {
 		return err
@@ -892,18 +898,19 @@ func (m *TableManager) updateIndexesForRow(table string, row Row, add bool) erro
 		return nil
 	}
 	tableKey := strings.ToLower(table)
-	tableSchema, schemaErr := m.schema.GetSchema(table)
-	if schemaErr == nil {
-		m.cacheMu.Lock()
-		if keys, initialized := m.rowKeyCache[tableKey]; initialized {
-			if add {
-				keys[rowid] = fmt.Sprintf("%v", row[tableSchema.PrimaryKey])
-			} else {
-				delete(keys, rowid)
-			}
+	m.cacheMu.Lock()
+	if keys, initialized := m.rowKeyCache[tableKey]; initialized {
+		pk := strings.TrimPrefix(actualKey, m.dataPrefix(table))
+		reverse := m.keyRowCache[tableKey]
+		if add {
+			keys[rowid] = pk
+			reverse[pk] = rowid
+		} else {
+			delete(keys, rowid)
+			delete(reverse, pk)
 		}
-		m.cacheMu.Unlock()
 	}
+	m.cacheMu.Unlock()
 
 	for _, idx := range indexes {
 		indexName := strings.ToLower(idx.Name)
@@ -1112,10 +1119,10 @@ func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, err
 	}
 
 	m.applyUniquePlan(uniquePlan, nil)
-	if err := m.updateIndexesForRow(table, oldRow, false); err != nil {
+	if err := m.updateIndexesForRow(table, oldRow, false, key); err != nil {
 		return nil, false, err
 	}
-	if err := m.updateIndexesForRow(table, row, true); err != nil {
+	if err := m.updateIndexesForRow(table, row, true, key); err != nil {
 		return nil, false, err
 	}
 	if err := m.bumpIndexPredicates(table, oldRow, row); err != nil {
@@ -1180,7 +1187,7 @@ func (m *TableManager) DeleteByPK(table, pk string) (Row, bool, error) {
 	}
 
 	m.removeUniqueKey(table, key)
-	if err := m.updateIndexesForRow(table, row, false); err != nil {
+	if err := m.updateIndexesForRow(table, row, false, key); err != nil {
 		return nil, false, err
 	}
 	if err := m.bumpIndexPredicates(table, row); err != nil {
@@ -1400,9 +1407,7 @@ func (m *TableManager) ensureIndex(index *Index) error {
 
 	m.cacheMu.Lock()
 	if _, initialized := m.indexCache[indexKey]; !initialized {
-		m.indexCache[indexKey] = values
-		m.indexTable[indexKey] = key
-		m.rowKeyCache[key] = rowKeys
+		m.installIndexCacheLocked(indexKey, key, values, rowKeys)
 	}
 	m.cacheMu.Unlock()
 
@@ -1428,6 +1433,10 @@ func (m *TableManager) AddIndexEntry(indexName string, colValue interface{}, row
 		}
 	}
 	values[valueKey] = append(rowids, rowid)
+	if m.indexRowValues[indexKey] == nil {
+		m.indexRowValues[indexKey] = make(map[int64]string)
+	}
+	m.indexRowValues[indexKey][rowid] = valueKey
 	return nil
 }
 
@@ -1439,24 +1448,7 @@ func (m *TableManager) RemoveIndexEntry(indexName string, colValue interface{},
 	m.cacheMu.Lock()
 	defer m.cacheMu.Unlock()
 
-	values, ok := m.indexCache[indexKey]
-	if !ok {
-		return nil
-	}
-	rowids := values[valueKey]
-	newRowids := make([]int64, 0, len(rowids))
-	for _, r := range rowids {
-		if r != rowid {
-			newRowids = append(newRowids, r)
-		}
-	}
-
-	if len(newRowids) == 0 {
-		delete(values, valueKey)
-		return nil
-	}
-
-	values[valueKey] = newRowids
+	m.removeIndexEntryLocked(indexKey, valueKey, rowid)
 	return nil
 }
 
@@ -1496,6 +1488,7 @@ func (m *TableManager) ClearIndex(indexName, tableName string, columns []string)
 	tableKey := strings.ToLower(tableName)
 	m.cacheMu.Lock()
 	delete(m.indexCache, indexKey)
+	delete(m.indexRowValues, indexKey)
 	delete(m.indexTable, indexKey)
 	delete(m.uniqueCache, indexKey)
 	m.disabledIndexes[indexKey] = true
@@ -1508,6 +1501,7 @@ func (m *TableManager) ClearIndex(indexName, tableName string, columns []string)
 	}
 	if !rowKeysNeeded {
 		delete(m.rowKeyCache, tableKey)
+		delete(m.keyRowCache, tableKey)
 	}
 	m.cacheMu.Unlock()
 	return nil
@@ -1519,6 +1513,7 @@ func (m *TableManager) BuildIndex(indexName, tableName string, columns []string)
 	m.cacheMu.Lock()
 	delete(m.disabledIndexes, indexKey)
 	delete(m.indexCache, indexKey)
+	delete(m.indexRowValues, indexKey)
 	delete(m.indexTable, indexKey)
 	m.cacheMu.Unlock()
 
@@ -1557,9 +1552,7 @@ func (m *TableManager) BuildIndex(indexName, tableName string, columns []string)
 	}
 
 	m.cacheMu.Lock()
-	m.indexCache[indexKey] = values
-	m.indexTable[indexKey] = strings.ToLower(tableName)
-	m.rowKeyCache[strings.ToLower(tableName)] = rowKeys
+	m.installIndexCacheLocked(indexKey, strings.ToLower(tableName), values, rowKeys)
 	m.cacheMu.Unlock()
 
 	return nil

pkg/storage/tx.go

@@ -206,6 +206,28 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 		uniquePlans[t] = plan
 	}
 
+	exclusiveTables := make(map[string]bool)
+	for table := range uniqueTables {
+		exclusiveTables[table] = true
+	}
+	for table := range s.scanGens {
+		exclusiveTables[table] = true
+	}
+	for _, predicate := range s.predicateGens {
+		exclusiveTables[predicate.table] = true
+	}
+	cachePlans := make(map[string]*tableCachePlan)
+	for table, entries := range s.overlay {
+		if exclusiveTables[table] {
+			plan, err := s.table.prepareTableCache(table, entries, s.reads)
+			if err != nil {
+				s.resetLocked()
+				return 0, err
+			}
+			cachePlans[table] = plan
+		}
+	}
+
 	// Build the compare set from every observed key.
 	checks := make([]CompareCheck, 0, len(s.reads))
 	for key, lsn := range s.reads {
@@ -261,9 +283,13 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 		return 0, ErrSerialization
 	}
 
-	// Advance generations and invalidate derived caches for written tables.
+	// Publish derived state before generations while retaining the table gates.
 	for t := range s.overlay {
-		s.table.invalidateCache(t)
+		if plan := cachePlans[t]; plan != nil {
+			s.table.applyTableCache(plan)
+		} else {
+			s.table.invalidateCache(t)
+		}
 		s.table.applyUniquePlan(uniquePlans[t], nil)
 		s.table.bumpIndexPredicateWildcard(t)
 		s.table.bumpGeneration(t)

pkg/storage/unique_cache.go

@@ -70,7 +70,7 @@ func (m *TableManager) ensureUniqueCaches(table string, indexes []*Index) (map[s
 	var missing []*Index
 	m.cacheMu.RLock()
 	for _, idx := range indexes {
-		if cache := m.uniqueCache[strings.ToLower(idx.Name)]; cache != nil && cacheableUniqueIndex(idx) {
+		if cache := m.uniqueCache[strings.ToLower(idx.Name)]; cache != nil && cacheableColumnIndex(idx) {
 			caches[idx.Name] = cache
 		} else {
 			missing = append(missing, idx)
@@ -123,7 +123,7 @@ func (m *TableManager) ensureUniqueCaches(table string, indexes []*Index) (map[s
 	}
 	m.cacheMu.Lock()
 	for _, idx := range missing {
-		if cacheableUniqueIndex(idx) {
+		if cacheableColumnIndex(idx) {
 			m.uniqueCache[strings.ToLower(idx.Name)] = caches[idx.Name]
 		}
 	}
@@ -133,7 +133,7 @@ func (m *TableManager) ensureUniqueCaches(table string, indexes []*Index) (map[s
 
 // Expression evaluators have no determinism contract. Rebuild these constraints
 // for each validation rather than freezing time/random/function-dependent values.
-func cacheableUniqueIndex(index *Index) bool {
+func cacheableColumnIndex(index *Index) bool {
 	for _, column := range index.Columns {
 		if column.Expression != "" {
 			return false