sqllogictest + views + index fixes

.gitignore

@@ -1,2 +1,5 @@
 .DS_Store
-.claude
+.claude
+testdata
+.db
+*.log

Makefile

@@ -1,4 +1,4 @@
-.PHONY: build test test-v test-cover bench clean fmt lint
+.PHONY: build test test-v test-cover bench clean fmt lint sqllogictest sqllogictest-basic sqllogictest-download build-sqllogictest
 
 # Build the project
 build:
@@ -52,3 +52,35 @@ test-race:
 # Quick test for development
 quick:
 	go test -short ./...
+
+# ── sqllogictest ──────────────────────────────────────────────────────────────
+
+# Run the sqllogictest suite against a running PizzaSQL server (default: localhost:8080).
+# Set URL= to point at a different server, e.g. make sqllogictest URL=http://host:9090
+sqllogictest:
+	go run ./cmd/sqllogictest -url $(or $(URL),http://localhost:8080) -dir testdata/sqllogictest -log sqllogictest-failures.log
+
+# Run only the built-in smoke test.
+sqllogictest-basic:
+	go run ./cmd/sqllogictest -url $(or $(URL),http://localhost:8080) -file testdata/sqllogictest/basic.test
+
+# Download the official SQLite sqllogictest corpus into testdata/sqllogictest/.
+# Requires curl and tar. Files are not committed to the repo.
+sqllogictest-download:
+	@echo "Downloading SQLite sqllogictest corpus..."
+	mkdir -p testdata/sqllogictest
+	mkdir -p /tmp/sqllogictest-dl
+	curl -fsSL https://github.com/gregrahn/sqllogictest/archive/refs/heads/master.tar.gz \
+	  | tar xz -C /tmp/sqllogictest-dl
+	find /tmp/sqllogictest-dl -name '*.test' | while read f; do \
+	  rel=$$(echo "$$f" | sed 's|.*/test/||'); \
+	  dir=testdata/sqllogictest/$$(dirname "$$rel"); \
+	  mkdir -p "$$dir"; \
+	  cp "$$f" "$$dir/"; \
+	done
+	rm -rf /tmp/sqllogictest-dl
+	@echo "Done. $$(find testdata/sqllogictest -name '*.test' | wc -l | tr -d ' ') test files in testdata/sqllogictest/"
+
+# Build the sqllogictest runner binary.
+build-sqllogictest:
+	go build -o ./bin/sqllogictest ./cmd/sqllogictest

cmd/sqllogictest/main.go

@@ -0,0 +1,745 @@
+// sqllogictest runner — sends SQL via HTTP to a running PizzaSQL server and
+// compares results against the expected output in .test files.
+//
+// File format: https://www.sqlite.org/sqllogictest/doc/trunk/about.wiki
+//
+// Usage:
+//
+//	go run ./cmd/sqllogictest -url http://localhost:8080 -dir testdata/sqllogictest
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"crypto/md5"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"math"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const engineName = "pizzasql"
+
+// ANSI color helpers
+const (
+	colorReset  = "\033[0m"
+	colorRed    = "\033[31m"
+	colorGreen  = "\033[32m"
+	colorYellow = "\033[33m"
+	colorCyan   = "\033[36m"
+	colorBold   = "\033[1m"
+	colorDim    = "\033[2m"
+)
+
+// ── types ────────────────────────────────────────────────────────────────────
+
+type lineInfo struct {
+	text string
+	num  int
+}
+
+type record struct {
+	isStatement bool
+	isQuery     bool
+	expectOK    bool     // statement: true → expect success
+	typeStr     string   // query: column type chars (I/R/T)
+	sortMode    string   // nosort | rowsort | valuesort
+	label       string
+	sql         string
+	expected    []string // flattened expected values, one per line
+	skip        bool
+	file        string
+	line        int
+}
+
+type queryRequest struct {
+	SQL string `json:"sql"`
+}
+
+type queryResponse struct {
+	Columns []struct {
+		Name string `json:"name"`
+		Type string `json:"type"`
+	} `json:"columns"`
+	Rows  [][]interface{} `json:"rows"`
+	Error *struct {
+		Code    string `json:"code"`
+		Message string `json:"message"`
+	} `json:"error"`
+}
+
+// ── runner ───────────────────────────────────────────────────────────────────
+
+type runner struct {
+	baseURL    string
+	client     *http.Client
+	verbose    bool
+	stopOnFail bool
+	passed     int
+	failed     int
+	skipped    int
+	total      int    // total files to run
+	filesDone  int    // files completed
+	logW       *bufio.Writer
+	logPath    string
+}
+
+func main() {
+	urlFlag     := flag.String("url", "http://localhost:8080", "PizzaSQL server URL")
+	dirFlag     := flag.String("dir", "testdata/sqllogictest", "Directory containing .test files")
+	fileFlag    := flag.String("file", "", "Single .test file to run (overrides -dir)")
+	verboseFlag := flag.Bool("v", false, "Print each passing record")
+	stopFlag    := flag.Bool("stop", false, "Stop on first failure")
+	logFlag     := flag.String("log", "sqllogictest-failures.log", "File to write failures to ('' to disable)")
+	flag.Parse()
+
+	r := &runner{
+		baseURL:    strings.TrimRight(*urlFlag, "/"),
+		client:     &http.Client{Timeout: 120 * time.Second},
+		verbose:    *verboseFlag,
+		stopOnFail: *stopFlag,
+		logPath:    *logFlag,
+	}
+
+	if *logFlag != "" {
+		lf, err := os.Create(*logFlag)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "cannot open log file: %v\n", err)
+			os.Exit(1)
+		}
+		defer lf.Close()
+		r.logW = bufio.NewWriter(lf)
+		defer r.logW.Flush()
+	}
+
+	var files []string
+	if *fileFlag != "" {
+		files = []string{*fileFlag}
+	} else {
+		err := filepath.WalkDir(*dirFlag, func(path string, d os.DirEntry, err error) error {
+			if err != nil {
+				return err
+			}
+			if !d.IsDir() && strings.HasSuffix(path, ".test") {
+				files = append(files, path)
+			}
+			return nil
+		})
+		if err != nil || len(files) == 0 {
+			fmt.Fprintf(os.Stderr, "no .test files found in %s\n", *dirFlag)
+			os.Exit(1)
+		}
+		sort.Strings(files)
+	}
+
+	r.total = len(files)
+	start := time.Now()
+	for _, f := range files {
+		if err := r.runFile(f, start); err != nil {
+			fmt.Fprintf(os.Stderr, "error in %s: %v\n", f, err)
+		}
+		if r.stopOnFail && r.failed > 0 {
+			break
+		}
+	}
+
+	// clear the progress line
+	fmt.Print("\r\033[K")
+
+	total := r.passed + r.failed
+	elapsed := time.Since(start).Round(time.Millisecond)
+
+	passColor, failColor := colorGreen, colorDim
+	if r.failed > 0 {
+		failColor = colorRed
+	}
+	pct := 0.0
+	if total > 0 {
+		pct = 100.0 * float64(r.passed) / float64(total)
+	}
+
+	fmt.Printf("%s--- Summary ---%s\n", colorBold, colorReset)
+	var summaryQPS string
+	if secs := elapsed.Seconds(); secs > 0 && total > 0 {
+		qps := float64(total) / secs
+		switch {
+		case qps >= 1_000_000:
+			summaryQPS = fmt.Sprintf("%.2fM q/s", qps/1_000_000)
+		case qps >= 1_000:
+			summaryQPS = fmt.Sprintf("%.2fk q/s", qps/1_000)
+		default:
+			summaryQPS = fmt.Sprintf("%.0f q/s", qps)
+		}
+	}
+
+	fmt.Printf("passed:  %s%d/%d (%.1f%%)%s\n", passColor, r.passed, total, pct, colorReset)
+	fmt.Printf("failed:  %s%d%s\n", failColor, r.failed, colorReset)
+	fmt.Printf("skipped: %d\n", r.skipped)
+	fmt.Printf("time:    %s\n", elapsed)
+	fmt.Printf("thru:    %s%s%s\n", colorCyan, summaryQPS, colorReset)
+	if r.failed > 0 && *logFlag != "" {
+		fmt.Printf("log:     %s%s%s\n", colorCyan, *logFlag, colorReset)
+	}
+	if r.failed > 0 {
+		os.Exit(1)
+	}
+}
+
+func (r *runner) runFile(path string, start time.Time) error {
+	f, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	records, err := parseFile(path, f)
+	if err != nil {
+		return err
+	}
+
+	// Drop any tables/views this file creates so it always runs against a clean state.
+	for _, tbl := range collectCreatedTables(records) {
+		r.execQuery("DROP TABLE IF EXISTS " + tbl) //nolint:errcheck
+	}
+	for _, v := range collectCreatedViews(records) {
+		r.execQuery("DROP VIEW IF EXISTS " + v) //nolint:errcheck
+	}
+
+	failsBefore := r.failed
+	for _, rec := range records {
+		if r.stopOnFail && r.failed > 0 {
+			break
+		}
+		if rec.skip {
+			r.skipped++
+			continue
+		}
+		r.runRecord(rec)
+		r.printProgress(path, start)
+	}
+
+	r.filesDone++
+	newFails := r.failed - failsBefore
+	rel, _ := filepath.Rel("testdata/sqllogictest", path)
+	if rel == "" {
+		rel = filepath.Base(path)
+	}
+	var statusStr string
+	if newFails == 0 {
+		statusStr = colorGreen + "ok" + colorReset
+	} else {
+		statusStr = fmt.Sprintf("%s%d FAILED%s", colorRed, newFails, colorReset)
+	}
+	fmt.Printf("\r\033[K%s[%d/%d]%s %-52s %s\n", colorDim, r.filesDone, r.total, colorReset, rel, statusStr)
+	return nil
+}
+
+func (r *runner) printProgress(currentFile string, start time.Time) {
+	rel, _ := filepath.Rel("testdata/sqllogictest", currentFile)
+	if rel == "" {
+		rel = filepath.Base(currentFile)
+	}
+
+	elapsed := time.Since(start)
+	elapsedStr := elapsed.Round(time.Second).String()
+
+	var etaStr string
+	if r.filesDone > 0 {
+		rate := float64(r.filesDone) / elapsed.Seconds()
+		eta := time.Duration(float64(r.total-r.filesDone)/rate * float64(time.Second)).Round(time.Second)
+		etaStr = "eta " + eta.String()
+	} else {
+		etaStr = "eta --"
+	}
+
+	checked := r.passed + r.failed
+	var rateStr string
+	if checked > 0 {
+		pct := 100.0 * float64(r.passed) / float64(checked)
+		color := colorRed
+		if r.failed == 0 {
+			color = colorGreen
+		} else if pct >= 90 {
+			color = colorYellow
+		}
+		rateStr = fmt.Sprintf("%s%.1f%%%s", color, pct, colorReset)
+	} else {
+		rateStr = "  --.--%"
+	}
+
+	var throughputStr string
+	if secs := elapsed.Seconds(); secs > 0 && checked > 0 {
+		qps := float64(checked) / secs
+		switch {
+		case qps >= 1_000_000:
+			throughputStr = fmt.Sprintf("%.1fM q/s", qps/1_000_000)
+		case qps >= 1_000:
+			throughputStr = fmt.Sprintf("%.1fk q/s", qps/1_000)
+		default:
+			throughputStr = fmt.Sprintf("%.0f q/s", qps)
+		}
+	} else {
+		throughputStr = "-- q/s"
+	}
+
+	fmt.Printf("\r\033[K%s[%d/%d]%s %-40s  %s  pass=%-6d %sfail=%-5d%s skip=%-5d  %s / %s  %s%s%s",
+		colorDim, r.filesDone+1, r.total, colorReset,
+		rel, rateStr,
+		r.passed,
+		colorRed, r.failed, colorReset,
+		r.skipped,
+		elapsedStr, etaStr,
+		colorCyan, throughputStr, colorReset,
+	)
+}
+
+// collectCreatedTables scans records for CREATE TABLE statements and returns
+// the table names so they can be pre-dropped before each test file runs.
+func collectCreatedViews(records []*record) []string {
+	seen := map[string]bool{}
+	var views []string
+	for _, rec := range records {
+		if !rec.isStatement {
+			continue
+		}
+		fields := strings.Fields(rec.sql)
+		if len(fields) < 3 {
+			continue
+		}
+		if !strings.EqualFold(fields[0], "CREATE") || !strings.EqualFold(fields[1], "VIEW") {
+			continue
+		}
+		idx := 2
+		if strings.EqualFold(fields[idx], "IF") && len(fields) > idx+2 {
+			idx = 5
+		}
+		if idx < len(fields) {
+			name := strings.TrimSuffix(fields[idx], ";")
+			if name != "" && !seen[name] {
+				seen[name] = true
+				views = append(views, name)
+			}
+		}
+	}
+	return views
+}
+
+func collectCreatedTables(records []*record) []string {
+	seen := map[string]bool{}
+	var tables []string
+	for _, rec := range records {
+		if !rec.isStatement {
+			continue
+		}
+		fields := strings.Fields(rec.sql)
+		if len(fields) < 3 {
+			continue
+		}
+		if !strings.EqualFold(fields[0], "CREATE") || !strings.EqualFold(fields[1], "TABLE") {
+			continue
+		}
+		idx := 2
+		if strings.EqualFold(fields[idx], "IF") && len(fields) > idx+2 {
+			idx = 5 // CREATE TABLE IF NOT EXISTS <name>
+		}
+		if idx < len(fields) {
+			name := strings.TrimSuffix(strings.TrimSuffix(fields[idx], "("), ";")
+			if name != "" && !seen[name] {
+				seen[name] = true
+				tables = append(tables, name)
+			}
+		}
+	}
+	return tables
+}
+
+func (r *runner) runRecord(rec *record) {
+	resp, err := r.execQuery(rec.sql)
+	if err != nil {
+		r.fail(rec, "http error: %v", err)
+		return
+	}
+
+	if rec.isStatement {
+		if rec.expectOK {
+			if resp.Error != nil {
+				r.fail(rec, "expected ok, got error: %s", resp.Error.Message)
+			} else {
+				r.pass(rec)
+			}
+		} else {
+			if resp.Error == nil {
+				r.fail(rec, "expected error, got ok")
+			} else {
+				r.pass(rec)
+			}
+		}
+		return
+	}
+
+	// query record
+	if resp.Error != nil {
+		r.fail(rec, "unexpected error: %s", resp.Error.Message)
+		return
+	}
+
+	// hash format: "N values hashing to <md5>"
+	if len(rec.expected) == 1 {
+		parts := strings.Fields(rec.expected[0])
+		if len(parts) == 5 && parts[1] == "values" && parts[2] == "hashing" && parts[3] == "to" {
+			wantCount, _ := strconv.Atoi(parts[0])
+			wantHash := parts[4]
+			got := r.formatResults(resp, rec.typeStr)
+			if len(got) != wantCount {
+				r.fail(rec, "hash record: want %d values got %d", wantCount, len(got))
+				return
+			}
+			ncols := len(rec.typeStr)
+			if ncols == 0 {
+				ncols = 1
+			}
+			switch rec.sortMode {
+			case "rowsort":
+				got = sortRows(got, ncols)
+			case "valuesort":
+				g := append([]string(nil), got...)
+				sort.Strings(g)
+				got = g
+			}
+			h := md5.Sum([]byte(strings.Join(got, "\n") + "\n"))
+			gotHash := fmt.Sprintf("%x", h)
+			if gotHash != wantHash {
+				r.fail(rec, "hash mismatch: want %s got %s", wantHash, gotHash)
+				return
+			}
+			r.pass(rec)
+			return
+		}
+	}
+
+	got := r.formatResults(resp, rec.typeStr)
+	exp := rec.expected
+
+	switch rec.sortMode {
+	case "rowsort":
+		ncols := len(rec.typeStr)
+		if ncols == 0 {
+			ncols = 1
+		}
+		got = sortRows(got, ncols)
+		exp = sortRows(exp, ncols)
+	case "valuesort":
+		g := append([]string(nil), got...)
+		e := append([]string(nil), exp...)
+		sort.Strings(g)
+		sort.Strings(e)
+		got, exp = g, e
+	}
+
+	if !equalSlices(got, exp) {
+		r.fail(rec, "result mismatch\n    want: %v\n    got:  %v", exp, got)
+	} else {
+		r.pass(rec)
+	}
+}
+
+// ── formatting ───────────────────────────────────────────────────────────────
+
+func (r *runner) formatResults(resp *queryResponse, typeStr string) []string {
+	var vals []string
+	for _, row := range resp.Rows {
+		for i, v := range row {
+			ct := byte('T')
+			if i < len(typeStr) {
+				ct = typeStr[i]
+			}
+			vals = append(vals, formatValue(v, ct))
+		}
+	}
+	return vals
+}
+
+// formatValue converts a JSON value to the string representation expected by
+// the sqllogictest format. Type chars: I=integer, R=real (%.3g), T=text.
+func formatValue(v interface{}, colType byte) string {
+	if v == nil {
+		return "NULL"
+	}
+	switch colType {
+	case 'I':
+		switch n := v.(type) {
+		case float64:
+			return strconv.FormatInt(int64(math.Round(n)), 10)
+		case int64:
+			return strconv.FormatInt(n, 10)
+		case int:
+			return strconv.Itoa(n)
+		case bool:
+			if n {
+				return "1"
+			}
+			return "0"
+		case string:
+			if i, err := strconv.ParseInt(n, 10, 64); err == nil {
+				return strconv.FormatInt(i, 10)
+			}
+			return n
+		default:
+			return fmt.Sprintf("%v", v)
+		}
+	case 'R':
+		switch n := v.(type) {
+		case float64:
+			return strconv.FormatFloat(n, 'g', 3, 64)
+		case int64:
+			return strconv.FormatFloat(float64(n), 'g', 3, 64)
+		case int:
+			return strconv.FormatFloat(float64(n), 'g', 3, 64)
+		case string:
+			if f, err := strconv.ParseFloat(n, 64); err == nil {
+				return strconv.FormatFloat(f, 'g', 3, 64)
+			}
+			return n
+		default:
+			return fmt.Sprintf("%v", v)
+		}
+	default: // T
+		switch s := v.(type) {
+		case string:
+			return s
+		case bool:
+			if s {
+				return "1"
+			}
+			return "0"
+		case float64:
+			if s == math.Trunc(s) && !math.IsInf(s, 0) {
+				return strconv.FormatInt(int64(s), 10)
+			}
+			return fmt.Sprintf("%g", s)
+		default:
+			return fmt.Sprintf("%v", v)
+		}
+	}
+}
+
+// ── helpers ───────────────────────────────────────────────────────────────────
+
+func sortRows(vals []string, ncols int) []string {
+	if ncols <= 0 || len(vals) == 0 {
+		return vals
+	}
+	nrows := len(vals) / ncols
+	rows := make([][]string, nrows)
+	for i := range rows {
+		s, e := i*ncols, i*ncols+ncols
+		if e > len(vals) {
+			e = len(vals)
+		}
+		rows[i] = vals[s:e]
+	}
+	sort.Slice(rows, func(i, j int) bool {
+		for k := 0; k < len(rows[i]) && k < len(rows[j]); k++ {
+			if rows[i][k] != rows[j][k] {
+				return rows[i][k] < rows[j][k]
+			}
+		}
+		return len(rows[i]) < len(rows[j])
+	})
+	out := make([]string, 0, len(vals))
+	for _, row := range rows {
+		out = append(out, row...)
+	}
+	return out
+}
+
+func equalSlices(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func (r *runner) execQuery(sql string) (*queryResponse, error) {
+	body, _ := json.Marshal(queryRequest{SQL: sql})
+	resp, err := r.client.Post(r.baseURL+"/query", "application/json", bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	var qr queryResponse
+	if err := json.NewDecoder(resp.Body).Decode(&qr); err != nil {
+		return nil, fmt.Errorf("decode response: %w", err)
+	}
+	return &qr, nil
+}
+
+func (r *runner) pass(rec *record) {
+	r.passed++
+	if r.verbose && r.logW != nil {
+		fmt.Fprintf(r.logW, "  ok   %s:%d\n", rec.file, rec.line)
+	}
+}
+
+func (r *runner) fail(rec *record, format string, args ...interface{}) {
+	r.failed++
+	msg := fmt.Sprintf(format, args...)
+	sql := strings.ReplaceAll(strings.TrimSpace(rec.sql), "\n", " ")
+	if len(sql) > 120 {
+		sql = sql[:117] + "..."
+	}
+	line := fmt.Sprintf("FAIL %s:%d: %s\n     SQL: %s\n", rec.file, rec.line, msg, sql)
+	if r.logW != nil {
+		fmt.Fprint(r.logW, line)
+		r.logW.Flush()
+	} else {
+		fmt.Print(line)
+	}
+}
+
+// ── parser ────────────────────────────────────────────────────────────────────
+
+// parseFile reads a sqllogictest file and returns all records.
+func parseFile(path string, f *os.File) ([]*record, error) {
+	scanner := bufio.NewScanner(f)
+
+	var lines []lineInfo
+	n := 0
+	for scanner.Scan() {
+		n++
+		text := scanner.Text()
+		if !strings.HasPrefix(strings.TrimSpace(text), "#") {
+			lines = append(lines, lineInfo{text: text, num: n})
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+
+	// split into blocks separated by blank lines
+	var blocks [][]lineInfo
+	var cur []lineInfo
+	for _, li := range lines {
+		if strings.TrimSpace(li.text) == "" {
+			if len(cur) > 0 {
+				blocks = append(blocks, cur)
+				cur = nil
+			}
+		} else {
+			cur = append(cur, li)
+		}
+	}
+	if len(cur) > 0 {
+		blocks = append(blocks, cur)
+	}
+
+	var records []*record
+	haltSeen := false
+	skipNext := false
+
+	for _, block := range blocks {
+		if haltSeen {
+			break
+		}
+
+		// consume skipif / onlyif lines at the top of the block
+		i := 0
+		for i < len(block) {
+			lower := strings.ToLower(strings.TrimSpace(block[i].text))
+			if strings.HasPrefix(lower, "skipif ") {
+				engine := strings.TrimSpace(block[i].text[7:])
+				if strings.EqualFold(engine, engineName) {
+					skipNext = true
+				}
+				i++
+			} else if strings.HasPrefix(lower, "onlyif ") {
+				engine := strings.TrimSpace(block[i].text[7:])
+				if !strings.EqualFold(engine, engineName) {
+					skipNext = true
+				}
+				i++
+			} else {
+				break
+			}
+		}
+
+		if i >= len(block) {
+			continue
+		}
+
+		directiveLine := block[i]
+		parts := strings.Fields(directiveLine.text)
+		if len(parts) == 0 {
+			continue
+		}
+
+		rec := &record{file: path, line: directiveLine.num, skip: skipNext}
+		skipNext = false
+		body := block[i+1:]
+
+		switch parts[0] {
+		case "halt":
+			haltSeen = true
+			continue
+
+		case "statement":
+			rec.isStatement = true
+			rec.expectOK = len(parts) > 1 && parts[1] == "ok"
+			var sqlLines []string
+			for _, li := range body {
+				sqlLines = append(sqlLines, li.text)
+			}
+			rec.sql = strings.Join(sqlLines, "\n")
+
+		case "query":
+			rec.isQuery = true
+			if len(parts) > 1 {
+				rec.typeStr = strings.ToUpper(parts[1])
+			}
+			if len(parts) > 2 {
+				rec.sortMode = parts[2]
+			} else {
+				rec.sortMode = "nosort"
+			}
+			if len(parts) > 3 {
+				rec.label = parts[3]
+			}
+			inResults := false
+			var sqlLines []string
+			for _, li := range body {
+				if strings.TrimSpace(li.text) == "----" {
+					inResults = true
+					continue
+				}
+				if inResults {
+					rec.expected = append(rec.expected, strings.TrimSpace(li.text))
+				} else {
+					sqlLines = append(sqlLines, li.text)
+				}
+			}
+			rec.sql = strings.Join(sqlLines, "\n")
+
+		default:
+			continue
+		}
+
+		if strings.TrimSpace(rec.sql) == "" {
+			continue
+		}
+		records = append(records, rec)
+	}
+
+	return records, nil
+}

main.go

@@ -33,6 +33,7 @@ var (
 	httpPort   = flag.Int("http-port", 8080, "HTTP server port")
 	httpCORS   = flag.Bool("http-cors", true, "Enable CORS")
 	httpAuth   = flag.Bool("http-auth", false, "Enable authentication")
+	httpQuiet  = flag.Bool("quiet", false, "Disable request logging")
 	apiKeys    = flag.String("api-keys", "", "Comma-separated API keys")
 
 	// Export/Import flags
@@ -723,6 +724,7 @@ func runHTTPServer() {
 	config.Port = *httpPort
 	config.EnableCORS = *httpCORS
 	config.EnableAuth = *httpAuth
+	config.EnableLogging = !*httpQuiet
 
 	if *apiKeys != "" {
 		config.APIKeys = strings.Split(*apiKeys, ",")

pkg/analyzer/analyzer.go

@@ -415,7 +415,7 @@ func (a *Analyzer) analyzeInsert(stmt *parser.InsertStmt) error {
 
 	// Analyze INSERT ... SELECT
 	if stmt.Select != nil {
-		a.scope.DefineTable(&TableInfo{Name: table.Name, Columns: table.Columns})
+		a.scope.DefineTable(&TableInfo{Name: table.Name, Columns: table.Columns, IsView: table.IsView})
 		if err := a.analyzeSelect(stmt.Select); err != nil {
 			return err
 		}
@@ -434,7 +434,7 @@ func (a *Analyzer) analyzeUpdate(stmt *parser.UpdateStmt) error {
 		}
 	}
 
-	a.scope.DefineTable(&TableInfo{Name: table.Name, Columns: table.Columns, Alias: stmt.Table.Alias})
+	a.scope.DefineTable(&TableInfo{Name: table.Name, Columns: table.Columns, Alias: stmt.Table.Alias, IsView: table.IsView})
 
 	// Validate SET assignments
 	for _, assign := range stmt.Set {
@@ -487,7 +487,7 @@ func (a *Analyzer) analyzeDelete(stmt *parser.DeleteStmt) error {
 		}
 	}
 
-	a.scope.DefineTable(&TableInfo{Name: table.Name, Columns: table.Columns})
+	a.scope.DefineTable(&TableInfo{Name: table.Name, Columns: table.Columns, IsView: table.IsView})
 
 	// Analyze WHERE clause
 	if stmt.Where != nil {
@@ -747,12 +747,15 @@ func (a *Analyzer) analyzeUnaryExpr(e *parser.UnaryExpr) (*ExprInfo, error) {
 	}
 
 	switch e.Op {
-	case lexer.TokenMinus, lexer.TokenPlus:
+	case lexer.TokenPlus:
+		// Unary + is a no-op in SQLite — passes any type through unchanged.
+		info.Type = operand.Type
+	case lexer.TokenMinus:
 		info.Type = operand.Type
 		if !operand.Type.IsNumeric() && operand.Type != TypeNull && operand.Type != TypeUnknown {
 			return nil, &AnalysisError{
 				Type:    ErrTypeMismatch,
-				Message: fmt.Sprintf("unary %s requires numeric type, got %s", e.Op, operand.Type),
+				Message: fmt.Sprintf("unary - requires numeric type, got %s", operand.Type),
 			}
 		}
 	case lexer.TokenNOT:

pkg/analyzer/types.go

@@ -258,10 +258,15 @@ type TableInfo struct {
 	Name    string
 	Columns []ColumnInfo
 	Alias   string // For query-local aliases
+	IsView  bool   // Views accept any column reference
 }
 
 // GetColumn returns a column by name.
+// For views (IsView=true), returns a wildcard ColumnInfo so column analysis passes.
 func (t *TableInfo) GetColumn(name string) (*ColumnInfo, bool) {
+	if t.IsView {
+		return &ColumnInfo{Name: name, TableName: t.Name, Type: TypeAny}, true
+	}
 	upper := strings.ToUpper(name)
 	for i := range t.Columns {
 		if strings.ToUpper(t.Columns[i].Name) == upper {

pkg/httpserver/server.go

@@ -22,6 +22,7 @@ type Config struct {
 	EnableCORS        bool
 	EnableAuth        bool
 	EnableCompression bool
+	EnableLogging     bool
 	APIKeys           []string
 	TLSCertFile       string
 	TLSKeyFile        string
@@ -38,6 +39,7 @@ func DefaultConfig() *Config {
 		EnableCORS:        true,
 		EnableAuth:        false,
 		EnableCompression: true,
+		EnableLogging:     true,
 		APIKeys:           []string{},
 	}
 }
@@ -138,7 +140,9 @@ func (s *Server) init() *Server {
 		handler = s.authMiddleware(handler)
 	}
 
-	handler = s.loggingMiddleware(handler)
+	if s.config.EnableLogging {
+		handler = s.loggingMiddleware(handler)
+	}
 
 	// Register routes
 	mux.HandleFunc("/query", s.handleQuery)

pkg/parser/ast.go

@@ -19,6 +19,29 @@ type Expr interface {
 	exprNode()
 }
 
+// SetOpType represents a set operation type.
+type SetOpType int
+
+const (
+	SetOpUnion SetOpType = iota
+	SetOpUnionAll
+	SetOpIntersect
+	SetOpExcept
+)
+
+// CompoundSelect chains two SELECT statements with a set operation.
+type CompoundSelect struct {
+	Left    *SelectStmt
+	Op      SetOpType
+	Right   *SelectStmt // may itself have Compound set for chained ops
+	OrderBy []OrderByItem
+	Limit   Expr
+	Offset  Expr
+}
+
+func (c *CompoundSelect) node()     {}
+func (c *CompoundSelect) stmtNode() {}
+
 // SelectStmt represents a SELECT statement.
 type SelectStmt struct {
 	Distinct bool
@@ -30,6 +53,8 @@ type SelectStmt struct {
 	OrderBy  []OrderByItem
 	Limit    Expr
 	Offset   Expr
+	// Compound chains a set operation onto this SELECT (UNION/INTERSECT/EXCEPT).
+	Compound *CompoundSelect
 }
 
 func (s *SelectStmt) node()     {}
@@ -217,6 +242,25 @@ type DropIndexStmt struct {
 func (s *DropIndexStmt) node()     {}
 func (s *DropIndexStmt) stmtNode() {}
 
+// CreateViewStmt represents a CREATE VIEW statement.
+type CreateViewStmt struct {
+	IfNotExists bool
+	View        *TableRef
+	Select      *SelectStmt
+}
+
+func (s *CreateViewStmt) node()     {}
+func (s *CreateViewStmt) stmtNode() {}
+
+// DropViewStmt represents a DROP VIEW statement.
+type DropViewStmt struct {
+	IfExists bool
+	Views    []*TableRef
+}
+
+func (s *DropViewStmt) node()     {}
+func (s *DropViewStmt) stmtNode() {}
+
 // AlterTableStmt represents an ALTER TABLE statement.
 type AlterTableStmt struct {
 	Table  string

pkg/parser/parser.go

@@ -141,13 +141,15 @@ func (p *Parser) parseStatement() (Statement, error) {
 	}
 }
 
-// parseSelect parses a SELECT statement.
-func (p *Parser) parseSelect() (*SelectStmt, error) {
+// parseSingleSelectTerm parses one SELECT body (SELECT … FROM … WHERE … GROUP BY … HAVING …)
+// but stops before any set operator, ORDER BY, LIMIT, or OFFSET.
+// Callers that want the full chain (including set ops) use parseSelect instead.
+func (p *Parser) parseSingleSelectTerm() (*SelectStmt, error) {
 	stmt := &SelectStmt{}
 
 	p.nextToken() // consume SELECT
 
-	// Check for DISTINCT
+	// Check for DISTINCT / ALL
 	if p.curTokenIs(lexer.TokenDISTINCT) {
 		stmt.Distinct = true
 		p.nextToken()
@@ -155,14 +157,12 @@ func (p *Parser) parseSelect() (*SelectStmt, error) {
 		p.nextToken()
 	}
 
-	// Parse select columns
 	cols, err := p.parseSelectColumns()
 	if err != nil {
 		return nil, err
 	}
 	stmt.Columns = cols
 
-	// Parse FROM clause
 	if p.curTokenIs(lexer.TokenFROM) {
 		p.nextToken()
 		tables, err := p.parseTableRefs()
@@ -172,7 +172,6 @@ func (p *Parser) parseSelect() (*SelectStmt, error) {
 		stmt.From = tables
 	}
 
-	// Parse WHERE clause
 	if p.curTokenIs(lexer.TokenWHERE) {
 		p.nextToken()
 		where, err := p.parseExpr()
@@ -182,7 +181,6 @@ func (p *Parser) parseSelect() (*SelectStmt, error) {
 		stmt.Where = where
 	}
 
-	// Parse GROUP BY clause
 	if p.curTokenIs(lexer.TokenGROUP) {
 		if err := p.expectPeek(lexer.TokenBY); err != nil {
 			return nil, err
@@ -195,7 +193,6 @@ func (p *Parser) parseSelect() (*SelectStmt, error) {
 		stmt.GroupBy = groupBy
 	}
 
-	// Parse HAVING clause
 	if p.curTokenIs(lexer.TokenHAVING) {
 		p.nextToken()
 		having, err := p.parseExpr()
@@ -205,6 +202,26 @@ func (p *Parser) parseSelect() (*SelectStmt, error) {
 		stmt.Having = having
 	}
 
+	return stmt, nil
+}
+
+// parseSelect parses a SELECT statement, including any trailing set operations
+// (UNION / INTERSECT / EXCEPT) and an optional ORDER BY / LIMIT / OFFSET.
+func (p *Parser) parseSelect() (*SelectStmt, error) {
+	stmt, err := p.parseSingleSelectTerm()
+	if err != nil {
+		return nil, err
+	}
+
+	// Handle set operations: UNION [ALL], INTERSECT, EXCEPT
+	if p.curTokenIs(lexer.TokenUNION) || p.curTokenIs(lexer.TokenINTERSECT) || p.curTokenIs(lexer.TokenEXCEPT) {
+		compound, err := p.parseCompoundChain(stmt)
+		if err != nil {
+			return nil, err
+		}
+		return &SelectStmt{Compound: compound}, nil
+	}
+
 	// Parse ORDER BY clause
 	if p.curTokenIs(lexer.TokenORDER) {
 		if err := p.expectPeek(lexer.TokenBY); err != nil {
@@ -241,6 +258,142 @@ func (p *Parser) parseSelect() (*SelectStmt, error) {
 	return stmt, nil
 }
 
+// setOpPrec returns the precedence of a set operator token.
+// INTERSECT binds more tightly than UNION/EXCEPT per SQL standard.
+func setOpPrec(t lexer.TokenType) int {
+	if t == lexer.TokenINTERSECT {
+		return 2
+	}
+	return 1
+}
+
+// parseCompoundChain collects all set-op legs and builds a left-associative tree
+// respecting INTERSECT > UNION/EXCEPT precedence.
+//
+// The algorithm is the standard precedence-climbing / Pratt approach:
+//
+//	parseMin(minPrec):
+//	  left = first already-parsed SELECT (passed in as `first`)
+//	  while curOp.prec >= minPrec:
+//	      op = curOp; consume op
+//	      right = parseSingleSelect()
+//	      while nextOp.prec > op.prec:   // right-bind tighter ops
+//	          right = parseMin(op.prec+1) using right as seed
+//	      left = Compound(left, op, right)
+//	  return left
+func (p *Parser) parseCompoundChain(first *SelectStmt) (*CompoundSelect, error) {
+	type leg struct {
+		op    SetOpType
+		query *SelectStmt
+	}
+
+	// Consume a set-op token and return its SetOpType + precedence.
+	consumeOp := func() (SetOpType, int, error) {
+		switch p.curToken.Type {
+		case lexer.TokenUNION:
+			p.nextToken()
+			if p.curTokenIs(lexer.TokenALL) {
+				p.nextToken()
+				return SetOpUnionAll, 1, nil
+			}
+			return SetOpUnion, 1, nil
+		case lexer.TokenINTERSECT:
+			p.nextToken()
+			return SetOpIntersect, 2, nil
+		case lexer.TokenEXCEPT:
+			p.nextToken()
+			return SetOpExcept, 1, nil
+		}
+		return 0, 0, p.curError("expected UNION, INTERSECT, or EXCEPT")
+	}
+
+	isSetOp := func() bool {
+		return p.curTokenIs(lexer.TokenUNION) ||
+			p.curTokenIs(lexer.TokenINTERSECT) ||
+			p.curTokenIs(lexer.TokenEXCEPT)
+	}
+
+	// parseSingleSelect parses the next SELECT term (no set-ops claimed).
+	parseSingleSelect := func() (*SelectStmt, error) {
+		if !p.curTokenIs(lexer.TokenSELECT) {
+			return nil, p.curError("expected SELECT after set operator")
+		}
+		return p.parseSingleSelectTerm()
+	}
+
+	// Precedence-climbing: build left-to-right tree, INTERSECT binds tighter.
+	var climb func(left *SelectStmt, minPrec int) (*CompoundSelect, error)
+	climb = func(left *SelectStmt, minPrec int) (*CompoundSelect, error) {
+		for isSetOp() && setOpPrec(p.curToken.Type) >= minPrec {
+			op, prec, err := consumeOp()
+			if err != nil {
+				return nil, err
+			}
+			right, err := parseSingleSelect()
+			if err != nil {
+				return nil, err
+			}
+			// If the right node itself resolved to a compound (via recursive parseSelect),
+			// unwrap and re-climb properly.
+			if right.Compound != nil {
+				// right was a sub-chain; treat it as already climbed.
+			} else {
+				// Absorb any higher-precedence ops on the right.
+				for isSetOp() && setOpPrec(p.curToken.Type) > prec {
+					sub, err := climb(right, prec+1)
+					if err != nil {
+						return nil, err
+					}
+					right = &SelectStmt{Compound: sub}
+					break
+				}
+			}
+			node := &CompoundSelect{Left: left, Op: op, Right: right}
+			left = &SelectStmt{Compound: node}
+		}
+		if left.Compound != nil {
+			return left.Compound, nil
+		}
+		return nil, p.curError("internal: no compound built")
+	}
+
+	compound, err := climb(first, 1)
+	if err != nil {
+		return nil, err
+	}
+
+	// Parse trailing ORDER BY / LIMIT / OFFSET that apply to the whole compound.
+	if p.curTokenIs(lexer.TokenORDER) {
+		if err := p.expectPeek(lexer.TokenBY); err != nil {
+			return nil, err
+		}
+		p.nextToken()
+		orderBy, err := p.parseOrderBy()
+		if err != nil {
+			return nil, err
+		}
+		compound.OrderBy = orderBy
+	}
+	if p.curTokenIs(lexer.TokenLIMIT) {
+		p.nextToken()
+		limit, err := p.parseExpr()
+		if err != nil {
+			return nil, err
+		}
+		compound.Limit = limit
+	}
+	if p.curTokenIs(lexer.TokenOFFSET) {
+		p.nextToken()
+		offset, err := p.parseExpr()
+		if err != nil {
+			return nil, err
+		}
+		compound.Offset = offset
+	}
+
+	return compound, nil
+}
+
 func (p *Parser) parseSelectColumns() ([]SelectColumn, error) {
 	var cols []SelectColumn
 
@@ -736,6 +889,8 @@ func (p *Parser) parseCreate() (Statement, error) {
 			return nil, p.curError("expected INDEX after UNIQUE")
 		}
 		return p.parseCreateIndex(true)
+	case lexer.TokenVIEW:
+		return p.parseCreateView()
 	default:
 		return nil, p.curError("expected TABLE or INDEX after CREATE")
 	}
@@ -1161,6 +1316,8 @@ func (p *Parser) parseDrop() (Statement, error) {
 		return p.parseDropTable()
 	case lexer.TokenINDEX:
 		return p.parseDropIndex()
+	case lexer.TokenVIEW:
+		return p.parseDropView()
 	default:
 		return nil, p.curError("expected TABLE or INDEX after DROP")
 	}
@@ -1223,6 +1380,76 @@ func (p *Parser) parseDropIndex() (*DropIndexStmt, error) {
 	return stmt, nil
 }
 
+func (p *Parser) parseCreateView() (*CreateViewStmt, error) {
+	stmt := &CreateViewStmt{}
+
+	p.nextToken() // consume VIEW
+
+	if p.curTokenIs(lexer.TokenIF) {
+		p.nextToken()
+		if !p.curTokenIs(lexer.TokenNOT) {
+			return nil, p.curError("expected NOT")
+		}
+		p.nextToken()
+		if !p.curTokenIs(lexer.TokenEXISTS) {
+			return nil, p.curError("expected EXISTS")
+		}
+		stmt.IfNotExists = true
+		p.nextToken()
+	}
+
+	// Parse view name directly — do NOT use parseTableRef here because it
+	// greedily interprets the AS keyword as an alias, consuming "AS SELECT".
+	if !p.curTokenIs(lexer.TokenIdent) {
+		return nil, p.curError("expected view name")
+	}
+	stmt.View = &TableRef{Name: p.curToken.Literal}
+	p.nextToken()
+
+	if !p.curTokenIs(lexer.TokenAS) {
+		return nil, p.curError("expected AS after view name")
+	}
+	p.nextToken() // consume AS
+
+	sel, err := p.parseSelect()
+	if err != nil {
+		return nil, err
+	}
+	stmt.Select = sel
+
+	return stmt, nil
+}
+
+func (p *Parser) parseDropView() (*DropViewStmt, error) {
+	stmt := &DropViewStmt{}
+
+	p.nextToken() // consume VIEW
+
+	if p.curTokenIs(lexer.TokenIF) {
+		p.nextToken()
+		if !p.curTokenIs(lexer.TokenEXISTS) {
+			return nil, p.curError("expected EXISTS")
+		}
+		stmt.IfExists = true
+		p.nextToken()
+	}
+
+	for {
+		view, err := p.parseTableRef()
+		if err != nil {
+			return nil, err
+		}
+		stmt.Views = append(stmt.Views, view)
+
+		if !p.curTokenIs(lexer.TokenComma) {
+			break
+		}
+		p.nextToken()
+	}
+
+	return stmt, nil
+}
+
 // parseAlter parses an ALTER statement.
 func (p *Parser) parseAlter() (Statement, error) {
 	p.nextToken() // consume ALTER
@@ -1690,8 +1917,8 @@ func (p *Parser) parseInExpr(left Expr, not bool) (Expr, error) {
 			return nil, err
 		}
 		expr.Subquery = sel
-	} else {
-		// Value list
+	} else if !p.curTokenIs(lexer.TokenRParen) {
+		// Value list (empty list is allowed — always false)
 		values, err := p.parseExprList()
 		if err != nil {
 			return nil, err

pkg/storage/schema.go

@@ -107,10 +107,10 @@ func (m *SchemaManager) CreateTable(schema *Schema) error {
 				break
 			}
 		}
-		// Default to first column if no primary key specified
-		if schema.PrimaryKey == "" && len(schema.Columns) > 0 {
-			schema.PrimaryKey = schema.Columns[0].Name
-			schema.Columns[0].PrimaryKey = true
+		// No explicit primary key declared — use synthetic _rowid_ so user
+		// columns remain unconstrained and can hold duplicate or NULL values.
+		if schema.PrimaryKey == "" {
+			schema.PrimaryKey = "_rowid_"
 		}
 	}
 

pkg/storage/table.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"sync"
 )
 
 // Row represents a database row.
@@ -14,6 +15,9 @@ type TableManager struct {
 	pool     *KVPool
 	schema   *SchemaManager
 	database string
+
+	cacheMu  sync.RWMutex
+	rowCache map[string][]Row // table name → all rows (nil means not loaded)
 }
 
 // NewTableManager creates a new table manager.
@@ -22,9 +26,22 @@ func NewTableManager(pool *KVPool, schema *SchemaManager, database string) *Tabl
 		pool:     pool,
 		schema:   schema,
 		database: database,
+		rowCache: make(map[string][]Row),
 	}
 }
 
+// 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))
+	m.cacheMu.Unlock()
+}
+
+// InvalidateCache is the exported version for use by the executor.
+func (m *TableManager) InvalidateCache(table string) {
+	m.invalidateCache(table)
+}
+
 // dataKey returns the key for a row.
 func (m *TableManager) dataKey(table, pk string) string {
 	return fmt.Sprintf("%s:_data:%s:%s", m.database, strings.ToLower(table), pk)
@@ -167,9 +184,179 @@ func (m *TableManager) Insert(table string, row Row) error {
 	// Update indexes
 	m.updateIndexesForRow(table, normalizedRow, true)
 
+	m.invalidateCache(table)
 	return nil
 }
 
+// InsertBulk inserts multiple rows efficiently, parallelizing KV writes across
+// the connection pool. Skips per-row duplicate checks (caller must ensure
+// uniqueness). Used by INSERT ... SELECT.
+func (m *TableManager) InsertBulk(table string, rows []Row) (int, error) {
+	if len(rows) == 0 {
+		return 0, nil
+	}
+
+	schema, err := m.schema.GetSchema(table)
+	if err != nil {
+		return 0, err
+	}
+
+	pkCol, _ := schema.GetColumn(schema.PrimaryKey)
+	isIntegerPK := pkCol != nil && isIntegerType(pkCol.Type)
+
+	// Normalize rows and assign _rowid_.
+	normalized := make([]Row, 0, len(rows))
+	var maxRowID int64
+	for _, row := range rows {
+		nr := make(Row)
+		for _, col := range schema.Columns {
+			for k, v := range row {
+				if strings.EqualFold(k, col.Name) {
+					nr[col.Name] = v
+					break
+				}
+			}
+		}
+		for _, col := range schema.Columns {
+			if _, ok := nr[col.Name]; !ok && col.Default != nil {
+				nr[col.Name] = col.Default
+			}
+		}
+		var rowid int64
+		var hasRowid bool
+		if isIntegerPK {
+			switch v := nr[schema.PrimaryKey].(type) {
+			case float64:
+				rowid = int64(v)
+				hasRowid = true
+			case int64:
+				rowid = v
+				hasRowid = true
+			case int:
+				rowid = int64(v)
+				hasRowid = true
+			}
+		}
+		if !hasRowid {
+			// Fall back to sequential insert for non-integer-pk rows.
+			if err := m.Insert(table, row); err != nil {
+				return len(normalized), err
+			}
+			continue
+		}
+		nr["_rowid_"] = rowid
+		if rowid > maxRowID {
+			maxRowID = rowid
+		}
+		normalized = append(normalized, nr)
+	}
+
+	if maxRowID > 0 {
+		m.schema.UpdateMaxRowID(table, maxRowID)
+	}
+
+	// Serialize all rows.
+	type kv struct{ key, val string }
+	rowKVs := make([]kv, 0, len(normalized))
+	for _, nr := range normalized {
+		pk := fmt.Sprintf("%v", nr[schema.PrimaryKey])
+		data, err := json.Marshal(nr)
+		if err != nil {
+			return 0, err
+		}
+		rowKVs = append(rowKVs, kv{m.dataKey(table, pk), string(data)})
+	}
+
+	// Write rows concurrently.
+	errs := make([]error, len(rowKVs))
+	var wg sync.WaitGroup
+	for i, w := range rowKVs {
+		wg.Add(1)
+		i, w := i, w
+		go func() {
+			defer wg.Done()
+			errs[i] = m.pool.WithClient(func(c *KVClient) error {
+				return c.Write(w.key, w.val)
+			})
+		}()
+	}
+	wg.Wait()
+	for _, e := range errs {
+		if e != nil {
+			return 0, e
+		}
+	}
+
+	// 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.
 func (m *TableManager) updateIndexesForRow(table string, row Row, add bool) {
 	indexes, err := m.schema.ListTableIndexes(table)
@@ -207,30 +394,52 @@ func (m *TableManager) Select(table string, filter func(Row) bool) ([]Row, error
 		return nil, fmt.Errorf("table not found: %s", table)
 	}
 
-	prefix := m.dataPrefix(table)
-	var values []string
+	key := strings.ToLower(table)
 
-	err := m.pool.WithClient(func(c *KVClient) error {
-		var err error
-		values, err = c.Reads(prefix)
-		return err
-	})
-	if err != nil {
-		return nil, err
-	}
+	m.cacheMu.RLock()
+	cached, ok := m.rowCache[key]
+	m.cacheMu.RUnlock()
 
-	rows := make([]Row, 0, len(values))
-	for _, data := range values {
-		var row Row
-		if err := json.Unmarshal([]byte(data), &row); err != nil {
-			continue // Skip invalid rows
+	if !ok {
+		prefix := m.dataPrefix(table)
+		var values []string
+		err := m.pool.WithClient(func(c *KVClient) error {
+			var err error
+			values, err = c.Reads(prefix)
+			return err
+		})
+		if err != nil {
+			return nil, err
 		}
 
-		if filter == nil || filter(row) {
-			rows = append(rows, row)
+		loaded := make([]Row, 0, len(values))
+		for _, data := range values {
+			var row Row
+			if err := json.Unmarshal([]byte(data), &row); err != nil {
+				continue
+			}
+			loaded = append(loaded, row)
 		}
+
+		m.cacheMu.Lock()
+		m.rowCache[key] = loaded
+		m.cacheMu.Unlock()
+
+		cached = loaded
+	}
+
+	if filter == nil {
+		result := make([]Row, len(cached))
+		copy(result, cached)
+		return result, nil
 	}
 
+	rows := make([]Row, 0, len(cached))
+	for _, row := range cached {
+		if filter(row) {
+			rows = append(rows, row)
+		}
+	}
 	return rows, nil
 }
 
@@ -308,6 +517,7 @@ func (m *TableManager) Update(table string, updates Row, filter func(Row) bool)
 		}
 	}
 
+	m.invalidateCache(table)
 	return count, nil
 }
 
@@ -369,6 +579,7 @@ func (m *TableManager) UpdateFunc(table string, updateFn func(Row) (Row, error),
 		}
 	}
 
+	m.invalidateCache(table)
 	return count, nil
 }
 
@@ -402,6 +613,7 @@ func (m *TableManager) Delete(table string, filter func(Row) bool) (int, error)
 		}
 	}
 
+	m.invalidateCache(table)
 	return count, nil
 }
 
@@ -669,42 +881,24 @@ func (m *TableManager) SelectByIndex(table, indexName string, colValue interface
 		return []Row{}, nil
 	}
 
-	schema, err := m.schema.GetSchema(table)
-	if err != nil {
-		return nil, err
+	// Build a set of target rowids for O(1) lookup.
+	rowidSet := make(map[int64]struct{}, len(rowids))
+	for _, rid := range rowids {
+		rowidSet[rid] = struct{}{}
 	}
-
-	// Check if primary key is INTEGER type (in which case rowid == pk)
-	pkCol, _ := schema.GetColumn(schema.PrimaryKey)
-	isPKInteger := pkCol != nil && isIntegerType(pkCol.Type)
-
-	rows := make([]Row, 0, len(rowids))
-	for _, rowid := range rowids {
-		var row Row
-
-		// For INTEGER PRIMARY KEY, the rowid IS the primary key
-		if isPKInteger {
-			row, err = m.GetByPK(table, fmt.Sprintf("%d", rowid))
-			if err == nil {
-				rows = append(rows, row)
-				continue
-			}
+	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
 		}
+		return false
+	})
 
-		// For non-INTEGER primary keys or if PK lookup fails, look up by _rowid_
-		allRows, _ := m.Select(table, func(r Row) bool {
-			if rid, ok := r["_rowid_"].(float64); ok {
-				return int64(rid) == rowid
-			}
-			if rid, ok := r["_rowid_"].(int64); ok {
-				return rid == rowid
-			}
-			return false
-		})
-		if len(allRows) > 0 {
-			rows = append(rows, allRows[0])
-		}
-	}
+	rows := allRows
 
 	return rows, nil
 }