Add regression tests for review findings; update plan docs

- SF golden MERGE SQL byte-identical to pre-refactor generator (fixture captured
  by running the merge-base generator); dotted per-component quoting + LegacyDotted
  toast-map keying
- BQ raw-name filter pinned to LegacyDotted; CH legacy-normalized dotted
  destination BuildQuery test
- Dotted CTID partitioning unit test (dotted schema/parent/children)
- E2e: dotted partitioned-parent ctid QRep, initial-snapshot-only dotted mirror,
  add-tables union validation rejections, skip-validation structural check,
  source dotted collision
- Snapshot clone workflow ID pinned to the legacy format; pua record.target/source
  legacy-dotted test; schema-helpers dotted overlap tests; Normalize*/Denormalize*
  helper round-trips
- SF dotted e2e namespace suffixed to avoid cross-run collisions
- Review findings doc + progress doc updated

Author: ilidemi

.claude/plans/qualified-table-identifiers-progress.md

@@ -83,7 +83,9 @@ Branch: `qualified-table-identifiers`
       continued syncing through upgrade (raw-table LegacyDotted round-trip live)
 - [x] D: per-component units — identifiers, BQ convertToDatasetTable matrix, EventHub
       scoped test, CH dotted-destination BuildQuery (headline), SF per-component
-      normalize, pua legacy-dotted
+      normalize; pua legacy-dotted + SF golden MERGE + schema-helpers tests were
+      MISSING (claimed done in error) — added during the post-implementation
+      adversarial review (see review-findings doc)
 
 ## Verification (after implementation)
 - [x] build: go build ./... clean (flow + pkg modules); cargo check (nexus) clean;
@@ -102,6 +104,14 @@ Branch: `qualified-table-identifiers`
 ## STATUS: COMPLETE (2026-06-12)
 All plan phases implemented and verified. 17 commits on qualified-table-identifiers.
 
+## Adversarial review (2026-06-12)
+
+An ultracode adversarial review of the finished branch found 14 distinct confirmed
+bugs (1 critical: ES dotted index rerouting; 3 high: CH dotted destination retarget,
+InitialLoadSummary namespace loss, add-tables collision validation) — all fixed, plus
+test gaps closed. Full findings, fixes, disprovals and the verification log:
+`.claude/plans/qualified-table-identifiers-review-findings.md`.
+
 ## Known acceptable gaps
 - Dotted-name resync e2e not written: resync `_resync`/`_peerdb_resync` suffix operates
   on `.Table` only (compile-checked struct logic), resync flows covered dotless by
@@ -111,6 +121,13 @@ All plan phases implemented and verified. 17 commits on qualified-table-identifi
 - EventHub names containing dots remain unsupported (documented; packing ambiguity,
   parity with pre-refactor).
 - Legacy strings in persisted configs to be dropped in release N+1 (follow-up).
+- Plan B.7 queue-destination dotted e2e (Kafka/PubSub/ES) not written — no local
+  Kafka/PubSub/ES infra; ES dotted routing is covered by the LegacyDotted fix + review,
+  kafka/pubsub already route via LegacyDotted. CI follow-up candidate.
+- Plan C.4 catalog-migration backfill Go tests replaced by manual verification on 30
+  real catalog rows + SQL-vs-Go-normalizer audit (see review-findings L7/T5).
+- Plan B.3 dotted table REMOVAL / cancel_table_addition dotted variants and full B.5/B.6
+  dotted QRep matrix (incremental, ctid parent, CH QRep, Mongo QRep) remain partial.
 
 ## Deviations from plan
 - Legacy proto fields annotated via comments instead of [deprecated = true] to avoid

.claude/plans/qualified-table-identifiers-review-findings.md

@@ -1,11 +1,15 @@
 package connbigquery
 
 import (
+	"fmt"
 	"reflect"
+	"strings"
 	"testing"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
+	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
 func TestGenerateUpdateStatement(t *testing.T) {
@@ -183,3 +187,31 @@ func TestGenerateUpdateStatement_WithUnchangedToastColsAndSoftDelete(t *testing.
 		t.Errorf("Unexpected result. Expected: %v,\nbut got: %v", expected, result)
 	}
 }
+
+// the raw-table filter must use the legacy dotted destination name — the exact string
+// the sync side writes into _peerdb_destination_table_name — including for dotted
+// name components arriving first-dot-split from legacy configs
+func TestGenerateFlattenedCTERawNameFilterIsLegacyDotted(t *testing.T) {
+	m := &mergeStmtGenerator{
+		shortColumn:     map[string]string{"id": "_c0"},
+		rawDatasetTable: datasetTable{dataset: "ds", table: "_peerdb_raw_my_mirror"},
+		mergeBatchId:    7,
+	}
+	schema := &protos.TableSchema{
+		Columns: []*protos.FieldDescription{{Name: "id", Type: string(types.QValueKindInt64)}},
+	}
+
+	for _, tc := range []struct {
+		dstTable common.QualifiedTable
+		rawName  string
+	}{
+		{dstTable: common.QualifiedTable{Namespace: "ds", Table: "my_table"}, rawName: "ds.my_table"},
+		{dstTable: common.QualifiedTable{Namespace: "sch.ema", Table: "ta.ble"}, rawName: "sch.ema.ta.ble"},
+	} {
+		cte := m.generateFlattenedCTE(tc.dstTable, schema)
+		want := fmt.Sprintf("_peerdb_destination_table_name='%s'", tc.rawName)
+		if !strings.Contains(cte, want) {
+			t.Errorf("flattened CTE filter for %v should contain %q, got: %s", tc.dstTable, want, cte)
+		}
+	}
+}

flow/connectors/bigquery/merge_stmt_generator_test.go

@@ -557,3 +557,42 @@ func TestGenerateCreateTableSQLForNormalizedTable(t *testing.T) {
 		})
 	}
 }
+
+// a pre-QualifiedTable config with destination string "dst.table" arrives first-dot-split
+// as {Namespace: "dst", Table: "table"}; the physical ClickHouse table and the raw-table
+// lookup must still both resolve to the original single-part name "dst.table"
+func TestBuildQuery_LegacyNormalizedDottedDestination(t *testing.T) {
+	ctx := t.Context()
+	tableName := common.QualifiedTable{Namespace: "dst", Table: "table"}
+	tableSchema := &protos.TableSchema{
+		Columns: []*protos.FieldDescription{
+			{Name: "id", Type: string(types.QValueKindInt64)},
+		},
+	}
+	g := NewNormalizeQueryGenerator(
+		tableName,
+		map[common.QualifiedTable]*protos.TableSchema{tableName: tableSchema},
+		[]*protos.TableMapping{
+			{
+				SourceTable:      &protos.QualifiedTable{Namespace: "public", Table: "src"},
+				DestinationTable: &protos.QualifiedTable{Namespace: tableName.Namespace, Table: tableName.Table},
+			},
+		},
+		10,
+		5,
+		false,
+		false,
+		map[string]string{},
+		"raw_my_table",
+		nil,
+		false,
+		"",
+		shared.InternalVersion_Latest,
+		nil,
+	)
+
+	query, err := g.BuildQuery(ctx)
+	require.NoError(t, err)
+	require.Contains(t, query, "INSERT INTO `dst.table`")
+	require.Contains(t, query, "_peerdb_destination_table_name = 'dst.table'")
+}

flow/connectors/clickhouse/normalize_test.go

@@ -22,6 +22,8 @@ func TestNewScopedEventhub(t *testing.T) {
 	require.Error(t, err, "missing namespace")
 	_, err = NewScopedEventhub(common.QualifiedTable{Namespace: "ns", Table: "hub"})
 	require.Error(t, err, "missing partition key column")
+	_, err = NewScopedEventhub(common.QualifiedTable{Namespace: "ns", Table: "hub.pk.extra"})
+	require.Error(t, err, "more than 3 parts was rejected before QualifiedTable too")
 }
 
 // legacy 3-part dotted destinations from Lua scripts keep parsing as before

flow/connectors/eventhub/scoped_eventhub_test.go

@@ -827,3 +827,82 @@ func prepareTestData(t *testing.T, pool *pgx.Conn, schema string, includeNulls b
 
 	return rowsCount
 }
+
+// CTID partitioning of a partitioned parent whose schema, parent and child names all
+// contain dots: ChildTableRange.child_table must carry exact QualifiedTable structs
+// and every child must map back to a partition (plan 7.D dotted watermark/child case).
+func TestCTIDPartitioningOnDottedPartitionedTable(t *testing.T) {
+	t.Parallel()
+	_, conn, connStr := setupTestSchema(t)
+
+	schemaName := "par.t_" + strings.ToLower(common.RandomString(8))
+	parent := common.QualifiedTable{Namespace: schemaName, Table: "pa.rent"}
+	parentQuoted := parent.String()
+
+	_, err := conn.Exec(t.Context(), "CREATE SCHEMA "+common.QuoteIdentifier(schemaName))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_, _ = conn.Exec(t.Context(), fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE", common.QuoteIdentifier(schemaName)))
+	})
+
+	_, err = conn.Exec(t.Context(), fmt.Sprintf(`
+		CREATE TABLE %s (
+			id SERIAL,
+			partition_key INT NOT NULL,
+			value TEXT,
+			PRIMARY KEY (partition_key, id)
+		) PARTITION BY RANGE (partition_key)
+	`, parentQuoted))
+	require.NoError(t, err)
+
+	rowsPerPartition := 25
+	numChildTables := 3
+	children := make([]common.QualifiedTable, numChildTables)
+	for i := range numChildTables {
+		children[i] = common.QualifiedTable{Namespace: schemaName, Table: fmt.Sprintf("chi.ld_%d", i)}
+		_, err = conn.Exec(t.Context(), fmt.Sprintf(
+			`CREATE TABLE %s PARTITION OF %s FOR VALUES FROM (%d) TO (%d)`,
+			children[i].String(), parentQuoted, i*rowsPerPartition, (i+1)*rowsPerPartition))
+		require.NoError(t, err)
+	}
+
+	for i := range numChildTables * rowsPerPartition {
+		_, err = conn.Exec(t.Context(), fmt.Sprintf(
+			`INSERT INTO %s (partition_key, value) VALUES ($1, $2)`, parentQuoted),
+			i, fmt.Sprintf("val_%d", i))
+		require.NoError(t, err)
+	}
+
+	c := &PostgresConnector{
+		connStr: connStr,
+		Config:  &protos.PostgresConfig{},
+		conn:    conn,
+		logger:  log.NewStructuredLogger(slog.With(slog.String(string(shared.FlowNameKey), "testCTIDDotted"))),
+	}
+	partitions, err := getQRepPartitions(t, c, &protos.QRepConfig{
+		FlowJobName:             "test_ctid_dotted_partitioned",
+		NumRowsPerPartition:     10,
+		NumPartitionsOverride:   8,
+		Query:                   fmt.Sprintf(`SELECT * FROM %s WHERE ctid BETWEEN {{.start}} AND {{.end}}`, parentQuoted),
+		QualifiedWatermarkTable: internal.QualifiedTableProto(parent),
+		WatermarkColumn:         "ctid",
+	}, nil)
+	require.NoError(t, err)
+	require.NotEmpty(t, partitions)
+
+	childTableCounts := make(map[common.QualifiedTable]int)
+	for _, p := range partitions {
+		require.Nil(t, p.Range)
+		require.NotEmpty(t, p.ChildTableRanges)
+		for _, ctr := range p.ChildTableRanges {
+			require.NotNil(t, ctr.ChildTable)
+			childTableCounts[internal.QualifiedTableFromProto(ctr.ChildTable)]++
+		}
+	}
+
+	require.Len(t, childTableCounts, numChildTables)
+	for _, child := range children {
+		require.Positive(t, childTableCounts[child],
+			"child %s should appear in at least one partition", child)
+	}
+}

flow/connectors/postgres/qrep_partition_test.go

@@ -2,10 +2,13 @@ package connsnowflake
 
 import (
 	"reflect"
+	"strings"
 	"testing"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
+	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
 func TestGenerateUpdateStatement(t *testing.T) {
@@ -148,3 +151,81 @@ func TestGenerateUpdateStatement_WithUnchangedToastColsAndSoftDelete(t *testing.
 		t.Errorf("Unexpected result. Expected: %v, but got: %v", expected, result)
 	}
 }
+
+// goldenDotlessMergeStmt is the exact statement the pre-QualifiedTable generator
+// (merge-base 0709374b) produced for destination "public.myTable" with the schema in
+// TestGenerateMergeStmtGolden; the refactored generator must reproduce it byte-for-byte
+// so existing mirrors see unchanged SQL.
+//
+//nolint:lll
+const goldenDotlessMergeStmt = `MERGE INTO "PUBLIC"."myTable" TARGET USING (WITH VARIANT_CONVERTED AS (
+		SELECT _PEERDB_UID,_PEERDB_TIMESTAMP,TO_VARIANT(PARSE_JSON(_PEERDB_DATA)) VAR_COLS,_PEERDB_RECORD_TYPE,
+		 _PEERDB_MATCH_DATA,_PEERDB_BATCH_ID,_PEERDB_UNCHANGED_TOAST_COLUMNS
+		FROM _PEERDB_INTERNAL._PEERDB_RAW_my_mirror WHERE _PEERDB_BATCH_ID = 42 AND
+		 _PEERDB_DATA != '' AND
+		 _PEERDB_DESTINATION_TABLE_NAME = ? ), FLATTENED AS
+		 (SELECT _PEERDB_UID,_PEERDB_TIMESTAMP,_PEERDB_RECORD_TYPE,_PEERDB_MATCH_DATA,_PEERDB_BATCH_ID,
+			_PEERDB_UNCHANGED_TOAST_COLUMNS,CAST(VAR_COLS:"id" AS INTEGER) AS "ID",CAST(VAR_COLS:"myData" AS STRING) AS "myData"
+		 FROM VARIANT_CONVERTED), DEDUPLICATED_FLATTENED AS (SELECT _PEERDB_RANKED.* FROM
+		 (SELECT RANK() OVER
+		 (PARTITION BY ("ID") ORDER BY _PEERDB_TIMESTAMP DESC) AS _PEERDB_RANK, * FROM FLATTENED)
+		 _PEERDB_RANKED WHERE _PEERDB_RANK = 1)
+		 SELECT * FROM DEDUPLICATED_FLATTENED) SOURCE ON TARGET."ID" = SOURCE."ID"
+		 WHEN NOT MATCHED AND (SOURCE._PEERDB_RECORD_TYPE != 2) THEN INSERT ("ID","myData","_PEERDB_SYNCED_AT") VALUES(SOURCE."ID",SOURCE."myData",CURRENT_TIMESTAMP)
+		 WHEN MATCHED AND
+		(SOURCE._PEERDB_RECORD_TYPE != 2) AND _PEERDB_UNCHANGED_TOAST_COLUMNS=''
+		THEN UPDATE SET "ID" = SOURCE."ID", "myData" = SOURCE."myData", "_PEERDB_SYNCED_AT" = CURRENT_TIMESTAMP, "_PEERDB_IS_DELETED" = FALSE  WHEN MATCHED AND
+			(SOURCE._PEERDB_RECORD_TYPE = 2) AND _PEERDB_UNCHANGED_TOAST_COLUMNS=''
+			THEN UPDATE SET "ID" = SOURCE."ID", "myData" = SOURCE."myData", "_PEERDB_SYNCED_AT" = CURRENT_TIMESTAMP, "_PEERDB_IS_DELETED" = TRUE  WHEN NOT MATCHED AND (SOURCE._PEERDB_RECORD_TYPE = 2) THEN INSERT ("ID","myData","_PEERDB_SYNCED_AT",_PEERDB_IS_DELETED) VALUES(SOURCE."ID",SOURCE."myData",CURRENT_TIMESTAMP,TRUE)
+		 WHEN MATCHED AND (SOURCE._PEERDB_RECORD_TYPE = 2) THEN UPDATE SET _PEERDB_IS_DELETED = TRUE, _PEERDB_SYNCED_AT = CURRENT_TIMESTAMP`
+
+func TestGenerateMergeStmtGolden(t *testing.T) {
+	schema := &protos.TableSchema{
+		Columns: []*protos.FieldDescription{
+			{Name: "id", Type: string(types.QValueKindInt64)},
+			{Name: "myData", Type: string(types.QValueKindString)},
+		},
+		PrimaryKeyColumns: []string{"id"},
+	}
+
+	newGenerator := func(dstTable common.QualifiedTable, rawName string) *mergeStmtGenerator {
+		return &mergeStmtGenerator{
+			tableSchemaMapping:       map[common.QualifiedTable]*protos.TableSchema{dstTable: schema},
+			unchangedToastColumnsMap: map[string][]string{rawName: {""}},
+			peerdbCols: &protos.PeerDBColumns{
+				SyncedAtColName:   "_PEERDB_SYNCED_AT",
+				SoftDeleteColName: "_PEERDB_IS_DELETED",
+			},
+			rawTableName: "_PEERDB_RAW_my_mirror",
+			mergeBatchId: 42,
+		}
+	}
+
+	t.Run("dotless byte-identical to pre-refactor", func(t *testing.T) {
+		dstTable := common.QualifiedTable{Namespace: "public", Table: "myTable"}
+		stmt, err := newGenerator(dstTable, "public.myTable").generateMergeStmt(t.Context(), nil, dstTable)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if stmt != goldenDotlessMergeStmt {
+			t.Errorf("merge statement diverged from pre-QualifiedTable output:\ngot:\n%s\nwant:\n%s", stmt, goldenDotlessMergeStmt)
+		}
+	})
+
+	t.Run("dotted components quoted per-component", func(t *testing.T) {
+		dstTable := common.QualifiedTable{Namespace: "sch.ema", Table: "ta.ble"}
+		stmt, err := newGenerator(dstTable, "sch.ema.ta.ble").generateMergeStmt(t.Context(), nil, dstTable)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !strings.HasPrefix(stmt, "MERGE INTO \"SCH.EMA\".\"TA.BLE\" TARGET") {
+			t.Errorf("dotted destination not quoted per-component: %s", stmt[:80])
+		}
+		// the unchanged-toast map is keyed by the raw-table string (LegacyDotted of the
+		// destination); a quoted-String() or .Table key would miss and silently drop
+		// the toast-handling UPDATE branch from the statement
+		if !strings.Contains(stmt, "_PEERDB_UNCHANGED_TOAST_COLUMNS=''") {
+			t.Error("LegacyDotted-keyed unchangedToastColumnsMap lookup missed; toast UPDATE branch absent")
+		}
+	})
+}