integrate durable PKBFI storage

PKBFI_STORAGE_MIGRATION.md

@@ -0,0 +1,41 @@
+# PKBFI Storage Migration
+
+PizzaSQL now requires the PKBFI protocol and PizzaKV's durable `.pkvdb` engine. The old text-delimited `.db` file cannot be opened directly by the new engine.
+
+## Upgrade
+
+1. Stop the PizzaSQL and PizzaKV processes that own the legacy file.
+2. Keep an immutable backup of the legacy `.db` file.
+3. Build or install the matching new PizzaKV and PizzaSQL binaries.
+4. Migrate into a new destination:
+
+```bash
+pizzakv -migrate=.db -path=.pkvdb
+```
+
+5. Start PizzaKV with the migrated file:
+
+```bash
+pizzakv -unix -path=.pkvdb
+```
+
+6. Start PizzaSQL against `unix:.pizzakv.sock`, or use `pizzasql -kv -kvflags="-path=.pkvdb"` to let PizzaSQL launch PizzaKV.
+7. Run point-read, bounded-scan, write, restart, and application smoke tests before removing the backup.
+
+The migration command exits after writing and verifying the destination. It refuses to overwrite an existing destination and does not modify the source file.
+
+## Compatibility
+
+- Deploy the new PizzaKV and PizzaSQL binaries together. New PizzaSQL does not fall back to the delimiter-based protocol.
+- Migration copies the final live key/value state. Legacy journal history is not imported, and imported records begin at the migration baseline LSN.
+- Existing PizzaSQL row values remain readable as legacy JSON after migration. New and updated rows use the versioned binary tuple encoding, so an eager row rewrite is unnecessary.
+- Schema and catalog values remain JSON because they are cold metadata.
+- Successful PKBFI writes are acknowledged only after PizzaKV's durability sync completes.
+
+## Rollback
+
+Stop the new processes and restart the old binaries against the untouched legacy `.db` backup. Writes accepted into `.pkvdb` after cutover are not copied back to the legacy file, so reconcile or discard them before rollback.
+
+## Production Sequence
+
+For each tenant, migrate and validate independently. Do not migrate a file while its owner process is running, and do not replace a live tenant's files as part of a benchmark or endurance experiment.

README.md

@@ -31,8 +31,8 @@ PizzaSQL passes **100% of the SQLite SQLLogicTest suite** — over 5 million ind
 ### Architecture
 
 - **Hand-Written Lexer & Parser** — Pure Go implementation
-- **PizzaKV Storage** — Custom high-performance Zig backend with radix trie indexes
-- **Unix Socket Transport** — Low-latency communication between PizzaSQL and PizzaKV via Unix domain sockets
+- **PizzaKV Storage** — Durable append-only `.pkvdb` backend implemented in Zig
+- **PKBFI Transport** — Checksummed binary frames over TCP or Unix domain sockets
 - **Thread-Safe** — Concurrent query execution with mutex-based locking
 
 ---
@@ -149,7 +149,9 @@ pizzasql -kv -http -pg
 pizzasql -kv
 ```
 
-The `-kv` flag auto-launches a PizzaKV storage process connected via Unix socket (`.pizzakv.sock` in the working directory). PizzaSQL writes its runtime state to `/tmp/pizzasql/<pid>/runtime.json` and cleans up on exit.
+The `-kv` flag auto-launches a PizzaKV storage process connected via Unix socket (`.pizzakv.sock` in the working directory). PizzaKV stores data in `.pkvdb` by default. PizzaSQL writes its runtime state to `/tmp/pizzasql/<pid>/runtime.json` and cleans up on exit.
+
+PizzaSQL requires a PKBFI-capable PizzaKV. Existing legacy `.db` files must be migrated once before upgrading. See [`PKBFI_STORAGE_MIGRATION.md`](PKBFI_STORAGE_MIGRATION.md).
 
 ### Your First Query
 
@@ -208,7 +210,7 @@ PizzaSQL uses a per-instance runtime directory at `/tmp/pizzasql/<pid>/` to trac
     runtime.json
 ```
 
-**Multiple instances** are supported as long as each runs from a different working directory (each needs its own `.db` and `.pizzakv.sock` file). If you try to launch `-kv` in a directory that already has a `.db` file and another instance is running, PizzaSQL will refuse and tell you the conflicting PID.
+**Multiple instances** are supported as long as each runs from a different working directory (each needs its own `.pkvdb` and `.pizzakv.sock` file).
 
 **Stale entries** (from crashed processes) are cleaned up automatically on the next startup.
 
@@ -241,9 +243,7 @@ Performance on an M2 MacBook Air, 10,000-row table, 200 repetitions.
 | **Category scan (indexed)** | 0.537 ms | 0.279 ms | **0.108 ms** |
 | Value range (indexed) | 2.456 ms | 0.828 ms | 17.854 ms |
 
-Raw PizzaKV single-key read: **0.024 ms**. Full-table prefix scan (10k rows): **0.791 ms**. The dominant cost for full-scan queries is JSON deserialization (~15 ms for 10k rows).
-
-Indexed equality lookups are faster than both SQLite and PostgreSQL because PizzaKV's radix trie resolves the index directly to rowids with no B-tree traversal overhead.
+These figures predate the `.pkvdb`/PKBFI storage integration and are retained as a legacy baseline. Current PizzaSQL uses paginated binary scans and versioned binary row tuples; rerun the suite on the target hardware before using these numbers for capacity planning.
 
 ### Run Benchmarks
 

pkg/executor/executor.go

@@ -2480,13 +2480,11 @@ func (e *Executor) executeDropTable(stmt *parser.DropTableStmt) (*Result, error)
 			e.schema.DropIndex(idx.Name)
 		}
 
-		// Then, truncate all data rows
-		e.table.Truncate(tableRef.Name)
-
-		// Finally, drop the table schema
+		// DropTable removes durable rows and schema state together.
 		if err := e.schema.DropTable(tableRef.Name); err != nil {
 			return nil, err
 		}
+		e.table.InvalidateCache(tableRef.Name)
 
 	}
 	if err := e.SyncCatalog(); err != nil {
@@ -2745,6 +2743,8 @@ func (e *Executor) executeAlterTableRename(table string, action *parser.RenameTa
 	if err := e.schema.RenameTable(table, action.NewName); err != nil {
 		return nil, err
 	}
+	e.table.InvalidateCache(table)
+	e.table.InvalidateCache(action.NewName)
 
 	// Update catalog
 	e.SyncCatalog()

pkg/storage/count_cache_test.go

@@ -126,9 +126,8 @@ func TestCountFastDerivedAfterRestart(t *testing.T) {
 	}
 }
 
-// TestIncrementalCacheAndIndexMaintenance verifies that writes keep the row
-// cache and already-built in-memory indexes coherent without full-table cache
-// invalidation.
+// TestIncrementalCacheAndIndexMaintenance verifies that writes keep
+// already-built in-memory indexes coherent without full-table invalidation.
 func TestIncrementalCacheAndIndexMaintenance(t *testing.T) {
 	kv := newTestKVServer(t)
 	defer kv.close()
@@ -168,7 +167,7 @@ func TestIncrementalCacheAndIndexMaintenance(t *testing.T) {
 		t.Fatalf("insert 2: %v", err)
 	}
 
-	// Load the row cache and the index.
+	// Build the in-memory index (Select only streams rows).
 	all, err := tables.Select("users", nil)
 	if err != nil || len(all) != 2 {
 		t.Fatalf("initial select: len=%d err=%v", len(all), err)
@@ -314,9 +313,9 @@ func TestConcurrentDuplicateInsertKeepsCountExact(t *testing.T) {
 	}
 }
 
-// TestConcurrentFirstLoadAndInsert runs the first cache load (Select on an
-// unloaded table) concurrently with inserts, then verifies the cache ends up
-// coherent with durable rows. Run under -race to detect map/slice data races.
+// TestConcurrentFirstLoadAndInsert runs a first scan (Select on a table)
+// concurrently with inserts, then verifies the table ends up coherent with
+// durable rows. Run under -race to detect map/slice data races.
 func TestConcurrentFirstLoadAndInsert(t *testing.T) {
 	kv := newTestKVServer(t)
 	defer kv.close()

pkg/storage/kv.go

@@ -4,38 +4,234 @@ import (
 	"bufio"
 	"errors"
 	"fmt"
+	"hash/crc32"
+	"io"
 	"net"
 	"strings"
 	"sync"
 	"time"
 )
 
-// KVClient represents a connection to PizzaKV.
+const (
+	headerSize    = 32
+	headerMagic   = "PKBF"
+	headerVersion = 1
+
+	opPing       = 1
+	opStatus     = 2
+	opGet        = 3
+	opPut        = 4
+	opDelete     = 5
+	opExists     = 6
+	opMultiGet   = 7
+	opBatchWrite = 8
+	opScanOpen   = 9
+	opScanNext   = 10
+	opScanClose  = 11
+
+	batchPut    = 1
+	batchDelete = 2
+
+	statusOK       = 0
+	statusNotFound = 1
+	statusError    = 2
+
+	maxKeySize         = 1024 * 1024
+	maxValueSize       = 64 * 1024 * 1024
+	maxTransactionSize = 64 * 1024 * 1024
+	maxOperations      = 65535
+	maxFrameSize       = maxKeySize + maxValueSize + 1024
+
+	scanPageSize       = 1024
+	existsPipelineSize = 128
+)
+
+var crc32cTable = crc32.MakeTable(crc32.Castagnoli)
+
+var (
+	ErrKeyNotFound = errors.New("key not found")
+	ErrProtocol    = errors.New("pkbfi protocol error")
+)
+
+func crc32c(p []byte) uint32 {
+	return crc32.Checksum(p, crc32cTable)
+}
+
+func putU16(b []byte, v uint16) {
+	b[0] = byte(v)
+	b[1] = byte(v >> 8)
+}
+
+func putU32(b []byte, v uint32) {
+	b[0] = byte(v)
+	b[1] = byte(v >> 8)
+	b[2] = byte(v >> 16)
+	b[3] = byte(v >> 24)
+}
+
+func putU64(b []byte, v uint64) {
+	b[0] = byte(v)
+	b[1] = byte(v >> 8)
+	b[2] = byte(v >> 16)
+	b[3] = byte(v >> 24)
+	b[4] = byte(v >> 32)
+	b[5] = byte(v >> 40)
+	b[6] = byte(v >> 48)
+	b[7] = byte(v >> 56)
+}
+
+func getU16(b []byte) uint16 {
+	return uint16(b[0]) | uint16(b[1])<<8
+}
+
+func getU32(b []byte) uint32 {
+	return uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24
+}
+
+func getU64(b []byte) uint64 {
+	return uint64(b[0]) | uint64(b[1])<<8 | uint64(b[2])<<16 | uint64(b[3])<<24 |
+		uint64(b[4])<<32 | uint64(b[5])<<40 | uint64(b[6])<<48 | uint64(b[7])<<56
+}
+
+func encodeFrame(opcode, flags uint16, requestID uint64, payload []byte) []byte {
+	frame := make([]byte, headerSize+len(payload))
+	copy(frame[0:4], headerMagic)
+	putU16(frame[4:6], headerVersion)
+	putU16(frame[6:8], 0)
+	putU16(frame[8:10], opcode)
+	putU16(frame[10:12], flags)
+	putU64(frame[12:20], requestID)
+	putU32(frame[20:24], uint32(len(payload)))
+	putU32(frame[24:28], crc32c(payload))
+	putU32(frame[28:32], 0)
+	putU32(frame[28:32], crc32c(frame[0:32]))
+	copy(frame[32:], payload)
+	return frame
+}
+
+func readFrame(r *bufio.Reader) (uint16, uint16, uint64, []byte, error) {
+	var header [headerSize]byte
+	if _, err := io.ReadFull(r, header[:]); err != nil {
+		return 0, 0, 0, nil, err
+	}
+	if string(header[0:4]) != headerMagic {
+		return 0, 0, 0, nil, fmt.Errorf("%w: invalid magic", ErrProtocol)
+	}
+	if getU16(header[4:6]) != headerVersion {
+		return 0, 0, 0, nil, fmt.Errorf("%w: incompatible version", ErrProtocol)
+	}
+	payloadLen := getU32(header[20:24])
+	if payloadLen > maxFrameSize {
+		return 0, 0, 0, nil, fmt.Errorf("%w: frame too large", ErrProtocol)
+	}
+	headerCRC := getU32(header[28:32])
+	var headerCopy [headerSize]byte
+	copy(headerCopy[:], header[:])
+	putU32(headerCopy[28:32], 0)
+	if crc32c(headerCopy[:]) != headerCRC {
+		return 0, 0, 0, nil, fmt.Errorf("%w: header checksum mismatch", ErrProtocol)
+	}
+	payload := make([]byte, payloadLen)
+	if _, err := io.ReadFull(r, payload); err != nil {
+		return 0, 0, 0, nil, err
+	}
+	if crc32c(payload) != getU32(header[24:28]) {
+		return 0, 0, 0, nil, fmt.Errorf("%w: payload checksum mismatch", ErrProtocol)
+	}
+	return getU16(header[8:10]), getU16(header[10:12]), getU64(header[12:20]), payload, nil
+}
+
+func encodeResponse(opcode uint16, requestID uint64, body []byte) []byte {
+	return encodeFrame(opcode|0x8000, 1, requestID, body)
+}
+
+func oneKeyPayload(key []byte) []byte {
+	payload := make([]byte, 4+len(key))
+	putU32(payload[0:4], uint32(len(key)))
+	copy(payload[4:], key)
+	return payload
+}
+
+func validateKey(key []byte) error {
+	if len(key) > maxKeySize {
+		return fmt.Errorf("pkbfi: key exceeds %d bytes", maxKeySize)
+	}
+	return nil
+}
+
+func validateValue(value []byte) error {
+	if len(value) > maxValueSize {
+		return fmt.Errorf("pkbfi: value exceeds %d bytes", maxValueSize)
+	}
+	return nil
+}
+
+func parseOneKey(payload []byte) ([]byte, bool) {
+	if len(payload) < 4 {
+		return nil, false
+	}
+	length := getU32(payload[0:4])
+	if length > maxKeySize || uint64(4)+uint64(length) != uint64(len(payload)) {
+		return nil, false
+	}
+	return payload[4:], true
+}
+
+func errorBody(message string) []byte {
+	body := make([]byte, 2+len(message))
+	putU16(body[0:2], statusError)
+	copy(body[2:], message)
+	return body
+}
+
 type KVClient struct {
-	conn   net.Conn
-	reader *bufio.Reader
-	writer *bufio.Writer
-	mu     sync.Mutex
+	conn           net.Conn
+	reader         *bufio.Reader
+	writer         *bufio.Writer
+	mu             sync.Mutex
+	nextID         uint64
+	requestTimeout time.Duration
+	lastUsed       time.Time
+}
+
+type KVResult struct {
+	Value []byte
+	LSN   uint64
+	Found bool
+}
+
+type KVEntry struct {
+	Key   []byte
+	Value []byte
+	LSN   uint64
+}
+
+type BatchOp struct {
+	Op    byte
+	Key   []byte
+	Value []byte
+}
+
+type ScanCursor struct {
+	client *KVClient
+	id     uint64
 }
 
-// NewKVClient creates a new KV client connected to the given address.
-// addr may be "host:port" for TCP or "unix:<path>" for a Unix socket.
 func NewKVClient(addr string) (*KVClient, error) {
 	network, target := parseAddr(addr)
 	conn, err := net.Dial(network, target)
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to PizzaKV: %w", err)
 	}
-
 	return &KVClient{
-		conn:   conn,
-		reader: bufio.NewReader(conn),
-		writer: bufio.NewWriter(conn),
+		conn:     conn,
+		reader:   bufio.NewReader(conn),
+		writer:   bufio.NewWriter(conn),
+		nextID:   1,
+		lastUsed: time.Now(),
 	}, nil
 }
 
-// parseAddr splits an addr string into (network, address).
-// "unix:<path>" → ("unix", "<path>"), anything else → ("tcp", addr).
 func parseAddr(addr string) (string, string) {
 	if strings.HasPrefix(addr, "unix:") {
 		return "unix", strings.TrimPrefix(addr, "unix:")
@@ -43,164 +239,465 @@ func parseAddr(addr string) (string, string) {
 	return "tcp", addr
 }
 
-// Close closes the connection.
 func (c *KVClient) Close() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-
 	if c.conn != nil {
-		return c.conn.Close()
+		err := c.conn.Close()
+		c.conn = nil
+		return err
 	}
 	return nil
 }
 
-// SetDeadline sets the read/write deadline.
 func (c *KVClient) SetDeadline(t time.Time) error {
+	if c.conn == nil {
+		return nil
+	}
 	return c.conn.SetDeadline(t)
 }
 
-// Write stores a key-value pair.
-func (c *KVClient) Write(key, value string) error {
-	c.mu.Lock()
-	defer c.mu.Unlock()
+func (c *KVClient) writeFrame(opcode, flags uint16, requestID uint64, payload []byte) error {
+	frame := encodeFrame(opcode, flags, requestID, payload)
+	if _, err := c.writer.Write(frame); err != nil {
+		return err
+	}
+	return c.writer.Flush()
+}
 
-	cmd := fmt.Sprintf("write %s|%s\r", key, value)
-	if _, err := c.writer.WriteString(cmd); err != nil {
-		return fmt.Errorf("write command failed: %w", err)
+func (c *KVClient) request(opcode uint16, payload []byte) (uint16, []byte, error) {
+	if c.requestTimeout > 0 {
+		if err := c.conn.SetDeadline(time.Now().Add(c.requestTimeout)); err != nil {
+			return 0, nil, err
+		}
 	}
-	if err := c.writer.Flush(); err != nil {
-		return fmt.Errorf("flush failed: %w", err)
+	requestID := c.nextID
+	c.nextID++
+	if err := c.writeFrame(opcode, 0, requestID, payload); err != nil {
+		return 0, nil, err
 	}
-
-	resp, err := c.reader.ReadString('\r')
+	respOpcode, respFlags, respID, body, err := readFrame(c.reader)
 	if err != nil {
-		return fmt.Errorf("read response failed: %w", err)
+		return 0, nil, err
 	}
-
-	resp = strings.TrimSuffix(resp, "\r")
-	if resp != "success" {
-		return fmt.Errorf("write failed: %s", resp)
+	if respOpcode != opcode|0x8000 {
+		return 0, nil, fmt.Errorf("%w: unexpected response opcode %d", ErrProtocol, respOpcode)
+	}
+	if respFlags != 1 {
+		return 0, nil, fmt.Errorf("%w: unexpected response flags %d", ErrProtocol, respFlags)
+	}
+	if respID != requestID {
+		return 0, nil, fmt.Errorf("%w: response id %d does not match request %d", ErrProtocol, respID, requestID)
 	}
+	c.lastUsed = time.Now()
+	if len(body) < 2 {
+		return 0, nil, fmt.Errorf("%w: response too short", ErrProtocol)
+	}
+	status := getU16(body[0:2])
+	if status == statusError {
+		return status, nil, fmt.Errorf("pkbfi server error: %s", body[2:])
+	}
+	return status, body[2:], nil
+}
 
-	return nil
+func (c *KVClient) Put(key, value []byte) (uint64, error) {
+	if err := validateKey(key); err != nil {
+		return 0, err
+	}
+	if err := validateValue(value); err != nil {
+		return 0, err
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	payload := make([]byte, 8+len(key)+len(value))
+	putU32(payload[0:4], uint32(len(key)))
+	putU32(payload[4:8], uint32(len(value)))
+	copy(payload[8:], key)
+	copy(payload[8+len(key):], value)
+	status, body, err := c.request(opPut, payload)
+	if err != nil {
+		return 0, err
+	}
+	if status != statusOK || len(body) != 8 {
+		return 0, fmt.Errorf("%w: malformed put response", ErrProtocol)
+	}
+	return getU64(body[0:8]), nil
 }
 
-// Read retrieves a value by key.
-func (c *KVClient) Read(key string) (string, error) {
+func (c *KVClient) Get(key []byte) (KVResult, error) {
+	if err := validateKey(key); err != nil {
+		return KVResult{}, err
+	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	status, body, err := c.request(opGet, oneKeyPayload(key))
+	if err != nil {
+		return KVResult{}, err
+	}
+	if status == statusNotFound {
+		return KVResult{}, ErrKeyNotFound
+	}
+	if status != statusOK || len(body) < 12 {
+		return KVResult{}, fmt.Errorf("%w: malformed get response", ErrProtocol)
+	}
+	lsn := getU64(body[0:8])
+	valueLen := getU32(body[8:12])
+	if valueLen > maxValueSize || uint64(len(body)) != 12+uint64(valueLen) {
+		return KVResult{}, fmt.Errorf("%w: malformed get value length", ErrProtocol)
+	}
+	return KVResult{Value: body[12:], LSN: lsn, Found: true}, nil
+}
 
-	cmd := fmt.Sprintf("read %s\r", key)
-	if _, err := c.writer.WriteString(cmd); err != nil {
-		return "", fmt.Errorf("read command failed: %w", err)
+func (c *KVClient) Del(key []byte) (bool, error) {
+	if err := validateKey(key); err != nil {
+		return false, err
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	status, body, err := c.request(opDelete, oneKeyPayload(key))
+	if err != nil {
+		return false, err
 	}
-	if err := c.writer.Flush(); err != nil {
-		return "", fmt.Errorf("flush failed: %w", err)
+	if status != statusOK || len(body) != 1 {
+		return false, fmt.Errorf("%w: malformed delete response", ErrProtocol)
 	}
+	return body[0] != 0, nil
+}
 
-	resp, err := c.reader.ReadString('\r')
+func (c *KVClient) Exists(key []byte) (bool, error) {
+	if err := validateKey(key); err != nil {
+		return false, err
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	status, body, err := c.request(opExists, oneKeyPayload(key))
 	if err != nil {
-		return "", fmt.Errorf("read response failed: %w", err)
+		return false, err
+	}
+	if status != statusOK || len(body) != 1 {
+		return false, fmt.Errorf("%w: malformed exists response", ErrProtocol)
 	}
+	return body[0] != 0, nil
+}
 
-	resp = strings.TrimSuffix(resp, "\r")
-	if resp == "error" {
-		return "", ErrKeyNotFound
+func (c *KVClient) ExistsMany(keys [][]byte) ([]bool, error) {
+	if len(keys) > maxOperations {
+		return nil, fmt.Errorf("pkbfi: too many keys")
+	}
+	for _, key := range keys {
+		if err := validateKey(key); err != nil {
+			return nil, err
+		}
 	}
 
-	return resp, nil
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	results := make([]bool, len(keys))
+	for start := 0; start < len(keys); start += existsPipelineSize {
+		end := start + existsPipelineSize
+		if end > len(keys) {
+			end = len(keys)
+		}
+		ids := make([]uint64, end-start)
+		if c.requestTimeout > 0 {
+			if err := c.conn.SetDeadline(time.Now().Add(c.requestTimeout)); err != nil {
+				return nil, err
+			}
+		}
+		for i, key := range keys[start:end] {
+			ids[i] = c.nextID
+			c.nextID++
+			if _, err := c.writer.Write(encodeFrame(opExists, 0, ids[i], oneKeyPayload(key))); err != nil {
+				return nil, err
+			}
+		}
+		if err := c.writer.Flush(); err != nil {
+			return nil, err
+		}
+		for i, requestID := range ids {
+			opcode, flags, responseID, body, err := readFrame(c.reader)
+			if err != nil {
+				return nil, err
+			}
+			if opcode != opExists|0x8000 || flags != 1 || responseID != requestID {
+				return nil, fmt.Errorf("%w: malformed exists response frame", ErrProtocol)
+			}
+			if len(body) < 2 {
+				return nil, fmt.Errorf("%w: response too short", ErrProtocol)
+			}
+			status := getU16(body[0:2])
+			if status == statusError {
+				return nil, fmt.Errorf("pkbfi server error: %s", body[2:])
+			}
+			if status != statusOK || len(body) != 3 {
+				return nil, fmt.Errorf("%w: malformed exists response", ErrProtocol)
+			}
+			results[start+i] = body[2] != 0
+			c.lastUsed = time.Now()
+		}
+	}
+	return results, nil
 }
 
-// Delete removes a key.
-func (c *KVClient) Delete(key string) error {
+func (c *KVClient) MultiGet(keys [][]byte) ([]KVResult, error) {
+	if len(keys) > maxOperations {
+		return nil, fmt.Errorf("pkbfi: too many keys")
+	}
+	payloadSize := 4
+	for _, key := range keys {
+		if err := validateKey(key); err != nil {
+			return nil, err
+		}
+		payloadSize += 4 + len(key)
+		if payloadSize > maxFrameSize {
+			return nil, fmt.Errorf("pkbfi: multi_get request exceeds frame limit")
+		}
+	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-
-	cmd := fmt.Sprintf("delete %s\r", key)
-	if _, err := c.writer.WriteString(cmd); err != nil {
-		return fmt.Errorf("delete command failed: %w", err)
+	payload := make([]byte, 4, payloadSize)
+	putU32(payload[0:4], uint32(len(keys)))
+	for _, key := range keys {
+		var length [4]byte
+		putU32(length[:], uint32(len(key)))
+		payload = append(payload, length[:]...)
+		payload = append(payload, key...)
+	}
+	status, body, err := c.request(opMultiGet, payload)
+	if err != nil {
+		return nil, err
 	}
-	if err := c.writer.Flush(); err != nil {
-		return fmt.Errorf("flush failed: %w", err)
+	if status != statusOK || len(body) < 4 {
+		return nil, fmt.Errorf("%w: malformed multi_get response", ErrProtocol)
 	}
+	count := getU32(body[0:4])
+	if count != uint32(len(keys)) {
+		return nil, fmt.Errorf("%w: multi_get count mismatch", ErrProtocol)
+	}
+	results := make([]KVResult, count)
+	pos := 4
+	for i := uint32(0); i < count; i++ {
+		if len(body)-pos < 16 {
+			return nil, fmt.Errorf("%w: truncated multi_get entry", ErrProtocol)
+		}
+		present := body[pos] != 0
+		valueLen := getU32(body[pos+4 : pos+8])
+		lsn := getU64(body[pos+8 : pos+16])
+		pos += 16
+		results[i] = KVResult{LSN: lsn, Found: present}
+		if present {
+			if valueLen > maxValueSize || len(body)-pos < int(valueLen) {
+				return nil, fmt.Errorf("%w: multi_get value length", ErrProtocol)
+			}
+			results[i].Value = body[pos : pos+int(valueLen)]
+			pos += int(valueLen)
+		}
+	}
+	if pos != len(body) {
+		return nil, fmt.Errorf("%w: multi_get trailing bytes", ErrProtocol)
+	}
+	return results, nil
+}
 
-	resp, err := c.reader.ReadString('\r')
+func (c *KVClient) BatchWrite(ops []BatchOp, metadata []byte) (uint64, error) {
+	if len(ops) == 0 || len(ops) > maxOperations {
+		return 0, fmt.Errorf("pkbfi: invalid operation count")
+	}
+	if len(metadata) > maxTransactionSize-8 {
+		return 0, fmt.Errorf("pkbfi: batch metadata exceeds transaction limit")
+	}
+	payloadSize := 8 + len(metadata)
+	for _, op := range ops {
+		if op.Op != batchPut && op.Op != batchDelete {
+			return 0, fmt.Errorf("pkbfi: invalid batch opcode %d", op.Op)
+		}
+		if err := validateKey(op.Key); err != nil {
+			return 0, err
+		}
+		if err := validateValue(op.Value); err != nil {
+			return 0, err
+		}
+		if op.Op == batchDelete && len(op.Value) != 0 {
+			return 0, fmt.Errorf("pkbfi: delete operation with value")
+		}
+		payloadSize += 12 + len(op.Key) + len(op.Value)
+		if payloadSize > maxTransactionSize {
+			return 0, fmt.Errorf("pkbfi: batch exceeds transaction limit")
+		}
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	payload := make([]byte, 8, payloadSize)
+	putU32(payload[0:4], uint32(len(ops)))
+	putU32(payload[4:8], uint32(len(metadata)))
+	payload = append(payload, metadata...)
+	for _, op := range ops {
+		var header [12]byte
+		header[0] = op.Op
+		putU32(header[4:8], uint32(len(op.Key)))
+		putU32(header[8:12], uint32(len(op.Value)))
+		payload = append(payload, header[:]...)
+		payload = append(payload, op.Key...)
+		payload = append(payload, op.Value...)
+	}
+	status, body, err := c.request(opBatchWrite, payload)
 	if err != nil {
-		return fmt.Errorf("read response failed: %w", err)
+		return 0, err
 	}
-
-	resp = strings.TrimSuffix(resp, "\r")
-	if resp != "success" && resp != "error" {
-		return fmt.Errorf("delete failed: %s", resp)
+	if status != statusOK || len(body) != 8 {
+		return 0, fmt.Errorf("%w: malformed batch_write response", ErrProtocol)
 	}
+	return getU64(body[0:8]), nil
+}
 
-	return nil
+func (c *KVClient) Scan(prefix []byte) (*ScanCursor, error) {
+	return c.openScan(prefix, true, scanPageSize)
 }
 
-// Reads retrieves all values with a key prefix.
-func (c *KVClient) Reads(prefix string) ([]string, error) {
+func (c *KVClient) ScanWithLimit(prefix []byte, pageSize uint32) (*ScanCursor, error) {
+	return c.openScan(prefix, true, pageSize)
+}
+
+// ScanKeys opens a key-only scan: the server omits values from the returned
+// pages. It is used when only the key set (or its size) is needed, such as the
+// COUNT(*) fast path or bulk deletion, so a full-table scan does not pull row
+// values across the wire.
+func (c *KVClient) ScanKeys(prefix []byte) (*ScanCursor, error) {
+	return c.openScan(prefix, false, scanPageSize)
+}
+
+func (c *KVClient) openScan(prefix []byte, includeValues bool, pageSize uint32) (*ScanCursor, error) {
+	if err := validateKey(prefix); err != nil {
+		return nil, err
+	}
+	if pageSize == 0 || pageSize > 4096 {
+		return nil, fmt.Errorf("pkbfi: scan page size must be between 1 and 4096")
+	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	payload := make([]byte, 12+len(prefix))
+	if includeValues {
+		payload[0] = 1
+	}
+	putU32(payload[4:8], pageSize)
+	putU32(payload[8:12], uint32(len(prefix)))
+	copy(payload[12:], prefix)
+	status, body, err := c.request(opScanOpen, payload)
+	if err != nil {
+		return nil, err
+	}
+	if status != statusOK || len(body) != 8 {
+		return nil, fmt.Errorf("%w: malformed scan_open response", ErrProtocol)
+	}
+	return &ScanCursor{client: c, id: getU64(body[0:8])}, nil
+}
 
-	cmd := fmt.Sprintf("reads %s\r", prefix)
-	if _, err := c.writer.WriteString(cmd); err != nil {
-		return nil, fmt.Errorf("reads command failed: %w", err)
+func (s *ScanCursor) Next() ([]KVEntry, bool, error) {
+	s.client.mu.Lock()
+	defer s.client.mu.Unlock()
+	var payload [12]byte
+	putU64(payload[0:8], s.id)
+	putU32(payload[8:12], 0)
+	status, body, err := s.client.request(opScanNext, payload[:])
+	if err != nil {
+		return nil, false, err
+	}
+	if status != statusOK || len(body) < 8 {
+		return nil, false, fmt.Errorf("%w: malformed scan_next response", ErrProtocol)
+	}
+	done := body[0] != 0
+	count := getU32(body[4:8])
+	entries := make([]KVEntry, 0, count)
+	pos := 8
+	for i := uint32(0); i < count; i++ {
+		if len(body)-pos < 16 {
+			return nil, false, fmt.Errorf("%w: truncated scan entry", ErrProtocol)
+		}
+		keyLen := getU32(body[pos : pos+4])
+		valueLen := getU32(body[pos+4 : pos+8])
+		lsn := getU64(body[pos+8 : pos+16])
+		pos += 16
+		if keyLen > maxKeySize || valueLen > maxValueSize || len(body)-pos < int(keyLen)+int(valueLen) {
+			return nil, false, fmt.Errorf("%w: scan entry length", ErrProtocol)
+		}
+		key := body[pos : pos+int(keyLen)]
+		pos += int(keyLen)
+		value := body[pos : pos+int(valueLen)]
+		pos += int(valueLen)
+		entries = append(entries, KVEntry{Key: key, Value: value, LSN: lsn})
 	}
-	if err := c.writer.Flush(); err != nil {
-		return nil, fmt.Errorf("flush failed: %w", err)
+	if pos != len(body) {
+		return nil, false, fmt.Errorf("%w: scan trailing bytes", ErrProtocol)
 	}
+	return entries, done, nil
+}
 
-	resp, err := c.reader.ReadString('\r')
+func (s *ScanCursor) Close() error {
+	s.client.mu.Lock()
+	defer s.client.mu.Unlock()
+	var payload [8]byte
+	putU64(payload[0:8], s.id)
+	status, body, err := s.client.request(opScanClose, payload[:])
 	if err != nil {
-		return nil, fmt.Errorf("read response failed: %w", err)
+		return err
+	}
+	if status != statusOK || len(body) != 1 {
+		return fmt.Errorf("%w: malformed scan_close response", ErrProtocol)
 	}
+	return nil
+}
+
+func (c *KVClient) Write(key, value string) error {
+	_, err := c.Put([]byte(key), []byte(value))
+	return err
+}
 
-	resp = strings.TrimSuffix(resp, "\r")
-	if resp == "" {
-		return nil, nil
+func (c *KVClient) Read(key string) (string, error) {
+	res, err := c.Get([]byte(key))
+	if err != nil {
+		return "", err
 	}
+	return string(res.Value), nil
+}
+
+func (c *KVClient) Delete(key string) error {
+	_, err := c.Del([]byte(key))
+	return err
+}
 
-	values := strings.Split(resp, "\n")
-	result := make([]string, 0, len(values))
-	for _, v := range values {
-		if v != "" {
-			result = append(result, v)
+func (c *KVClient) Reads(prefix string) ([]string, error) {
+	scan, err := c.Scan([]byte(prefix))
+	if err != nil {
+		return nil, err
+	}
+	defer scan.Close()
+	values := make([]string, 0)
+	for {
+		entries, done, err := scan.Next()
+		if err != nil {
+			return nil, err
+		}
+		for _, entry := range entries {
+			values = append(values, string(entry.Value))
+		}
+		if done {
+			return values, nil
 		}
 	}
-
-	return result, nil
 }
 
-// IsAlive checks if the connection is still alive.
 func (c *KVClient) IsAlive() bool {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-
 	if c.conn == nil {
 		return false
 	}
-
-	// Try to set a short deadline and do a no-op check
-	c.conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
-	defer c.conn.SetReadDeadline(time.Time{})
-
-	one := make([]byte, 1)
-	c.conn.SetReadDeadline(time.Now().Add(1 * time.Millisecond))
-	_, err := c.conn.Read(one)
-
-	if err != nil {
-		if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
-			return true // Timeout is expected
-		}
-		return false
-	}
-	return true
+	c.conn.SetDeadline(time.Now().Add(500 * time.Millisecond))
+	defer c.conn.SetDeadline(time.Time{})
+	_, _, err := c.request(opPing, nil)
+	return err == nil
 }
 
-// ErrKeyNotFound is returned when a key doesn't exist.
-var ErrKeyNotFound = fmt.Errorf("key not found")
-
-// KVPool manages a pool of KV client connections.
 type KVPool struct {
 	addr    string
 	pool    chan *KVClient
@@ -210,7 +707,23 @@ type KVPool struct {
 	closed  bool
 }
 
-// NewKVPool creates a new connection pool.
+func (p *KVPool) replacementClient() (*KVClient, error) {
+	client, err := NewKVClient(p.addr)
+	if err == nil {
+		client.requestTimeout = p.timeout
+		return client, nil
+	}
+	p.mu.Lock()
+	if !p.closed {
+		select {
+		case p.pool <- nil:
+		default:
+		}
+	}
+	p.mu.Unlock()
+	return nil, err
+}
+
 func NewKVPool(addr string, size int, timeout time.Duration) (*KVPool, error) {
 	p := &KVPool{
 		addr:    addr,
@@ -218,22 +731,17 @@ func NewKVPool(addr string, size int, timeout time.Duration) (*KVPool, error) {
 		size:    size,
 		timeout: timeout,
 	}
-
-	// Pre-create connections
 	for i := 0; i < size; i++ {
 		client, err := NewKVClient(addr)
 		if err != nil {
-			// Close any created connections
 			p.Close()
 			return nil, fmt.Errorf("failed to create connection pool: %w", err)
 		}
 		p.pool <- client
 	}
-
 	return p, nil
 }
 
-// Get retrieves a connection from the pool.
 func (p *KVPool) Get() (*KVClient, error) {
 	p.mu.Lock()
 	if p.closed {
@@ -244,21 +752,22 @@ func (p *KVPool) Get() (*KVClient, error) {
 
 	select {
 	case client := <-p.pool:
-		// Validate connection
 		if client != nil && client.conn != nil {
-			if p.timeout > 0 {
-				client.SetDeadline(time.Now().Add(p.timeout))
+			if client.lastUsed.IsZero() {
+				client.lastUsed = time.Now()
+			} else if time.Since(client.lastUsed) >= 20*time.Second && !client.IsAlive() {
+				client.Close()
+				return p.replacementClient()
 			}
+			client.requestTimeout = p.timeout
 			return client, nil
 		}
-		// Stale connection — replace with a fresh one
-		return NewKVClient(p.addr)
+		return p.replacementClient()
 	case <-time.After(30 * time.Second):
 		return nil, fmt.Errorf("kv pool timeout: no connection available after 30s")
 	}
 }
 
-// Put returns a connection to the pool.
 func (p *KVPool) Put(client *KVClient) {
 	if client == nil {
 		return
@@ -272,19 +781,16 @@ func (p *KVPool) Put(client *KVClient) {
 	}
 	p.mu.Unlock()
 
-	// Clear deadline
+	client.requestTimeout = 0
 	client.SetDeadline(time.Time{})
 
 	select {
 	case p.pool <- client:
-		// Returned to pool
 	default:
-		// Pool full, close connection
 		client.Close()
 	}
 }
 
-// Close closes all connections in the pool.
 func (p *KVPool) Close() error {
 	p.mu.Lock()
 	if p.closed {
@@ -303,7 +809,6 @@ func (p *KVPool) Close() error {
 	return nil
 }
 
-// WithClient executes a function with a pooled connection.
 func (p *KVPool) WithClient(fn func(*KVClient) error) error {
 	client, err := p.Get()
 	if err != nil {
@@ -314,8 +819,6 @@ func (p *KVPool) WithClient(fn func(*KVClient) error) error {
 			p.Put(client)
 			return err
 		}
-		// A timeout or short response can leave an acknowledgement buffered on
-		// this connection. Never let the next request consume that response.
 		client.Close()
 		p.mu.Lock()
 		if !p.closed {
@@ -336,8 +839,8 @@ func isConnectionError(err error) bool {
 	if errors.As(err, &netErr) {
 		return true
 	}
-	message := err.Error()
-	return strings.Contains(message, "write command failed:") ||
-		strings.Contains(message, "flush failed:") ||
-		strings.Contains(message, "read response failed:")
+	if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+		return true
+	}
+	return errors.Is(err, ErrProtocol)
 }

pkg/storage/kv_test.go

@@ -0,0 +1,639 @@
+package storage
+
+import (
+	"bufio"
+	"bytes"
+	"errors"
+	"fmt"
+	"net"
+	"strings"
+	"testing"
+	"time"
+)
+
+func getResponseBody(value []byte, lsn uint64) []byte {
+	body := make([]byte, 14+len(value))
+	putU16(body[0:2], statusOK)
+	putU64(body[2:10], lsn)
+	putU32(body[10:14], uint32(len(value)))
+	copy(body[14:], value)
+	return body
+}
+
+func pipeClient(conn net.Conn) *KVClient {
+	return &KVClient{
+		conn:     conn,
+		reader:   bufio.NewReader(conn),
+		writer:   bufio.NewWriter(conn),
+		nextID:   1,
+		lastUsed: time.Now(),
+	}
+}
+
+func TestCRC32C(t *testing.T) {
+	if got := crc32c([]byte("123456789")); got != 0xe3069283 {
+		t.Fatalf("crc32c = %#x, want 0xe3069283", got)
+	}
+}
+
+func TestEncodeReadFrameRoundTrip(t *testing.T) {
+	payload := []byte{0x00, 0x01, 0xfe, '\n', '\r'}
+	frame := encodeFrame(opPut, 0, 42, payload)
+	r := bufio.NewReader(bytes.NewReader(frame))
+	opcode, flags, requestID, body, err := readFrame(r)
+	if err != nil {
+		t.Fatalf("readFrame: %v", err)
+	}
+	if opcode != opPut || flags != 0 || requestID != 42 {
+		t.Fatalf("opcode=%d flags=%d requestID=%d", opcode, flags, requestID)
+	}
+	if !bytes.Equal(body, payload) {
+		t.Fatalf("body = %x, want %x", body, payload)
+	}
+}
+
+func TestReadFrameRejectsOversizedFrame(t *testing.T) {
+	var header [headerSize]byte
+	copy(header[0:4], headerMagic)
+	putU16(header[4:6], headerVersion)
+	putU32(header[20:24], maxFrameSize+1)
+	r := bufio.NewReader(strings.NewReader(string(header[:])))
+	if _, _, _, _, err := readFrame(r); !errors.Is(err, ErrProtocol) {
+		t.Fatalf("err = %v, want ErrProtocol", err)
+	}
+}
+
+func TestClientRejectsOversizedRequests(t *testing.T) {
+	c := &KVClient{}
+	largeKey := make([]byte, maxKeySize+1)
+	largeValue := make([]byte, maxValueSize+1)
+
+	if _, err := c.Put(largeKey, nil); err == nil {
+		t.Fatal("Put accepted an oversized key")
+	}
+	if _, err := c.Put(nil, largeValue); err == nil {
+		t.Fatal("Put accepted an oversized value")
+	}
+	if _, err := c.Get(largeKey); err == nil {
+		t.Fatal("Get accepted an oversized key")
+	}
+	if _, err := c.MultiGet([][]byte{largeKey}); err == nil {
+		t.Fatal("MultiGet accepted an oversized key")
+	}
+	if _, err := c.BatchWrite([]BatchOp{{Op: batchPut, Key: []byte("k"), Value: largeValue}}, nil); err == nil {
+		t.Fatal("BatchWrite accepted an oversized value")
+	}
+	if _, err := c.Scan(largeKey); err == nil {
+		t.Fatal("Scan accepted an oversized prefix")
+	}
+	if _, err := c.ScanWithLimit(nil, 0); err == nil {
+		t.Fatal("ScanWithLimit accepted a zero page size")
+	}
+	if _, err := c.ScanWithLimit(nil, 4097); err == nil {
+		t.Fatal("ScanWithLimit accepted an oversized page size")
+	}
+}
+
+func TestPutGetBinaryRoundTrip(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	c := kv.client()
+	defer c.Close()
+
+	key := []byte{0x00, 0x01, 0x02, 'k', '\n'}
+	value := []byte{0xff, 0x00, '\r', '\n', 'v', 0x80}
+	lsn, err := c.Put(key, value)
+	if err != nil {
+		t.Fatalf("put: %v", err)
+	}
+	if lsn == 0 {
+		t.Fatalf("put returned zero lsn")
+	}
+	res, err := c.Get(key)
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if !res.Found {
+		t.Fatalf("expected found")
+	}
+	if res.LSN != lsn {
+		t.Fatalf("lsn = %d, want %d", res.LSN, lsn)
+	}
+	if !bytes.Equal(res.Value, value) {
+		t.Fatalf("value = %x, want %x", res.Value, value)
+	}
+}
+
+func TestGetNotFound(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	c := kv.client()
+	defer c.Close()
+
+	if _, err := c.Get([]byte("missing")); err != ErrKeyNotFound {
+		t.Fatalf("err = %v, want ErrKeyNotFound", err)
+	}
+	if _, err := c.Read("missing"); err != ErrKeyNotFound {
+		t.Fatalf("read err = %v, want ErrKeyNotFound", err)
+	}
+}
+
+func TestDeleteAndExists(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	c := kv.client()
+	defer c.Close()
+
+	if _, err := c.Put([]byte("k"), []byte("v")); err != nil {
+		t.Fatalf("put: %v", err)
+	}
+	found, err := c.Exists([]byte("k"))
+	if err != nil || !found {
+		t.Fatalf("exists = %v, %v", found, err)
+	}
+	deleted, err := c.Del([]byte("k"))
+	if err != nil || !deleted {
+		t.Fatalf("del = %v, %v", deleted, err)
+	}
+	found, err = c.Exists([]byte("k"))
+	if err != nil || found {
+		t.Fatalf("exists after delete = %v, %v", found, err)
+	}
+	deleted, err = c.Del([]byte("k"))
+	if err != nil || deleted {
+		t.Fatalf("del missing = %v, %v", deleted, err)
+	}
+}
+
+func TestExistsManyPipelinesRequests(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+	go func() {
+		conn, err := ln.Accept()
+		if err == nil {
+			kv.handle(conn)
+		}
+	}()
+	c, err := NewKVClient(ln.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close()
+
+	keys := make([][]byte, 300)
+	for i := range keys {
+		keys[i] = []byte(fmt.Sprintf("key:%03d", i))
+		if i%2 == 0 {
+			if _, err := c.Put(keys[i], []byte("v")); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	exists, err := c.ExistsMany(keys)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i, found := range exists {
+		if found != (i%2 == 0) {
+			t.Fatalf("exists[%d]=%v", i, found)
+		}
+	}
+}
+
+func TestMultiGet(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	c := kv.client()
+	defer c.Close()
+
+	if _, err := c.Put([]byte("a"), []byte("1")); err != nil {
+		t.Fatalf("put a: %v", err)
+	}
+	if _, err := c.Put([]byte("b"), []byte{0x00, 0x02}); err != nil {
+		t.Fatalf("put b: %v", err)
+	}
+	results, err := c.MultiGet([][]byte{[]byte("a"), []byte("missing"), []byte("b")})
+	if err != nil {
+		t.Fatalf("multi_get: %v", err)
+	}
+	if len(results) != 3 {
+		t.Fatalf("len = %d", len(results))
+	}
+	if !results[0].Found || string(results[0].Value) != "1" {
+		t.Fatalf("results[0] = %+v", results[0])
+	}
+	if results[1].Found {
+		t.Fatalf("results[1] should be missing")
+	}
+	if !results[2].Found || !bytes.Equal(results[2].Value, []byte{0x00, 0x02}) {
+		t.Fatalf("results[2] = %+v", results[2])
+	}
+}
+
+func TestBatchWrite(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	c := kv.client()
+	defer c.Close()
+
+	if _, err := c.Put([]byte("d"), []byte("old")); err != nil {
+		t.Fatalf("put d: %v", err)
+	}
+	lsn, err := c.BatchWrite([]BatchOp{
+		{Op: batchPut, Key: []byte("a"), Value: []byte("1")},
+		{Op: batchPut, Key: []byte("b"), Value: []byte("2")},
+		{Op: batchDelete, Key: []byte("d")},
+	}, []byte("meta"))
+	if err != nil {
+		t.Fatalf("batch_write: %v", err)
+	}
+	if lsn == 0 {
+		t.Fatalf("zero lsn")
+	}
+	ra, err := c.Get([]byte("a"))
+	if err != nil || !ra.Found || string(ra.Value) != "1" {
+		t.Fatalf("a = %+v, %v", ra, err)
+	}
+	rb, err := c.Get([]byte("b"))
+	if err != nil || !rb.Found || string(rb.Value) != "2" {
+		t.Fatalf("b = %+v, %v", rb, err)
+	}
+	rd, err := c.Get([]byte("d"))
+	if err != ErrKeyNotFound {
+		t.Fatalf("d err = %v, want ErrKeyNotFound", err)
+	}
+	if rd.Found {
+		t.Fatalf("d should be deleted")
+	}
+}
+
+func TestScanPagination(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	kv.maxScanPage = 2
+	c := kv.client()
+	defer c.Close()
+
+	expected := make(map[string]string)
+	for i := 0; i < 5; i++ {
+		key := fmt.Sprintf("pre:%02d", i)
+		value := fmt.Sprintf("v%d", i)
+		if _, err := c.Put([]byte(key), []byte(value)); err != nil {
+			t.Fatalf("put %s: %v", key, err)
+		}
+		expected[key] = value
+	}
+	scan, err := c.Scan([]byte("pre:"))
+	if err != nil {
+		t.Fatalf("scan: %v", err)
+	}
+	defer scan.Close()
+
+	var entries []KVEntry
+	pages := 0
+	for {
+		batch, done, err := scan.Next()
+		if err != nil {
+			t.Fatalf("scan next: %v", err)
+		}
+		pages++
+		entries = append(entries, batch...)
+		if done {
+			break
+		}
+	}
+	if pages < 3 {
+		t.Fatalf("expected pagination across multiple pages, got %d", pages)
+	}
+	if len(entries) != 5 {
+		t.Fatalf("entries = %d, want 5", len(entries))
+	}
+	for _, e := range entries {
+		if want := expected[string(e.Key)]; string(e.Value) != want {
+			t.Fatalf("key %q value = %q, want %q", e.Key, e.Value, want)
+		}
+		if e.LSN == 0 {
+			t.Fatalf("key %q has zero lsn", e.Key)
+		}
+	}
+}
+
+func TestScanBinaryValues(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	c := kv.client()
+	defer c.Close()
+
+	key := []byte{0x00, 0x01, 'p'}
+	value := []byte{0xff, 0x00, '\n', '\r'}
+	if _, err := c.Put(key, value); err != nil {
+		t.Fatalf("put: %v", err)
+	}
+	scan, err := c.Scan([]byte{0x00})
+	if err != nil {
+		t.Fatalf("scan: %v", err)
+	}
+	defer scan.Close()
+	batch, done, err := scan.Next()
+	if err != nil {
+		t.Fatalf("next: %v", err)
+	}
+	if !done || len(batch) != 1 {
+		t.Fatalf("done=%v len=%d", done, len(batch))
+	}
+	if !bytes.Equal(batch[0].Key, key) {
+		t.Fatalf("key = %x, want %x", batch[0].Key, key)
+	}
+	if !bytes.Equal(batch[0].Value, value) {
+		t.Fatalf("value = %x, want %x", batch[0].Value, value)
+	}
+}
+
+func TestReadsNoDelimiterAssumptions(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	c := kv.client()
+	defer c.Close()
+
+	values := []string{"a\nb", "c\r\nd", "", "e\rf\ng"}
+	for i, v := range values {
+		if _, err := c.Put([]byte(fmt.Sprintf("p:%d", i)), []byte(v)); err != nil {
+			t.Fatalf("put %d: %v", i, err)
+		}
+	}
+	got, err := c.Reads("p:")
+	if err != nil {
+		t.Fatalf("reads: %v", err)
+	}
+	if len(got) != len(values) {
+		t.Fatalf("reads returned %d values, want %d", len(got), len(values))
+	}
+	for i, v := range values {
+		if got[i] != v {
+			t.Fatalf("got[%d] = %q, want %q", i, got[i], v)
+		}
+	}
+}
+
+func TestHeaderCRCError(t *testing.T) {
+	clientConn, serverConn := net.Pipe()
+	defer clientConn.Close()
+	defer serverConn.Close()
+	c := pipeClient(clientConn)
+
+	go func() {
+		r := bufio.NewReader(serverConn)
+		_, _, requestID, _, err := readFrame(r)
+		if err != nil {
+			return
+		}
+		resp := encodeResponse(opGet, requestID, getResponseBody([]byte("v"), 1))
+		resp[6] ^= 0xff
+		serverConn.Write(resp)
+	}()
+
+	if _, err := c.Read("k"); !errors.Is(err, ErrProtocol) {
+		t.Fatalf("err = %v, want ErrProtocol", err)
+	}
+}
+
+func TestPayloadCRCError(t *testing.T) {
+	clientConn, serverConn := net.Pipe()
+	defer clientConn.Close()
+	defer serverConn.Close()
+	c := pipeClient(clientConn)
+
+	go func() {
+		r := bufio.NewReader(serverConn)
+		_, _, requestID, _, err := readFrame(r)
+		if err != nil {
+			return
+		}
+		resp := encodeResponse(opGet, requestID, getResponseBody([]byte("v"), 1))
+		resp[headerSize] ^= 0xff
+		serverConn.Write(resp)
+	}()
+
+	if _, err := c.Read("k"); !errors.Is(err, ErrProtocol) {
+		t.Fatalf("err = %v, want ErrProtocol", err)
+	}
+}
+
+func TestServerErrorStatus(t *testing.T) {
+	clientConn, serverConn := net.Pipe()
+	defer clientConn.Close()
+	defer serverConn.Close()
+	c := pipeClient(clientConn)
+
+	go func() {
+		r := bufio.NewReader(serverConn)
+		_, _, requestID, _, err := readFrame(r)
+		if err != nil {
+			return
+		}
+		serverConn.Write(encodeResponse(opGet, requestID, errorBody("Boom")))
+	}()
+
+	_, err := c.Read("k")
+	if err == nil {
+		t.Fatalf("expected error")
+	}
+	if errors.Is(err, ErrProtocol) {
+		t.Fatalf("server error should not be ErrProtocol: %v", err)
+	}
+	if !strings.Contains(err.Error(), "Boom") {
+		t.Fatalf("err = %v", err)
+	}
+}
+
+func TestRequestIDMismatch(t *testing.T) {
+	clientConn, serverConn := net.Pipe()
+	defer clientConn.Close()
+	defer serverConn.Close()
+	c := pipeClient(clientConn)
+
+	go func() {
+		r := bufio.NewReader(serverConn)
+		_, _, requestID, _, err := readFrame(r)
+		if err != nil {
+			return
+		}
+		serverConn.Write(encodeResponse(opGet, requestID+1, getResponseBody([]byte("v"), 1)))
+	}()
+
+	if _, err := c.Read("k"); !errors.Is(err, ErrProtocol) {
+		t.Fatalf("err = %v, want ErrProtocol", err)
+	}
+}
+
+func TestOpcodeMismatch(t *testing.T) {
+	clientConn, serverConn := net.Pipe()
+	defer clientConn.Close()
+	defer serverConn.Close()
+	c := pipeClient(clientConn)
+
+	go func() {
+		r := bufio.NewReader(serverConn)
+		_, _, requestID, _, err := readFrame(r)
+		if err != nil {
+			return
+		}
+		serverConn.Write(encodeResponse(opPut, requestID, getResponseBody([]byte("v"), 1)))
+	}()
+
+	if _, err := c.Read("k"); !errors.Is(err, ErrProtocol) {
+		t.Fatalf("err = %v, want ErrProtocol", err)
+	}
+}
+
+func TestPartialReads(t *testing.T) {
+	clientConn, serverConn := net.Pipe()
+	defer clientConn.Close()
+	defer serverConn.Close()
+	c := pipeClient(clientConn)
+
+	go func() {
+		r := bufio.NewReader(serverConn)
+		_, _, requestID, _, err := readFrame(r)
+		if err != nil {
+			return
+		}
+		resp := encodeResponse(opGet, requestID, getResponseBody([]byte("hello"), 7))
+		for _, b := range resp {
+			if _, err := serverConn.Write([]byte{b}); err != nil {
+				return
+			}
+		}
+	}()
+
+	res, err := c.Get([]byte("k"))
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if !res.Found || string(res.Value) != "hello" || res.LSN != 7 {
+		t.Fatalf("res = %+v", res)
+	}
+}
+
+func TestMonotonicRequestIDs(t *testing.T) {
+	clientConn, serverConn := net.Pipe()
+	defer clientConn.Close()
+	defer serverConn.Close()
+	c := pipeClient(clientConn)
+
+	ids := make(chan uint64, 4)
+	go func() {
+		r := bufio.NewReader(serverConn)
+		for i := 0; i < 4; i++ {
+			_, _, requestID, _, err := readFrame(r)
+			if err != nil {
+				return
+			}
+			ids <- requestID
+			serverConn.Write(encodeResponse(opPing, requestID, []byte{0, 0}))
+		}
+	}()
+
+	for i := 0; i < 4; i++ {
+		if _, _, err := c.request(opPing, nil); err != nil {
+			t.Fatalf("ping %d: %v", i, err)
+		}
+	}
+	var prev uint64
+	for i := 0; i < 4; i++ {
+		id := <-ids
+		if i > 0 && id <= prev {
+			t.Fatalf("request id not monotonic: %d then %d", prev, id)
+		}
+		prev = id
+	}
+}
+
+func TestPoolReconnectsAfterBrokenConnection(t *testing.T) {
+	s := newTestKVServer(t)
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	t.Cleanup(func() { ln.Close() })
+
+	accepted := make(chan net.Conn, 8)
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			accepted <- conn
+			go s.handle(conn)
+		}
+	}()
+
+	pool, err := NewKVPool(ln.Addr().String(), 1, 2*time.Second)
+	if err != nil {
+		t.Fatalf("pool: %v", err)
+	}
+	defer pool.Close()
+
+	if err := pool.WithClient(func(c *KVClient) error { return c.Write("k", "v") }); err != nil {
+		t.Fatalf("first write: %v", err)
+	}
+	if err := pool.WithClient(func(c *KVClient) error {
+		v, err := c.Read("k")
+		if err != nil {
+			return err
+		}
+		if v != "v" {
+			return fmt.Errorf("read = %q", v)
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("first read: %v", err)
+	}
+
+	first := <-accepted
+	first.Close()
+
+	if err := pool.WithClient(func(c *KVClient) error { return c.Write("k", "v") }); err == nil {
+		t.Fatalf("expected write on broken connection to fail")
+	}
+
+	if err := pool.WithClient(func(c *KVClient) error {
+		v, err := c.Read("k")
+		if err != nil {
+			return err
+		}
+		if v != "v" {
+			return fmt.Errorf("read = %q", v)
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("read after reconnect: %v", err)
+	}
+
+	client, err := pool.Get()
+	if err != nil {
+		t.Fatal(err)
+	}
+	client.lastUsed = time.Now().Add(-31 * time.Second)
+	pool.Put(client)
+	second := <-accepted
+	second.Close()
+	if err := pool.WithClient(func(c *KVClient) error {
+		v, err := c.Read("k")
+		if err != nil {
+			return err
+		}
+		if v != "v" {
+			return fmt.Errorf("read = %q", v)
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("stale idle connection was not replaced before use: %v", err)
+	}
+}

pkg/storage/pizzakv_integration_test.go

@@ -0,0 +1,253 @@
+package storage
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/goccy/go-json"
+)
+
+func startPizzaKVTest(t testing.TB, binary, socket, database string) func() {
+	t.Helper()
+	cmd := exec.Command(binary, "-unix="+socket, "-path="+database)
+	cmd.Stdout = os.Stderr
+	cmd.Stderr = os.Stderr
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start PizzaKV: %v", err)
+	}
+	var once sync.Once
+	stop := func() {
+		once.Do(func() {
+			_ = cmd.Process.Kill()
+			_ = cmd.Wait()
+		})
+	}
+	t.Cleanup(stop)
+	return stop
+}
+
+func waitPizzaKVPool(t testing.TB, socket string) *KVPool {
+	t.Helper()
+	addr := "unix:" + socket
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		pool, err := NewKVPool(addr, 2, 5*time.Second)
+		if err == nil {
+			return pool
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatal("PizzaKV did not become ready")
+	return nil
+}
+
+func shortPizzaKVSocket(t testing.TB) string {
+	t.Helper()
+	dir, err := os.MkdirTemp("/tmp", "pkv-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = os.RemoveAll(dir) })
+	return filepath.Join(dir, "s")
+}
+
+func TestPizzaKVIntegration(t *testing.T) {
+	binary := os.Getenv("PIZZAKV_BIN")
+	if binary == "" {
+		t.Skip("PIZZAKV_BIN is not set")
+	}
+
+	dir := t.TempDir()
+	socket := shortPizzaKVSocket(t)
+	database := filepath.Join(dir, "integration.pkvdb")
+	stop := startPizzaKVTest(t, binary, socket, database)
+	pool := waitPizzaKVPool(t, socket)
+
+	schemas := NewSchemaManager(pool, "integration")
+	tables := NewTableManager(pool, schemas, "integration")
+	if err := schemas.CreateTable(&Schema{
+		Name: "events",
+		Columns: []Column{
+			{Name: "id", Type: "INTEGER", Nullable: false, PrimaryKey: true},
+			{Name: "payload", Type: "BLOB", Nullable: true},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	for i := int64(1); i <= 20; i++ {
+		payload := []byte{byte(i), 0, '|', '\r', '\n'}
+		if err := tables.Insert("events", Row{"id": i, "payload": payload}); err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+	}
+	largePayload := bytes.Repeat([]byte{0xab}, 2*1024*1024)
+	if err := tables.Insert("events", Row{"id": int64(21), "payload": largePayload}); err != nil {
+		t.Fatalf("insert large row: %v", err)
+	}
+	largeRows, err := tables.Select("events", func(row Row) bool { return row["id"] == int64(21) })
+	if err != nil || len(largeRows) != 1 {
+		t.Fatalf("scan large row: len=%d err=%v", len(largeRows), err)
+	}
+	if got, ok := largeRows[0]["payload"].([]byte); !ok || !bytes.Equal(got, largePayload) {
+		t.Fatalf("large binary payload mismatch: len=%d type=%T", len(got), largeRows[0]["payload"])
+	}
+	rows, err := tables.SelectWithLimit("events", nil, 3, 2)
+	if err != nil || len(rows) != 3 {
+		t.Fatalf("limited select: len=%d err=%v", len(rows), err)
+	}
+	row, err := tables.GetByPK("events", "1")
+	if err != nil {
+		t.Fatalf("point read: %v", err)
+	}
+	if got, ok := row["payload"].([]byte); !ok || !bytes.Equal(got, []byte{1, 0, '|', '\r', '\n'}) {
+		t.Fatalf("binary payload = %v (%T)", row["payload"], row["payload"])
+	}
+	if count, err := tables.CountFast("events"); err != nil || count != 21 {
+		t.Fatalf("count = %d, err=%v", count, err)
+	}
+
+	if err := pool.Close(); err != nil {
+		t.Fatalf("close pool: %v", err)
+	}
+	stop()
+	startPizzaKVTest(t, binary, socket, database)
+	pool = waitPizzaKVPool(t, socket)
+	defer pool.Close()
+	schemas = NewSchemaManager(pool, "integration")
+	tables = NewTableManager(pool, schemas, "integration")
+	row, err = tables.GetByPK("events", "1")
+	if err != nil {
+		t.Fatalf("point read after restart: %v", err)
+	}
+	if got, ok := row["payload"].([]byte); !ok || !bytes.Equal(got, []byte{1, 0, '|', '\r', '\n'}) {
+		t.Fatalf("binary payload after restart = %v (%T)", row["payload"], row["payload"])
+	}
+	if count, err := tables.CountFast("events"); err != nil || count != 21 {
+		t.Fatalf("count after restart = %d, err=%v", count, err)
+	}
+	if err := schemas.DropTable("events"); err != nil {
+		t.Fatalf("drop table: %v", err)
+	}
+	if schemas.TableExists("events") {
+		t.Fatal("table still exists")
+	}
+}
+
+func TestPizzaKVLegacyMigrationIntegration(t *testing.T) {
+	binary := os.Getenv("PIZZAKV_BIN")
+	if binary == "" {
+		t.Skip("PIZZAKV_BIN is not set")
+	}
+
+	dir := t.TempDir()
+	source := filepath.Join(dir, "legacy.db")
+	destination := filepath.Join(dir, "legacy.pkvdb")
+	socket := shortPizzaKVSocket(t)
+	schema, err := json.Marshal(&Schema{
+		Name:       "events",
+		Columns:    []Column{{Name: "id", Type: "INTEGER", PrimaryKey: true}, {Name: "name", Type: "TEXT", Nullable: true}},
+		PrimaryKey: "id",
+		NextRowID:  2,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	legacy := []byte(fmt.Sprintf(
+		"W|legacy:_schema:events|%s\rW|legacy:_sys:tables|[\"events\"]\rW|legacy:_data:events:1|{\"id\":1,\"name\":\"old\",\"_rowid_\":1}\r",
+		schema,
+	))
+	if err := os.WriteFile(source, legacy, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if output, err := exec.Command(binary, "-migrate="+source, "-path="+destination).CombinedOutput(); err != nil {
+		t.Fatalf("migrate legacy database: %v\n%s", err, output)
+	}
+	unchanged, err := os.ReadFile(source)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(unchanged, legacy) {
+		t.Fatal("migration modified the legacy source")
+	}
+
+	startPizzaKVTest(t, binary, socket, destination)
+	pool := waitPizzaKVPool(t, socket)
+	defer pool.Close()
+	schemas := NewSchemaManager(pool, "legacy")
+	tables := NewTableManager(pool, schemas, "legacy")
+	row, err := tables.GetByPK("events", "1")
+	if err != nil || row["name"] != "old" {
+		t.Fatalf("read migrated row: row=%v err=%v", row, err)
+	}
+	if err := tables.Insert("events", Row{"id": int64(2), "name": "new"}); err != nil {
+		t.Fatalf("insert binary row after migration: %v", err)
+	}
+	rows, err := tables.Select("events", nil)
+	if err != nil || len(rows) != 2 {
+		t.Fatalf("mixed legacy/binary scan: len=%d err=%v", len(rows), err)
+	}
+}
+
+func BenchmarkPizzaKVStorage(b *testing.B) {
+	binary := os.Getenv("PIZZAKV_BIN")
+	if binary == "" {
+		b.Skip("PIZZAKV_BIN is not set")
+	}
+
+	dir := b.TempDir()
+	socket := shortPizzaKVSocket(b)
+	startPizzaKVTest(b, binary, socket, filepath.Join(dir, "benchmark.pkvdb"))
+	pool := waitPizzaKVPool(b, socket)
+	defer pool.Close()
+	schemas := NewSchemaManager(pool, "benchmark")
+	tables := NewTableManager(pool, schemas, "benchmark")
+	if err := schemas.CreateTable(&Schema{
+		Name: "events",
+		Columns: []Column{
+			{Name: "id", Type: "INTEGER", Nullable: false, PrimaryKey: true},
+			{Name: "symbol", Type: "TEXT", Nullable: false},
+			{Name: "price", Type: "REAL", Nullable: false},
+		},
+	}); err != nil {
+		b.Fatal(err)
+	}
+	rows := make([]Row, 10_000)
+	for i := range rows {
+		rows[i] = Row{"id": int64(i + 1), "symbol": fmt.Sprintf("PIZZA-%03d", i%100), "price": float64(i) / 100}
+	}
+	if n, err := tables.InsertBulk("events", rows); err != nil || n != len(rows) {
+		b.Fatalf("seed: n=%d err=%v", n, err)
+	}
+
+	b.Run("point_read", func(b *testing.B) {
+		b.ReportAllocs()
+		for b.Loop() {
+			if _, err := tables.GetByPK("events", "5000"); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+	b.Run("limit_10", func(b *testing.B) {
+		b.ReportAllocs()
+		for b.Loop() {
+			if _, err := tables.SelectWithLimit("events", nil, 10, 0); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+	b.Run("scan_10000", func(b *testing.B) {
+		b.ReportAllocs()
+		for b.Loop() {
+			if rows, err := tables.Select("events", nil); err != nil || len(rows) != 10_000 {
+				b.Fatalf("scan: len=%d err=%v", len(rows), err)
+			}
+		}
+	})
+}

pkg/storage/rowcodec.go

@@ -0,0 +1,301 @@
+package storage
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"math"
+	"sort"
+
+	"github.com/goccy/go-json"
+)
+
+// rowMagic prefixes every versioned binary row value. It is chosen so it can
+// never be the first bytes of a legacy JSON row (which always begins with '{',
+// '[', '"', a digit, 't', 'f', or 'n'), so decodeRow can disambiguate the two
+// encodings unambiguously.
+const rowMagic = "PZSQLROW"
+
+// rowVersion is the format version. It must be bumped whenever the binary
+// layout changes in a way that would make old bytes undecodable.
+const rowVersion = 1
+
+// rowHeaderLen is the fixed size of the binary header: magic + version + count.
+const rowHeaderLen = len(rowMagic) + 1 + 4
+
+// maxRowFieldLen caps the encoded length of a field name or a variable-length
+// value (string, bytes, json.Number). It is far larger than any value the KV
+// layer can return (64 MiB), so it only ever rejects adversarial lengths.
+const maxRowFieldLen = 1 << 30
+
+// minFieldEncodedSize is the smallest possible on-disk size of a single field:
+// a 4-byte name length, an empty name, and a 1-byte type tag.
+const minFieldEncodedSize = 4 + 1
+
+// Value type tags. A tag occupies one byte and precedes the value payload.
+const (
+	tagNil     = 0x00
+	tagFalse   = 0x01
+	tagTrue    = 0x02
+	tagInt     = 0x03 // signed integer, normalized to int64
+	tagUint    = 0x04 // unsigned integer, normalized to uint64
+	tagFloat32 = 0x05
+	tagFloat64 = 0x06
+	tagString  = 0x07
+	tagBytes   = 0x08
+	tagNumber  = 0x09 // json.Number, preserved verbatim as decimal bytes
+)
+
+var errMalformedRow = errors.New("malformed row encoding")
+
+// encodeRow serializes a row into a deterministic, compact, versioned binary
+// value. Field names are sorted so identical rows always encode to identical
+// bytes. If any value cannot be represented exactly in the binary format
+// (e.g. a slice, map, struct, or time), the entire row is encoded as legacy
+// JSON instead so no data is lost.
+func encodeRow(row Row) ([]byte, error) {
+	if len(row) > math.MaxUint32 {
+		return nil, fmt.Errorf("row has too many fields")
+	}
+	names := make([]string, 0, len(row))
+	for name := range row {
+		if len(name) > maxRowFieldLen {
+			return nil, fmt.Errorf("row field name is too long")
+		}
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	// Encode values first; fall back to JSON if any is unrepresentable.
+	encoded := make([][]byte, len(names))
+	for i, name := range names {
+		enc, ok := encodeValue(row[name])
+		if !ok {
+			return json.Marshal(row)
+		}
+		encoded[i] = enc
+	}
+
+	buf := make([]byte, 0, rowHeaderLen+len(names)*8)
+	buf = append(buf, rowMagic...)
+	buf = append(buf, rowVersion)
+	buf = appendU32(buf, uint32(len(names)))
+	for i, name := range names {
+		buf = appendU32(buf, uint32(len(name)))
+		buf = append(buf, name...)
+		buf = append(buf, encoded[i]...)
+	}
+	return buf, nil
+}
+
+// encodeValue returns the type tag plus payload for v, and reports whether v
+// can be represented exactly. All Go integer widths are normalized to their
+// fixed-width equivalents; every other supported type is self-describing.
+func encodeValue(v interface{}) ([]byte, bool) {
+	switch t := v.(type) {
+	case nil:
+		return []byte{tagNil}, true
+	case bool:
+		if t {
+			return []byte{tagTrue}, true
+		}
+		return []byte{tagFalse}, true
+	case int:
+		return appendU64([]byte{tagInt}, uint64(int64(t))), true
+	case int8:
+		return appendU64([]byte{tagInt}, uint64(int64(t))), true
+	case int16:
+		return appendU64([]byte{tagInt}, uint64(int64(t))), true
+	case int32:
+		return appendU64([]byte{tagInt}, uint64(int64(t))), true
+	case int64:
+		return appendU64([]byte{tagInt}, uint64(t)), true
+	case uint:
+		return appendU64([]byte{tagUint}, uint64(t)), true
+	case uint8:
+		return appendU64([]byte{tagUint}, uint64(t)), true
+	case uint16:
+		return appendU64([]byte{tagUint}, uint64(t)), true
+	case uint32:
+		return appendU64([]byte{tagUint}, uint64(t)), true
+	case uint64:
+		return appendU64([]byte{tagUint}, t), true
+	case uintptr:
+		return appendU64([]byte{tagUint}, uint64(t)), true
+	case float32:
+		var b [5]byte
+		b[0] = tagFloat32
+		binary.LittleEndian.PutUint32(b[1:], math.Float32bits(t))
+		return b[:], true
+	case float64:
+		var b [9]byte
+		b[0] = tagFloat64
+		binary.LittleEndian.PutUint64(b[1:], math.Float64bits(t))
+		return b[:], true
+	case string:
+		if len(t) > maxRowFieldLen {
+			return nil, false
+		}
+		return appendBytesField([]byte{tagString}, []byte(t)), true
+	case []byte:
+		if len(t) > maxRowFieldLen {
+			return nil, false
+		}
+		return appendBytesField([]byte{tagBytes}, t), true
+	case json.Number:
+		if len(t) > maxRowFieldLen {
+			return nil, false
+		}
+		return appendBytesField([]byte{tagNumber}, []byte(string(t))), true
+	default:
+		return nil, false
+	}
+}
+
+// decodeRow decodes a row value in either the versioned binary format or the
+// legacy untagged JSON format. The two are distinguished solely by the magic
+// prefix: bytes carrying the magic are always parsed as binary and never fall
+// back to JSON, while anything else is parsed as legacy JSON for backward
+// compatibility.
+func decodeRow(data []byte) (Row, error) {
+	if len(data) >= len(rowMagic) && string(data[:len(rowMagic)]) == rowMagic {
+		return decodeBinaryRow(data)
+	}
+
+	var row Row
+	if err := json.Unmarshal(data, &row); err != nil {
+		return nil, err
+	}
+	return row, nil
+}
+
+// decodeBinaryRow parses a versioned binary row, validating the magic,
+// version, field count, per-field length bounds, and that no trailing bytes
+// remain once every field has been consumed.
+func decodeBinaryRow(data []byte) (Row, error) {
+	if len(data) < rowHeaderLen {
+		return nil, errMalformedRow
+	}
+	if string(data[:len(rowMagic)]) != rowMagic {
+		return nil, errMalformedRow
+	}
+	version := data[len(rowMagic)]
+	if version != rowVersion {
+		return nil, fmt.Errorf("%w: unsupported version %d", errMalformedRow, version)
+	}
+
+	count := binary.LittleEndian.Uint32(data[len(rowMagic)+1 : len(rowMagic)+5])
+	pos := rowHeaderLen
+	remaining := len(data) - pos
+
+	// Reject impossible field counts up front so a hostile count cannot drive
+	// an unbounded loop: every field occupies at least minFieldEncodedSize.
+	if uint64(count)*minFieldEncodedSize > uint64(remaining) {
+		return nil, errMalformedRow
+	}
+
+	row := make(Row, count)
+	for i := uint32(0); i < count; i++ {
+		if remaining < 4 {
+			return nil, errMalformedRow
+		}
+		nameLen := binary.LittleEndian.Uint32(data[pos : pos+4])
+		pos += 4
+		remaining -= 4
+		if nameLen > maxRowFieldLen || uint64(nameLen) > uint64(remaining) {
+			return nil, errMalformedRow
+		}
+		name := string(data[pos : pos+int(nameLen)])
+		pos += int(nameLen)
+		remaining -= int(nameLen)
+
+		if remaining < 1 {
+			return nil, errMalformedRow
+		}
+		tag := data[pos]
+		pos++
+		remaining--
+
+		value, n, err := decodeValue(tag, data[pos:])
+		if err != nil {
+			return nil, err
+		}
+		pos += n
+		remaining -= n
+		row[name] = value
+	}
+
+	if pos != len(data) {
+		return nil, errMalformedRow
+	}
+	return row, nil
+}
+
+// decodeValue decodes a single tagged value from data, returning the value and
+// the number of payload bytes consumed. Variable-length payloads are validated
+// against both the global cap and the actual remaining input.
+func decodeValue(tag byte, data []byte) (interface{}, int, error) {
+	switch tag {
+	case tagNil:
+		return nil, 0, nil
+	case tagFalse:
+		return false, 0, nil
+	case tagTrue:
+		return true, 0, nil
+	case tagInt:
+		if len(data) < 8 {
+			return nil, 0, errMalformedRow
+		}
+		return int64(binary.LittleEndian.Uint64(data[:8])), 8, nil
+	case tagUint:
+		if len(data) < 8 {
+			return nil, 0, errMalformedRow
+		}
+		return binary.LittleEndian.Uint64(data[:8]), 8, nil
+	case tagFloat32:
+		if len(data) < 4 {
+			return nil, 0, errMalformedRow
+		}
+		return math.Float32frombits(binary.LittleEndian.Uint32(data[:4])), 4, nil
+	case tagFloat64:
+		if len(data) < 8 {
+			return nil, 0, errMalformedRow
+		}
+		return math.Float64frombits(binary.LittleEndian.Uint64(data[:8])), 8, nil
+	case tagString, tagBytes, tagNumber:
+		if len(data) < 4 {
+			return nil, 0, errMalformedRow
+		}
+		l := binary.LittleEndian.Uint32(data[:4])
+		if l > maxRowFieldLen || uint64(l) > uint64(len(data)-4) {
+			return nil, 0, errMalformedRow
+		}
+		content := data[4 : 4+int(l)]
+		switch tag {
+		case tagString:
+			return string(content), 4 + int(l), nil
+		case tagBytes:
+			return append([]byte(nil), content...), 4 + int(l), nil
+		case tagNumber:
+			return json.Number(string(content)), 4 + int(l), nil
+		}
+	}
+	return nil, 0, fmt.Errorf("%w: unknown tag %d", errMalformedRow, tag)
+}
+
+func appendU32(dst []byte, v uint32) []byte {
+	var b [4]byte
+	binary.LittleEndian.PutUint32(b[:], v)
+	return append(dst, b[:]...)
+}
+
+func appendU64(dst []byte, v uint64) []byte {
+	var b [8]byte
+	binary.LittleEndian.PutUint64(b[:], v)
+	return append(dst, b[:]...)
+}
+
+func appendBytesField(dst []byte, data []byte) []byte {
+	dst = appendU32(dst, uint32(len(data)))
+	return append(dst, data...)
+}

pkg/storage/rowcodec_test.go

@@ -0,0 +1,421 @@
+package storage
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"reflect"
+	"testing"
+
+	"github.com/goccy/go-json"
+)
+
+func TestEncodeRowDeterministic(t *testing.T) {
+	row := Row{
+		"b": int64(2),
+		"a": int64(1),
+		"c": int64(3),
+	}
+	first, err := encodeRow(row)
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	// Rebuild with the same pairs in a different insertion order.
+	rowAgain := Row{}
+	rowAgain["c"] = int64(3)
+	rowAgain["a"] = int64(1)
+	rowAgain["b"] = int64(2)
+	second, err := encodeRow(rowAgain)
+	if err != nil {
+		t.Fatalf("encodeRow again: %v", err)
+	}
+	if !bytes.Equal(first, second) {
+		t.Fatalf("encoding is not deterministic:\n%x\n%x", first, second)
+	}
+}
+
+func TestEncodeRowBinaryBytes(t *testing.T) {
+	got, err := encodeRow(Row{"a": int64(1)})
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	want := []byte{
+		'P', 'Z', 'S', 'Q', 'L', 'R', 'O', 'W', // magic
+		0x01,                   // version
+		0x01, 0x00, 0x00, 0x00, // count = 1
+		0x01, 0x00, 0x00, 0x00, // nameLen = 1
+		'a',                                            // name
+		0x03,                                           // tagInt
+		0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // int64(1)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("binary bytes = %x, want %x", got, want)
+	}
+}
+
+func TestEncodeDecodeRoundTripAllTypes(t *testing.T) {
+	in := Row{
+		"nil":   nil,
+		"bt":    true,
+		"bf":    false,
+		"i":     int(42),
+		"i8":    int8(-8),
+		"i16":   int16(-1600),
+		"i32":   int32(-70000),
+		"i64":   int64(-9000000000000000000),
+		"u":     uint(7),
+		"u8":    uint8(200),
+		"u16":   uint16(60000),
+		"u32":   uint32(4000000000),
+		"u64":   uint64(18446744073709551615),
+		"f32":   float32(1.5),
+		"f64":   float64(-2.25),
+		"str":   "hello",
+		"bytes": []byte{0x00, 0xff, 0x01, '\n'},
+		"num":   json.Number("12345678901234567890"),
+	}
+	want := Row{
+		"nil":   nil,
+		"bt":    true,
+		"bf":    false,
+		"i":     int64(42),
+		"i8":    int64(-8),
+		"i16":   int64(-1600),
+		"i32":   int64(-70000),
+		"i64":   int64(-9000000000000000000),
+		"u":     uint64(7),
+		"u8":    uint64(200),
+		"u16":   uint64(60000),
+		"u32":   uint64(4000000000),
+		"u64":   uint64(18446744073709551615),
+		"f32":   float32(1.5),
+		"f64":   float64(-2.25),
+		"str":   "hello",
+		"bytes": []byte{0x00, 0xff, 0x01, '\n'},
+		"num":   json.Number("12345678901234567890"),
+	}
+
+	data, err := encodeRow(in)
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	if len(data) < len(rowMagic) || string(data[:len(rowMagic)]) != rowMagic {
+		t.Fatalf("binary row missing magic prefix: %x", data)
+	}
+
+	got, err := decodeRow(data)
+	if err != nil {
+		t.Fatalf("decodeRow: %v", err)
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("round trip mismatch:\n got = %#v\nwant = %#v", got, want)
+	}
+}
+
+func TestEncodeRowJSONNumberExact(t *testing.T) {
+	// A decimal that would lose precision as a float64 must round-trip exactly.
+	row := Row{"n": json.Number("0.123456789012345678901234567890")}
+	data, err := encodeRow(row)
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	got, err := decodeRow(data)
+	if err != nil {
+		t.Fatalf("decodeRow: %v", err)
+	}
+	if got["n"] != json.Number("0.123456789012345678901234567890") {
+		t.Fatalf("number = %#v, want exact json.Number", got["n"])
+	}
+}
+
+func TestDecodeLegacyJSON(t *testing.T) {
+	legacy := []byte(`{"_rowid_":7,"name":"alice","score":12.5,"active":true,"extra":null}`)
+	got, err := decodeRow(legacy)
+	if err != nil {
+		t.Fatalf("decodeRow: %v", err)
+	}
+	if got["_rowid_"] != float64(7) {
+		t.Fatalf("_rowid_ = %#v, want float64(7)", got["_rowid_"])
+	}
+	if got["name"] != "alice" {
+		t.Fatalf("name = %#v", got["name"])
+	}
+	if got["score"] != float64(12.5) {
+		t.Fatalf("score = %#v", got["score"])
+	}
+	if got["active"] != true {
+		t.Fatalf("active = %#v", got["active"])
+	}
+	if got["extra"] != nil {
+		t.Fatalf("extra = %#v", got["extra"])
+	}
+}
+
+func TestEncodeRowUnsupportedValueFallsBackToJSON(t *testing.T) {
+	row := Row{"id": int64(1), "tags": []string{"a", "b"}}
+	data, err := encodeRow(row)
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	if len(data) >= len(rowMagic) && string(data[:len(rowMagic)]) == rowMagic {
+		t.Fatalf("expected JSON fallback, got binary magic: %x", data)
+	}
+
+	var decoded Row
+	if err := json.Unmarshal(data, &decoded); err != nil {
+		t.Fatalf("fallback is not valid JSON: %v", err)
+	}
+	if decoded["id"] != float64(1) {
+		t.Fatalf("id = %#v", decoded["id"])
+	}
+	tags, ok := decoded["tags"].([]interface{})
+	if !ok || len(tags) != 2 || tags[0] != "a" || tags[1] != "b" {
+		t.Fatalf("tags = %#v", decoded["tags"])
+	}
+}
+
+func TestEncodeRowJSONFallbackRoundTripsThroughDecode(t *testing.T) {
+	row := Row{"nested": map[string]interface{}{"x": 1, "y": []interface{}{true, nil}}}
+	data, err := encodeRow(row)
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	got, err := decodeRow(data)
+	if err != nil {
+		t.Fatalf("decodeRow: %v", err)
+	}
+	if _, ok := got["nested"].(map[string]interface{}); !ok {
+		t.Fatalf("nested = %#v, want map", got["nested"])
+	}
+}
+
+func TestDecodeBinaryRowTruncated(t *testing.T) {
+	data, err := encodeRow(Row{"name": "alice", "id": int64(5), "payload": []byte("data")})
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	for _, n := range []int{1, len(rowMagic), rowHeaderLen, rowHeaderLen + 1, len(data) - 1} {
+		trunc := data[:n]
+		if _, err := decodeRow(trunc); err == nil {
+			t.Fatalf("decodeRow(truncated to %d bytes) succeeded, want error", n)
+		}
+	}
+}
+
+func TestDecodeBinaryRowTrailingBytes(t *testing.T) {
+	data, err := encodeRow(Row{"id": int64(1)})
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	withTrailing := append(append([]byte(nil), data...), 0x00, 0x01, 0x02)
+	if _, err := decodeRow(withTrailing); err == nil {
+		t.Fatalf("decodeRow with trailing bytes succeeded, want error")
+	}
+}
+
+func TestDecodeBinaryRowUnknownTag(t *testing.T) {
+	var buf []byte
+	buf = append(buf, rowMagic...)
+	buf = append(buf, rowVersion)
+	buf = appendU32(buf, 1)
+	buf = appendU32(buf, 2)
+	buf = append(buf, "id"...)
+	buf = append(buf, 0x7f) // unknown tag
+	buf = append(buf, 0, 0, 0, 0, 0, 0, 0, 0)
+	if _, err := decodeRow(buf); !errors.Is(err, errMalformedRow) {
+		t.Fatalf("err = %v, want errMalformedRow", err)
+	}
+}
+
+func TestDecodeBinaryRowUnknownVersion(t *testing.T) {
+	data, err := encodeRow(Row{"id": int64(1)})
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	corrupted := append([]byte(nil), data...)
+	corrupted[len(rowMagic)] = 0x7f
+	if _, err := decodeRow(corrupted); !errors.Is(err, errMalformedRow) {
+		t.Fatalf("err = %v, want errMalformedRow", err)
+	}
+}
+
+func TestDecodeBinaryRowOversizedNameLength(t *testing.T) {
+	var buf []byte
+	buf = append(buf, rowMagic...)
+	buf = append(buf, rowVersion)
+	buf = appendU32(buf, 1)
+	buf = appendU32(buf, uint32(maxRowFieldLen+1)) // oversized name length
+	buf = append(buf, 'x')
+	if _, err := decodeRow(buf); err == nil {
+		t.Fatalf("decodeRow with oversized name length succeeded, want error")
+	}
+}
+
+func TestDecodeBinaryRowOversizedValueLength(t *testing.T) {
+	var buf []byte
+	buf = append(buf, rowMagic...)
+	buf = append(buf, rowVersion)
+	buf = appendU32(buf, 1)
+	buf = appendU32(buf, 1)
+	buf = append(buf, 'a')
+	buf = append(buf, tagString)
+	buf = appendU32(buf, uint32(maxRowFieldLen+1)) // oversized string length
+	if _, err := decodeRow(buf); err == nil {
+		t.Fatalf("decodeRow with oversized value length succeeded, want error")
+	}
+}
+
+func TestDecodeBinaryRowStringLengthExceedsInput(t *testing.T) {
+	var buf []byte
+	buf = append(buf, rowMagic...)
+	buf = append(buf, rowVersion)
+	buf = appendU32(buf, 1)
+	buf = appendU32(buf, 1)
+	buf = append(buf, 'a')
+	buf = append(buf, tagString)
+	buf = appendU32(buf, 100) // claims 100 bytes but only 0 follow
+	if _, err := decodeRow(buf); err == nil {
+		t.Fatalf("decodeRow with lying string length succeeded, want error")
+	}
+}
+
+func TestDecodeBinaryRowImpossibleFieldCount(t *testing.T) {
+	var buf []byte
+	buf = append(buf, rowMagic...)
+	buf = append(buf, rowVersion)
+	buf = appendU32(buf, 0xffffffff) // far more fields than bytes available
+	if _, err := decodeRow(buf); err == nil {
+		t.Fatalf("decodeRow with impossible field count succeeded, want error")
+	}
+}
+
+func TestDecodeMalformedTaggedBinaryNotReinterpretedAsJSON(t *testing.T) {
+	// Bytes carrying the magic prefix must never fall back to the JSON path,
+	// even if the tail happens to look JSON-ish.
+	corrupted := append([]byte(nil), rowMagic...)
+	corrupted = append(corrupted, rowVersion)
+	corrupted = append(corrupted, 0xff, 0xff, 0xff, 0xff) // bogus count
+	corrupted = append(corrupted, 'g', 'a', 'r', 'b', 'a', 'g', 'e')
+
+	if _, err := decodeRow(corrupted); err == nil {
+		t.Fatalf("decodeRow succeeded on malformed tagged binary, want error")
+	}
+}
+
+func TestDecodeNonJSONNonBinaryInput(t *testing.T) {
+	// No magic prefix and not valid JSON must fail rather than panic or return
+	// a partial row.
+	if _, err := decodeRow([]byte{0x01, 0x02, 0x03, 0x04}); err == nil {
+		t.Fatalf("decodeRow on garbage succeeded, want error")
+	}
+}
+
+func TestEncodeDecodeEmptyRow(t *testing.T) {
+	data, err := encodeRow(Row{})
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	got, err := decodeRow(data)
+	if err != nil {
+		t.Fatalf("decodeRow: %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("empty row decoded to %#v", got)
+	}
+}
+
+func TestEncodeDecodeNilRow(t *testing.T) {
+	data, err := encodeRow(nil)
+	if err != nil {
+		t.Fatalf("encodeRow(nil): %v", err)
+	}
+	got, err := decodeRow(data)
+	if err != nil {
+		t.Fatalf("decodeRow: %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("nil row decoded to %#v", got)
+	}
+}
+
+func TestEncodeRowSortedFieldNames(t *testing.T) {
+	data, err := encodeRow(Row{"z": int64(3), "a": int64(1), "m": int64(2)})
+	if err != nil {
+		t.Fatalf("encodeRow: %v", err)
+	}
+	// Verify the field names appear in sorted order by walking the encoding.
+	pos := rowHeaderLen
+	count := binary.LittleEndian.Uint32(data[pos-4 : pos])
+	names := make([]string, 0, count)
+	for i := uint32(0); i < count; i++ {
+		nameLen := binary.LittleEndian.Uint32(data[pos : pos+4])
+		pos += 4
+		names = append(names, string(data[pos:pos+int(nameLen)]))
+		pos += int(nameLen)
+		pos++ // skip tag
+		switch data[pos-1] {
+		case tagInt, tagUint, tagFloat64:
+			pos += 8
+		case tagFloat32:
+			pos += 4
+		case tagString, tagBytes, tagNumber:
+			l := binary.LittleEndian.Uint32(data[pos : pos+4])
+			pos += 4 + int(l)
+		}
+	}
+	if !reflect.DeepEqual(names, []string{"a", "m", "z"}) {
+		t.Fatalf("field names = %v, want [a m z]", names)
+	}
+}
+
+func BenchmarkRowCodec(b *testing.B) {
+	row := Row{
+		"_rowid_": int64(4812),
+		"id":      int64(4812),
+		"symbol":  "PIZZA",
+		"price":   104.25,
+		"active":  true,
+		"payload": []byte{0, 1, 2, '|', '\r', '\n'},
+	}
+	binaryRow, err := encodeRow(row)
+	if err != nil {
+		b.Fatal(err)
+	}
+	jsonRow, err := json.Marshal(row)
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	b.Run("encode_binary", func(b *testing.B) {
+		b.ReportAllocs()
+		for b.Loop() {
+			if _, err := encodeRow(row); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+	b.Run("encode_json", func(b *testing.B) {
+		b.ReportAllocs()
+		for b.Loop() {
+			if _, err := json.Marshal(row); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+	b.Run("decode_binary", func(b *testing.B) {
+		b.ReportAllocs()
+		for b.Loop() {
+			if _, err := decodeRow(binaryRow); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+	b.Run("decode_json", func(b *testing.B) {
+		b.ReportAllocs()
+		for b.Loop() {
+			if _, err := decodeRow(jsonRow); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+}

pkg/storage/schema.go

@@ -54,6 +54,8 @@ type SchemaManager struct {
 	version          uint64
 	mu               sync.RWMutex
 	txMu             sync.RWMutex
+	tableLocksMu     sync.Mutex
+	tableLocks       map[string]*sync.RWMutex
 }
 
 // BeginTransaction prevents other connections from observing intermediate
@@ -76,9 +78,22 @@ func NewSchemaManager(pool *KVPool, database string) *SchemaManager {
 		database:         database,
 		cache:            make(map[string]*Schema),
 		rowIDInitialized: make(map[string]bool),
+		tableLocks:       make(map[string]*sync.RWMutex),
 	}
 }
 
+func (m *SchemaManager) tableLock(table string) *sync.RWMutex {
+	key := strings.ToLower(table)
+	m.tableLocksMu.Lock()
+	lock, ok := m.tableLocks[key]
+	if !ok {
+		lock = &sync.RWMutex{}
+		m.tableLocks[key] = lock
+	}
+	m.tableLocksMu.Unlock()
+	return lock
+}
+
 // GetDatabaseName returns the database name.
 func (m *SchemaManager) GetDatabaseName() string {
 	return m.database
@@ -184,6 +199,10 @@ func (m *SchemaManager) CreateTable(schema *Schema) error {
 
 // DropTable drops a table.
 func (m *SchemaManager) DropTable(name string) error {
+	tableLock := m.tableLock(name)
+	tableLock.Lock()
+	defer tableLock.Unlock()
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -198,20 +217,11 @@ func (m *SchemaManager) DropTable(name string) error {
 		return fmt.Errorf("table not found: %s", name)
 	}
 
-	// Delete all rows
-	dataPrefix := fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(name))
-	err = m.pool.WithClient(func(c *KVClient) error {
-		// Get all keys with this prefix and delete them
-		// Note: This is a simplified version - in production you'd want batch delete
-		values, err := c.Reads(dataPrefix)
-		if err != nil {
-			return err
-		}
-		// The Reads command returns values, not keys, so we can't delete them directly
-		// In a real implementation, we'd need a keys scan command
-		_ = values
-		return nil
-	})
+	// Delete all rows by scanning their actual keys and batch-deleting them, so
+	// a table drop no longer leaks durable rows.
+	if err := m.deleteKeysWithPrefix([]byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(name)))); err != nil {
+		return err
+	}
 
 	// Delete schema
 	err = m.pool.WithClient(func(c *KVClient) error {
@@ -482,32 +492,95 @@ func (m *SchemaManager) getNextRowIDLocked(schema *Schema) (int64, error) {
 	return schema.NextRowID, nil
 }
 
-// deriveNextRowIDLocked scans durable row values to recover max(rowid)+1.
+// deriveNextRowIDLocked scans durable row keys to recover max(rowid)+1,
+// streaming each page through decodeRow instead of materializing every value.
 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 {
-		var err error
-		values, err = c.Reads(prefix)
-		return err
+	prefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(schema.Name)))
+	var maxRowID int64
+	err := m.pool.WithClient(func(client *KVClient) (retErr error) {
+		cursor, err := client.Scan(prefix)
+		if err != nil {
+			return err
+		}
+		defer func() {
+			if err := cursor.Close(); retErr == nil {
+				retErr = err
+			}
+		}()
+		for {
+			entries, done, err := cursor.Next()
+			if err != nil {
+				return err
+			}
+			for _, e := range entries {
+				row, err := decodeRow(e.Value)
+				if err != nil {
+					return fmt.Errorf("failed to parse row while deriving ROWID: %w", err)
+				}
+				if rowid, ok := valueAsInt64(row["_rowid_"]); ok && rowid > maxRowID {
+					maxRowID = rowid
+				}
+			}
+			if done {
+				return nil
+			}
+		}
 	})
 	if err != nil {
 		return 0, err
 	}
+	return maxRowID + 1, nil
+}
 
-	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)
+// deleteKeysWithPrefix streams the keys with the given prefix one page at a
+// time and atomically batch-deletes them, so a bulk operation never leaves
+// durable rows behind.
+func (m *SchemaManager) deleteKeysWithPrefix(prefix []byte) error {
+	return m.pool.WithClient(func(client *KVClient) (retErr error) {
+		cursor, err := client.ScanKeys(prefix)
+		if err != nil {
+			return err
 		}
+		defer func() {
+			if err := cursor.Close(); retErr == nil {
+				retErr = err
+			}
+		}()
 
-		if rowid, ok := valueAsInt64(row["_rowid_"]); ok && rowid > maxRowID {
-			maxRowID = rowid
+		ops := make([]BatchOp, 0, scanPageSize)
+		batchBytes := 8
+		flush := func() error {
+			if len(ops) == 0 {
+				return nil
+			}
+			if _, err := client.BatchWrite(ops, nil); err != nil {
+				return err
+			}
+			ops = ops[:0]
+			batchBytes = 8
+			return nil
 		}
-	}
 
-	return maxRowID + 1, nil
+		for {
+			entries, done, err := cursor.Next()
+			if err != nil {
+				return err
+			}
+			for _, e := range entries {
+				opBytes := 12 + len(e.Key)
+				if len(ops) == maxOperations || batchBytes+opBytes > bulkBatchByteBudget {
+					if err := flush(); err != nil {
+						return err
+					}
+				}
+				ops = append(ops, BatchOp{Op: batchDelete, Key: append([]byte(nil), e.Key...)})
+				batchBytes += opBytes
+			}
+			if done {
+				return flush()
+			}
+		}
+	})
 }
 
 func valueAsInt64(value interface{}) (int64, bool) {

pkg/storage/schema_test.go

@@ -4,24 +4,40 @@ import (
 	"bufio"
 	"fmt"
 	"net"
+	"sort"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 )
 
+type testScan struct {
+	keys     []string
+	offset   int
+	limit    uint32
+	keysOnly bool
+}
+
 type testKVServer struct {
-	mu      sync.Mutex
-	data    map[string]string
-	writes  map[string]int
-	closers []net.Conn
+	mu           sync.Mutex
+	data         map[string][]byte
+	lsns         map[string]uint64
+	writes       map[string]int
+	nextLSN      uint64
+	maxScanPage  uint32
+	scanOpens    int
+	scanNexts    int
+	scanCloses   int
+	keyOnlyOpens int
+	closers      []net.Conn
 }
 
 func newTestKVServer(t *testing.T) *testKVServer {
 	t.Helper()
 
 	return &testKVServer{
-		data:   make(map[string]string),
+		data:   make(map[string][]byte),
+		lsns:   make(map[string]uint64),
 		writes: make(map[string]int),
 	}
 }
@@ -57,9 +73,11 @@ func (s *testKVServer) client() *KVClient {
 
 	go s.handle(serverConn)
 	return &KVClient{
-		conn:   clientConn,
-		reader: bufio.NewReader(clientConn),
-		writer: bufio.NewWriter(clientConn),
+		conn:     clientConn,
+		reader:   bufio.NewReader(clientConn),
+		writer:   bufio.NewWriter(clientConn),
+		nextID:   1,
+		lastUsed: time.Now(),
 	}
 }
 
@@ -84,60 +102,361 @@ func (s *testKVServer) hasKey(key string) bool {
 	return ok
 }
 
+func (s *testKVServer) countKeys(prefix string) int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var count int
+	for key := range s.data {
+		if strings.HasPrefix(key, prefix) {
+			count++
+		}
+	}
+	return count
+}
+
+func (s *testKVServer) scanStats() (opens, nexts, closes int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.scanOpens, s.scanNexts, s.scanCloses
+}
+
+func (s *testKVServer) keyOnlyOpenCount() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.keyOnlyOpens
+}
+
 func (s *testKVServer) handle(conn net.Conn) {
 	defer conn.Close()
 
 	r := bufio.NewReader(conn)
+	scans := make(map[uint64]*testScan)
+	var nextScan uint64 = 1
 	for {
-		cmd, err := r.ReadString('\r')
+		opcode, _, requestID, payload, err := readFrame(r)
 		if err != nil {
 			return
 		}
-		cmd = strings.TrimSuffix(cmd, "\r")
-
-		resp := s.execute(cmd)
-		if _, err := fmt.Fprintf(conn, "%s\r", resp); err != nil {
+		body := s.execute(opcode, payload, scans, &nextScan)
+		if _, err := conn.Write(encodeResponse(opcode, requestID, body)); err != nil {
 			return
 		}
 	}
 }
 
-func (s *testKVServer) execute(cmd string) string {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	switch {
-	case strings.HasPrefix(cmd, "write "):
-		parts := strings.SplitN(strings.TrimPrefix(cmd, "write "), "|", 2)
-		if len(parts) != 2 {
-			return "error"
-		}
-		s.data[parts[0]] = parts[1]
-		s.writes[parts[0]]++
-		return "success"
-	case strings.HasPrefix(cmd, "read "):
-		key := strings.TrimPrefix(cmd, "read ")
-		value, ok := s.data[key]
+func (s *testKVServer) execute(opcode uint16, payload []byte, scans map[uint64]*testScan, nextScan *uint64) []byte {
+	switch opcode {
+	case opPing:
+		body := make([]byte, 2+len(payload))
+		putU16(body[0:2], statusOK)
+		copy(body[2:], payload)
+		return body
+	case opGet:
+		key, ok := parseOneKey(payload)
+		if !ok {
+			return errorBody("InvalidPayload")
+		}
+		s.mu.Lock()
+		value, found := s.data[string(key)]
+		lsn := s.lsns[string(key)]
+		s.mu.Unlock()
+		if !found {
+			body := make([]byte, 2)
+			putU16(body[0:2], statusNotFound)
+			return body
+		}
+		body := make([]byte, 14+len(value))
+		putU16(body[0:2], statusOK)
+		putU64(body[2:10], lsn)
+		putU32(body[10:14], uint32(len(value)))
+		copy(body[14:], value)
+		return body
+	case opPut:
+		key, value, ok := parsePut(payload)
+		if !ok {
+			return errorBody("InvalidPayload")
+		}
+		s.mu.Lock()
+		s.nextLSN++
+		lsn := s.nextLSN
+		s.data[string(key)] = append([]byte(nil), value...)
+		s.lsns[string(key)] = lsn
+		s.writes[string(key)]++
+		s.mu.Unlock()
+		body := make([]byte, 10)
+		putU16(body[0:2], statusOK)
+		putU64(body[2:10], lsn)
+		return body
+	case opDelete:
+		key, ok := parseOneKey(payload)
+		if !ok {
+			return errorBody("InvalidPayload")
+		}
+		s.mu.Lock()
+		_, found := s.data[string(key)]
+		delete(s.data, string(key))
+		delete(s.lsns, string(key))
+		s.mu.Unlock()
+		body := make([]byte, 3)
+		putU16(body[0:2], statusOK)
+		if found {
+			body[2] = 1
+		}
+		return body
+	case opExists:
+		key, ok := parseOneKey(payload)
+		if !ok {
+			return errorBody("InvalidPayload")
+		}
+		s.mu.Lock()
+		_, found := s.data[string(key)]
+		s.mu.Unlock()
+		body := make([]byte, 3)
+		putU16(body[0:2], statusOK)
+		if found {
+			body[2] = 1
+		}
+		return body
+	case opMultiGet:
+		keys, ok := parseMultiGetKeys(payload)
+		if !ok {
+			return errorBody("InvalidPayload")
+		}
+		body := make([]byte, 6)
+		putU16(body[0:2], statusOK)
+		putU32(body[2:6], uint32(len(keys)))
+		s.mu.Lock()
+		for _, key := range keys {
+			value, found := s.data[string(key)]
+			if !found {
+				body = append(body, make([]byte, 16)...)
+				continue
+			}
+			lsn := s.lsns[string(key)]
+			entry := make([]byte, 16+len(value))
+			entry[0] = 1
+			putU32(entry[4:8], uint32(len(value)))
+			putU64(entry[8:16], lsn)
+			copy(entry[16:], value)
+			body = append(body, entry...)
+		}
+		s.mu.Unlock()
+		return body
+	case opBatchWrite:
+		ops, ok := parseBatchOps(payload)
+		if !ok {
+			return errorBody("InvalidPayload")
+		}
+		s.mu.Lock()
+		s.nextLSN++
+		lsn := s.nextLSN
+		for _, op := range ops {
+			if op.Op == batchPut {
+				s.data[string(op.Key)] = append([]byte(nil), op.Value...)
+				s.lsns[string(op.Key)] = lsn
+				s.writes[string(op.Key)]++
+			} else {
+				delete(s.data, string(op.Key))
+				delete(s.lsns, string(op.Key))
+			}
+		}
+		s.mu.Unlock()
+		body := make([]byte, 10)
+		putU16(body[0:2], statusOK)
+		putU64(body[2:10], lsn)
+		return body
+	case opScanOpen:
+		includeValues, limit, prefix, ok := parseScanOpen(payload)
 		if !ok {
-			return "error"
-		}
-		return value
-	case strings.HasPrefix(cmd, "delete "):
-		key := strings.TrimPrefix(cmd, "delete ")
-		delete(s.data, key)
-		return "success"
-	case strings.HasPrefix(cmd, "reads "):
-		prefix := strings.TrimPrefix(cmd, "reads ")
-		values := make([]string, 0)
-		for key, value := range s.data {
-			if strings.HasPrefix(key, prefix) {
-				values = append(values, value)
+			return errorBody("InvalidPayload")
+		}
+		s.mu.Lock()
+		s.scanOpens++
+		if !includeValues {
+			s.keyOnlyOpens++
+		}
+		s.mu.Unlock()
+		s.mu.Lock()
+		keys := make([]string, 0)
+		for key := range s.data {
+			if strings.HasPrefix(key, string(prefix)) {
+				keys = append(keys, key)
 			}
 		}
-		return strings.Join(values, "\n")
+		s.mu.Unlock()
+		sort.Strings(keys)
+		if s.maxScanPage > 0 && limit > s.maxScanPage {
+			limit = s.maxScanPage
+		}
+		id := *nextScan
+		*nextScan = id + 1
+		scans[id] = &testScan{keys: keys, limit: limit, keysOnly: !includeValues}
+		body := make([]byte, 10)
+		putU16(body[0:2], statusOK)
+		putU64(body[2:10], id)
+		return body
+	case opScanNext:
+		id, ok := parseScanID(payload)
+		if !ok {
+			return errorBody("InvalidPayload")
+		}
+		scan := scans[id]
+		if scan == nil {
+			return errorBody("ScanNotFound")
+		}
+		s.mu.Lock()
+		s.scanNexts++
+		s.mu.Unlock()
+		remaining := len(scan.keys) - scan.offset
+		count := int(scan.limit)
+		if count > remaining {
+			count = remaining
+		}
+		end := scan.offset + count
+		body := make([]byte, 10)
+		putU16(body[0:2], statusOK)
+		if end >= len(scan.keys) {
+			body[2] = 1
+		}
+		putU32(body[6:10], uint32(count))
+		s.mu.Lock()
+		for _, key := range scan.keys[scan.offset:end] {
+			value := s.data[key]
+			if scan.keysOnly {
+				value = nil
+			}
+			lsn := s.lsns[key]
+			entry := make([]byte, 16+len(key)+len(value))
+			putU32(entry[0:4], uint32(len(key)))
+			putU32(entry[4:8], uint32(len(value)))
+			putU64(entry[8:16], lsn)
+			copy(entry[16:], key)
+			copy(entry[16+len(key):], value)
+			body = append(body, entry...)
+		}
+		s.mu.Unlock()
+		scan.offset = end
+		return body
+	case opScanClose:
+		id, ok := parseScanID(payload)
+		if !ok {
+			return errorBody("InvalidPayload")
+		}
+		s.mu.Lock()
+		s.scanCloses++
+		s.mu.Unlock()
+		delete(scans, id)
+		body := make([]byte, 3)
+		putU16(body[0:2], statusOK)
+		body[2] = 1
+		return body
 	default:
-		return "error"
+		return errorBody("UnknownOpcode")
+	}
+}
+
+func parsePut(payload []byte) ([]byte, []byte, bool) {
+	if len(payload) < 8 {
+		return nil, nil, false
+	}
+	keyLen := getU32(payload[0:4])
+	valueLen := getU32(payload[4:8])
+	if keyLen > maxKeySize || valueLen > maxValueSize {
+		return nil, nil, false
+	}
+	if uint64(8)+uint64(keyLen)+uint64(valueLen) != uint64(len(payload)) {
+		return nil, nil, false
+	}
+	return payload[8 : 8+keyLen], payload[8+keyLen:], true
+}
+
+func parseMultiGetKeys(payload []byte) ([][]byte, bool) {
+	if len(payload) < 4 {
+		return nil, false
+	}
+	count := getU32(payload[0:4])
+	if count > maxOperations {
+		return nil, false
+	}
+	keys := make([][]byte, 0, count)
+	pos := 4
+	for i := uint32(0); i < count; i++ {
+		if len(payload)-pos < 4 {
+			return nil, false
+		}
+		length := getU32(payload[pos : pos+4])
+		pos += 4
+		if length > maxKeySize || len(payload)-pos < int(length) {
+			return nil, false
+		}
+		keys = append(keys, payload[pos:pos+int(length)])
+		pos += int(length)
+	}
+	return keys, pos == len(payload)
+}
+
+func parseBatchOps(payload []byte) ([]BatchOp, bool) {
+	if len(payload) < 8 {
+		return nil, false
+	}
+	count := getU32(payload[0:4])
+	metadataLen := getU32(payload[4:8])
+	if count == 0 || count > maxOperations || uint64(metadataLen) > uint64(len(payload)-8) {
+		return nil, false
+	}
+	pos := 8 + int(metadataLen)
+	ops := make([]BatchOp, 0, count)
+	for i := uint32(0); i < count; i++ {
+		if len(payload)-pos < 12 {
+			return nil, false
+		}
+		opcode := payload[pos]
+		keyLen := getU32(payload[pos+4 : pos+8])
+		valueLen := getU32(payload[pos+8 : pos+12])
+		pos += 12
+		if opcode != batchPut && opcode != batchDelete {
+			return nil, false
+		}
+		if keyLen > maxKeySize || valueLen > maxValueSize {
+			return nil, false
+		}
+		if opcode == batchDelete && valueLen != 0 {
+			return nil, false
+		}
+		if len(payload)-pos < int(keyLen)+int(valueLen) {
+			return nil, false
+		}
+		key := payload[pos : pos+int(keyLen)]
+		pos += int(keyLen)
+		value := payload[pos : pos+int(valueLen)]
+		pos += int(valueLen)
+		ops = append(ops, BatchOp{Op: opcode, Key: key, Value: value})
+	}
+	return ops, pos == len(payload)
+}
+
+func parseScanOpen(payload []byte) (bool, uint32, []byte, bool) {
+	if len(payload) < 12 {
+		return false, 0, nil, false
+	}
+	includeValues := payload[0] != 0
+	limit := getU32(payload[4:8])
+	prefixLen := getU32(payload[8:12])
+	if limit == 0 || limit > 4096 || prefixLen > maxKeySize {
+		return false, 0, nil, false
+	}
+	if uint64(12)+uint64(prefixLen) != uint64(len(payload)) {
+		return false, 0, nil, false
+	}
+	return includeValues, limit, payload[12:], true
+}
+
+func parseScanID(payload []byte) (uint64, bool) {
+	if len(payload) < 8 {
+		return 0, false
 	}
+	return getU64(payload[0:8]), true
 }
 
 func TestInsertDoesNotRewriteSchemaForRowIDUpdates(t *testing.T) {

pkg/storage/table_test.go

@@ -0,0 +1,295 @@
+package storage
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+// TestSelectWithLimitStopsBeforeAllPages verifies that a limited scan stops
+// consuming pages as soon as the limit is satisfied instead of reading the
+// whole table.
+func TestSelectWithLimitStopsBeforeAllPages(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+
+	pool := newTestKVPool(kv, 4, 5*time.Second)
+	defer pool.Close()
+
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+
+	if err := schemas.CreateTable(&Schema{
+		Name: "t",
+		Columns: []Column{
+			{Name: "id", Type: "INTEGER", Nullable: false, PrimaryKey: true},
+			{Name: "name", Type: "TEXT", Nullable: true},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	const n = 20
+	for i := int64(1); i <= n; i++ {
+		if err := tables.Insert("t", Row{"id": i, "name": fmt.Sprintf("n%d", i)}); err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+	}
+
+	// Force small pages so early termination is observable.
+	kv.maxScanPage = 2
+
+	_, nextsBefore, _ := kv.scanStats()
+	rows, err := tables.SelectWithLimit("t", nil, 3, 0)
+	if err != nil {
+		t.Fatalf("SelectWithLimit: %v", err)
+	}
+	_, nextsAfter, _ := kv.scanStats()
+
+	if len(rows) != 3 {
+		t.Fatalf("SelectWithLimit returned %d rows, want 3", len(rows))
+	}
+
+	fullPages := (n + 1) / 2 // ceil(n / pageSize)
+	if got := nextsAfter - nextsBefore; got >= fullPages {
+		t.Fatalf("SelectWithLimit consumed %d pages, want < %d (should stop early)", got, fullPages)
+	}
+}
+
+// TestDropTableDeletesDurableRows verifies that a direct DropTable removes all
+// durable row keys, not just the schema entry.
+func TestDropTableDeletesDurableRows(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+
+	pool := newTestKVPool(kv, 4, 5*time.Second)
+	defer pool.Close()
+
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+
+	if err := schemas.CreateTable(&Schema{
+		Name: "t",
+		Columns: []Column{
+			{Name: "id", Type: "INTEGER", Nullable: false, PrimaryKey: true},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	for i := int64(1); i <= 5; i++ {
+		if err := tables.Insert("t", Row{"id": i}); err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+	}
+	if got := kv.countKeys("testdb:_data:t:"); got != 5 {
+		t.Fatalf("expected 5 durable rows before drop, got %d", got)
+	}
+
+	if err := schemas.DropTable("t"); err != nil {
+		t.Fatalf("DropTable: %v", err)
+	}
+
+	if got := kv.countKeys("testdb:_data:t:"); got != 0 {
+		t.Fatalf("expected 0 durable rows after drop, got %d", got)
+	}
+	if kv.hasKey("testdb:_schema:t") {
+		t.Fatalf("schema key still present after drop")
+	}
+	if kv.hasKey("testdb:_sys:rowid:t") {
+		t.Fatalf("rowid counter key still present after drop")
+	}
+}
+
+// TestScanKeysReturnsKeysOnly verifies that the key-only scan constructor
+// returns keys with empty values.
+func TestScanKeysReturnsKeysOnly(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+
+	c := kv.client()
+	defer c.Close()
+
+	if _, err := c.Put([]byte("p:a"), []byte("value-a")); err != nil {
+		t.Fatalf("put a: %v", err)
+	}
+	if _, err := c.Put([]byte("p:b"), []byte("value-b")); err != nil {
+		t.Fatalf("put b: %v", err)
+	}
+
+	scan, err := c.ScanKeys([]byte("p:"))
+	if err != nil {
+		t.Fatalf("ScanKeys: %v", err)
+	}
+	defer scan.Close()
+
+	entries, done, err := scan.Next()
+	if err != nil {
+		t.Fatalf("next: %v", err)
+	}
+	if !done || len(entries) != 2 {
+		t.Fatalf("done=%v len=%d, want done and 2 entries", done, len(entries))
+	}
+	for _, e := range entries {
+		if len(e.Key) == 0 {
+			t.Fatalf("expected non-empty key")
+		}
+		if len(e.Value) != 0 {
+			t.Fatalf("key-only scan returned a value %q for key %q", e.Value, e.Key)
+		}
+	}
+}
+
+// TestCountFastFirstDerivationUsesKeyOnlyScan verifies that the first-time
+// COUNT(*) derivation issues a key-only scan rather than pulling row values.
+func TestCountFastFirstDerivationUsesKeyOnlyScan(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+
+	pool := newTestKVPool(kv, 4, 5*time.Second)
+	defer pool.Close()
+
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+
+	if err := schemas.CreateTable(&Schema{
+		Name: "t",
+		Columns: []Column{
+			{Name: "id", Type: "INTEGER", Nullable: false, PrimaryKey: true},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	for i := int64(1); i <= 5; i++ {
+		if err := tables.Insert("t", Row{"id": i}); err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+	}
+
+	before := kv.keyOnlyOpenCount()
+	got, err := tables.CountFast("t")
+	if err != nil {
+		t.Fatalf("CountFast: %v", err)
+	}
+	after := kv.keyOnlyOpenCount()
+
+	if got != 5 {
+		t.Fatalf("CountFast = %d, want 5", got)
+	}
+	if after-before != 1 {
+		t.Fatalf("expected CountFast first derivation to use one key-only scan, got %d", after-before)
+	}
+}
+
+func TestInsertBulkStringPrimaryKeyCount(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+
+	pool := newTestKVPool(kv, 4, 5*time.Second)
+	defer pool.Close()
+
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+	if err := schemas.CreateTable(&Schema{
+		Name: "labels",
+		Columns: []Column{
+			{Name: "id", Type: "TEXT", Nullable: false, PrimaryKey: true},
+			{Name: "value", Type: "BLOB", Nullable: true},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	n, err := tables.InsertBulk("labels", []Row{
+		{"id": "a", "value": []byte{0, 1, 2}},
+		{"id": "b", "value": []byte{'|', '\r', '\n'}},
+	})
+	if err != nil {
+		t.Fatalf("InsertBulk: %v", err)
+	}
+	if n != 2 {
+		t.Fatalf("InsertBulk count = %d, want 2", n)
+	}
+	rows, err := tables.Select("labels", nil)
+	if err != nil {
+		t.Fatalf("Select: %v", err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("Select returned %d rows, want 2", len(rows))
+	}
+	if n, err := tables.InsertBulk("labels", []Row{{"id": "a", "value": "duplicate"}}); err == nil || n != 0 {
+		t.Fatalf("existing duplicate: n=%d err=%v", n, err)
+	}
+	if n, err := tables.InsertBulk("labels", []Row{{"id": "c"}, {"id": "c"}}); err == nil || n != 0 {
+		t.Fatalf("batch duplicate: n=%d err=%v", n, err)
+	}
+	if _, err := tables.GetByPK("labels", "c"); err == nil {
+		t.Fatal("duplicate batch persisted a row")
+	}
+}
+
+func TestSelectByIndexUsesPointReadsAfterBuild(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	pool := newTestKVPool(kv, 4, 5*time.Second)
+	defer pool.Close()
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+	if err := schemas.CreateTable(&Schema{
+		Name: "items",
+		Columns: []Column{
+			{Name: "id", Type: "INTEGER", PrimaryKey: true},
+			{Name: "kind", Type: "TEXT"},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := schemas.CreateIndex(&Index{Name: "idx_kind", Table: "items", Columns: []IndexColumn{{Name: "kind"}}}); err != nil {
+		t.Fatal(err)
+	}
+	for i := int64(1); i <= 20; i++ {
+		if err := tables.Insert("items", Row{"id": i, "kind": fmt.Sprintf("k%d", i%2)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if rows, err := tables.SelectByIndex("items", "idx_kind", "k1"); err != nil || len(rows) != 10 {
+		t.Fatalf("initial indexed select: len=%d err=%v", len(rows), err)
+	}
+	opensBefore, _, _ := kv.scanStats()
+	if rows, err := tables.SelectByIndex("items", "idx_kind", "k1"); err != nil || len(rows) != 10 {
+		t.Fatalf("cached indexed select: len=%d err=%v", len(rows), err)
+	}
+	opensAfter, _, _ := kv.scanStats()
+	if opensAfter != opensBefore {
+		t.Fatalf("indexed select opened %d table scans after index build", opensAfter-opensBefore)
+	}
+}
+
+func TestCountFastResetsAfterDirectDropAndRecreate(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")
+	create := func() {
+		if err := schemas.CreateTable(&Schema{Name: "events", Columns: []Column{{Name: "id", Type: "INTEGER", PrimaryKey: true}}}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	create()
+	if err := tables.Insert("events", Row{"id": int64(1)}); err != nil {
+		t.Fatal(err)
+	}
+	if count, err := tables.CountFast("events"); err != nil || count != 1 {
+		t.Fatalf("initial count=%d err=%v", count, err)
+	}
+	if err := schemas.DropTable("events"); err != nil {
+		t.Fatal(err)
+	}
+	create()
+	if count, err := tables.CountFast("events"); err != nil || count != 0 {
+		t.Fatalf("recreated count=%d err=%v", count, err)
+	}
+}