Merge remote-tracking branch 'origin/master' into feat/compat-suite

pkg/executor/alter_table_test.go

@@ -0,0 +1,143 @@
+package executor
+
+import (
+	"sync"
+	"testing"
+)
+
+// ADD COLUMN only appends schema metadata. Existing rows are not rewritten, so
+// a declared DEFAULT applies to later INSERTs only.
+func TestAlterTableAddColumnDoesNotBackfillPopulatedTable(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (1), (2)")
+
+	execMust(t, e, "ALTER TABLE t ADD COLUMN note TEXT DEFAULT 'later'")
+
+	res := execMust(t, e, "SELECT id, note FROM t ORDER BY id")
+	if res.RowCount != 2 {
+		t.Fatalf("row count = %d, want 2", res.RowCount)
+	}
+	for _, row := range res.Rows {
+		if row[1] != nil {
+			t.Fatalf("existing row %v note = %v, want NULL (no backfill)", row[0], row[1])
+		}
+	}
+
+	execMust(t, e, "INSERT INTO t (id) VALUES (3)")
+	res = execMust(t, e, "SELECT note FROM t WHERE id = 3")
+	if got := res.Rows[0][0]; got != "later" {
+		t.Fatalf("new row note = %v, want later", got)
+	}
+}
+
+func TestAlterTableAddColumnIfNotExistsIsIdempotent(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+	execMust(t, e, "ALTER TABLE t ADD COLUMN IF NOT EXISTS c TEXT")
+	execMust(t, e, "ALTER TABLE t ADD COLUMN IF NOT EXISTS c TEXT")
+
+	s, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	count := 0
+	for _, col := range s.Columns {
+		if col.Name == "c" {
+			count++
+		}
+	}
+	if count != 1 {
+		t.Fatalf("column c appears %d times, want 1", count)
+	}
+}
+
+// Concurrent IF NOT EXISTS ADD COLUMN on one engine instance must not duplicate
+// metadata; the schema manager serializes the check-and-append.
+func TestAlterTableAddColumnIfNotExistsConcurrent(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if _, err := execSQL(e, "ALTER TABLE t ADD COLUMN IF NOT EXISTS c TEXT"); err != nil {
+				t.Errorf("add column: %v", err)
+			}
+		}()
+	}
+	wg.Wait()
+
+	s, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	count := 0
+	for _, col := range s.Columns {
+		if col.Name == "c" {
+			count++
+		}
+	}
+	if count != 1 {
+		t.Fatalf("column c appears %d times, want 1", count)
+	}
+}
+
+// DROP COLUMN rewrites every remaining row so the removed value does not stay
+// in durable storage.
+func TestAlterTableDropColumnRewritesPopulatedRows(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, keep TEXT, dropme TEXT)")
+	execMust(t, e, "INSERT INTO t (id, keep, dropme) VALUES (1, 'a', 'secret'), (2, 'b', 'secret')")
+
+	execMust(t, e, "ALTER TABLE t DROP COLUMN dropme")
+
+	res := execMust(t, e, "SELECT * FROM t ORDER BY id")
+	if len(res.Columns) != 2 || res.Columns[0] != "id" || res.Columns[1] != "keep" {
+		t.Fatalf("columns after drop = %v, want [id keep]", res.Columns)
+	}
+	if res.RowCount != 2 || res.Rows[0][1] != "a" || res.Rows[1][1] != "b" {
+		t.Fatalf("rows after drop = %v", res.Rows)
+	}
+	if _, err := execSQL(e, "SELECT dropme FROM t"); err == nil {
+		t.Fatal("expected dropped column to be unknown")
+	}
+}
+
+// RENAME COLUMN rewrites row keys but deliberately leaves index metadata
+// unchanged. Documented guidance: drop and recreate indexes after a rename.
+func TestAlterTableRenameColumnLeavesIndexMetadataStale(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT)")
+	execMust(t, e, "INSERT INTO t (id, a) VALUES (1, 'x')")
+	execMust(t, e, "CREATE INDEX idx_a ON t(a)")
+
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN a TO aa")
+
+	res := execMust(t, e, "SELECT aa FROM t WHERE id = 1")
+	if got := res.Rows[0][0]; got != "x" {
+		t.Fatalf("renamed value = %v, want x", got)
+	}
+
+	names, err := schema.ListIndexes()
+	if err != nil {
+		t.Fatalf("list indexes: %v", err)
+	}
+	found := false
+	for _, name := range names {
+		if name == "idx_a" {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("indexes = %v, want idx_a to still exist (rename does not rewrite index metadata)", names)
+	}
+}

pkg/executor/result_types.go

@@ -50,6 +50,11 @@ func projectionColumnType(col parser.SelectColumn, schema *storage.Schema) strin
 			return c.Type
 		}
 	}
+	if storage.IsRowIDColumn(ref.Column) {
+		// The hidden rowid alias is a signed 64-bit integer, not an unknown
+		// expression, so report it as BIGINT (int8) on the wire.
+		return "BIGINT"
+	}
 	return "TEXT"
 }
 
@@ -103,6 +108,9 @@ func (e *Executor) joinedSelectColumnTypes(stmt *parser.SelectStmt, refs []parse
 				break
 			}
 		}
+		if !found && storage.IsRowIDColumn(ref.Column) {
+			columnType = "BIGINT"
+		}
 		types = append(types, columnType)
 	}
 	return types

pkg/executor/rowid_cursor_test.go

@@ -0,0 +1,200 @@
+package executor
+
+import (
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+func rowIDs(t *testing.T, e *Executor, sql string) []int64 {
+	t.Helper()
+	res := execMust(t, e, sql)
+	ids := make([]int64, 0, res.RowCount)
+	for _, row := range res.Rows {
+		id, ok := row[0].(int64)
+		if !ok {
+			t.Fatalf("rowid %v has type %T, want int64", row[0], row[0])
+		}
+		ids = append(ids, id)
+	}
+	return ids
+}
+
+func TestRowIDSelectableStableAcrossUpdateAndTypedBigint(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE items (id TEXT PRIMARY KEY, name TEXT)")
+	execMust(t, e, "INSERT INTO items (id, name) VALUES ('a', 'a'), ('b', 'b'), ('c', 'c')")
+
+	res := execMust(t, e, "SELECT _rowid_, rowid, oid FROM items ORDER BY _rowid_")
+	if res.RowCount != 3 {
+		t.Fatalf("row count = %d, want 3", res.RowCount)
+	}
+	if got := res.GetColumnType(0); got != "BIGINT" {
+		t.Fatalf("_rowid_ type = %q, want BIGINT", got)
+	}
+	if got := res.Rows[0][0]; got != int64(1) {
+		t.Fatalf("first _rowid_ = %v, want 1", got)
+	}
+	for i := 1; i < res.RowCount; i++ {
+		if res.Rows[i][0].(int64) <= res.Rows[i-1][0].(int64) {
+			t.Fatalf("_rowid_ is not monotonically increasing: %v", res.Rows)
+		}
+		if res.Rows[i][0] != res.Rows[i][1] || res.Rows[i][0] != res.Rows[i][2] {
+			t.Fatalf("rowid/oid aliases disagree: %v", res.Rows[i])
+		}
+	}
+
+	execMust(t, e, "UPDATE items SET name = 'updated' WHERE id = 'b'")
+	execMust(t, e, "UPDATE items SET id = 'bb' WHERE id = 'b'")
+	ids := rowIDs(t, e, "SELECT _rowid_ FROM items WHERE id = 'bb'")
+	if len(ids) != 1 || ids[0] != 2 {
+		t.Fatalf("rowid after UPDATE = %v, want [2]", ids)
+	}
+}
+
+func TestRowIDExplicitColumnTakesPrecedence(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, rowid TEXT)")
+	execMust(t, e, "INSERT INTO t (id, rowid) VALUES (1, 'explicit')")
+
+	res := execMust(t, e, "SELECT rowid, _rowid_ FROM t")
+	if got := res.Rows[0][0]; got != "explicit" {
+		t.Fatalf("explicit rowid column = %v, want explicit", got)
+	}
+	if got := res.Rows[0][1]; got != int64(1) {
+		t.Fatalf("hidden _rowid_ = %v, want 1", got)
+	}
+}
+
+func TestRowIDAllocationSurvivesManagerRestart(t *testing.T) {
+	pool, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
+	execMust(t, e, "INSERT INTO t (id, v) VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+
+	// A new manager pair reads the durable schema and rowid allocator state.
+	reopenedSchema := storage.NewSchemaManager(pool, "testdb")
+	reopenedTable := storage.NewTableManager(pool, reopenedSchema, "testdb")
+	reopened := newExec(reopenedSchema, reopenedTable)
+	execMust(t, reopened, "INSERT INTO t (id, v) VALUES (4, 'd')")
+
+	ids := rowIDs(t, reopened, "SELECT _rowid_ FROM t ORDER BY _rowid_")
+	want := []int64{1, 2, 3, 4}
+	if len(ids) != len(want) {
+		t.Fatalf("rowids = %v, want %v", ids, want)
+	}
+	for i := range want {
+		if ids[i] != want[i] {
+			t.Fatalf("rowids = %v, want %v", ids, want)
+		}
+	}
+}
+
+func TestRowIDKeysetPaginationDoesNotSkipOrRepeat(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE events (id INTEGER PRIMARY KEY, payload TEXT)")
+	for i := 1; i <= 5; i++ {
+		execMust(t, e, "INSERT INTO events (id, payload) VALUES ("+strconv.Itoa(i)+", 'x')")
+	}
+
+	// Paginate with literals so the test exercises the executor filter and order
+	// path directly; insert new rows between pages to prove the cursor stays
+	// monotonic.
+	var seen []int64
+	cursor := int64(0)
+	for {
+		res := execMust(t, e, "SELECT _rowid_ FROM events WHERE _rowid_ > "+strconv.FormatInt(cursor, 10)+" ORDER BY _rowid_ LIMIT 3")
+		if res.RowCount == 0 {
+			break
+		}
+		for _, row := range res.Rows {
+			id := row[0].(int64)
+			if len(seen) > 0 && id <= seen[len(seen)-1] {
+				t.Fatalf("rowid repeated or out of order: seen=%v next=%d", seen, id)
+			}
+			seen = append(seen, id)
+		}
+		cursor = seen[len(seen)-1]
+		if cursor == 3 {
+			// New rows arriving after the first page must appear later.
+			execMust(t, e, "INSERT INTO events (id, payload) VALUES (100, 'late')")
+		}
+	}
+	if len(seen) != 6 {
+		t.Fatalf("paginated %v, want 6 rows including the late insert", seen)
+	}
+}
+
+func TestRowIDUniqueAndMonotonicUnderConcurrentInsert(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
+
+	const workers, perWorker = 4, 20
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	var ids []int64
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func(worker int) {
+			defer wg.Done()
+			for i := 0; i < perWorker; i++ {
+				id := strconv.Itoa(worker*perWorker + i + 1)
+				if _, err := execSQL(e, "INSERT INTO t (id, v) VALUES ("+id+", 'x')"); err != nil {
+					t.Errorf("insert %s: %v", id, err)
+					return
+				}
+				res, err := execSQL(e, "SELECT _rowid_ FROM t WHERE id = "+id)
+				if err != nil {
+					t.Errorf("select %s: %v", id, err)
+					return
+				}
+				mu.Lock()
+				ids = append(ids, res.Rows[0][0].(int64))
+				mu.Unlock()
+			}
+		}(w)
+	}
+	wg.Wait()
+
+	seen := make(map[int64]bool, len(ids))
+	for _, id := range ids {
+		if seen[id] {
+			t.Fatalf("duplicate rowid %d", id)
+		}
+		seen[id] = true
+	}
+	if len(seen) != workers*perWorker {
+		t.Fatalf("got %d unique rowids, want %d", len(seen), workers*perWorker)
+	}
+}
+
+// INTEGER PRIMARY KEY is SQLite's rowid alias: a row inserted with id=10 gets
+// _rowid_=10. UPDATE never rewrites the stored _rowid_, so a keyset cursor is
+// stable across value and primary-key updates even though the rowid is not
+// independent of the key. Clients must not insert descending integer keys if
+// they rely on rowid monotonicity.
+func TestRowIDIntegerPrimaryKeyAliasIsStableAcrossUpdate(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
+	execMust(t, e, "INSERT INTO t (id, name) VALUES (10, 'a'), (20, 'b'), (30, 'c')")
+
+	res := execMust(t, e, "SELECT id, _rowid_ FROM t ORDER BY id")
+	for _, row := range res.Rows {
+		if row[0] != row[1] {
+			t.Fatalf("integer primary key %v does not alias _rowid_ %v", row[0], row[1])
+		}
+	}
+
+	execMust(t, e, "UPDATE t SET id = 200, name = 'updated' WHERE id = 20")
+	res = execMust(t, e, "SELECT id, _rowid_ FROM t WHERE id = 200")
+	if got := res.Rows[0][1]; got != int64(20) {
+		t.Fatalf("rowid after primary-key update = %v, want the original 20", got)
+	}
+}

pkg/pgserver/alter_table_test.go

@@ -0,0 +1,46 @@
+package pgserver
+
+import (
+	"testing"
+)
+
+func TestAlterTableAddColumnOverSimpleQuery(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE t (id INTEGER PRIMARY KEY)"), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t (id) VALUES (1)"), 0)})
+
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("ALTER TABLE t ADD COLUMN IF NOT EXISTS note TEXT DEFAULT 'later'"), 0)})
+	// Idempotent second boot-time migration.
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("ALTER TABLE t ADD COLUMN IF NOT EXISTS note TEXT DEFAULT 'later'"), 0)})
+
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("SELECT id, note FROM t ORDER BY id"), 0)})
+	values := dataRowValues(t, msgs[1])
+	if len(values) != 2 || values[0] != "1" || values[1] != "<null>" {
+		t.Fatalf("existing row after ADD COLUMN = %v, want [1 <null>]", values)
+	}
+
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t (id) VALUES (2)"), 0)})
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("SELECT note FROM t WHERE id = 2"), 0)})
+	if got := dataRowValues(t, msgs[1]); len(got) != 1 || got[0] != "later" {
+		t.Fatalf("new row note = %v, want [later]", got)
+	}
+}
+
+func TestAlterTableDropAndRenameColumnOverSimpleQuery(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE t (id INTEGER PRIMARY KEY, keep TEXT, dropme TEXT)"), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t (id, keep, dropme) VALUES (1, 'a', 'gone')"), 0)})
+
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("ALTER TABLE t DROP COLUMN dropme"), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("ALTER TABLE t RENAME COLUMN keep TO kept"), 0)})
+
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("SELECT kept FROM t WHERE id = 1"), 0)})
+	if got := dataRowValues(t, msgs[1]); len(got) != 1 || got[0] != "a" {
+		t.Fatalf("renamed column value = %v, want [a]", got)
+	}
+
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("SELECT dropme FROM t"), 0)})
+	if code := errorCode(msgs[0]); code == "" {
+		t.Fatalf("expected an error selecting a dropped column, got %v", msgs)
+	}
+}

pkg/pgserver/json_contract_test.go

@@ -0,0 +1,91 @@
+package pgserver
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+)
+
+// jsonErrorField extracts a named field from an ErrorResponse.
+func jsonErrorField(t *testing.T, msg *Message, want byte) string {
+	t.Helper()
+	if msg == nil || msg.Type != MsgErrorResponse {
+		t.Fatalf("message type = %v, want ErrorResponse", msg)
+	}
+	data := msg.Data
+	for i := 0; i < len(data); {
+		if data[i] == 0 {
+			break
+		}
+		field := data[i]
+		i++
+		end := bytes.IndexByte(data[i:], 0)
+		if end < 0 {
+			break
+		}
+		value := string(data[i : i+end])
+		if field == want {
+			return value
+		}
+		i += end + 1
+	}
+	return ""
+}
+
+// The supported JSON contract is TEXT storage plus the JSON1 function surface.
+// JSON/JSONB type names are accepted as text-affinity aliases; there is no
+// native binary jsonb type or ->> operator.
+func TestJSONTextContractOverSimpleQuery(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE settings (id INTEGER PRIMARY KEY, body TEXT)"), 0)})
+
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`INSERT INTO settings (id, body) VALUES (1, '{"theme":"dark","widgets":[1,2]}')`), 0)})
+
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT json_extract(body, '$.theme'), json_type(body), json_valid(body) FROM settings WHERE id = 1`), 0)})
+	values := dataRowValues(t, msgs[1])
+	if len(values) != 3 || values[0] != "dark" || values[1] != "object" || values[2] != "1" {
+		t.Fatalf("json round trip = %v, want [dark object 1]", values)
+	}
+
+	// Partial update through json_set, without rewriting the whole document.
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`UPDATE settings SET body = json_set(body, '$.theme', 'light') WHERE id = 1`), 0)})
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT json_extract(body, '$.theme'), json_extract(body, '$.widgets[1]') FROM settings WHERE id = 1`), 0)})
+	values = dataRowValues(t, msgs[1])
+	if len(values) != 2 || values[0] != "light" || values[1] != "2" {
+		t.Fatalf("json partial update = %v, want [light 2]", values)
+	}
+
+	// JSON and JSONB column type names are text-affinity aliases.
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE docs (id INTEGER PRIMARY KEY, body JSON)"), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`INSERT INTO docs (id, body) VALUES (1, '{"a":1}')`), 0)})
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT json_extract(body, '$.a') FROM docs`), 0)})
+	if got := dataRowValues(t, msgs[1]); len(got) != 1 || got[0] != "1" {
+		t.Fatalf("JSON column extract = %v, want [1]", got)
+	}
+}
+
+func TestJSONInvalidInputOverSimpleQuery(t *testing.T) {
+	c, client := newDataConnection(t)
+
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT json_extract('not json', '$.a')`), 0)})
+	if len(msgs) != 2 || msgs[0].Type != MsgErrorResponse {
+		t.Fatalf("expected ErrorResponse+ReadyForQuery, got %v", msgs)
+	}
+	if msg := jsonErrorField(t, msgs[0], ErrorFieldMessage); !strings.Contains(msg, "malformed JSON") {
+		t.Fatalf("error message = %q, want a malformed-JSON error", msg)
+	}
+
+	// json_valid is the non-throwing way to probe a value.
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT json_valid('not json')`), 0)})
+	if got := dataRowValues(t, msgs[1]); len(got) != 1 || got[0] != "0" {
+		t.Fatalf("json_valid = %v, want [0]", got)
+	}
+}
+
+func TestJSONArrowOperatorUnsupported(t *testing.T) {
+	c, client := newDataConnection(t)
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(`SELECT '{"a":1}' -> '$.a'`), 0)})
+	if len(msgs) == 0 || msgs[0].Type != MsgErrorResponse {
+		t.Fatalf("expected a parse error for the -> operator, got %v", msgs)
+	}
+}

pkg/pgserver/rowid_test.go

@@ -0,0 +1,83 @@
+package pgserver
+
+import (
+	"testing"
+)
+
+// RowDescription for the hidden rowid alias must report an integer OID, not
+// TEXT, because the value is a signed 64-bit integer.
+func TestRowIDWireTypeIsBigint(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE t (id TEXT PRIMARY KEY)"), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t (id) VALUES ('a')"), 0)})
+
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("SELECT _rowid_ FROM t"), 0)})
+	if msgs[0].Type != MsgRowDescription {
+		t.Fatalf("first message = %c, want RowDescription", msgs[0].Type)
+	}
+	oids := rowDescriptionOIDs(t, msgs[0].Data)
+	if len(oids) != 1 || oids[0] != 20 { // INT8OID
+		t.Fatalf("rowid OIDs = %v, want [20]", oids)
+	}
+	if got := dataRowValues(t, msgs[1]); len(got) != 1 || got[0] != "1" {
+		t.Fatalf("rowid value = %v, want [1]", got)
+	}
+}
+
+// A keyset cursor built from _rowid_ must paginate without repeating or
+// skipping rows, including rows inserted after the first page is read.
+func TestRowIDKeysetPaginationOverSimpleQuery(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE events (id TEXT PRIMARY KEY, payload TEXT)"), 0)})
+	for _, id := range []string{"a", "b", "c", "d", "e"} {
+		runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO events (id, payload) VALUES ('"+id+"', 'x')"), 0)})
+	}
+
+	page := func(cursor string) []string {
+		msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("SELECT _rowid_ FROM events WHERE _rowid_ > "+cursor+" ORDER BY _rowid_ LIMIT 2"), 0)})
+		if len(msgs) < 2 || msgs[0].Type != MsgRowDescription {
+			t.Fatalf("unexpected page response: %v", msgs)
+		}
+		var ids []string
+		for _, msg := range msgs[1:] {
+			if msg.Type != MsgDataRow {
+				break
+			}
+			ids = append(ids, dataRowValues(t, msg)[0])
+		}
+		return ids
+	}
+
+	first := page("0")
+	if len(first) != 2 || first[0] != "1" || first[1] != "2" {
+		t.Fatalf("first page = %v, want [1 2]", first)
+	}
+	// A late insert must not move the cursor backwards or duplicate page 1.
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO events (id, payload) VALUES ('late', 'x')"), 0)})
+
+	second := page(first[1])
+	if len(second) != 2 || second[0] != "3" || second[1] != "4" {
+		t.Fatalf("second page = %v, want [3 4]", second)
+	}
+	third := page(second[1])
+	if len(third) != 2 || third[0] != "5" || third[1] != "6" {
+		t.Fatalf("third page = %v, want [5 6] (late row appended)", third)
+	}
+	if fourth := page(third[1]); len(fourth) != 0 {
+		t.Fatalf("final page = %v, want empty", fourth)
+	}
+}
+
+// INSERT ... RETURNING _rowid_ must expose the assigned cursor value.
+func TestRowIDReturningOverSimpleQuery(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE t (id TEXT PRIMARY KEY)"), 0)})
+
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t (id) VALUES ('a') RETURNING _rowid_"), 0)})
+	if len(msgs) < 3 {
+		t.Fatalf("expected RowDescription+DataRow+CommandComplete, got %v", msgs)
+	}
+	if got := dataRowValues(t, msgs[1]); len(got) != 1 || got[0] != "1" {
+		t.Fatalf("returned rowid = %v, want [1]", got)
+	}
+}