fix transactional DDL review findings (rename orphans, savepoint aliasing, rowid, batch ceiling)

P0: RENAME followed by UPDATE/DELETE left orphaned rows. Two defects combined:

- Session.getByPKLocked (the point read behind UPDATE and DELETE) read the
  staged table name's data prefix, so after ALTER TABLE RENAME the committed
  row was not found and the statement silently affected 0 rows. It now reads
  the durable source prefix (durableDataKeyLocked) and keeps the staged key
  for the overlay lookup.
- schema_tx.commitDataOps skipped a durable row whose post-rename key was
  already held by the overlay without emitting the delete of its committed
  key, leaving an unreachable row under the dropped name; a later CREATE
  TABLE with that name resurrected it. The rename move is now built by
  renameRowOps: an overlay-owned row deletes only its stale committed key,
  every other row is put under the destination prefix and deleted from the
  source prefix.

P1: repeated ROLLBACK TO SAVEPOINT re-installed post-rollback state because
schemaTx.restore and rollbackToSavepointLocked assigned the retained
savepoint's maps by reference. Both now copy on restore (fresh maps/slices;
staged definitions are immutable so values are shared).

P1: a committed transactional DDL reset every table's ROWID allocator
(invalidateCommittedDDL). A concurrent transaction that had staged a
TEXT/composite-key insert then had its ROWID derived again, publishing two
durable rows with the same _rowid_. invalidateCommittedDDL now drops the
allocator only for the touched tables.

P1: commitDataOps buffered every row op and only failed inside
CompareBatchWrite past the store's per-batch ceiling. A ddlBatchBudget aborts
the scan as soon as the operation/byte ceiling is crossed with an explicit
limit error, CommitWithLSN pre-flights the assembled batch, and the DDL
capability note documents the ceiling (65535 operations / 64 MiB).

P2:
- ADD COLUMN pins the table generation before the durable-row validation scan
  so a concurrent committed row cannot slip into the recorded window.
- RENAME away and back cancels the pending move (no put+delete for the same
  key); commitCatalogOps and commitDataOps also skip no-op rename pairs.
- A DDL commit transport error or compare failure now invalidates the shared
  catalog caches for the touched tables, so a lost response cannot leave a
  permanently stale cache.
- pgserver's information_schema/pg_tables/pg_indexes/table_constraints
  emulation reads through the connection's session, so introspection sees
  staged DDL inside a transaction and keeps seeing the committed catalog
  elsewhere.
- refreshCommitted reloads the table list as well as the index list.

Tests: RENAME+UPDATE and RENAME+DELETE orphan regressions, repeated ROLLBACK
TO SAVEPOINT (DDL and rows), rename away and back (executor behavior plus a
storage-level no-duplicate-op/pending-move property), the T1/T2/T3 ROWID
interleaving, an oversized RENAME failing at COMMIT with the limit error and
publishing nothing, the ddlBatchBudget unit boundary, and pgserver catalog
visibility across two connections and ROLLBACK.

pkg/capabilities/capabilities.go

@@ -176,7 +176,7 @@ func Current() *Document {
 		},
 
 		DDL: DDL{
-			Transactional:          Feature{Supported: true, Status: "staged_until_commit", Notes: "inside an explicit transaction CREATE TABLE, DROP TABLE, ALTER TABLE (add/drop/rename column, rename table), and CREATE/DROP INDEX are staged in a session-scoped catalog overlay: the issuing session reads its own DDL, other sessions keep reading the committed catalog, ROLLBACK discards the overlay and the transaction's rows together, and COMMIT publishes both in one atomic compare-batch write; validations use the staged catalog, and a staged catalog write that conflicts with a concurrently committed DDL fails with 40001; a failed DDL statement leaves the transaction's earlier schema intact; autocommit DDL keeps the same immediate behavior"},
+			Transactional:          Feature{Supported: true, Status: "staged_until_commit", Notes: "inside an explicit transaction CREATE TABLE, DROP TABLE, ALTER TABLE (add/drop/rename column, rename table), and CREATE/DROP INDEX are staged in a session-scoped catalog overlay: the issuing session reads its own DDL, other sessions keep reading the committed catalog, ROLLBACK discards the overlay and the transaction's rows together, and COMMIT publishes both in one atomic compare-batch write; validations use the staged catalog, and a staged catalog write that conflicts with a concurrently committed DDL fails with 40001; a failed DDL statement leaves the transaction's earlier schema intact; autocommit DDL keeps the same immediate behavior; because a transaction publishes one compare-batch, a DDL row rewrite (ALTER TABLE backfill, DROP TABLE, RENAME TABLE) is bounded by the storage batch ceiling (65535 operations or 64 MiB) and an oversized migration fails at COMMIT with an explicit limit error instead of being split, while the same statement outside a transaction pages its writes"},
 			AddColumn:              Feature{Supported: true, Status: "materializes_defaults", Notes: "adds one nullable column; a constant DEFAULT is written into every existing row, and a whitelisted clock default is evaluated once per ALTER statement, so the rewrite cost is O(rows) and holds the table's exclusive gate"},
 			AddColumnBackfill:      Feature{Supported: true, Status: "rewrites_rows", Notes: "a DEFAULT is materialized into existing rows during the rewrite: constants as their stored value, whitelisted clock defaults evaluated at one instant per ALTER statement; ADD COLUMN ... NOT NULL without a DEFAULT fails with 23502 on a non-empty table, ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row (a constant default is predicted before the rewrite, a clock default is validated against the values the rewrite writes), ADD COLUMN ... CHECK fails with 23514 when an existing row or the backfilled default violates the predicate, and the column is published together with its CHECK constraints and UNIQUE index in one atomic schema write"},
 			DropColumn:             Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint references the column or a FOREIGN KEY depends on it (child column or referenced parent column); the stored constraint text cannot be rewritten"},

pkg/executor/executor.go

@@ -163,6 +163,13 @@ func (e *Executor) NewSessionExecutor() *Executor {
 	return exec
 }
 
+// Session exposes the executor's catalog and row session. Protocol layers that
+// emulate system catalogs must read through it so their introspection observes
+// DDL staged by an open transaction, exactly like the SQL executor does.
+func (e *Executor) Session() *storage.Session {
+	return e.session
+}
+
 // SyncCatalog synchronizes the analyzer catalog with the storage schema.
 func (e *Executor) SyncCatalog() error {
 	tables, err := e.session.ListTables()

pkg/executor/transactional_ddl_test.go

@@ -657,3 +657,197 @@ func TestTransactionalDDLIndexCachesSurviveRollback(t *testing.T) {
 		t.Fatalf("index unusable after committing a re-created index: %v", err)
 	}
 }
+
+// Transactional DDL: a RENAME followed by an UPDATE must still delete the
+// committed row under the old prefix. The updated image lives in the session
+// overlay under the new prefix, so the commit has to move-or-delete every
+// durable old key, not only the rows whose image it rewrites.
+func TestTransactionalDDLRenameThenUpdateKeepsNoOrphanRows(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE a (id INTEGER PRIMARY KEY, v TEXT)")
+	execMust(t, e, "INSERT INTO a VALUES (1, 'one'), (2, 'two')")
+
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "ALTER TABLE a RENAME TO b")
+	execMust(t, e, "UPDATE b SET v = 'changed' WHERE id = 1")
+	execMust(t, e, "COMMIT")
+
+	if schema.TableExists("a") {
+		t.Fatal("renamed table is still addressable under its old name")
+	}
+	if res := execMust(t, e, "SELECT id, v FROM b ORDER BY id"); res.RowCount != 2 || res.Rows[0][1] != "changed" || res.Rows[1][1] != "two" {
+		t.Fatalf("rows after committed rename+update = %v", res.Rows)
+	}
+	// The decisive check: a table re-created with the old name must start empty.
+	// Any durable row left under _data:a: would resurface here as a ghost.
+	execMust(t, e, "CREATE TABLE a (id INTEGER PRIMARY KEY, v TEXT)")
+	if res := execMust(t, e, "SELECT COUNT(*) FROM a"); res.Rows[0][0] != int64(0) {
+		t.Fatalf("orphaned rows resurfaced under the re-created old name: %v", res.Rows)
+	}
+	if rows, err := table.Select("b", nil); err != nil || len(rows) != 2 {
+		t.Fatalf("durable rows after rename+update = %v err=%v", rows, err)
+	}
+}
+
+// Transactional DDL: the same guarantee for a staged DELETE after a RENAME.
+func TestTransactionalDDLRenameThenDeleteKeepsNoOrphanRows(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE a (id INTEGER PRIMARY KEY, v TEXT)")
+	execMust(t, e, "INSERT INTO a VALUES (1, 'one'), (2, 'two')")
+
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "ALTER TABLE a RENAME TO b")
+	execMust(t, e, "DELETE FROM b WHERE id = 2")
+	execMust(t, e, "COMMIT")
+
+	if res := execMust(t, e, "SELECT id FROM b"); res.RowCount != 1 || res.Rows[0][0] != int64(1) {
+		t.Fatalf("rows after committed rename+delete = %v", res.Rows)
+	}
+	execMust(t, e, "CREATE TABLE a (id INTEGER PRIMARY KEY, v TEXT)")
+	if res := execMust(t, e, "SELECT COUNT(*) FROM a"); res.Rows[0][0] != int64(0) {
+		t.Fatalf("deleted row or its sibling resurfaced under the re-created old name: %v", res.Rows)
+	}
+	if rows, err := table.Select("b", nil); err != nil || len(rows) != 1 {
+		t.Fatalf("durable rows after rename+delete = %v err=%v", rows, err)
+	}
+}
+
+// Transactional DDL: renaming a table away and back inside one transaction must
+// not publish a put+delete for the same key; all rows and the schema survive.
+func TestTransactionalDDLRenameAwayAndBack(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE a (id INTEGER PRIMARY KEY, v TEXT)")
+	execMust(t, e, "INSERT INTO a VALUES (1, 'one'), (2, 'two')")
+
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "ALTER TABLE a RENAME TO b")
+	execMust(t, e, "ALTER TABLE b RENAME TO a")
+	execMust(t, e, "INSERT INTO a VALUES (3, 'three')")
+	execMust(t, e, "COMMIT")
+
+	if schema.TableExists("b") {
+		t.Fatal("intermediate rename name survived the commit")
+	}
+	if res := execMust(t, e, "SELECT id, v FROM a ORDER BY id"); res.RowCount != 3 {
+		t.Fatalf("rows after rename away and back = %v", res.Rows)
+	}
+	if rows, err := table.Select("a", nil); err != nil || len(rows) != 3 {
+		t.Fatalf("durable rows after rename away and back = %v err=%v", rows, err)
+	}
+}
+
+// Transactional DDL: ROLLBACK TO SAVEPOINT can be repeated. The savepoint is
+// retained, so DDL and rows staged after the first rollback must not leak into
+// the state a second rollback installs.
+func TestTransactionalDDLRepeatedRollbackToSavepoint(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "SAVEPOINT s")
+	execMust(t, e, "CREATE TABLE x (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t VALUES (1)")
+	execMust(t, e, "ROLLBACK TO SAVEPOINT s")
+
+	execMust(t, e, "CREATE TABLE y (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t VALUES (2)")
+	execMust(t, e, "ROLLBACK TO SAVEPOINT s")
+	execMust(t, e, "COMMIT")
+
+	if schema.TableExists("x") || schema.TableExists("y") {
+		t.Fatalf("DDL staged between rollbacks survived: x=%v y=%v", schema.TableExists("x"), schema.TableExists("y"))
+	}
+	if res := execMust(t, e, "SELECT COUNT(*) FROM t"); res.Rows[0][0] != int64(0) {
+		t.Fatalf("rows staged between rollbacks survived: %v", res.Rows)
+	}
+}
+
+// Transactional DDL: committing DDL on one table must not reset another table's
+// ROWID allocator. A concurrent transaction that staged a TEXT-keyed insert
+// holds a ROWID that is not durable yet; if the allocator is wiped, a later
+// insert derives that same value again and both rows become durable with the
+// same _rowid_, mis-associating every rowid-keyed index cache entry.
+func TestTransactionalDDLCommitKeepsUnrelatedRowIDAllocators(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	stager := newExec(schema, table)
+	ddl := newExec(schema, table)
+	other := newExec(schema, table)
+
+	execMust(t, stager, "CREATE TABLE events (name TEXT PRIMARY KEY, v TEXT)")
+
+	execMust(t, stager, "BEGIN")
+	execMust(t, stager, "INSERT INTO events VALUES ('staged', 'a')")
+
+	// An unrelated transactional DDL commit must not touch the events allocator.
+	execMust(t, ddl, "BEGIN")
+	execMust(t, ddl, "CREATE TABLE side (id INTEGER PRIMARY KEY)")
+	execMust(t, ddl, "COMMIT")
+
+	// Another session allocates its own ROWID while the first insert is staged.
+	execMust(t, other, "INSERT INTO events VALUES ('later', 'b')")
+	execMust(t, stager, "COMMIT")
+
+	rows, err := table.Select("events", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("events rows = %d, want 2", len(rows))
+	}
+	seen := make(map[int64]bool, len(rows))
+	for _, row := range rows {
+		rowid, ok := row["_rowid_"].(int64)
+		if !ok {
+			t.Fatalf("row without a rowid: %v", row)
+		}
+		if seen[rowid] {
+			t.Fatalf("duplicate durable _rowid_ %d after an unrelated DDL commit: %v", rowid, rows)
+		}
+		seen[rowid] = true
+	}
+}
+
+// Transactional DDL: a RENAME whose row move cannot fit in the commit's single
+// storage transaction fails at COMMIT with an explicit limit error instead of
+// buffering the whole table and failing inside the storage protocol, and the
+// failed commit publishes nothing.
+func TestTransactionalDDLRenameExceedingBatchCeilingFails(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE big (id INTEGER PRIMARY KEY, v TEXT)")
+
+	// A rename publishes a put and a delete per row, so just over half of the
+	// store's 65,535-operation ceiling is enough to exceed it.
+	const rows = 33000
+	const chunk = 5000
+	for start := 0; start < rows; start += chunk {
+		batch := make([]storage.Row, 0, chunk)
+		for i := start; i < start+chunk && i < rows; i++ {
+			batch = append(batch, storage.Row{"id": int64(i + 1), "v": "x"})
+		}
+		if _, err := table.InsertBulk("big", batch); err != nil {
+			t.Fatalf("seed rows %d-%d: %v", start, start+len(batch), err)
+		}
+	}
+
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "ALTER TABLE big RENAME TO big_renamed")
+	_, err := execSQL(e, "COMMIT")
+	if err == nil {
+		t.Fatal("commit of an oversized rename succeeded")
+	}
+	if !strings.Contains(err.Error(), "storage batch limit") {
+		t.Fatalf("commit error = %v, want the storage batch limit", err)
+	}
+	// Nothing was published: the old name, its schema, and every row survive.
+	if !schema.TableExists("big") || schema.TableExists("big_renamed") {
+		t.Fatalf("failed commit published the rename: big=%v big_renamed=%v", schema.TableExists("big"), schema.TableExists("big_renamed"))
+	}
+	if count, err := table.CountFast("big"); err != nil || count != rows {
+		t.Fatalf("rows after failed commit = %d err=%v, want %d", count, err, rows)
+	}
+}

pkg/pgserver/connection.go

@@ -32,6 +32,7 @@ type Connection struct {
 	writer         *bufio.Writer
 	executor       *executor.Executor
 	schema         *storage.SchemaManager
+	table          *storage.TableManager
 	dbManager      *storage.DatabaseManager
 	database       string
 	params         map[string]string
@@ -45,6 +46,9 @@ type Connection struct {
 	// next Sync. Notifications are not interleaved into an extended-protocol
 	// sequence; they are held until Sync.
 	extendedCycle bool
+	// catalogFallback serves catalog emulation before the executor exists, when
+	// a connection has managers but has not completed startup.
+	catalogFallback *storage.Session
 }
 
 // backendPIDSeq gives each connection a distinct synthetic backend PID. It is
@@ -276,6 +280,7 @@ func (c *Connection) initDatabase(dbName string) error {
 	}
 
 	c.schema = db.Schema
+	c.table = db.Table
 	c.executor = executor.New(db.Schema, db.Table)
 	c.executor.SetBackendPID(c.backendPID)
 	c.executor.SetNotificationHub(db.Hub)
@@ -737,6 +742,24 @@ func (c *Connection) handleExecute(msg *Message) error {
 	return nil
 }
 
+// catalogSession is the catalog view this connection serves. The executor's
+// session is authoritative because it resolves against the transaction-scoped
+// DDL overlay, so catalog emulation inside a transaction sees staged tables,
+// columns, and indexes. The fallback only covers a connection that has not
+// completed startup yet.
+func (c *Connection) catalogSession() *storage.Session {
+	if c.executor != nil {
+		return c.executor.Session()
+	}
+	if c.schema == nil || c.table == nil {
+		return nil
+	}
+	if c.catalogFallback == nil {
+		c.catalogFallback = storage.NewSession(c.schema, c.table)
+	}
+	return c.catalogFallback
+}
+
 var catalogFilterPattern = regexp.MustCompile(`(?i)\b(table_name|tablename|table_schema|schemaname|constraint_name|indexname)\s*=\s*'((?:''|[^'])*)'`)
 
 func (c *Connection) catalogResult(sql string) (*executor.Result, bool, error) {
@@ -747,12 +770,12 @@ func (c *Connection) catalogResult(sql string) (*executor.Result, bool, error) {
 	if source == "" {
 		return nil, false, nil
 	}
-	if c.txStatus == TxStatusIdle {
-		c.schema.LockStatement()
-		defer c.schema.UnlockStatement()
+	session := c.catalogSession()
+	if session == nil {
+		return nil, false, nil
 	}
 
-	tables, err := c.schema.ListTables()
+	tables, err := session.ListTables()
 	if err != nil {
 		return nil, true, err
 	}
@@ -766,7 +789,7 @@ func (c *Connection) catalogResult(sql string) (*executor.Result, bool, error) {
 		}
 	case "PG_TABLES":
 		for _, table := range tables {
-			indexes, _ := c.schema.ListTableIndexes(table)
+			indexes, _ := session.ListTableIndexes(table)
 			rows = append(rows, map[string]interface{}{
 				"schemaname": "public", "tablename": table, "tableowner": c.params["user"], "tablespace": nil,
 				"hasindexes": len(indexes) > 0, "hasrules": false, "hastriggers": false, "rowsecurity": false,
@@ -774,7 +797,7 @@ func (c *Connection) catalogResult(sql string) (*executor.Result, bool, error) {
 		}
 	case "INFORMATION_SCHEMA.COLUMNS":
 		for _, table := range tables {
-			schema, schemaErr := c.schema.GetSchema(table)
+			schema, schemaErr := session.GetSchema(table)
 			if schemaErr != nil {
 				continue
 			}
@@ -788,7 +811,7 @@ func (c *Connection) catalogResult(sql string) (*executor.Result, bool, error) {
 		}
 	case "INFORMATION_SCHEMA.TABLE_CONSTRAINTS", "INFORMATION_SCHEMA.KEY_COLUMN_USAGE":
 		for _, table := range tables {
-			schema, schemaErr := c.schema.GetSchema(table)
+			schema, schemaErr := session.GetSchema(table)
 			if schemaErr != nil || schema.UsesSyntheticRowID() {
 				continue
 			}
@@ -812,7 +835,7 @@ func (c *Connection) catalogResult(sql string) (*executor.Result, bool, error) {
 		}
 	case "PG_INDEXES":
 		for _, table := range tables {
-			indexes, _ := c.schema.ListTableIndexes(table)
+			indexes, _ := session.ListTableIndexes(table)
 			for _, index := range indexes {
 				columns := make([]string, len(index.Columns))
 				for i, column := range index.Columns {

pkg/pgserver/transactional_catalog_test.go

@@ -0,0 +1,106 @@
+package pgserver
+
+import (
+	"net"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+	"github.com/danfragoso/pizzasql-next/pkg/testkv"
+)
+
+// catalogRows decodes every DataRow of a completed simple-query exchange.
+func catalogRows(t *testing.T, msgs []*Message) [][]string {
+	t.Helper()
+	var rows [][]string
+	for _, msg := range msgs {
+		if msg.Type == MsgDataRow {
+			rows = append(rows, dataRowValues(t, msg))
+		}
+	}
+	return rows
+}
+
+func newSharedDataConnections(t *testing.T) (*Connection, net.Conn, *Connection, net.Conn) {
+	t.Helper()
+	kv := testkv.New(t)
+	pool := kv.Pool(8)
+	t.Cleanup(func() { pool.Close() })
+	schema := storage.NewSchemaManager(pool, "pg_tx_catalog")
+	table := storage.NewTableManager(pool, schema, "pg_tx_catalog")
+	writer, writerClient := newDataConnectionFor(t, schema, table)
+	observer, observerClient := newDataConnectionFor(t, schema, table)
+	return writer, writerClient, observer, observerClient
+}
+
+// The system-catalog emulation must read through the connection's session so it
+// sees DDL staged by an open transaction, and must keep showing the committed
+// catalog to other connections until that transaction commits.
+func TestCatalogEmulationSeesStagedDDLInTransaction(t *testing.T) {
+	writer, writerClient, observer, observerClient := newSharedDataConnections(t)
+
+	runQuery(t, writer, writerClient, simpleQueryMessage("CREATE TABLE base (id INTEGER PRIMARY KEY)"))
+	runQuery(t, writer, writerClient, simpleQueryMessage("BEGIN"))
+	runQuery(t, writer, writerClient, simpleQueryMessage("CREATE TABLE staged (id INTEGER PRIMARY KEY)"))
+	runQuery(t, writer, writerClient, simpleQueryMessage("ALTER TABLE base ADD COLUMN note TEXT"))
+	runQuery(t, writer, writerClient, simpleQueryMessage("CREATE INDEX idx_base_note ON base (note)"))
+
+	hasTable := func(msgs []*Message, want string) bool {
+		for _, row := range catalogRows(t, msgs) {
+			if len(row) > 0 && row[0] == want {
+				return true
+			}
+		}
+		return false
+	}
+
+	// The transaction's own session sees its staged catalog.
+	tables := runQuery(t, writer, writerClient, simpleQueryMessage("SELECT table_name FROM information_schema.tables"))
+	if !hasTable(tables, "staged") {
+		t.Fatal("transactional catalog emulation did not report a staged CREATE TABLE")
+	}
+	columns := runQuery(t, writer, writerClient, simpleQueryMessage("SELECT column_name FROM information_schema.columns"))
+	if !hasTable(columns, "note") {
+		t.Fatal("transactional catalog emulation did not report a staged ADD COLUMN")
+	}
+	indexes := runQuery(t, writer, writerClient, simpleQueryMessage("SELECT indexname FROM pg_indexes"))
+	if !hasTable(indexes, "idx_base_note") {
+		t.Fatal("transactional catalog emulation did not report a staged CREATE INDEX")
+	}
+
+	// Another connection keeps reading the committed catalog.
+	observerTables := runQuery(t, observer, observerClient, simpleQueryMessage("SELECT table_name FROM information_schema.tables"))
+	if hasTable(observerTables, "staged") {
+		t.Fatal("observer saw a table created in another connection's transaction")
+	}
+	observerColumns := runQuery(t, observer, observerClient, simpleQueryMessage("SELECT column_name FROM information_schema.columns"))
+	if hasTable(observerColumns, "note") {
+		t.Fatal("observer saw a column added in another connection's transaction")
+	}
+
+	runQuery(t, writer, writerClient, simpleQueryMessage("COMMIT"))
+
+	// After COMMIT the committed catalog reflects the migration for everyone.
+	observerTables = runQuery(t, observer, observerClient, simpleQueryMessage("SELECT table_name FROM information_schema.tables"))
+	if !hasTable(observerTables, "staged") {
+		t.Fatal("observer did not see the committed CREATE TABLE")
+	}
+	observerColumns = runQuery(t, observer, observerClient, simpleQueryMessage("SELECT column_name FROM information_schema.columns"))
+	if !hasTable(observerColumns, "note") {
+		t.Fatal("observer did not see the committed ADD COLUMN")
+	}
+	observerIndexes := runQuery(t, observer, observerClient, simpleQueryMessage("SELECT indexname FROM pg_indexes"))
+	if !hasTable(observerIndexes, "idx_base_note") {
+		t.Fatal("observer did not see the committed CREATE INDEX")
+	}
+
+	// A ROLLBACK leaves the emulation on the committed catalog again.
+	runQuery(t, writer, writerClient, simpleQueryMessage("BEGIN"))
+	runQuery(t, writer, writerClient, simpleQueryMessage("CREATE TABLE rolled_back (id INTEGER PRIMARY KEY)"))
+	if tables := runQuery(t, writer, writerClient, simpleQueryMessage("SELECT table_name FROM information_schema.tables")); !hasTable(tables, "rolled_back") {
+		t.Fatal("transactional catalog emulation did not report a staged table before ROLLBACK")
+	}
+	runQuery(t, writer, writerClient, simpleQueryMessage("ROLLBACK"))
+	if tables := runQuery(t, writer, writerClient, simpleQueryMessage("SELECT table_name FROM information_schema.tables")); hasTable(tables, "rolled_back") {
+		t.Fatal("catalog emulation kept a rolled-back table")
+	}
+}

pkg/storage/ddl_budget_test.go

@@ -0,0 +1,36 @@
+package storage
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+// The DDL commit publishes one storage transaction, so the rewrite budget must
+// reject an oversized table before the batch is encoded. The boundary is
+// inclusive: exactly the store's operation ceiling is still sendable.
+func TestDDLBatchBudgetRejectsOversizedRewrite(t *testing.T) {
+	budget := &ddlBatchBudget{}
+	if err := budget.reserve("t", maxOperations, 0); err != nil {
+		t.Fatalf("reserve at the operation ceiling = %v, want nil", err)
+	}
+	err := budget.reserve("t", 1, 0)
+	var limit *ddlBatchLimitError
+	if !errors.As(err, &limit) {
+		t.Fatalf("reserve past the operation ceiling = %v, want *ddlBatchLimitError", err)
+	}
+	if !strings.Contains(limit.Error(), "storage batch limit") {
+		t.Fatalf("limit error message = %q, want the storage batch limit named", limit.Error())
+	}
+	if limit.Table != "t" {
+		t.Fatalf("limit error table = %q, want t", limit.Table)
+	}
+
+	bytes := &ddlBatchBudget{}
+	if err := bytes.reserve("t", 1, maxTransactionSize); err != nil {
+		t.Fatalf("reserve at the byte ceiling = %v, want nil", err)
+	}
+	if err := bytes.reserve("t", 1, 1); !errors.As(err, &limit) {
+		t.Fatalf("reserve past the byte ceiling = %v, want *ddlBatchLimitError", err)
+	}
+}

pkg/storage/schema.go

@@ -394,7 +394,12 @@ func (m *SchemaManager) indexListDurable() ([]string, uint64, error) {
 // readers: every cache derived from the catalog is dropped and the version is
 // bumped so the foreign-key metadata and session analyzers reload. The caches
 // are rebuilt lazily from the durable catalog, which the commit already wrote.
-func (m *SchemaManager) invalidateCommittedDDL() {
+//
+// Only the touched tables lose their ROWID allocator. Wiping every allocator
+// would let a concurrent transaction that staged a TEXT/composite-key insert
+// (whose ROWID is not durable yet) derive the same ROWID again and publish a
+// duplicate, so untouched tables keep their derived counters.
+func (m *SchemaManager) invalidateCommittedDDL(tables []string) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.cache = make(map[string]*Schema)
@@ -402,7 +407,9 @@ func (m *SchemaManager) invalidateCommittedDDL() {
 	m.indexListCache = nil
 	m.indexListCached = false
 	m.rowIDMu.Lock()
-	m.rowIDAlloc = make(map[string]*rowIDAllocator)
+	for _, table := range tables {
+		delete(m.rowIDAlloc, strings.ToLower(table))
+	}
 	m.rowIDMu.Unlock()
 	m.fkCatalogMu.Lock()
 	m.fkCatalog = nil

pkg/storage/schema_tx.go

@@ -161,23 +161,67 @@ func (tx *SchemaTx) snapshot() schemaTxSnapshot {
 }
 
 // restore returns the overlay to a snapshot and undoes the overlay re-keyings
-// performed after it. The caller holds the session lock.
+// performed after it. Every collection is copied rather than aliased: the
+// snapshot is retained for a later ROLLBACK TO SAVEPOINT, and DDL staged after
+// this restore must not mutate the retained state. Staged schema and index
+// definitions are immutable once stored, so a shallow copy of the maps is
+// enough. The caller holds the session lock.
 func (tx *SchemaTx) restore(snap schemaTxSnapshot) {
 	for i := len(tx.rekeys) - 1; i >= snap.rekeys; i-- {
 		tx.sess.rekeyOverlayLocked(tx.rekeys[i].to, tx.rekeys[i].from)
 	}
 	tx.rekeys = tx.rekeys[:snap.rekeys]
 	tx.version = snap.version
-	tx.staged = snap.staged
-	tx.stagedIndexes = snap.stagedIndexes
-	tx.tables = snap.tables
-	tx.indexList = snap.indexList
+	tx.staged = copySchemaMap(snap.staged)
+	tx.stagedIndexes = copyIndexMap(snap.stagedIndexes)
+	tx.tables = append([]string(nil), snap.tables...)
+	tx.indexList = append([]string(nil), snap.indexList...)
 	tx.catalogDirty = snap.catalogDirty
 	tx.indexListDirty = snap.indexListDirty
-	tx.rewrites = snap.rewrites
-	tx.dataSources = snap.dataSources
-	tx.drops = snap.drops
-	tx.renames = snap.renames
+	tx.rewrites = copyRewriteMap(snap.rewrites)
+	tx.dataSources = copyStringMap(snap.dataSources)
+	tx.drops = copyBoolMap(snap.drops)
+	tx.renames = copyStringMap(snap.renames)
+}
+
+func copySchemaMap(source map[string]*Schema) map[string]*Schema {
+	cloned := make(map[string]*Schema, len(source))
+	for key, value := range source {
+		cloned[key] = value
+	}
+	return cloned
+}
+
+func copyIndexMap(source map[string]*Index) map[string]*Index {
+	cloned := make(map[string]*Index, len(source))
+	for key, value := range source {
+		cloned[key] = value
+	}
+	return cloned
+}
+
+func copyRewriteMap(source map[string][]rowRewrite) map[string][]rowRewrite {
+	cloned := make(map[string][]rowRewrite, len(source))
+	for key, value := range source {
+		cloned[key] = append([]rowRewrite(nil), value...)
+	}
+	return cloned
+}
+
+func copyStringMap(source map[string]string) map[string]string {
+	cloned := make(map[string]string, len(source))
+	for key, value := range source {
+		cloned[key] = value
+	}
+	return cloned
+}
+
+func copyBoolMap(source map[string]bool) map[string]bool {
+	cloned := make(map[string]bool, len(source))
+	for key, value := range source {
+		cloned[key] = value
+	}
+	return cloned
 }
 
 // bump advances the transaction's catalog version so the executor's analyzer
@@ -201,6 +245,9 @@ func (tx *SchemaTx) recordRead(key string, lsn uint64) {
 func (tx *SchemaTx) refreshCommitted() {
 	tx.committedSchemas = make(map[string]*Schema)
 	tx.committedIndexes = make(map[string]*Index)
+	if !tx.catalogDirty {
+		tx.tables = nil
+	}
 	if !tx.indexListDirty {
 		tx.indexList = nil
 	}
@@ -520,6 +567,11 @@ func (tx *SchemaTx) addColumn(table string, column Column, checks []CheckConstra
 	}
 
 	now := tx.base.defaultNow()
+	// Pin the table generation before scanning: a concurrent commit that lands
+	// between the scan and the generation capture would otherwise be invisible,
+	// so a durable row the scan did not validate could violate the new
+	// constraint. Recording first turns that interleaving into 40001 at COMMIT.
+	tx.sess.markTableScanned(strings.ToLower(table))
 	// Validation covers the committed rows, mirroring the autocommit DDL path:
 	// rows this transaction staged are re-validated against the staged schema at
 	// commit (applyOverlaySchemaDefaults), because a staged row may still be
@@ -599,7 +651,6 @@ func (tx *SchemaTx) addColumn(table string, column Column, checks []CheckConstra
 			})
 		}
 	}
-	tx.sess.markTableScanned(strings.ToLower(table))
 	return nil
 }
 
@@ -742,7 +793,14 @@ func (tx *SchemaTx) renameTable(oldName, newName string) error {
 		tx.dataSources[newLower] = ""
 	} else {
 		tx.dataSources[newLower] = source
-		tx.renames[source] = newName
+		if strings.EqualFold(source, newLower) {
+			// Renaming back to the committed name cancels the pending move: the
+			// rows already live under the destination prefix, so emitting a
+			// put+delete for the same key would be order-dependent.
+			delete(tx.renames, source)
+		} else {
+			tx.renames[source] = newName
+		}
 	}
 	delete(tx.dataSources, oldLower)
 	// A drop staged before the rename keeps its durable source; the rename
@@ -842,13 +900,15 @@ func (tx *SchemaTx) commitCatalogOps() ([]BatchOp, []CompareCheck, error) {
 		keys = append(keys, key)
 	}
 	// A staged table definition whose name replaced an older one deletes the
-	// old record in the same batch.
+	// old record in the same batch. A rename whose destination is the committed
+	// name (renamed away and back) has no move to publish.
 	for source, newName := range tx.renames {
+		if strings.EqualFold(source, newName) {
+			continue
+		}
 		ops = append(ops, BatchOp{Op: batchDelete, Key: []byte(tx.base.schemaKey(source))})
 		keys = append(keys, tx.base.schemaKey(source))
-		if !strings.EqualFold(source, newName) {
-			ops = append(ops, BatchOp{Op: batchDelete, Key: []byte(tx.base.rowIDKey(source))})
-		}
+		ops = append(ops, BatchOp{Op: batchDelete, Key: []byte(tx.base.rowIDKey(source))})
 	}
 
 	for lower, index := range tx.stagedIndexes {
@@ -901,7 +961,7 @@ func (tx *SchemaTx) commitCatalogOps() ([]BatchOp, []CompareCheck, error) {
 // session staged in its overlay are published from the overlay instead, so they
 // are skipped here. The caller holds the session lock and the exclusive table
 // gates of every affected table.
-func (tx *SchemaTx) commitDataOps() ([]BatchOp, map[string]map[string]*overlayEntry, error) {
+func (tx *SchemaTx) commitDataOps(budget *ddlBatchBudget) ([]BatchOp, map[string]map[string]*overlayEntry, error) {
 	entries := make(map[string]map[string]*overlayEntry)
 	var ops []BatchOp
 	if len(tx.drops) == 0 && len(tx.renames) == 0 && len(tx.rewrites) == 0 {
@@ -928,7 +988,7 @@ func (tx *SchemaTx) commitDataOps() ([]BatchOp, map[string]map[string]*overlayEn
 
 	for _, source := range names {
 		if tx.drops[source] {
-			dropped, err := tx.scanRowsForDDL(source, func(key string, row Row) (bool, Row, error) {
+			dropped, err := tx.scanRowsForDDL(source, budget, func(key string, row Row) (bool, Row, error) {
 				if tx.sess.overlayHolds(strings.ToLower(source), key) {
 					return false, nil, nil
 				}
@@ -942,17 +1002,13 @@ func (tx *SchemaTx) commitDataOps() ([]BatchOp, map[string]map[string]*overlayEn
 		}
 
 		newName, renamed := tx.renames[source]
+		if renamed && strings.EqualFold(source, newName) {
+			// A rename that returned to the committed name has no row move; the
+			// table still owes its pending ALTER rewrites.
+			renamed = false
+		}
 		if renamed {
-			updated, err := tx.scanRowsForDDL(source, func(key string, row Row) (bool, Row, error) {
-				if _, err := tx.rewriteRow(newName, row); err != nil {
-					return false, nil, err
-				}
-				newKey := tx.sess.table.dataKey(newName, mustRowPK(tx, newName, row))
-				if tx.sess.overlayHolds(strings.ToLower(newName), newKey) {
-					return false, nil, nil
-				}
-				return true, row, nil
-			})
+			moved, movedEntries, err := tx.renameRowOps(source, newName, budget)
 			if err != nil {
 				return nil, nil, err
 			}
@@ -960,20 +1016,10 @@ func (tx *SchemaTx) commitDataOps() ([]BatchOp, map[string]map[string]*overlayEn
 			if entries[lower] == nil {
 				entries[lower] = make(map[string]*overlayEntry)
 			}
-			for _, op := range updated {
-				oldKey := op.Key
-				suffix := strings.TrimPrefix(string(oldKey), tx.sess.table.dataPrefix(source))
-				newKey := []byte(tx.sess.table.dataPrefix(newName) + suffix)
-				row, err := decodeRow(op.Value)
-				if err != nil {
-					return nil, nil, err
-				}
-				entries[lower][string(newKey)] = &overlayEntry{row: row}
-				ops = append(ops,
-					BatchOp{Op: batchPut, Key: newKey, Value: op.Value},
-					BatchOp{Op: batchDelete, Key: oldKey},
-				)
+			for key, entry := range movedEntries {
+				entries[lower][key] = entry
 			}
+			ops = append(ops, moved...)
 			continue
 		}
 
@@ -981,7 +1027,7 @@ func (tx *SchemaTx) commitDataOps() ([]BatchOp, map[string]map[string]*overlayEn
 		if len(rewrites) == 0 {
 			continue
 		}
-		updated, err := tx.scanRowsForDDL(source, func(key string, row Row) (bool, Row, error) {
+		updated, err := tx.scanRowsForDDL(source, budget, func(key string, row Row) (bool, Row, error) {
 			if tx.sess.overlayHolds(strings.ToLower(source), key) {
 				return false, nil, nil
 			}
@@ -1010,24 +1056,94 @@ func (tx *SchemaTx) commitDataOps() ([]BatchOp, map[string]map[string]*overlayEn
 	return ops, entries, nil
 }
 
-// mustRowPK derives the primary-key string of a row the transaction's staged
-// schema describes.
-func mustRowPK(tx *SchemaTx, table string, row Row) string {
-	schema, err := tx.getSchema(table)
-	if err != nil {
-		return ""
-	}
-	pk, err := schema.PrimaryKeyForRow(row)
-	if err != nil {
-		return ""
+// renameRowOps builds the row move of one RENAME TABLE: every committed row of
+// source is written under the destination prefix and its old key deleted. A row
+// the transaction already re-staged under the destination (an UPDATE, DELETE,
+// or re-INSERT after the rename) is owned by the session overlay, so only the
+// stale committed key is deleted; without that delete the rename would leave an
+// unreachable row under the dropped table name.
+func (tx *SchemaTx) renameRowOps(source, newName string, budget *ddlBatchBudget) ([]BatchOp, map[string]*overlayEntry, error) {
+	entries := make(map[string]*overlayEntry)
+	var ops []BatchOp
+	oldPrefix := tx.sess.table.dataPrefix(source)
+	newPrefix := tx.sess.table.dataPrefix(newName)
+	lowerNew := strings.ToLower(newName)
+	err := tx.sess.table.scanRowsAtPrefix(oldPrefix, func(oldKey string, row Row, _ uint64) (bool, error) {
+		if _, err := tx.rewriteRow(newName, row); err != nil {
+			return true, err
+		}
+		suffix := strings.TrimPrefix(oldKey, oldPrefix)
+		newKey := newPrefix + suffix
+		if tx.sess.overlayHolds(lowerNew, newKey) {
+			deleted := BatchOp{Op: batchDelete, Key: []byte(oldKey)}
+			if err := budget.reserve(source, 1, batchOpBytes(deleted)); err != nil {
+				return true, err
+			}
+			ops = append(ops, deleted)
+			return false, nil
+		}
+		value, err := encodeRow(row)
+		if err != nil {
+			return true, err
+		}
+		put := BatchOp{Op: batchPut, Key: []byte(newKey), Value: value}
+		del := BatchOp{Op: batchDelete, Key: []byte(oldKey)}
+		if err := budget.reserve(source, 2, batchOpBytes(put)+batchOpBytes(del)); err != nil {
+			return true, err
+		}
+		entries[newKey] = &overlayEntry{row: row}
+		ops = append(ops, put, del)
+		return false, nil
+	})
+	return ops, entries, err
+}
+
+// ddlBatchBudget bounds the row operations one transactional DDL commit may add
+// to its single compare-batch. The whole transaction is published atomically as
+// one storage transaction, so it cannot page its writes the way autocommit DDL
+// does: a rewrite that would exceed the operation or byte ceiling fails with an
+// explicit limit error while the table is still being scanned, instead of
+// buffering every row and failing inside CompareBatchWrite.
+type ddlBatchBudget struct {
+	ops   int
+	bytes int
+}
+
+// reserve counts one table's pending operations and fails as soon as the batch
+// ceiling is crossed.
+func (b *ddlBatchBudget) reserve(table string, ops, bytes int) error {
+	b.ops += ops
+	b.bytes += bytes
+	if b.ops > maxOperations || b.bytes > maxTransactionSize {
+		return &ddlBatchLimitError{Table: table, Operations: b.ops, Bytes: b.bytes}
 	}
-	return pk
+	return nil
+}
+
+// ddlBatchLimitError reports that a transactional DDL row rewrite cannot fit in
+// the single storage transaction the commit publishes.
+type ddlBatchLimitError struct {
+	Table      string
+	Operations int
+	Bytes      int
+}
+
+func (e *ddlBatchLimitError) Error() string {
+	return fmt.Sprintf(
+		"transactional DDL on %s cannot be published atomically: the rewrite needs more than the storage batch limit (%d operations / %d bytes, reached %d operations / %d bytes); run the statement outside a transaction so it can page its writes, or split the migration",
+		e.Table, maxOperations, maxTransactionSize, e.Operations, e.Bytes)
+}
+
+// batchOpBytes is the wire size of one operation (header, key, and value), used
+// to bound a commit batch before it is encoded.
+func batchOpBytes(op BatchOp) int {
+	return 12 + len(op.Key) + len(op.Value)
 }
 
 // scanRowsForDDL streams a durable table's rows through transform. When
 // transform reports a change the row is re-encoded and returned as a put under
 // its existing key; when it reports removal the key is returned as a delete.
-func (tx *SchemaTx) scanRowsForDDL(table string, transform func(key string, row Row) (write bool, replacement Row, err error)) ([]BatchOp, error) {
+func (tx *SchemaTx) scanRowsForDDL(table string, budget *ddlBatchBudget, transform func(key string, row Row) (write bool, replacement Row, err error)) ([]BatchOp, error) {
 	var ops []BatchOp
 	prefix := []byte(tx.sess.table.dataPrefix(table))
 	err := tx.base.pool.WithClient(func(client *KVClient) (retErr error) {
@@ -1058,14 +1174,22 @@ func (tx *SchemaTx) scanRowsForDDL(table string, transform func(key string, row
 					continue
 				}
 				if replacement == nil {
-					ops = append(ops, BatchOp{Op: batchDelete, Key: append([]byte(nil), entry.Key...)})
+					op := BatchOp{Op: batchDelete, Key: append([]byte(nil), entry.Key...)}
+					if err := budget.reserve(table, 1, batchOpBytes(op)); err != nil {
+						return err
+					}
+					ops = append(ops, op)
 					continue
 				}
 				value, err := encodeRow(replacement)
 				if err != nil {
 					return err
 				}
-				ops = append(ops, BatchOp{Op: batchPut, Key: append([]byte(nil), entry.Key...), Value: value})
+				op := BatchOp{Op: batchPut, Key: append([]byte(nil), entry.Key...), Value: value}
+				if err := budget.reserve(table, 1, batchOpBytes(op)); err != nil {
+					return err
+				}
+				ops = append(ops, op)
 			}
 			if done {
 				return nil

pkg/storage/schema_tx_rename_test.go

@@ -0,0 +1,56 @@
+package storage
+
+import "testing"
+
+// A RENAME that returns to the committed name must not leave a pending move:
+// publishing a put and a delete for the same key in one batch is order-dependent
+// and can lose the row or the schema record. Assert the property directly on the
+// transaction's catalog and row ops.
+func TestSchemaTxRenameAwayAndBackHasNoDuplicateOps(t *testing.T) {
+	_, _, schemas, tables := newTestSession(t)
+	createTestTable(t, schemas, "a", []Column{
+		{Name: "id", Type: "INTEGER", PrimaryKey: true},
+		{Name: "v", Type: "TEXT"},
+	})
+	if err := tables.Insert("a", Row{"id": int64(1), "v": "one"}); err != nil {
+		t.Fatal(err)
+	}
+
+	s := NewSession(schemas, tables)
+	if err := s.Begin(); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RenameTable("a", "b"); err != nil {
+		t.Fatalf("rename a->b: %v", err)
+	}
+	if err := s.RenameTable("b", "a"); err != nil {
+		t.Fatalf("rename b->a: %v", err)
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	// Renaming back to the committed name cancels the pending move, so the
+	// overlay has nothing left to publish for it.
+	if len(s.schemaTx.renames) != 0 {
+		t.Fatalf("pending renames after renaming away and back = %v, want none", s.schemaTx.renames)
+	}
+	if source := s.schemaTx.dataSources["a"]; source != "a" {
+		t.Fatalf("data source for a = %q, want a", source)
+	}
+	catalogOps, _, err := s.schemaTx.commitCatalogOps()
+	if err != nil {
+		t.Fatalf("catalog ops: %v", err)
+	}
+	dataOps, _, err := s.schemaTx.commitDataOps(&ddlBatchBudget{})
+	if err != nil {
+		t.Fatalf("data ops: %v", err)
+	}
+	seen := make(map[string]bool, len(catalogOps)+len(dataOps))
+	for _, op := range append(catalogOps, dataOps...) {
+		key := string(op.Key)
+		if seen[key] {
+			t.Fatalf("rename away and back emitted more than one operation for key %q", key)
+		}
+		seen[key] = true
+	}
+}

pkg/storage/tx.go

@@ -196,11 +196,17 @@ func (s *Session) RollbackToSavepoint(sp *SessionSavepoint) {
 // rollbackToSavepointLocked restores a savepoint (caller holds the session
 // lock).
 func (s *Session) rollbackToSavepointLocked(sp *SessionSavepoint) {
-	s.overlay = sp.overlay
-	s.log = sp.mutationLog
-	s.reads = sp.reads
-	s.scanGens = sp.scanGens
-	s.predicateGens = sp.predicateGens
+	// Copy rather than alias: the savepoint stays installed for a later
+	// ROLLBACK TO SAVEPOINT, and writes staged after this restore must not
+	// mutate the retained state.
+	s.overlay = copyOverlay(sp.overlay)
+	s.log = append([]txMutation(nil), sp.mutationLog...)
+	s.reads = copyLSNMap(sp.reads)
+	s.scanGens = copyLSNMap(sp.scanGens)
+	s.predicateGens = make(map[string]predicateRead, len(sp.predicateGens))
+	for key, predicate := range sp.predicateGens {
+		s.predicateGens[key] = predicate
+	}
 	if s.schemaTx != nil && sp.schema != nil {
 		s.schemaTx.restore(*sp.schema)
 	}
@@ -208,6 +214,21 @@ func (s *Session) rollbackToSavepointLocked(sp *SessionSavepoint) {
 	s.fkVersion = 0
 }
 
+// copyOverlay copies the two-level staged row map. Overlay entries are
+// immutable once staged (rows are cloned before they are stored), so the inner
+// maps are copied but the entries themselves are shared.
+func copyOverlay(source map[string]map[string]*overlayEntry) map[string]map[string]*overlayEntry {
+	cloned := make(map[string]map[string]*overlayEntry, len(source))
+	for table, entries := range source {
+		staged := make(map[string]*overlayEntry, len(entries))
+		for key, entry := range entries {
+			staged[key] = entry
+		}
+		cloned[table] = staged
+	}
+	return cloned
+}
+
 func copyLSNMap(source map[string]uint64) map[string]uint64 {
 	cloned := make(map[string]uint64, len(source))
 	for key, value := range source {
@@ -351,7 +372,7 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 			s.resetLocked()
 			return 0, err
 		}
-		dataOps, dataEntries, err := s.schemaTx.commitDataOps()
+		dataOps, dataEntries, err := s.schemaTx.commitDataOps(&ddlBatchBudget{})
 		if err != nil {
 			s.resetLocked()
 			return 0, err
@@ -449,6 +470,21 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 		return 0, nil
 	}
 
+	// Pre-flight the assembled batch so an oversized commit fails with an
+	// actionable error instead of the storage protocol's generic limit error.
+	if len(ops) > maxOperations {
+		s.resetLocked()
+		return 0, fmt.Errorf("transaction cannot be published atomically: %d operations exceed the storage batch limit of %d; split the transaction or run the DDL outside a transaction so it can page its writes", len(ops), maxOperations)
+	}
+	batchBytes := 0
+	for _, op := range ops {
+		batchBytes += batchOpBytes(op)
+	}
+	if batchBytes > maxTransactionSize {
+		s.resetLocked()
+		return 0, fmt.Errorf("transaction cannot be published atomically: %d bytes exceed the storage batch limit of %d; split the transaction or run the DDL outside a transaction so it can page its writes", batchBytes, maxTransactionSize)
+	}
+
 	var committed bool
 	var commitLSN uint64
 	err = s.table.pool.WithClient(func(c *KVClient) error {
@@ -463,6 +499,15 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 		for table := range s.overlay {
 			s.table.invalidateWriteFailure(table)
 		}
+		if hasDDL {
+			// Staged DDL may have become durable even though the response was
+			// lost, so the shared catalog view cannot be trusted either: drop it
+			// for every touched table and reload from storage on the next read.
+			s.schema.invalidateCommittedDDL(touchedTables)
+			for _, table := range touchedTables {
+				s.table.invalidateWriteFailure(table)
+			}
+		}
 	}
 	if err != nil {
 		s.resetLocked()
@@ -507,7 +552,7 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 // reload lazily from the committed records. The caller still holds the table
 // gates, which is the documented lock order (table gate, then catalog mutex).
 func (s *Session) publishCommittedCatalog(tables []string) {
-	s.schema.invalidateCommittedDDL()
+	s.schema.invalidateCommittedDDL(tables)
 	for _, table := range tables {
 		s.table.InvalidateCache(table)
 		s.table.bumpIndexPredicateWildcard(table)
@@ -1326,25 +1371,29 @@ func (s *Session) deleteByPKLocked(table, pk string) (Row, bool, error) {
 	return row, true, nil
 }
 
-// getByPKLocked reads a row observing the overlay (caller holds s.mu).
+// getByPKLocked reads a row observing the overlay (caller holds s.mu). The
+// staged key addresses the overlay; the durable read follows the committed name
+// when RENAME TABLE has staged a move, so a point UPDATE/DELETE issued after a
+// rename still finds the row instead of silently matching nothing.
 func (s *Session) getByPKLocked(table, pk string) (Row, error) {
-	key := s.table.dataKey(table, pk)
+	stagedKey := s.table.dataKey(table, pk)
 	tl := strings.ToLower(table)
-	if e, ok := s.overlay[tl][key]; ok {
+	if e, ok := s.overlay[tl][stagedKey]; ok {
 		if e.absent {
 			return nil, ErrKeyNotFound
 		}
 		return cloneRow(e.row), nil
 	}
-	row, lsn, err := s.table.getByPKWithLSN(table, pk)
+	durableKey := s.durableDataKeyLocked(table, pk)
+	row, lsn, err := s.table.getKeyWithLSN(durableKey)
 	if err != nil {
 		if err == ErrKeyNotFound {
-			s.reads[key] = 0
+			s.reads[durableKey] = 0
 			return nil, ErrKeyNotFound
 		}
 		return nil, err
 	}
-	s.reads[key] = lsn
+	s.reads[durableKey] = lsn
 	if err := s.applyStagedRewritesLocked(table, row); err != nil {
 		return nil, err
 	}