agg cache

pkg/executor/executor.go

@@ -1,6 +1,7 @@
 package executor
 
 import (
+	"errors"
 	"fmt"
 	"math/rand"
 	"sort"
@@ -19,6 +20,8 @@ type Executor struct {
 	table    *storage.TableManager
 	analyzer *analyzer.Analyzer
 	catalog  *analyzer.Catalog
+	// Last SchemaManager version reflected in catalog.
+	catalogVersion uint64
 
 	// Multi-database support
 	attachedDatabases map[string]*DatabaseConnection // alias -> connection
@@ -36,10 +39,25 @@ type Executor struct {
 	// Keyed by subquery AST pointer; valid for one top-level Execute call.
 	subqueryCache map[*parser.SelectStmt]*Result
 
+	// Per-query cache for decorrelated scalar aggregate subqueries.
+	// Keyed by subquery AST pointer; valid for one top-level Execute call.
+	correlatedAggCache map[*parser.SelectStmt]*correlatedAggCache
+
 	// In-memory view registry: view name (lowercase) → SELECT AST.
 	views map[string]*parser.SelectStmt
 }
 
+type correlatedAggCache struct {
+	values       map[string]interface{}
+	defaultValue interface{}
+}
+
+type correlatedAggSpec struct {
+	innerKey parser.Expr
+	outerKey *parser.ColumnRef
+	aggExpr  parser.Expr
+}
+
 // DatabaseConnection represents an attached database.
 type DatabaseConnection struct {
 	Alias  string
@@ -87,7 +105,9 @@ func (e *Executor) SyncCatalog() error {
 		return err
 	}
 
+	storageTables := make(map[string]struct{}, len(tables))
 	for _, tableName := range tables {
+		storageTables[strings.ToUpper(tableName)] = struct{}{}
 		schema, err := e.schema.GetSchema(tableName)
 		if err != nil {
 			continue
@@ -96,14 +116,27 @@ func (e *Executor) SyncCatalog() error {
 		e.catalog.DropTable(tableName)
 		e.catalog.CreateTable(schema.ToAnalyzerTableInfo())
 	}
+	for _, table := range e.catalog.GetTables() {
+		if table.IsView {
+			continue
+		}
+		if _, exists := storageTables[strings.ToUpper(table.Name)]; !exists {
+			e.catalog.DropTable(table.Name)
+		}
+	}
 
+	e.catalogVersion = e.schema.Version()
 	return nil
 }
 
 // Execute executes a SQL statement.
 func (e *Executor) Execute(stmt parser.Statement) (*Result, error) {
 	e.subqueryCache = make(map[*parser.SelectStmt]*Result)
-	defer func() { e.subqueryCache = nil }()
+	e.correlatedAggCache = make(map[*parser.SelectStmt]*correlatedAggCache)
+	defer func() {
+		e.subqueryCache = nil
+		e.correlatedAggCache = nil
+	}()
 
 	// PRAGMA doesn't need analysis
 	if pragma, ok := stmt.(*parser.PragmaStmt); ok {
@@ -141,10 +174,10 @@ func (e *Executor) Execute(stmt parser.Statement) (*Result, error) {
 		return e.executeDetach(s)
 	}
 
-	// Analyze first — create a fresh analyzer per call so concurrent requests
-	// don't share mutable scope state (e.analyzer.scope would race otherwise).
-	a := analyzer.New(e.catalog)
-	if err := a.Analyze(stmt); err != nil {
+	// Analyze first. If the cached analyzer catalog is stale because schema was
+	// changed through another executor/API path, resync from storage and retry
+	// once before returning table/column-not-found errors.
+	if err := e.analyzeWithCatalogRetry(stmt); err != nil {
 		return nil, err
 	}
 
@@ -172,6 +205,39 @@ func (e *Executor) Execute(stmt parser.Statement) (*Result, error) {
 	}
 }
 
+func (e *Executor) analyzeWithCatalogRetry(stmt parser.Statement) error {
+	if e.catalogVersion != e.schema.Version() {
+		if err := e.SyncCatalog(); err != nil {
+			return err
+		}
+	}
+
+	a := analyzer.New(e.catalog)
+	err := a.Analyze(stmt)
+	if err == nil {
+		return nil
+	}
+	if !isCatalogMiss(err) {
+		return err
+	}
+
+	if syncErr := e.SyncCatalog(); syncErr != nil {
+		return err
+	}
+
+	a = analyzer.New(e.catalog)
+	return a.Analyze(stmt)
+}
+
+func isCatalogMiss(err error) bool {
+	var analysisErr *analyzer.AnalysisError
+	if !errors.As(err, &analysisErr) {
+		return false
+	}
+	return analysisErr.Type == analyzer.ErrTableNotFound ||
+		analysisErr.Type == analyzer.ErrColumnNotFound
+}
+
 // executeSelect executes a SELECT statement (or compound SELECT).
 func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 	if stmt.Compound != nil {
@@ -1346,20 +1412,12 @@ func (e *Executor) executeAggregateSelect(stmt *parser.SelectStmt, rows []storag
 
 // executeGroupBy executes a GROUP BY query.
 func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, schema *storage.Schema) (*Result, error) {
-	// Group rows
-	groups := make(map[string][]storage.Row)
-	for _, row := range rows {
-		key := e.buildGroupKey(stmt.GroupBy, row)
-		groups[key] = append(groups[key], row)
-	}
-
 	result := NewResult("SELECT")
 
 	// Expand SELECT * if present
-	expandedColumns := make([]parser.SelectColumn, 0)
+	expandedColumns := make([]parser.SelectColumn, 0, len(stmt.Columns))
 	for _, col := range stmt.Columns {
 		if col.Star {
-			// Expand * to all columns from schema
 			for _, c := range schema.Columns {
 				expandedColumns = append(expandedColumns, parser.SelectColumn{
 					Expr: &parser.ColumnRef{Column: c.Name},
@@ -1370,7 +1428,7 @@ func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, s
 		}
 	}
 
-	// Determine columns
+	// Determine column names
 	columnNames := make([]string, len(expandedColumns))
 	for i, col := range expandedColumns {
 		if col.Alias != "" {
@@ -1385,16 +1443,27 @@ func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, s
 		}
 	}
 
-	// Process each group
+	// Fast path: use running accumulators instead of collecting rows per group.
+	// Applicable when there is no HAVING clause and all aggregate SELECT columns
+	// are direct FunctionCalls (COUNT/SUM/AVG/MIN/MAX).
+	if e.canUseGroupAccum(stmt, expandedColumns) {
+		return e.executeGroupByAccum(stmt, rows, result, expandedColumns, columnNames)
+	}
+
+	// Slow path: collect full rows per group then evaluate aggregates over them.
+	groups := make(map[string][]storage.Row)
+	for _, row := range rows {
+		key := e.buildGroupKey(stmt.GroupBy, row)
+		groups[key] = append(groups[key], row)
+	}
+
 	for _, groupRows := range groups {
-		// Apply HAVING
 		if stmt.Having != nil {
 			val, err := e.evalAggregateExpr(stmt.Having, groupRows)
 			if err != nil || val == nil || !toBool(val) {
 				continue
 			}
 		}
-
 		values := make([]interface{}, len(expandedColumns))
 		for i, col := range expandedColumns {
 			if e.isAggregate(col.Expr) {
@@ -1404,7 +1473,6 @@ func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, s
 				}
 				values[i] = val
 			} else {
-				// Use first row's value for non-aggregate columns
 				val, err := e.evalExpr(col.Expr, groupRows[0])
 				if err != nil {
 					return nil, err
@@ -1415,17 +1483,221 @@ func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, s
 		result.AddRow(values...)
 	}
 
-	// Apply DISTINCT
+	return e.finalizeGroupResult(stmt, result, expandedColumns, columnNames)
+}
+
+// canUseGroupAccum returns true when the fast accumulator path can handle the query.
+func (e *Executor) canUseGroupAccum(stmt *parser.SelectStmt, expandedColumns []parser.SelectColumn) bool {
+	if stmt.Having != nil {
+		return false
+	}
+	for _, col := range expandedColumns {
+		if !e.isAggregate(col.Expr) {
+			continue
+		}
+		fn, ok := col.Expr.(*parser.FunctionCall)
+		if !ok {
+			return false
+		}
+		switch strings.ToUpper(fn.Name) {
+		case "COUNT", "SUM", "AVG", "MIN", "MAX":
+		default:
+			return false
+		}
+	}
+	return true
+}
+
+// aggColInfo pairs a SELECT column index with its aggregate FunctionCall.
+type aggColInfo struct {
+	colIdx int
+	fn     *parser.FunctionCall
+}
+
+// aggAccum holds running state for a single aggregate function.
+type aggAccum struct {
+	count   int64
+	sumI    int64
+	sumF    float64
+	allInt  bool
+	hasVal  bool
+	extreme interface{}
+	seen    map[interface{}]struct{} // for DISTINCT
+}
+
+// groupAccumState holds per-group state for the fast accumulator path.
+type groupAccumState struct {
+	firstRow storage.Row
+	accums   []*aggAccum
+}
+
+// executeGroupByAccum is the fast GROUP BY path: increments per-group counters as rows
+// arrive rather than materialising row slices, keeping O(1) state per group.
+func (e *Executor) executeGroupByAccum(stmt *parser.SelectStmt, rows []storage.Row, result *Result, expandedColumns []parser.SelectColumn, columnNames []string) (*Result, error) {
+	var aggCols []aggColInfo
+	for i, col := range expandedColumns {
+		if e.isAggregate(col.Expr) {
+			aggCols = append(aggCols, aggColInfo{i, col.Expr.(*parser.FunctionCall)})
+		}
+	}
+
+	states := make(map[string]*groupAccumState, 64)
+	var keyOrder []string
+
+	for _, row := range rows {
+		key := e.buildGroupKey(stmt.GroupBy, row)
+		state, exists := states[key]
+		if !exists {
+			accums := make([]*aggAccum, len(aggCols))
+			for j, ac := range aggCols {
+				a := &aggAccum{allInt: true}
+				if ac.fn.Distinct {
+					a.seen = make(map[interface{}]struct{})
+				}
+				accums[j] = a
+			}
+			state = &groupAccumState{firstRow: row, accums: accums}
+			states[key] = state
+			keyOrder = append(keyOrder, key)
+		}
+		for j, ac := range aggCols {
+			e.feedAggAccum(state.accums[j], ac.fn, row)
+		}
+	}
+
+	for _, key := range keyOrder {
+		state := states[key]
+		values := make([]interface{}, len(expandedColumns))
+		for i, col := range expandedColumns {
+			if e.isAggregate(col.Expr) {
+				for j, ac := range aggCols {
+					if ac.colIdx == i {
+						values[i] = finalizeAggAccum(state.accums[j], ac.fn)
+						break
+					}
+				}
+			} else {
+				val, _ := e.evalExpr(col.Expr, state.firstRow)
+				values[i] = val
+			}
+		}
+		result.AddRow(values...)
+	}
+
+	return e.finalizeGroupResult(stmt, result, expandedColumns, columnNames)
+}
+
+// feedAggAccum updates a running accumulator with one row.
+func (e *Executor) feedAggAccum(a *aggAccum, fn *parser.FunctionCall, row storage.Row) {
+	switch strings.ToUpper(fn.Name) {
+	case "COUNT":
+		if fn.Star {
+			a.count++
+			return
+		}
+		if len(fn.Args) == 0 {
+			return
+		}
+		val, _ := e.evalExpr(fn.Args[0], row)
+		if val == nil {
+			return
+		}
+		if fn.Distinct {
+			k := fmt.Sprintf("%v", val)
+			if _, exists := a.seen[k]; exists {
+				return
+			}
+			a.seen[k] = struct{}{}
+		}
+		a.count++
+
+	case "SUM":
+		if len(fn.Args) == 0 {
+			return
+		}
+		val, _ := e.evalExpr(fn.Args[0], row)
+		if val == nil {
+			return
+		}
+		if fn.Distinct {
+			k := fmt.Sprintf("%v", val)
+			if _, exists := a.seen[k]; exists {
+				return
+			}
+			a.seen[k] = struct{}{}
+		}
+		if isIntVal(val) {
+			a.sumI += toInt64(val)
+		} else {
+			a.allInt = false
+			a.sumF += toFloat(val)
+		}
+		a.hasVal = true
+
+	case "AVG":
+		if len(fn.Args) == 0 {
+			return
+		}
+		val, _ := e.evalExpr(fn.Args[0], row)
+		if val == nil {
+			return
+		}
+		a.sumF += toFloat(val)
+		a.count++
+		a.hasVal = true
+
+	case "MIN":
+		if len(fn.Args) == 0 {
+			return
+		}
+		val, _ := e.evalExpr(fn.Args[0], row)
+		if val != nil && (a.extreme == nil || compare(val, a.extreme) < 0) {
+			a.extreme = val
+		}
+
+	case "MAX":
+		if len(fn.Args) == 0 {
+			return
+		}
+		val, _ := e.evalExpr(fn.Args[0], row)
+		if val != nil && (a.extreme == nil || compare(val, a.extreme) > 0) {
+			a.extreme = val
+		}
+	}
+}
+
+// finalizeAggAccum computes the final aggregate value from a running accumulator.
+func finalizeAggAccum(a *aggAccum, fn *parser.FunctionCall) interface{} {
+	switch strings.ToUpper(fn.Name) {
+	case "COUNT":
+		return a.count
+	case "SUM":
+		if !a.hasVal {
+			return nil
+		}
+		if a.allInt {
+			return a.sumI
+		}
+		return a.sumF + float64(a.sumI)
+	case "AVG":
+		if !a.hasVal || a.count == 0 {
+			return nil
+		}
+		return a.sumF / float64(a.count)
+	case "MIN", "MAX":
+		return a.extreme
+	}
+	return nil
+}
+
+// finalizeGroupResult applies DISTINCT, ORDER BY, and LIMIT/OFFSET to a GROUP BY result.
+func (e *Executor) finalizeGroupResult(stmt *parser.SelectStmt, result *Result, expandedColumns []parser.SelectColumn, columnNames []string) (*Result, error) {
 	if stmt.Distinct {
 		result.Rows = e.applyDistinct(result.Rows)
 	}
-
-	// Apply ORDER BY
 	if len(stmt.OrderBy) > 0 {
 		e.sortResultRows(result, stmt.OrderBy, expandedColumns, columnNames)
 	}
-
-	// Apply LIMIT/OFFSET
 	if stmt.Offset != nil {
 		offset := e.evalIntExpr(stmt.Offset)
 		if offset < len(result.Rows) {
@@ -1442,12 +1714,15 @@ func (e *Executor) executeGroupBy(stmt *parser.SelectStmt, rows []storage.Row, s
 		}
 		result.RowCount = len(result.Rows)
 	}
-
 	return result, nil
 }
 
 // executeJoins recursively processes all JOIN clauses in a table reference.
 func (e *Executor) executeJoins(tableRef parser.TableRef, leftRows []storage.Row) ([]storage.Row, error) {
+	return e.executeJoinsWithMode(tableRef, leftRows, true)
+}
+
+func (e *Executor) executeJoinsWithMode(tableRef parser.TableRef, leftRows []storage.Row, qualifyLeft bool) ([]storage.Row, error) {
 	if tableRef.Join == nil || tableRef.Join.Table == nil {
 		return leftRows, nil
 	}
@@ -1477,69 +1752,100 @@ func (e *Executor) executeJoins(tableRef parser.TableRef, leftRows []storage.Row
 	if rightAlias == "" {
 		rightAlias = rightTable
 	}
+	leftAliasForMerge := leftAlias
+	if !qualifyLeft {
+		leftAliasForMerge = ""
+	}
+
+	// Build a synthetic TableRef so we can reuse extractEqualityJoinKeys.
+	syntheticLeft := parser.TableRef{Name: leftTableName, Alias: leftAlias}
+	syntheticJoin := &parser.JoinClause{
+		Type:      tableRef.Join.Type,
+		Table:     &parser.TableRef{Name: rightTable, Alias: rightAlias},
+		Condition: tableRef.Join.Condition,
+	}
+	leftKey, rightKey, canHash := extractEqualityJoinKeys(tableRef.Join.Condition, syntheticLeft, syntheticJoin)
 
 	switch tableRef.Join.Type {
 	case parser.JoinInner:
-		for _, left := range leftRows {
+		if canHash {
+			hashTable := make(map[string][]storage.Row, len(rightRows))
 			for _, right := range rightRows {
-				merged := e.mergeRows(left, right, leftAlias, rightAlias)
-				if tableRef.Join.Condition != nil {
-					match, _ := e.evalExpr(tableRef.Join.Condition, merged)
-					if toBool(match) {
+				k := joinKeyString(right, rightKey)
+				hashTable[k] = append(hashTable[k], right)
+			}
+			for _, left := range leftRows {
+				k := joinKeyString(left, leftKey)
+				for _, right := range hashTable[k] {
+					result = append(result, e.mergeRows(left, right, leftAliasForMerge, rightAlias))
+				}
+			}
+		} else {
+			for _, left := range leftRows {
+				for _, right := range rightRows {
+					merged := e.mergeRows(left, right, leftAliasForMerge, rightAlias)
+					if tableRef.Join.Condition != nil {
+						match, _ := e.evalExpr(tableRef.Join.Condition, merged)
+						if toBool(match) {
+							result = append(result, merged)
+						}
+					} else {
 						result = append(result, merged)
 					}
-				} else {
-					result = append(result, merged)
 				}
 			}
 		}
 
 	case parser.JoinLeft:
-		for _, left := range leftRows {
-			matched := false
+		if canHash {
+			hashTable := make(map[string][]storage.Row, len(rightRows))
 			for _, right := range rightRows {
-				merged := e.mergeRows(left, right, leftAlias, rightAlias)
-				if tableRef.Join.Condition != nil {
-					match, _ := e.evalExpr(tableRef.Join.Condition, merged)
-					if toBool(match) {
-						result = append(result, merged)
-						matched = true
+				k := joinKeyString(right, rightKey)
+				hashTable[k] = append(hashTable[k], right)
+			}
+			nullRight := makeNullRow(rightRows, rightTable, e)
+			for _, left := range leftRows {
+				k := joinKeyString(left, leftKey)
+				matches := hashTable[k]
+				if len(matches) == 0 {
+					result = append(result, e.mergeRows(left, nullRight, leftAliasForMerge, rightAlias))
+				} else {
+					for _, right := range matches {
+						result = append(result, e.mergeRows(left, right, leftAliasForMerge, rightAlias))
 					}
 				}
 			}
-			if !matched {
-				// Add left row with nulls for right table columns
-				// Create a null row for the right table
-				nullRight := make(storage.Row)
-				if len(rightRows) > 0 {
-					// Use the first right row as a template to get column names
-					for k := range rightRows[0] {
-						nullRight[k] = nil
-					}
-				} else {
-					// If right table is empty, get schema to determine columns
-					rightSchema, err := e.schema.GetSchema(rightTable)
-					if err == nil {
-						for _, col := range rightSchema.Columns {
-							nullRight[col.Name] = nil
+		} else {
+			for _, left := range leftRows {
+				matched := false
+				for _, right := range rightRows {
+					merged := e.mergeRows(left, right, leftAliasForMerge, rightAlias)
+					if tableRef.Join.Condition != nil {
+						match, _ := e.evalExpr(tableRef.Join.Condition, merged)
+						if toBool(match) {
+							result = append(result, merged)
+							matched = true
 						}
 					}
 				}
-				result = append(result, e.mergeRows(left, nullRight, leftAlias, rightAlias))
+				if !matched {
+					nullRight := makeNullRow(rightRows, rightTable, e)
+					result = append(result, e.mergeRows(left, nullRight, leftAliasForMerge, rightAlias))
+				}
 			}
 		}
 
 	case parser.JoinCross:
 		for _, left := range leftRows {
 			for _, right := range rightRows {
-				result = append(result, e.mergeRows(left, right, leftAlias, rightAlias))
+				result = append(result, e.mergeRows(left, right, leftAliasForMerge, rightAlias))
 			}
 		}
 	}
 
 	// Recursively process any additional joins
 	if rightTableRef.Join != nil {
-		return e.executeJoins(*rightTableRef, result)
+		return e.executeJoinsWithMode(*rightTableRef, result, false)
 	}
 
 	return result, nil
@@ -1562,52 +1868,73 @@ func (e *Executor) executeJoin(tableRef parser.TableRef, leftRows []storage.Row)
 
 	switch join.Type {
 	case parser.JoinInner:
-		for _, left := range leftRows {
+		leftKey, rightKey, canHash := extractEqualityJoinKeys(join.Condition, tableRef, join)
+		if canHash {
+			// Hash join: build phase on right, probe phase on left — O(N+M) vs O(N*M)
+			hashTable := make(map[string][]storage.Row, len(rightRows))
 			for _, right := range rightRows {
-				merged := e.mergeRows(left, right, tableRef.Alias, join.Table.Alias)
-				if join.Condition != nil {
-					match, _ := e.evalExpr(join.Condition, merged)
-					if toBool(match) {
+				k := joinKeyString(right, rightKey)
+				hashTable[k] = append(hashTable[k], right)
+			}
+			for _, left := range leftRows {
+				k := joinKeyString(left, leftKey)
+				for _, right := range hashTable[k] {
+					result = append(result, e.mergeRows(left, right, tableRef.Alias, join.Table.Alias))
+				}
+			}
+		} else {
+			for _, left := range leftRows {
+				for _, right := range rightRows {
+					merged := e.mergeRows(left, right, tableRef.Alias, join.Table.Alias)
+					if join.Condition != nil {
+						match, _ := e.evalExpr(join.Condition, merged)
+						if toBool(match) {
+							result = append(result, merged)
+						}
+					} else {
 						result = append(result, merged)
 					}
-				} else {
-					result = append(result, merged)
 				}
 			}
 		}
 
 	case parser.JoinLeft:
-		for _, left := range leftRows {
-			matched := false
+		leftKey, rightKey, canHash := extractEqualityJoinKeys(join.Condition, tableRef, join)
+		if canHash {
+			hashTable := make(map[string][]storage.Row, len(rightRows))
 			for _, right := range rightRows {
-				merged := e.mergeRows(left, right, tableRef.Alias, join.Table.Alias)
-				if join.Condition != nil {
-					match, _ := e.evalExpr(join.Condition, merged)
-					if toBool(match) {
-						result = append(result, merged)
-						matched = true
+				k := joinKeyString(right, rightKey)
+				hashTable[k] = append(hashTable[k], right)
+			}
+			nullRight := makeNullRow(rightRows, rightTable, e)
+			for _, left := range leftRows {
+				k := joinKeyString(left, leftKey)
+				matches := hashTable[k]
+				if len(matches) == 0 {
+					result = append(result, e.mergeRows(left, nullRight, tableRef.Alias, join.Table.Alias))
+				} else {
+					for _, right := range matches {
+						result = append(result, e.mergeRows(left, right, tableRef.Alias, join.Table.Alias))
 					}
 				}
 			}
-			if !matched {
-				// Add left row with nulls for right table columns
-				// Create a null row for the right table
-				nullRight := make(storage.Row)
-				if len(rightRows) > 0 {
-					// Use the first right row as a template to get column names
-					for k := range rightRows[0] {
-						nullRight[k] = nil
-					}
-				} else {
-					// If right table is empty, get schema to determine columns
-					rightSchema, err := e.schema.GetSchema(rightTable)
-					if err == nil {
-						for _, col := range rightSchema.Columns {
-							nullRight[col.Name] = nil
+		} else {
+			for _, left := range leftRows {
+				matched := false
+				for _, right := range rightRows {
+					merged := e.mergeRows(left, right, tableRef.Alias, join.Table.Alias)
+					if join.Condition != nil {
+						match, _ := e.evalExpr(join.Condition, merged)
+						if toBool(match) {
+							result = append(result, merged)
+							matched = true
 						}
 					}
 				}
-				result = append(result, e.mergeRows(left, nullRight, tableRef.Alias, join.Table.Alias))
+				if !matched {
+					nullRight := makeNullRow(rightRows, rightTable, e)
+					result = append(result, e.mergeRows(left, nullRight, tableRef.Alias, join.Table.Alias))
+				}
 			}
 		}
 
@@ -1622,6 +1949,85 @@ func (e *Executor) executeJoin(tableRef parser.TableRef, leftRows []storage.Row)
 	return result, nil
 }
 
+// extractEqualityJoinKeys checks if a JOIN condition is a simple col = col equality
+// and returns the key names to probe in left rows and build from right rows.
+func extractEqualityJoinKeys(condition parser.Expr, leftRef parser.TableRef, join *parser.JoinClause) (leftKey, rightKey string, ok bool) {
+	if condition == nil {
+		return "", "", false
+	}
+	bin, isBin := condition.(*parser.BinaryExpr)
+	if !isBin || bin.Op != lexer.TokenEq {
+		return "", "", false
+	}
+	lRef, leftIsCol := bin.Left.(*parser.ColumnRef)
+	rRef, rightIsCol := bin.Right.(*parser.ColumnRef)
+	if !leftIsCol || !rightIsCol {
+		return "", "", false
+	}
+
+	leftAlias := leftRef.Alias
+	leftName := leftRef.Name
+	rightAlias := join.Table.Alias
+	rightName := join.Table.Name
+
+	leftJoinKey := func(r *parser.ColumnRef) (string, bool) {
+		if r.Table == "" || r.Table == leftAlias || r.Table == leftName {
+			return r.Column, true
+		}
+		// In a chained explicit JOIN, the left row already contains every table
+		// joined so far. Preserve qualified references such as "o.id" so joins
+		// against earlier tables can still use the hash path.
+		if r.Table != rightAlias && r.Table != rightName {
+			return r.Table + "." + r.Column, true
+		}
+		return "", false
+	}
+	rightJoinKey := func(r *parser.ColumnRef) (string, bool) {
+		if r.Table == "" || r.Table == rightAlias || r.Table == rightName {
+			return r.Column, true
+		}
+		return "", false
+	}
+
+	if lk, leftOK := leftJoinKey(lRef); leftOK {
+		if rk, rightOK := rightJoinKey(rRef); rightOK {
+			return lk, rk, true
+		}
+	}
+	if lk, leftOK := leftJoinKey(rRef); leftOK {
+		if rk, rightOK := rightJoinKey(lRef); rightOK {
+			return lk, rk, true
+		}
+	}
+	return "", "", false
+}
+
+// joinKeyString returns a string representation of a row's join key for hashing.
+func joinKeyString(row storage.Row, col string) string {
+	if v, ok := row[col]; ok {
+		return fmt.Sprintf("%v", v)
+	}
+	return "\x00"
+}
+
+// makeNullRow builds a null-valued row based on the right table's rows or schema.
+func makeNullRow(rightRows []storage.Row, rightTable string, e *Executor) storage.Row {
+	nullRight := make(storage.Row)
+	if len(rightRows) > 0 {
+		for k := range rightRows[0] {
+			nullRight[k] = nil
+		}
+	} else {
+		rightSchema, err := e.schema.GetSchema(rightTable)
+		if err == nil {
+			for _, col := range rightSchema.Columns {
+				nullRight[col.Name] = nil
+			}
+		}
+	}
+	return nullRight
+}
+
 // mergeRows merges two rows with optional table aliases.
 func (e *Executor) mergeRows(left, right storage.Row, leftAlias, rightAlias string) storage.Row {
 	result := make(storage.Row)
@@ -3708,6 +4114,12 @@ func (e *Executor) evalCastExpr(expr *parser.CastExpr, row storage.Row) (interfa
 // - NULL if the subquery returns no rows
 // - Error if the subquery returns more than one row (for strict SQL compliance)
 func (e *Executor) evalSubqueryExpr(expr *parser.SubqueryExpr, row storage.Row) (interface{}, error) {
+	if row != nil && e.correlatedAggCache != nil {
+		if val, ok, err := e.evalDecorrelatedAggSubquery(expr.Query, row); ok || err != nil {
+			return val, err
+		}
+	}
+
 	// Save and set outer row context for correlated subqueries
 	savedOuter := e.outerRow
 	e.outerRow = row
@@ -3739,6 +4151,149 @@ func (e *Executor) evalSubqueryExpr(expr *parser.SubqueryExpr, row storage.Row)
 	return nil, nil
 }
 
+func (e *Executor) evalDecorrelatedAggSubquery(query *parser.SelectStmt, outerRow storage.Row) (interface{}, bool, error) {
+	spec, ok := e.correlatedAggSpec(query)
+	if !ok {
+		return nil, false, nil
+	}
+
+	outerVal, err := e.evalExpr(spec.outerKey, outerRow)
+	if err != nil {
+		return nil, true, err
+	}
+	cache, exists := e.correlatedAggCache[query]
+	if !exists {
+		cache, err = e.buildCorrelatedAggCache(query, spec)
+		if err != nil {
+			return nil, true, err
+		}
+		e.correlatedAggCache[query] = cache
+	}
+	if outerVal == nil {
+		return cache.defaultValue, true, nil
+	}
+	if val, exists := cache.values[fmt.Sprintf("%v", outerVal)]; exists {
+		return val, true, nil
+	}
+	return cache.defaultValue, true, nil
+}
+
+func (e *Executor) correlatedAggSpec(query *parser.SelectStmt) (correlatedAggSpec, bool) {
+	if query == nil ||
+		query.Compound != nil ||
+		len(query.Columns) != 1 ||
+		len(query.From) == 0 ||
+		query.Where == nil ||
+		len(query.GroupBy) > 0 ||
+		query.Having != nil ||
+		query.Limit != nil ||
+		query.Offset != nil {
+		return correlatedAggSpec{}, false
+	}
+	if query.Columns[0].Star {
+		return correlatedAggSpec{}, false
+	}
+	agg, ok := query.Columns[0].Expr.(*parser.FunctionCall)
+	if !ok {
+		return correlatedAggSpec{}, false
+	}
+	switch strings.ToUpper(agg.Name) {
+	case "COUNT", "SUM", "AVG", "MIN", "MAX":
+	default:
+		return correlatedAggSpec{}, false
+	}
+
+	innerAliases := collectFromAliases(query.From)
+	bin, ok := query.Where.(*parser.BinaryExpr)
+	if !ok || bin.Op != lexer.TokenEq {
+		return correlatedAggSpec{}, false
+	}
+	leftRef, leftIsRef := bin.Left.(*parser.ColumnRef)
+	rightRef, rightIsRef := bin.Right.(*parser.ColumnRef)
+	if !leftIsRef || !rightIsRef {
+		return correlatedAggSpec{}, false
+	}
+
+	leftInner := refBelongsToAliases(leftRef, innerAliases)
+	rightInner := refBelongsToAliases(rightRef, innerAliases)
+	if leftInner == rightInner {
+		return correlatedAggSpec{}, false
+	}
+	if leftInner {
+		return correlatedAggSpec{innerKey: leftRef, outerKey: rightRef, aggExpr: agg}, true
+	}
+	return correlatedAggSpec{innerKey: rightRef, outerKey: leftRef, aggExpr: agg}, true
+}
+
+func (e *Executor) buildCorrelatedAggCache(query *parser.SelectStmt, spec correlatedAggSpec) (*correlatedAggCache, error) {
+	grouped := *query
+	grouped.Where = nil
+	grouped.GroupBy = []parser.Expr{spec.innerKey}
+	grouped.Having = nil
+	grouped.OrderBy = nil
+	grouped.Limit = nil
+	grouped.Offset = nil
+	grouped.Columns = []parser.SelectColumn{
+		{Expr: spec.innerKey, Alias: "__corr_key"},
+		{Expr: spec.aggExpr, Alias: "__corr_value"},
+	}
+
+	savedOuter := e.outerRow
+	e.outerRow = nil
+	result, err := e.executeSelect(&grouped)
+	e.outerRow = savedOuter
+	if err != nil {
+		return nil, fmt.Errorf("decorrelated aggregate subquery error: %w", err)
+	}
+
+	cache := &correlatedAggCache{
+		values:       make(map[string]interface{}, len(result.Rows)),
+		defaultValue: correlatedAggDefault(spec.aggExpr),
+	}
+	for _, row := range result.Rows {
+		if len(row) < 2 || row[0] == nil {
+			continue
+		}
+		cache.values[fmt.Sprintf("%v", row[0])] = row[1]
+	}
+	return cache, nil
+}
+
+func correlatedAggDefault(expr parser.Expr) interface{} {
+	if fn, ok := expr.(*parser.FunctionCall); ok && strings.EqualFold(fn.Name, "COUNT") {
+		return int64(0)
+	}
+	return nil
+}
+
+func collectFromAliases(from []parser.TableRef) map[string]struct{} {
+	aliases := make(map[string]struct{})
+	var addRef func(parser.TableRef)
+	addRef = func(ref parser.TableRef) {
+		if ref.Name != "" {
+			aliases[strings.ToLower(ref.Name)] = struct{}{}
+		}
+		if ref.Alias != "" {
+			aliases[strings.ToLower(ref.Alias)] = struct{}{}
+		}
+		if ref.Join != nil && ref.Join.Table != nil {
+			addRef(*ref.Join.Table)
+		}
+	}
+	for _, ref := range from {
+		addRef(ref)
+	}
+	return aliases
+}
+
+func refBelongsToAliases(ref *parser.ColumnRef, aliases map[string]struct{}) bool {
+	if ref == nil || ref.Table == "" {
+		return false
+	}
+	_, ok := aliases[strings.ToLower(ref.Table)]
+	return ok
+}
+
 // evalExistsExpr evaluates an EXISTS expression.
 // Returns true if the subquery returns at least one row, false otherwise.
 func (e *Executor) evalExistsExpr(expr *parser.ExistsExpr, row storage.Row) (interface{}, error) {

pkg/executor/executor_test.go

@@ -800,6 +800,115 @@ func TestEvalSubqueryExpr(t *testing.T) {
 	execSQL(exec, "DROP TABLE IF EXISTS categories")
 }
 
+func TestChainedJoinCanHashAgainstEarlierTable(t *testing.T) {
+	pool, err := storage.NewKVPool("localhost:8085", 5, 5*time.Second)
+	if err != nil {
+		t.Skipf("PizzaKV not available: %v", err)
+	}
+	defer pool.Close()
+
+	schema := storage.NewSchemaManager(pool, "test_chained_join_hash_db")
+	table := storage.NewTableManager(pool, schema, "test_chained_join_hash_db")
+	exec := New(schema, table)
+
+	for _, sql := range []string{
+		"DROP TABLE IF EXISTS order_items",
+		"DROP TABLE IF EXISTS orders",
+		"DROP TABLE IF EXISTS addresses",
+		"CREATE TABLE orders (id INTEGER PRIMARY KEY, shipping_address_id INTEGER)",
+		"CREATE TABLE addresses (id INTEGER PRIMARY KEY, state TEXT)",
+		"CREATE TABLE order_items (id INTEGER PRIMARY KEY, order_id INTEGER, line_total REAL)",
+		"INSERT INTO addresses VALUES (1, 'CA')",
+		"INSERT INTO addresses VALUES (2, 'NY')",
+		"INSERT INTO orders VALUES (10, 1)",
+		"INSERT INTO orders VALUES (11, 2)",
+		"INSERT INTO order_items VALUES (100, 10, 25.0)",
+		"INSERT INTO order_items VALUES (101, 10, 30.0)",
+		"INSERT INTO order_items VALUES (102, 11, 10.0)",
+	} {
+		if _, err := execSQL(exec, sql); err != nil {
+			t.Fatalf("%s: %v", sql, err)
+		}
+	}
+	defer execSQL(exec, "DROP TABLE IF EXISTS order_items")
+	defer execSQL(exec, "DROP TABLE IF EXISTS orders")
+	defer execSQL(exec, "DROP TABLE IF EXISTS addresses")
+
+	result, err := execSQL(exec, `
+		SELECT a.state, COUNT(oi.id) AS lines, SUM(oi.line_total) AS revenue
+		FROM orders o
+		JOIN addresses a ON o.shipping_address_id = a.id
+		JOIN order_items oi ON oi.order_id = o.id
+		GROUP BY a.state
+		ORDER BY a.state
+	`)
+	if err != nil {
+		t.Fatalf("query failed: %v", err)
+	}
+	if len(result.Rows) != 2 {
+		t.Fatalf("expected 2 rows, got %d: %#v", len(result.Rows), result.Rows)
+	}
+	if result.Rows[0][0] != "CA" || result.Rows[0][1] != int64(2) {
+		t.Fatalf("unexpected CA row: %#v", result.Rows[0])
+	}
+	if result.Rows[1][0] != "NY" || result.Rows[1][1] != int64(1) {
+		t.Fatalf("unexpected NY row: %#v", result.Rows[1])
+	}
+}
+
+func TestCorrelatedAggregateSubqueryUsesGroupedResult(t *testing.T) {
+	pool, err := storage.NewKVPool("localhost:8085", 5, 5*time.Second)
+	if err != nil {
+		t.Skipf("PizzaKV not available: %v", err)
+	}
+	defer pool.Close()
+
+	schema := storage.NewSchemaManager(pool, "test_correlated_agg_cache_db")
+	table := storage.NewTableManager(pool, schema, "test_correlated_agg_cache_db")
+	exec := New(schema, table)
+
+	for _, sql := range []string{
+		"DROP TABLE IF EXISTS orders",
+		"DROP TABLE IF EXISTS users",
+		"CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT)",
+		"CREATE TABLE orders (id INTEGER PRIMARY KEY, user_id INTEGER)",
+		"INSERT INTO users VALUES (1, 'a@example.com')",
+		"INSERT INTO users VALUES (2, 'b@example.com')",
+		"INSERT INTO users VALUES (3, 'c@example.com')",
+		"INSERT INTO orders VALUES (10, 1)",
+		"INSERT INTO orders VALUES (11, 1)",
+		"INSERT INTO orders VALUES (12, 3)",
+		"INSERT INTO orders VALUES (13, 3)",
+		"INSERT INTO orders VALUES (14, 3)",
+	} {
+		if _, err := execSQL(exec, sql); err != nil {
+			t.Fatalf("%s: %v", sql, err)
+		}
+	}
+	defer execSQL(exec, "DROP TABLE IF EXISTS orders")
+	defer execSQL(exec, "DROP TABLE IF EXISTS users")
+
+	result, err := execSQL(exec, `
+		SELECT u.id, u.email
+		FROM users u
+		WHERE (
+			SELECT COUNT(*)
+			FROM orders o
+			WHERE o.user_id = u.id
+		) >= 2
+		ORDER BY u.id
+	`)
+	if err != nil {
+		t.Fatalf("query failed: %v", err)
+	}
+	if len(result.Rows) != 2 {
+		t.Fatalf("expected 2 rows, got %d: %#v", len(result.Rows), result.Rows)
+	}
+	if result.Rows[0][0] != int64(1) || result.Rows[1][0] != int64(3) {
+		t.Fatalf("unexpected result rows: %#v", result.Rows)
+	}
+}
+
 // Benchmark
 func BenchmarkEvalExpr(b *testing.B) {
 	exec := &Executor{}
@@ -1350,6 +1459,75 @@ func TestAlterTable(t *testing.T) {
 	execSQL(exec, "DROP TABLE IF EXISTS test_renamed")
 }
 
+func TestExecutorResyncsCatalogAfterExternalCreateTable(t *testing.T) {
+	pool, err := storage.NewKVPool("localhost:8085", 5, 5*time.Second)
+	if err != nil {
+		t.Skip("PizzaKV not available, skipping catalog resync tests")
+	}
+	defer pool.Close()
+
+	dbName := fmt.Sprintf("test_catalog_create_%d", time.Now().UnixNano())
+	schema := storage.NewSchemaManager(pool, dbName)
+	table := storage.NewTableManager(pool, schema, dbName)
+
+	staleExec := New(schema, table)
+	if err := staleExec.SyncCatalog(); err != nil {
+		t.Fatalf("initial sync: %v", err)
+	}
+
+	schemaWriter := New(schema, table)
+	if _, err := execSQL(schemaWriter, "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)"); err != nil {
+		t.Fatalf("create table through second executor: %v", err)
+	}
+	if _, err := execSQL(schemaWriter, "INSERT INTO users (id, name) VALUES (1, 'Alice')"); err != nil {
+		t.Fatalf("insert through second executor: %v", err)
+	}
+
+	result, err := execSQL(staleExec, "SELECT name FROM users WHERE id = 1")
+	if err != nil {
+		t.Fatalf("stale executor should resync and query new table: %v", err)
+	}
+	if len(result.Rows) != 1 || len(result.Rows[0]) != 1 || result.Rows[0][0] != "Alice" {
+		t.Fatalf("unexpected rows after catalog resync: %#v", result.Rows)
+	}
+}
+
+func TestExecutorResyncsCatalogAfterExternalAlterTable(t *testing.T) {
+	pool, err := storage.NewKVPool("localhost:8085", 5, 5*time.Second)
+	if err != nil {
+		t.Skip("PizzaKV not available, skipping catalog resync tests")
+	}
+	defer pool.Close()
+
+	dbName := fmt.Sprintf("test_catalog_alter_%d", time.Now().UnixNano())
+	schema := storage.NewSchemaManager(pool, dbName)
+	table := storage.NewTableManager(pool, schema, dbName)
+
+	staleExec := New(schema, table)
+	if _, err := execSQL(staleExec, "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)"); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+	if _, err := execSQL(staleExec, "INSERT INTO users (id, name) VALUES (1, 'Alice')"); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+	if err := staleExec.SyncCatalog(); err != nil {
+		t.Fatalf("sync after create: %v", err)
+	}
+
+	schemaWriter := New(schema, table)
+	if _, err := execSQL(schemaWriter, "ALTER TABLE users ADD COLUMN status TEXT DEFAULT 'active'"); err != nil {
+		t.Fatalf("alter table through second executor: %v", err)
+	}
+
+	result, err := execSQL(staleExec, "SELECT status FROM users WHERE id = 1")
+	if err != nil {
+		t.Fatalf("stale executor should resync and query new column: %v", err)
+	}
+	if len(result.Rows) != 1 {
+		t.Fatalf("expected one row after catalog resync, got %#v", result.Rows)
+	}
+}
+
 // Test ATTACH/DETACH DATABASE statements
 func TestAttachDetach(t *testing.T) {
 	pool, err := storage.NewKVPool("localhost:8085", 5, 5*time.Second)

pkg/storage/schema.go

@@ -47,18 +47,21 @@ type IndexColumn struct {
 
 // SchemaManager manages table schemas.
 type SchemaManager struct {
-	pool     *KVPool
-	database string
-	cache    map[string]*Schema
-	mu       sync.RWMutex
+	pool             *KVPool
+	database         string
+	cache            map[string]*Schema
+	rowIDInitialized map[string]bool
+	version          uint64
+	mu               sync.RWMutex
 }
 
 // NewSchemaManager creates a new schema manager.
 func NewSchemaManager(pool *KVPool, database string) *SchemaManager {
 	return &SchemaManager{
-		pool:     pool,
-		database: database,
-		cache:    make(map[string]*Schema),
+		pool:             pool,
+		database:         database,
+		cache:            make(map[string]*Schema),
+		rowIDInitialized: make(map[string]bool),
 	}
 }
 
@@ -72,6 +75,19 @@ func (m *SchemaManager) GetPool() *KVPool {
 	return m.pool
 }
 
+// Version returns the in-process schema catalog version. It is incremented for
+// schema/index definition changes so cached executors can resync their analyzer
+// catalogs without scanning storage on every query.
+func (m *SchemaManager) Version() uint64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.version
+}
+
+func (m *SchemaManager) bumpVersionLocked() {
+	m.version++
+}
+
 // schemaKey returns the key for a table schema.
 func (m *SchemaManager) schemaKey(table string) string {
 	return fmt.Sprintf("%s:_schema:%s", m.database, strings.ToLower(table))
@@ -145,6 +161,7 @@ func (m *SchemaManager) CreateTable(schema *Schema) error {
 
 	// Update cache
 	m.cache[strings.ToLower(schema.Name)] = schema
+	m.bumpVersionLocked()
 
 	return nil
 }
@@ -199,7 +216,10 @@ func (m *SchemaManager) DropTable(name string) error {
 	}
 
 	// Update cache
-	delete(m.cache, strings.ToLower(name))
+	tableLower := strings.ToLower(name)
+	delete(m.cache, tableLower)
+	delete(m.rowIDInitialized, tableLower)
+	m.bumpVersionLocked()
 
 	return nil
 }
@@ -328,7 +348,9 @@ func (m *SchemaManager) removeFromCatalog(name string) error {
 func (m *SchemaManager) InvalidateCache(name string) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	delete(m.cache, strings.ToLower(name))
+	tableLower := strings.ToLower(name)
+	delete(m.cache, tableLower)
+	delete(m.rowIDInitialized, tableLower)
 }
 
 // ToAnalyzerTableInfo converts a Schema to analyzer.TableInfo.
@@ -376,9 +398,7 @@ func (m *SchemaManager) GetNextRowID(table string) (int64, error) {
 		return 0, err
 	}
 
-	if err := m.saveNextRowIDLocked(schema.Name, nextRowID+1); err != nil {
-		return 0, err
-	}
+	schema.NextRowID = nextRowID + 1
 
 	return nextRowID, nil
 }
@@ -399,58 +419,79 @@ func (m *SchemaManager) UpdateMaxRowID(table string, rowid int64) error {
 	}
 
 	if rowid >= nextRowID {
-		return m.saveNextRowIDLocked(schema.Name, rowid+1)
+		schema.NextRowID = rowid + 1
 	}
 
 	return nil
 }
 
-// getNextRowIDLocked reads a table's next ROWID counter (must hold lock).
+// getNextRowIDLocked returns a table's in-memory ROWID counter (must hold lock).
+// On first use after startup, the counter is derived from durable row data so
+// ROWID movement does not add a separate WAL entry.
 func (m *SchemaManager) getNextRowIDLocked(schema *Schema) (int64, error) {
-	key := m.rowIDKey(schema.Name)
-	var data string
-	err := m.pool.WithClient(func(c *KVClient) error {
-		var err error
-		data, err = c.Read(key)
-		return err
-	})
-	if err == nil {
-		var nextRowID int64
-		if _, scanErr := fmt.Sscanf(data, "%d", &nextRowID); scanErr != nil {
-			return 0, fmt.Errorf("failed to parse rowid counter: %w", scanErr)
+	tableLower := strings.ToLower(schema.Name)
+	if m.rowIDInitialized[tableLower] {
+		if schema.NextRowID < 1 {
+			schema.NextRowID = 1
 		}
-		if nextRowID < 1 {
-			nextRowID = 1
-		}
-		return nextRowID, nil
+		return schema.NextRowID, nil
 	}
-	if err != ErrKeyNotFound {
+
+	nextRowID, err := m.deriveNextRowIDLocked(schema)
+	if err != nil {
 		return 0, err
 	}
-
-	if schema.NextRowID > 0 {
-		return schema.NextRowID, nil
+	if schema.NextRowID > nextRowID {
+		nextRowID = schema.NextRowID
 	}
-	return 1, nil
-}
-
-// saveNextRowIDLocked saves a table's next ROWID counter (must hold lock).
-func (m *SchemaManager) saveNextRowIDLocked(table string, nextRowID int64) error {
 	if nextRowID < 1 {
 		nextRowID = 1
 	}
 
+	schema.NextRowID = nextRowID
+	m.rowIDInitialized[tableLower] = true
+	return schema.NextRowID, nil
+}
+
+// deriveNextRowIDLocked scans durable row values to recover max(rowid)+1.
+func (m *SchemaManager) deriveNextRowIDLocked(schema *Schema) (int64, error) {
+	prefix := fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(schema.Name))
+	var values []string
 	err := m.pool.WithClient(func(c *KVClient) error {
-		return c.Write(m.rowIDKey(table), fmt.Sprintf("%d", nextRowID))
+		var err error
+		values, err = c.Reads(prefix)
+		return err
 	})
 	if err != nil {
-		return fmt.Errorf("failed to write rowid counter: %w", err)
+		return 0, err
 	}
 
-	if schema, ok := m.cache[strings.ToLower(table)]; ok {
-		schema.NextRowID = nextRowID
+	var maxRowID int64
+	for _, value := range values {
+		var row Row
+		if err := json.Unmarshal([]byte(value), &row); err != nil {
+			return 0, fmt.Errorf("failed to parse row while deriving ROWID: %w", err)
+		}
+
+		if rowid, ok := valueAsInt64(row["_rowid_"]); ok && rowid > maxRowID {
+			maxRowID = rowid
+		}
+	}
+
+	return maxRowID + 1, nil
+}
+
+func valueAsInt64(value interface{}) (int64, bool) {
+	switch v := value.(type) {
+	case int64:
+		return v, true
+	case int:
+		return int64(v), true
+	case float64:
+		return int64(v), true
+	default:
+		return 0, false
 	}
-	return nil
 }
 
 // getSchemaLocked retrieves schema (must hold lock).
@@ -548,7 +589,11 @@ func (m *SchemaManager) CreateIndex(index *Index) error {
 	}
 
 	// Add to index list
-	return m.addToIndexList(index.Name)
+	if err := m.addToIndexList(index.Name); err != nil {
+		return err
+	}
+	m.bumpVersionLocked()
+	return nil
 }
 
 // DropIndex drops an index.
@@ -564,7 +609,11 @@ func (m *SchemaManager) DropIndex(name string) error {
 		return fmt.Errorf("failed to delete index: %w", err)
 	}
 
-	return m.removeFromIndexList(name)
+	if err := m.removeFromIndexList(name); err != nil {
+		return err
+	}
+	m.bumpVersionLocked()
+	return nil
 }
 
 // IndexExists checks if an index exists.
@@ -781,15 +830,7 @@ func (m *SchemaManager) RenameTable(oldName, newName string) error {
 	// Update schema name
 	schema.Name = newName
 
-	var nextRowID string
 	rowIDKey := m.rowIDKey(oldName)
-	m.pool.WithClient(func(c *KVClient) error {
-		data, err := c.Read(rowIDKey)
-		if err == nil {
-			nextRowID = data
-		}
-		return nil
-	})
 
 	// Delete old schema
 	oldKey := m.schemaKey(oldName)
@@ -807,17 +848,13 @@ func (m *SchemaManager) RenameTable(oldName, newName string) error {
 	m.pool.WithClient(func(c *KVClient) error {
 		return c.Delete(rowIDKey)
 	})
-	if nextRowID != "" {
-		err = m.pool.WithClient(func(c *KVClient) error {
-			return c.Write(m.rowIDKey(newName), nextRowID)
-		})
-		if err != nil {
-			return err
-		}
-	}
 
 	// Update cache
-	delete(m.cache, strings.ToLower(oldName))
+	oldLower := strings.ToLower(oldName)
+	newLower := strings.ToLower(newName)
+	wasInitialized := m.rowIDInitialized[oldLower]
+	delete(m.cache, oldLower)
+	delete(m.rowIDInitialized, oldLower)
 
 	// Write new schema
 	newKey := m.schemaKey(newName)
@@ -833,7 +870,11 @@ func (m *SchemaManager) RenameTable(oldName, newName string) error {
 	m.addToCatalog(newName)
 
 	// Update cache
-	m.cache[strings.ToLower(newName)] = schema
+	m.cache[newLower] = schema
+	if wasInitialized {
+		m.rowIDInitialized[newLower] = true
+	}
+	m.bumpVersionLocked()
 
 	return nil
 }
@@ -922,5 +963,6 @@ func (m *SchemaManager) updateSchemaUnsafe(schema *Schema) error {
 
 	// Update cache
 	m.cache[strings.ToLower(schema.Name)] = schema
+	m.bumpVersionLocked()
 	return nil
 }

pkg/storage/schema_test.go

@@ -76,6 +76,14 @@ func (s *testKVServer) writeCount(prefix string) int {
 	return count
 }
 
+func (s *testKVServer) hasKey(key string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	_, ok := s.data[key]
+	return ok
+}
+
 func (s *testKVServer) handle(conn net.Conn) {
 	defer conn.Close()
 
@@ -168,7 +176,160 @@ func TestInsertDoesNotRewriteSchemaForRowIDUpdates(t *testing.T) {
 	if got := kv.writeCount(":_schema:"); got != initialSchemaWrites {
 		t.Fatalf("expected inserts not to rewrite schema, got %d schema writes", got)
 	}
-	if got := kv.writeCount(":_sys:rowid:"); got != 3 {
-		t.Fatalf("expected rowid counter writes for inserts, got %d", got)
+	if got := kv.writeCount(":_sys:rowid:"); got != 0 {
+		t.Fatalf("expected no rowid counter writes for inserts, got %d", got)
+	}
+}
+
+func TestRowIDIsDerivedFromRowsAfterRestart(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+
+	pool := newTestKVPool(kv, 2, 5*time.Second)
+	defer pool.Close()
+
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+
+	err := schemas.CreateTable(&Schema{
+		Name: "events",
+		Columns: []Column{
+			{Name: "name", Type: "TEXT", Nullable: true},
+		},
+	})
+	if err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	for i := 1; i <= 2; i++ {
+		err := tables.Insert("events", Row{"name": fmt.Sprintf("event-%d", i)})
+		if err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+	}
+
+	// Simulate a process restart: new managers have empty in-memory ROWID state
+	// but the same durable KV rows.
+	restartedSchemas := NewSchemaManager(pool, "testdb")
+	restartedTables := NewTableManager(pool, restartedSchemas, "testdb")
+	if err := restartedTables.Insert("events", Row{"name": "event-3"}); err != nil {
+		t.Fatalf("insert after restart: %v", err)
+	}
+
+	if !kv.hasKey("testdb:_data:events:3") {
+		t.Fatalf("expected restart insert to continue at rowid 3")
+	}
+	if got := kv.writeCount(":_sys:rowid:"); got != 0 {
+		t.Fatalf("expected no rowid counter writes, got %d", got)
+	}
+}
+
+func TestInsertDoesNotWriteDurableIndexEntries(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+
+	pool := newTestKVPool(kv, 2, 5*time.Second)
+	defer pool.Close()
+
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+
+	err := schemas.CreateTable(&Schema{
+		Name: "users",
+		Columns: []Column{
+			{Name: "id", Type: "INTEGER", Nullable: false, PrimaryKey: true},
+			{Name: "status", Type: "TEXT", Nullable: false},
+		},
+	})
+	if err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+	err = schemas.CreateIndex(&Index{
+		Name:  "idx_users_status",
+		Table: "users",
+		Columns: []IndexColumn{
+			{Name: "status"},
+		},
+	})
+	if err != nil {
+		t.Fatalf("create index: %v", err)
+	}
+
+	for i := int64(1); i <= 3; i++ {
+		status := "active"
+		if i == 2 {
+			status = "inactive"
+		}
+		err := tables.Insert("users", Row{"id": i, "status": status})
+		if err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+	}
+
+	if got := kv.writeCount(":idx:"); got != 0 {
+		t.Fatalf("expected no durable index entry writes, got %d", got)
+	}
+
+	rows, err := tables.SelectByIndex("users", "idx_users_status", "active")
+	if err != nil {
+		t.Fatalf("select by index: %v", err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("expected 2 active rows from derived index, got %d", len(rows))
+	}
+}
+
+func TestIndexIsDerivedFromRowsAfterRestart(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+
+	pool := newTestKVPool(kv, 2, 5*time.Second)
+	defer pool.Close()
+
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+
+	err := schemas.CreateTable(&Schema{
+		Name: "users",
+		Columns: []Column{
+			{Name: "id", Type: "INTEGER", Nullable: false, PrimaryKey: true},
+			{Name: "status", Type: "TEXT", Nullable: false},
+		},
+	})
+	if err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+	err = schemas.CreateIndex(&Index{
+		Name:  "idx_users_status",
+		Table: "users",
+		Columns: []IndexColumn{
+			{Name: "status"},
+		},
+	})
+	if err != nil {
+		t.Fatalf("create index: %v", err)
+	}
+	for i := int64(1); i <= 3; i++ {
+		status := "active"
+		if i == 3 {
+			status = "inactive"
+		}
+		if err := tables.Insert("users", Row{"id": i, "status": status}); err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+	}
+
+	restartedSchemas := NewSchemaManager(pool, "testdb")
+	restartedTables := NewTableManager(pool, restartedSchemas, "testdb")
+
+	rows, err := restartedTables.SelectByIndex("users", "idx_users_status", "active")
+	if err != nil {
+		t.Fatalf("select by index after restart: %v", err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("expected 2 active rows from restart-derived index, got %d", len(rows))
+	}
+	if got := kv.writeCount(":idx:"); got != 0 {
+		t.Fatalf("expected no durable index entry writes, got %d", got)
 	}
 }

pkg/storage/table.go

@@ -19,22 +19,37 @@ type TableManager struct {
 
 	cacheMu  sync.RWMutex
 	rowCache map[string][]Row // table name → all rows (nil means not loaded)
+	rowIDMap map[string]map[int64]Row
+
+	indexCache map[string]map[string][]int64 // index name → indexed value → rowids
+	indexTable map[string]string             // index name → table name
 }
 
 // NewTableManager creates a new table manager.
 func NewTableManager(pool *KVPool, schema *SchemaManager, database string) *TableManager {
 	return &TableManager{
-		pool:     pool,
-		schema:   schema,
-		database: database,
-		rowCache: make(map[string][]Row),
+		pool:       pool,
+		schema:     schema,
+		database:   database,
+		rowCache:   make(map[string][]Row),
+		rowIDMap:   make(map[string]map[int64]Row),
+		indexCache: make(map[string]map[string][]int64),
+		indexTable: make(map[string]string),
 	}
 }
 
 // invalidateCache removes a table's rows from the in-memory cache.
 func (m *TableManager) invalidateCache(table string) {
 	m.cacheMu.Lock()
-	delete(m.rowCache, strings.ToLower(table))
+	key := strings.ToLower(table)
+	delete(m.rowCache, key)
+	delete(m.rowIDMap, key)
+	for indexName, tableName := range m.indexTable {
+		if tableName == key {
+			delete(m.indexCache, indexName)
+			delete(m.indexTable, indexName)
+		}
+	}
 	m.cacheMu.Unlock()
 }
 
@@ -182,7 +197,7 @@ func (m *TableManager) Insert(table string, row Row) error {
 		return err
 	}
 
-	// Update indexes
+	// Update in-memory indexes only. Durable index entries are derived from rows.
 	m.updateIndexesForRow(table, normalizedRow, true)
 
 	m.invalidateCache(table)
@@ -288,93 +303,33 @@ func (m *TableManager) InsertBulk(table string, rows []Row) (int, error) {
 		}
 	}
 
-	// Build index entries grouped by key (to avoid read-merge races).
-	indexes, _ := m.schema.ListTableIndexes(table)
-	if len(indexes) > 0 {
-		// For each index, gather {indexKey → []rowid} from the new rows.
-		type indexEntry struct {
-			key    string
-			rowids []int64
-		}
-		var entries []indexEntry
-
-		for _, idx := range indexes {
-			cols := make([]string, len(idx.Columns))
-			for i, c := range idx.Columns {
-				cols[i] = c.Name
-			}
-			byKey := make(map[string][]int64)
-			for _, nr := range normalized {
-				colVal := m.buildIndexValue(nr, cols)
-				ikey := m.indexEntryKey(idx.Name, colVal)
-				var rowid int64
-				switch v := nr["_rowid_"].(type) {
-				case int64:
-					rowid = v
-				case float64:
-					rowid = int64(v)
-				}
-				byKey[ikey] = append(byKey[ikey], rowid)
-			}
-			for k, rs := range byKey {
-				entries = append(entries, indexEntry{k, rs})
-			}
-		}
-
-		// Write index entries concurrently; each key is handled by exactly
-		// one goroutine so there's no merge race.
-		ieErrs := make([]error, len(entries))
-		var iwg sync.WaitGroup
-		for i, e := range entries {
-			iwg.Add(1)
-			i, e := i, e
-			go func() {
-				defer iwg.Done()
-				// Merge with any existing rowids for this key.
-				var existing []int64
-				m.pool.WithClient(func(c *KVClient) error {
-					data, err := c.Read(e.key)
-					if err == nil {
-						json.Unmarshal([]byte(data), &existing)
-					}
-					return nil
-				})
-				merged := append(existing, e.rowids...)
-				data, _ := json.Marshal(merged)
-				ieErrs[i] = m.pool.WithClient(func(c *KVClient) error {
-					return c.Write(e.key, string(data))
-				})
-			}()
-		}
-		iwg.Wait()
-		for _, e := range ieErrs {
-			if e != nil {
-				return 0, e
-			}
-		}
-	}
-
 	m.invalidateCache(table)
 	return len(normalized), nil
 }
 
-// updateIndexesForRow adds or removes index entries for a row.
+// 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) {
 	indexes, err := m.schema.ListTableIndexes(table)
 	if err != nil || len(indexes) == 0 {
 		return
 	}
 
-	rowid, ok := row["_rowid_"].(float64)
+	rowid, ok := rowIDFromRow(row)
 	if !ok {
-		if rid, ok := row["_rowid_"].(int64); ok {
-			rowid = float64(rid)
-		} else {
-			return
-		}
+		return
 	}
 
 	for _, idx := range indexes {
+		indexName := strings.ToLower(idx.Name)
+		m.cacheMu.RLock()
+		_, initialized := m.indexCache[indexName]
+		m.cacheMu.RUnlock()
+		if !initialized {
+			continue
+		}
+
 		columns := make([]string, len(idx.Columns))
 		for i, col := range idx.Columns {
 			columns[i] = col.Name
@@ -382,9 +337,9 @@ func (m *TableManager) updateIndexesForRow(table string, row Row, add bool) {
 		colValue := m.buildIndexValue(row, columns)
 
 		if add {
-			m.AddIndexEntry(idx.Name, colValue, int64(rowid))
+			m.AddIndexEntry(idx.Name, colValue, rowid)
 		} else {
-			m.RemoveIndexEntry(idx.Name, colValue, int64(rowid))
+			m.RemoveIndexEntry(idx.Name, colValue, rowid)
 		}
 	}
 }
@@ -414,16 +369,21 @@ func (m *TableManager) Select(table string, filter func(Row) bool) ([]Row, error
 		}
 
 		loaded := make([]Row, 0, len(values))
+		byRowID := make(map[int64]Row, len(values))
 		for _, data := range values {
 			var row Row
 			if err := json.Unmarshal([]byte(data), &row); err != nil {
 				continue
 			}
 			loaded = append(loaded, row)
+			if rowid, ok := valueAsInt64(row["_rowid_"]); ok {
+				byRowID[rowid] = row
+			}
 		}
 
 		m.cacheMu.Lock()
 		m.rowCache[key] = loaded
+		m.rowIDMap[key] = byRowID
 		m.cacheMu.Unlock()
 
 		cached = loaded
@@ -682,82 +642,118 @@ func IsRowIDColumn(name string) bool {
 
 // indexEntryKey returns the key for an index entry.
 func (m *TableManager) indexEntryKey(indexName string, colValue interface{}) string {
-	// Format the value without scientific notation
-	var valueStr string
-	switch v := colValue.(type) {
+	return fmt.Sprintf("%s:idx:%s:%s", m.database, strings.ToLower(indexName), formatIndexValue(colValue))
+}
+
+// indexPrefix returns the prefix for all entries of an index.
+func (m *TableManager) indexPrefix(indexName string) string {
+	return fmt.Sprintf("%s:idx:%s:", m.database, strings.ToLower(indexName))
+}
+
+func formatIndexValue(value interface{}) string {
+	switch v := value.(type) {
 	case float64:
-		// Check if it's actually an integer value
 		if v == float64(int64(v)) {
-			valueStr = fmt.Sprintf("%d", int64(v))
-		} else {
-			valueStr = fmt.Sprintf("%f", v)
+			return fmt.Sprintf("%d", int64(v))
 		}
+		return fmt.Sprintf("%f", v)
 	case int64:
-		valueStr = fmt.Sprintf("%d", v)
+		return fmt.Sprintf("%d", v)
 	case int:
-		valueStr = fmt.Sprintf("%d", v)
+		return fmt.Sprintf("%d", v)
 	default:
-		valueStr = fmt.Sprintf("%v", v)
+		return fmt.Sprintf("%v", v)
 	}
-	return fmt.Sprintf("%s:idx:%s:%s", m.database, strings.ToLower(indexName), valueStr)
 }
 
-// indexPrefix returns the prefix for all entries of an index.
-func (m *TableManager) indexPrefix(indexName string) string {
-	return fmt.Sprintf("%s:idx:%s:", m.database, strings.ToLower(indexName))
+func rowIDFromRow(row Row) (int64, bool) {
+	switch v := row["_rowid_"].(type) {
+	case int64:
+		return v, true
+	case int:
+		return int64(v), true
+	case float64:
+		return int64(v), true
+	default:
+		return 0, false
+	}
 }
 
-// AddIndexEntry adds a rowid to an index entry.
-func (m *TableManager) AddIndexEntry(indexName string, colValue interface{}, rowid int64) error {
-	key := m.indexEntryKey(indexName, colValue)
+func (m *TableManager) ensureIndex(index *Index) error {
+	indexKey := strings.ToLower(index.Name)
+	m.cacheMu.RLock()
+	_, initialized := m.indexCache[indexKey]
+	m.cacheMu.RUnlock()
+	if initialized {
+		return nil
+	}
 
-	// Read existing rowids
-	var rowids []int64
-	err := m.pool.WithClient(func(c *KVClient) error {
-		data, err := c.Read(key)
-		if err == nil && data != "" {
-			json.Unmarshal([]byte(data), &rowids)
-		}
-		return nil // Ignore not found errors
-	})
+	columns := make([]string, len(index.Columns))
+	for i, col := range index.Columns {
+		columns[i] = col.Name
+	}
+
+	rows, err := m.Select(index.Table, nil)
 	if err != nil {
 		return err
 	}
 
-	// Add new rowid if not already present
+	values := make(map[string][]int64)
+	for _, row := range rows {
+		rowid, ok := rowIDFromRow(row)
+		if !ok {
+			continue
+		}
+		colValue := m.buildIndexValue(row, columns)
+		valueKey := formatIndexValue(colValue)
+		values[valueKey] = append(values[valueKey], rowid)
+	}
+
+	m.cacheMu.Lock()
+	if _, initialized := m.indexCache[indexKey]; !initialized {
+		m.indexCache[indexKey] = values
+		m.indexTable[indexKey] = strings.ToLower(index.Table)
+	}
+	m.cacheMu.Unlock()
+
+	return nil
+}
+
+// AddIndexEntry adds a rowid to an in-memory index entry.
+func (m *TableManager) AddIndexEntry(indexName string, colValue interface{}, rowid int64) error {
+	indexKey := strings.ToLower(indexName)
+	valueKey := formatIndexValue(colValue)
+
+	m.cacheMu.Lock()
+	defer m.cacheMu.Unlock()
+
+	values, ok := m.indexCache[indexKey]
+	if !ok {
+		return nil
+	}
+	rowids := values[valueKey]
 	for _, r := range rowids {
 		if r == rowid {
-			return nil // Already exists
+			return nil
 		}
 	}
-	rowids = append(rowids, rowid)
-
-	// Write back
-	data, _ := json.Marshal(rowids)
-	return m.pool.WithClient(func(c *KVClient) error {
-		return c.Write(key, string(data))
-	})
+	values[valueKey] = append(rowids, rowid)
+	return nil
 }
 
-// RemoveIndexEntry removes a rowid from an index entry.
+// RemoveIndexEntry removes a rowid from an in-memory index entry.
 func (m *TableManager) RemoveIndexEntry(indexName string, colValue interface{}, rowid int64) error {
-	key := m.indexEntryKey(indexName, colValue)
+	indexKey := strings.ToLower(indexName)
+	valueKey := formatIndexValue(colValue)
 
-	// Read existing rowids
-	var rowids []int64
-	err := m.pool.WithClient(func(c *KVClient) error {
-		data, err := c.Read(key)
-		if err != nil {
-			return err
-		}
-		json.Unmarshal([]byte(data), &rowids)
+	m.cacheMu.Lock()
+	defer m.cacheMu.Unlock()
+
+	values, ok := m.indexCache[indexKey]
+	if !ok {
 		return nil
-	})
-	if err != nil {
-		return nil // Entry doesn't exist
 	}
-
-	// Remove rowid
+	rowids := values[valueKey]
 	newRowids := make([]int64, 0, len(rowids))
 	for _, r := range rowids {
 		if r != rowid {
@@ -766,35 +762,31 @@ func (m *TableManager) RemoveIndexEntry(indexName string, colValue interface{},
 	}
 
 	if len(newRowids) == 0 {
-		// Delete the entry entirely
-		return m.pool.WithClient(func(c *KVClient) error {
-			return c.Delete(key)
-		})
+		delete(values, valueKey)
+		return nil
 	}
 
-	// Write back
-	data, _ := json.Marshal(newRowids)
-	return m.pool.WithClient(func(c *KVClient) error {
-		return c.Write(key, string(data))
-	})
+	values[valueKey] = newRowids
+	return nil
 }
 
 // LookupIndex returns rowids matching a column value using the index.
 func (m *TableManager) LookupIndex(indexName string, colValue interface{}) ([]int64, error) {
-	key := m.indexEntryKey(indexName, colValue)
-
-	var rowids []int64
-	err := m.pool.WithClient(func(c *KVClient) error {
-		data, err := c.Read(key)
-		if err != nil {
-			return err
-		}
-		return json.Unmarshal([]byte(data), &rowids)
-	})
+	index, err := m.schema.GetIndex(indexName)
 	if err != nil {
-		return nil, nil // Return empty if not found
+		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
 }
 
@@ -818,24 +810,33 @@ func (m *TableManager) ClearIndex(indexName, tableName string, columns []string)
 
 // BuildIndex builds index entries for all existing rows in a table.
 func (m *TableManager) BuildIndex(indexName, tableName string, columns []string) error {
+	index, err := m.schema.GetIndex(indexName)
+	if err == nil {
+		return m.ensureIndex(index)
+	}
+
 	rows, err := m.Select(tableName, nil)
 	if err != nil {
 		return err
 	}
 
+	values := make(map[string][]int64)
 	for _, row := range rows {
-		rowid, ok := row["_rowid_"].(float64)
+		rowid, ok := rowIDFromRow(row)
 		if !ok {
 			continue
 		}
 
-		// Build composite key value for multi-column indexes
 		colValue := m.buildIndexValue(row, columns)
-		if err := m.AddIndexEntry(indexName, colValue, int64(rowid)); err != nil {
-			return err
-		}
+		values[formatIndexValue(colValue)] = append(values[formatIndexValue(colValue)], rowid)
 	}
 
+	indexKey := strings.ToLower(indexName)
+	m.cacheMu.Lock()
+	m.indexCache[indexKey] = values
+	m.indexTable[indexKey] = strings.ToLower(tableName)
+	m.cacheMu.Unlock()
+
 	return nil
 }
 
@@ -883,23 +884,30 @@ func (m *TableManager) SelectByIndex(table, indexName string, colValue interface
 	}
 
 	// Build a set of target rowids for O(1) lookup.
-	rowidSet := make(map[int64]struct{}, len(rowids))
-	for _, rid := range rowids {
-		rowidSet[rid] = struct{}{}
-	}
-	allRows, _ := m.Select(table, func(r Row) bool {
-		switch v := r["_rowid_"].(type) {
-		case float64:
-			_, ok := rowidSet[int64(v)]
-			return ok
-		case int64:
-			_, ok := rowidSet[v]
-			return ok
+	key := strings.ToLower(table)
+	m.cacheMu.RLock()
+	byRowID, ok := m.rowIDMap[key]
+	m.cacheMu.RUnlock()
+	if !ok {
+		if _, err := m.Select(table, nil); err != nil {
+			return nil, err
 		}
-		return false
-	})
+		m.cacheMu.RLock()
+		byRowID = m.rowIDMap[key]
+		m.cacheMu.RUnlock()
+	}
 
-	rows := allRows
+	rows := make([]Row, 0, len(rowids))
+	seen := make(map[int64]struct{}, len(rowids))
+	for _, rid := range rowids {
+		if _, duplicate := seen[rid]; duplicate {
+			continue
+		}
+		seen[rid] = struct{}{}
+		if row, ok := byRowID[rid]; ok {
+			rows = append(rows, row)
+		}
+	}
 
 	return rows, nil
 }