Merge pull request #3 from database-pizza/fix/rowid-cursor-contract

fix(rowid): type _rowid_ as BIGINT and pin the keyset-cursor contract

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/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)
+	}
+}