fix: distinguish unavailable index caches from empty results (#21)

pkg/executor/index_cache_lifecycle_test.go

@@ -0,0 +1,20 @@
+package executor
+
+import "testing"
+
+func TestDisabledIndexFallsBackRatherThanReturningEmpty(t *testing.T) {
+	_, schemas, tables := newTestDB(t)
+	e := newExec(schemas, tables)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
+	execMust(t, e, "INSERT INTO t VALUES (1,7)")
+	execMust(t, e, "CREATE INDEX idx ON t(v)")
+	execMust(t, e, "SELECT id FROM t WHERE v=7")
+	// DROP clears derived entries before removing metadata. A concurrent SELECT
+	// in that interval must fall back, not mistake an unavailable cache for empty.
+	if err := tables.ClearIndex("idx", "t", []string{"v"}); err != nil {
+		t.Fatal(err)
+	}
+	if got := execMust(t, e, "SELECT id FROM t WHERE v=7").Rows; len(got) != 1 || got[0][0] != int64(1) {
+		t.Fatalf("disabled cache changed query result: %v", got)
+	}
+}

pkg/storage/count_cache_test.go

@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"testing"
@@ -259,8 +260,8 @@ func TestClearIndexDropsInMemoryEntries(t *testing.T) {
 	if cached || mapped {
 		t.Fatalf("cleared index remains cached: cache=%v table=%v", cached, mapped)
 	}
-	if rows, err := tables.SelectByIndex("users", "idx_status", "old"); err != nil || len(rows) != 0 {
-		t.Fatalf("cleared index rebuilt during drop: len=%d err=%v", len(rows), err)
+	if rows, err := tables.SelectByIndex("users", "idx_status", "old"); !errors.Is(err, ErrIndexNotFound) || len(rows) != 0 {
+		t.Fatalf("cleared index must report unavailable so SQL falls back: len=%d err=%v", len(rows), err)
 	}
 
 	if err := tables.BuildIndex("idx_status", "users", []string{"status"}); err != nil {

pkg/storage/index_cache_lifecycle_test.go

@@ -0,0 +1,61 @@
+package storage
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+)
+
+func TestIndexedReadWhileSharedGateCommitsInvalidate(t *testing.T) {
+	_, _, schemas, tables := newTestSession(t)
+	createTestTable(t, schemas, "t", []Column{{Name: "id", Type: "INTEGER", PrimaryKey: true}, {Name: "v", Type: "INTEGER"}})
+	if _, err := tables.InsertBulk("t", []Row{{"id": int64(1), "v": int64(1)}, {"id": int64(2), "v": int64(2)}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := schemas.CreateIndex(&Index{Name: "idx", Table: "t", Columns: []IndexColumn{{Name: "v"}}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := tables.BuildIndex("idx", "t", []string{"v"}); err != nil {
+		t.Fatal(err)
+	}
+	errors := make(chan error, 2)
+	var workers sync.WaitGroup
+	workers.Add(2)
+	go func() {
+		defer workers.Done()
+		session := NewSession(schemas, tables)
+		for i := 0; i < 500; i++ {
+			if err := session.Begin(); err != nil {
+				errors <- err
+				return
+			}
+			if _, _, err := session.UpdateByPK("t", "2", func(Row) (Row, error) { return Row{"v": int64(2)}, nil }); err != nil {
+				errors <- err
+				return
+			}
+			if err := session.Commit(); err != nil {
+				errors <- err
+				return
+			}
+		}
+	}()
+	go func() {
+		defer workers.Done()
+		for i := 0; i < 500; i++ {
+			rows, err := tables.SelectByIndex("t", "idx", int64(1))
+			if err != nil {
+				errors <- err
+				return
+			}
+			if len(rows) != 1 || rows[0]["id"] != int64(1) {
+				errors <- fmt.Errorf("unchanged indexed row disappeared: %v", rows)
+				return
+			}
+		}
+	}()
+	workers.Wait()
+	close(errors)
+	for err := range errors {
+		t.Error(err)
+	}
+}

pkg/storage/table.go

@@ -1466,18 +1466,25 @@ func (m *TableManager) LookupIndex(indexName string, colValue interface{}) ([]in
 	if err != nil {
 		return nil, err
 	}
-	if err := m.ensureIndex(index); err != nil {
-		return nil, err
-	}
-
 	indexKey := strings.ToLower(indexName)
 	valueKey := formatIndexValue(colValue)
-
-	m.cacheMu.RLock()
-	rowids := append([]int64(nil), m.indexCache[indexKey][valueKey]...)
-	m.cacheMu.RUnlock()
-
-	return rowids, nil
+	for {
+		if err := m.ensureIndex(index); err != nil {
+			return nil, err
+		}
+		m.cacheMu.RLock()
+		values, initialized := m.indexCache[indexKey]
+		disabled := m.disabledIndexes[indexKey]
+		rowids := append([]int64(nil), values[valueKey]...)
+		m.cacheMu.RUnlock()
+		if disabled {
+			return nil, fmt.Errorf("%w: %s is disabled", ErrIndexNotFound, indexName)
+		}
+		if initialized {
+			return rowids, nil
+		}
+		// A shared-gate commit invalidated the cache after ensureIndex.
+	}
 }
 
 // ClearIndex removes an index's in-memory entries and marks it disabled so a
@@ -1848,25 +1855,38 @@ func (m *TableManager) selectByIndexKeyWithLSN(table, indexName, valueKey string
 	if err != nil {
 		return nil, indexPredicateSnapshot{}, err
 	}
-	if err := m.ensureIndex(index); err != nil {
-		return nil, indexPredicateSnapshot{}, err
-	}
-
 	tableKey := strings.ToLower(table)
 	tl := m.tableLock(tableKey)
-	tl.RLock()
-	defer tl.RUnlock()
-	if !m.schema.TableExists(table) {
-		return nil, indexPredicateSnapshot{}, fmt.Errorf("table not found: %s", table)
-	}
-
 	indexKey := strings.ToLower(indexName)
+	for {
+		if err := m.ensureIndex(index); err != nil {
+			return nil, indexPredicateSnapshot{}, err
+		}
+		tl.RLock()
+		if !m.schema.TableExists(table) {
+			tl.RUnlock()
+			return nil, indexPredicateSnapshot{}, fmt.Errorf("table not found: %s", table)
+		}
+		m.cacheMu.RLock()
+		if m.disabledIndexes[indexKey] {
+			m.cacheMu.RUnlock()
+			tl.RUnlock()
+			return nil, indexPredicateSnapshot{}, fmt.Errorf("%w: %s is disabled", ErrIndexNotFound, indexName)
+		}
+		if _, initialized := m.indexCache[indexKey]; initialized {
+			break // Keep cacheMu through the complete candidate/key snapshot.
+		}
+		m.cacheMu.RUnlock()
+		tl.RUnlock()
+		// Commits can invalidate under a shared table gate. Retry the build,
+		// never interpret a missing cache as a legitimately empty index.
+	}
+	defer tl.RUnlock()
 	predicateKey, predicateGen, wildcardKey, wildcardGen := m.predicateSnapshot(table, indexName, valueKey)
 	snapshot := indexPredicateSnapshot{
 		valueKey: predicateKey, valueGen: predicateGen,
 		wildcardKey: wildcardKey, wildcardGen: wildcardGen,
 	}
-	m.cacheMu.RLock()
 	rowids := append([]int64(nil), m.indexCache[indexKey][valueKey]...)
 	primaryKeys := make([]string, 0, len(rowids))
 	missingRowID := int64(0)