perf: bound simple scans and use indexed equality conjuncts (#19)

README.md

@@ -526,6 +526,12 @@ no partial statement effects; in an explicit transaction, earlier successful
 statements remain staged. Mutations exceeding a single storage transaction's
 limits fail rather than being split into independently committed chunks.
 
+Simple autocommit scans with a literal non-negative `LIMIT`/`OFFSET` can stop
+after enough matching rows. Queries requiring sorting, grouping, DISTINCT,
+windows, subquery projections, or explicit-transaction overlay validation keep
+the full execution path. Equality predicates inside `AND` can use a primary or
+single-column index lookup; the original predicate is still checked on candidates.
+
 **Indexes:**
 ```sql
 CREATE INDEX idx_users_email ON users(email);

pkg/executor/bounded_scan_test.go

@@ -0,0 +1,87 @@
+package executor
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+	"github.com/danfragoso/pizzasql-next/pkg/testkv"
+)
+
+func TestBoundedScanAndConjunctionLookup(t *testing.T) {
+	kv := testkv.New(t)
+	pool := kv.Pool(8)
+	defer pool.Close()
+	schema := storage.NewSchemaManager(pool, "bounded")
+	table := storage.NewTableManager(pool, schema, "bounded")
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
+	var rows []string
+	for i := 1; i <= 2048; i++ {
+		rows = append(rows, fmt.Sprintf("(%d,%d)", i, i))
+	}
+	execMust(t, e, "INSERT INTO t VALUES "+strings.Join(rows, ","))
+	full := execMust(t, e, "SELECT id FROM t WHERE v>=0").Rows
+	for _, tc := range []struct {
+		sql   string
+		want  [][]interface{}
+		pages int
+	}{
+		{"SELECT id FROM t LIMIT 3", full[:3], 1},
+		{"SELECT id FROM t WHERE v>=0 LIMIT 3 OFFSET 2", full[2:5], 1},
+		{"SELECT id FROM t LIMIT 0", nil, 0},
+		{"SELECT id FROM t WHERE v>=0 AND id=1000", [][]interface{}{{int64(1000)}}, 0},
+		{"SELECT id FROM t WHERE id=1000 AND v<0", nil, 0},
+	} {
+		before := kv.RequestCount(10) // PKBFI SCAN_NEXT
+		got := execMust(t, e, tc.sql).Rows
+		if len(got) != len(tc.want) || (len(got) > 0 && !reflect.DeepEqual(got, tc.want)) {
+			t.Fatalf("%s: got %v want %v", tc.sql, got, tc.want)
+		}
+		if pages := kv.RequestCount(10) - before; pages != tc.pages {
+			t.Fatalf("%s: scanned %d pages want %d", tc.sql, pages, tc.pages)
+		}
+	}
+	execMust(t, e, "CREATE INDEX t_v ON t(v)")
+	before := kv.RequestCount(10)
+	if got := execMust(t, e, "SELECT id FROM t WHERE id>0 AND v=1000").Rows; !reflect.DeepEqual(got, [][]interface{}{{int64(1000)}}) {
+		t.Fatal(got)
+	}
+	if kv.RequestCount(10) != before {
+		t.Fatal("indexed conjunct fell back to scan")
+	}
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "INSERT INTO t VALUES (9999,9999)")
+	if got := execMust(t, e, "SELECT id FROM t WHERE v>9000 LIMIT 1").Rows; !reflect.DeepEqual(got, [][]interface{}{{int64(9999)}}) {
+		t.Fatalf("bounded scan lost transaction overlay: %v", got)
+	}
+	execMust(t, e, "SELECT id FROM t LIMIT 1")
+	other := newExec(schema, table)
+	execMust(t, other, "UPDATE t SET v=0 WHERE id=2048")
+	if _, err := execSQL(e, "COMMIT"); !errors.Is(err, storage.ErrSerialization) {
+		t.Fatalf("limited transaction lost scan validation: %v", err)
+	}
+}
+
+func TestScanLimitRejectsUnsafeShapes(t *testing.T) {
+	for _, sql := range []string{
+		"SELECT DISTINCT v FROM t LIMIT 1",
+		"SELECT COUNT(*) FROM t LIMIT 1",
+		"SELECT v FROM t GROUP BY v LIMIT 1",
+		"SELECT v FROM t ORDER BY v LIMIT 1",
+		"SELECT ROW_NUMBER() OVER (ORDER BY v) FROM t LIMIT 1",
+		"SELECT (SELECT 1) FROM t LIMIT 1",
+		"SELECT v FROM t LIMIT -1",
+		"SELECT v FROM t LIMIT 1+1",
+	} {
+		stmt := parse(t, sql)
+		// Parsing is enough: eligibility must not execute expressions or scan data.
+		if _, ok := scanLimit(stmt.(*parser.SelectStmt)); ok {
+			t.Fatalf("unsafe bound for %s", sql)
+		}
+	}
+}

pkg/executor/executor.go

@@ -529,36 +529,41 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 	// A lookup may only restrict a correctly bound, preserved table. The
 	// original predicate is still evaluated after the lookup/join.
 	if effectiveWhere != nil && !isMultiTable {
-		colName, colValue, isEquality := e.extractSelectLookupCondition(stmt.Where, stmt.From[0])
-		if isEquality {
-			if strings.EqualFold(schema.PrimaryKey, colName) && safePrimaryKeyLiteral(colValue) {
-				row, getErr := e.session.GetByPK(tableName, fmt.Sprintf("%v", colValue))
-				if getErr == nil {
-					normalizeRowBySchema(row, schema)
-					rows = []storage.Row{row}
-					usedIndex = true
-				} else if getErr == storage.ErrKeyNotFound {
-					rows = []storage.Row{}
-					usedIndex = true
+		for _, condition := range splitANDClauses(effectiveWhere) {
+			colName, colValue, isEquality := e.extractSelectLookupCondition(condition, stmt.From[0])
+			if isEquality {
+				if strings.EqualFold(schema.PrimaryKey, colName) && safePrimaryKeyLiteral(colValue) {
+					row, getErr := e.session.GetByPK(tableName, fmt.Sprintf("%v", colValue))
+					if getErr == nil {
+						normalizeRowBySchema(row, schema)
+						rows = []storage.Row{row}
+						usedIndex = true
+					} else if getErr == storage.ErrKeyNotFound {
+						rows = []storage.Row{}
+						usedIndex = true
+					} else {
+						return nil, getErr
+					}
 				} else {
-					return nil, getErr
-				}
-			} else {
-				// Look for an index on this column
-				indexes, _ := e.schema.ListTableIndexes(tableName)
-				for _, idx := range indexes {
-					if len(idx.Columns) == 1 && strings.EqualFold(idx.Columns[0].Name, colName) {
-						rows, err = e.session.SelectByIndex(tableName, idx.Name, colValue)
-						if err == nil {
-							usedIndex = true
-							for i := range rows {
-								normalizeRowBySchema(rows[i], schema)
+					// Look for an index on this column
+					indexes, _ := e.schema.ListTableIndexes(tableName)
+					for _, idx := range indexes {
+						if len(idx.Columns) == 1 && strings.EqualFold(idx.Columns[0].Name, colName) {
+							rows, err = e.session.SelectByIndex(tableName, idx.Name, colValue)
+							if err == nil {
+								usedIndex = true
+								for i := range rows {
+									normalizeRowBySchema(rows[i], schema)
+								}
 							}
+							break
 						}
-						break
 					}
 				}
 			}
+			if usedIndex {
+				break
+			}
 		}
 	}
 
@@ -577,7 +582,11 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 				return toBool(val)
 			}
 		}
-		rows, err = e.session.Select(tableName, filter)
+		if limit, bounded := scanLimit(stmt); bounded && (effectiveWhere == nil || canFilterInScan) {
+			rows, err = e.session.SelectLimited(tableName, filter, limit)
+		} else {
+			rows, err = e.session.Select(tableName, filter)
+		}
 		if filterErr != nil {
 			return nil, filterErr
 		}
@@ -4424,6 +4433,48 @@ func (e *Executor) extractIndexableCondition(expr parser.Expr) (string, interfac
 	return "", nil, false
 }
 
+// Bound only simple scans. Final OFFSET/LIMIT is still applied once by the
+// normal result path; transactions conservatively retain their full read set.
+func scanLimit(stmt *parser.SelectStmt) (int, bool) {
+	if stmt.Limit == nil || len(stmt.From) != 1 || stmt.From[0].Join != nil ||
+		stmt.From[0].Alias != "" || stmt.Distinct || len(stmt.GroupBy) != 0 ||
+		stmt.Having != nil || len(stmt.OrderBy) != 0 || stmt.Compound != nil || len(stmt.With) != 0 {
+		return 0, false
+	}
+	for _, column := range stmt.Columns {
+		switch column.Expr.(type) {
+		case *parser.ColumnRef, *parser.LiteralExpr:
+		default:
+			if !column.Star {
+				return 0, false
+			}
+		}
+	}
+	literalInt := func(expr parser.Expr) (int, bool) {
+		literal, ok := expr.(*parser.LiteralExpr)
+		if !ok || literal.Type != lexer.TokenNumber {
+			return 0, false
+		}
+		n, err := strconv.Atoi(literal.Value)
+		return n, err == nil && n >= 0
+	}
+	limit, ok := literalInt(stmt.Limit)
+	if !ok {
+		return 0, false
+	}
+	if limit == 0 {
+		return 0, true
+	}
+	if stmt.Offset != nil {
+		offset, valid := literalInt(stmt.Offset)
+		if !valid || offset > int(^uint(0)>>1)-limit {
+			return 0, false
+		}
+		limit += offset
+	}
+	return limit, true
+}
+
 func (e *Executor) extractSelectLookupCondition(expr parser.Expr, table parser.TableRef) (string, interface{}, bool) {
 	for join := table.Join; join != nil; join = join.Table.Join {
 		if join.Type == parser.JoinRight || join.Type == parser.JoinFull {

pkg/storage/tx.go

@@ -451,6 +451,22 @@ func (s *Session) Select(table string, filter func(Row) bool) ([]Row, error) {
 	return s.selectLocked(table, filter)
 }
 
+// SelectLimited bounds an autocommit scan. Explicit transactions retain the
+// complete overlay/read-set behavior until bounded transactional scans have a
+// separate phantom/overlay contract; their caller still applies LIMIT later.
+func (s *Session) SelectLimited(table string, filter func(Row) bool, maxRows int) ([]Row, error) {
+	s.mu.Lock()
+	if s.inTx {
+		defer s.mu.Unlock()
+		return s.selectLocked(table, filter)
+	}
+	s.mu.Unlock()
+	if maxRows == 0 {
+		return []Row{}, nil
+	}
+	return s.table.SelectWithLimit(table, filter, maxRows, 0)
+}
+
 func (s *Session) selectLocked(table string, filter func(Row) bool) ([]Row, error) {
 	schema, err := s.schema.GetSchema(table)
 	if err != nil {

pkg/testkv/testkv.go

@@ -53,10 +53,11 @@ type scan struct {
 
 // Server is an in-memory KV server.
 type Server struct {
-	mu      sync.Mutex
-	data    map[string][]byte
-	lsns    map[string]uint64
-	nextLSN uint64
+	mu       sync.Mutex
+	data     map[string][]byte
+	lsns     map[string]uint64
+	nextLSN  uint64
+	requests map[uint16]int
 
 	ln       net.Listener
 	wg       sync.WaitGroup
@@ -139,7 +140,20 @@ func (s *Server) handle(conn net.Conn) {
 	}
 }
 
+// RequestCount reports observed PKBFI requests for bounded-work assertions.
+func (s *Server) RequestCount(opcode uint16) int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.requests[opcode]
+}
+
 func (s *Server) execute(opcode uint16, payload []byte) []byte {
+	s.mu.Lock()
+	if s.requests == nil {
+		s.requests = make(map[uint16]int)
+	}
+	s.requests[opcode]++
+	s.mu.Unlock()
 	switch opcode {
 	case opPing:
 		return statusOKBody(nil)