Fix adversarial review findings

- Table-only destinations (ClickHouse, Elasticsearch) derive physical names via
  LegacyDotted so legacy dotted destinations keep targeting the same table/index
- Add-tables signal path validates additions against the flow's existing tables;
  structural identifier validation runs even with skip_validation; source dotted
  collisions rejected (duplicate clone workflow IDs)
- Missing entry normalization added: RemoveTablesFromRawTable, QRepHasNewRows,
  SetupQRepMetadataTables activities and QRepWaitForNewRowsWorkflow
- Legacy fallbacks when consuming activity outputs recorded by old releases
- Continue-As-New payloads denormalized for rollback safety (continue_as_new.go)
- InitialLoadSummary recomposes LegacyDotted names from qrep_runs split columns
- CancelTableAddition response denormalized; telemetry reads structs;
  reset_sequences handles empty-namespace destinations; ScopedEventhub rejects
  extra dots; QRep watermark validation; UI QRep destination re-parsed at submit;
  UI removals reuse canonical mappings; delete ParseTableIdentifier (plan 1.3)

Author: ilidemi

flow/activities/flowable.go

@@ -546,6 +546,7 @@ func (a *FlowableActivity) pullAndSyncPg(
 
 // SetupQRepMetadataTables sets up the metadata tables for QReplication.
 func (a *FlowableActivity) SetupQRepMetadataTables(ctx context.Context, config *protos.QRepConfig) error {
+	internal.NormalizeQRepConfig(config)
 	conn, connClose, err := connectors.GetByNameAs[connectors.QRepSyncConnector](ctx, config.Env, a.CatalogPool, config.DestinationName)
 	if err != nil {
 		return a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to get connector: %w", err))
@@ -1597,6 +1598,8 @@ var activeFlowStatuses = map[protos.FlowStatus]struct{}{
 func (a *FlowableActivity) QRepHasNewRows(ctx context.Context,
 	config *protos.QRepConfig, last *protos.QRepPartition,
 ) (bool, error) {
+	internal.NormalizeQRepConfig(config)
+	internal.NormalizeQRepPartition(last)
 	shutdown := common.HeartbeatRoutine(ctx, func() string {
 		return "scanning for new rows"
 	})
@@ -1893,6 +1896,7 @@ func (a *FlowableActivity) RemoveTablesFromRawTable(
 	cfg *protos.FlowConnectionConfigsCore,
 	tablesToRemove []*protos.TableMapping,
 ) error {
+	internal.NormalizeTableMappings(tablesToRemove)
 	shutdown := common.HeartbeatRoutine(ctx, func() string {
 		return "removing tables from raw table"
 	})

flow/cmd/cancel_table_addition.go

@@ -60,5 +60,8 @@ func (h *FlowRequestHandler) CancelTableAddition(
 		slog.String("flowJobName", req.FlowJobName),
 		slog.String("workflowID", workflowID))
 
+	// populate legacy string identifiers for clients that predate QualifiedTable
+	internal.DenormalizeTableMappings(output.GetTablesAfterCancellation())
+
 	return output, nil
 }

flow/cmd/handler.go

@@ -218,6 +218,11 @@ func (h *FlowRequestHandler) CreateCDCFlow(
 			slog.ErrorContext(ctx, "validate mirror error", slog.Any("error", err))
 			return nil, NewInternalApiError(fmt.Errorf("invalid mirror: %w", err))
 		}
+	} else if err := validateTableMappingIdentifiers(connectionConfigsCore.TableMappings); err != nil {
+		// SkipValidation skips environmental checks only; structural identifier
+		// invariants (duplicates, dotted-name collisions, empty components) are
+		// relied on by the raw-table round-trip and must always hold
+		return nil, NewInvalidArgumentApiError(err)
 	}
 
 	if resp, err := h.createCDCFlow(ctx, connectionConfigsCore, workflowID); err != nil {
@@ -467,6 +472,20 @@ func (h *FlowRequestHandler) FlowStateChange(
 		if err := validateTableMappingIdentifiers(cdcUpdate.AdditionalTables); err != nil {
 			return nil, NewInvalidArgumentApiError(err)
 		}
+		if len(cdcUpdate.AdditionalTables) > 0 {
+			// added tables must not duplicate or dotted-collide with the flow's
+			// EXISTING tables either, not just within the additions batch
+			config, err := h.getFlowConfigFromCatalog(ctx, req.FlowJobName)
+			if err != nil {
+				return nil, NewInternalApiError(fmt.Errorf("unable to load flow config to validate additional tables: %w", err))
+			}
+			combined := make([]*protos.TableMapping, 0, len(config.TableMappings)+len(cdcUpdate.AdditionalTables))
+			combined = append(combined, config.TableMappings...)
+			combined = append(combined, cdcUpdate.AdditionalTables...)
+			if err := validateTableMappingIdentifiers(combined); err != nil {
+				return nil, NewInvalidArgumentApiError(fmt.Errorf("additional tables conflict with existing tables: %w", err))
+			}
+		}
 	}
 
 	if req.FlowConfigUpdate != nil && req.FlowConfigUpdate.GetCdcFlowConfigUpdate() != nil &&

flow/cmd/mirror_status.go

@@ -18,6 +18,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/connectors"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 	"github.com/PeerDB-io/peerdb/flow/workflows/cdc_state"
@@ -300,7 +301,9 @@ func (h *FlowRequestHandler) InitialLoadSummary(
 	q := `
 	SELECT
 		distinct qr.flow_name,
+		qr.destination_table_namespace,
 		qr.destination_table,
+		qr.source_table_namespace,
 		qr.source_table,
 		qr.start_time AS StartTime,
 		qr.fetch_complete as FetchCompleted,
@@ -312,10 +315,13 @@ func (h *FlowRequestHandler) InitialLoadSummary(
 	FROM peerdb_stats.qrep_partitions qp
 	RIGHT JOIN peerdb_stats.qrep_runs qr ON qp.flow_name = qr.flow_name
 	WHERE qr.parent_mirror_name = $1
-	GROUP BY qr.flow_name, qr.destination_table, qr.source_table, qr.start_time, qr.fetch_complete, qr.consolidate_complete;
+	GROUP BY qr.flow_name, qr.destination_table_namespace, qr.destination_table,
+		qr.source_table_namespace, qr.source_table, qr.start_time, qr.fetch_complete, qr.consolidate_complete;
 	`
 	var flowName pgtype.Text
+	var destinationTableNamespace pgtype.Text
 	var destinationTable pgtype.Text
+	var sourceTableNamespace pgtype.Text
 	var sourceTable pgtype.Text
 	var fetchCompleted pgtype.Bool
 	var consolidateCompleted pgtype.Bool
@@ -338,7 +344,9 @@ func (h *FlowRequestHandler) InitialLoadSummary(
 	for rows.Next() {
 		if err := rows.Scan(
 			&flowName,
+			&destinationTableNamespace,
 			&destinationTable,
+			&sourceTableNamespace,
 			&sourceTable,
 			&startTime,
 			&fetchCompleted,
@@ -360,11 +368,11 @@ func (h *FlowRequestHandler) InitialLoadSummary(
 		}
 
 		if destinationTable.Valid {
-			res.TableName = destinationTable.String
+			res.TableName = common.QualifiedTable{Namespace: destinationTableNamespace.String, Table: destinationTable.String}.LegacyDotted()
 		}
 
 		if sourceTable.Valid {
-			res.SourceTable = sourceTable.String
+			res.SourceTable = common.QualifiedTable{Namespace: sourceTableNamespace.String, Table: sourceTable.String}.LegacyDotted()
 		}
 
 		if startTime.Valid {

flow/cmd/reset_sequences.go

@@ -10,6 +10,7 @@ import (
 	connpostgres "github.com/PeerDB-io/peerdb/flow/connectors/postgres"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 )
 
 func (h *FlowRequestHandler) ResetMirrorSequences(
@@ -48,7 +49,13 @@ func (h *FlowRequestHandler) ResetMirrorSequences(
 	for _, tm := range config.TableMappings {
 		// quoted form: pg_get_serial_sequence parses its argument as a possibly-quoted
 		// qualified name, so this stays correct for names containing dots
-		destTables = append(destTables, internal.QualifiedTableFromProto(tm.DestinationTable).String())
+		destTable := internal.QualifiedTableFromProto(tm.DestinationTable)
+		if destTable.Namespace == "" {
+			// unqualified names resolve via search_path; `""."t"` would be invalid
+			destTables = append(destTables, common.QuoteIdentifier(destTable.Table))
+		} else {
+			destTables = append(destTables, destTable.String())
+		}
 	}
 
 	quotedTables := make([]string, 0, len(destTables))

flow/cmd/validate_mirror.go

@@ -38,6 +38,9 @@ func validateQRepIdentifiers(cfg *protos.QRepConfig) error {
 	if cfg.DestinationTable.GetTable() == "" {
 		return errors.New("destination table name is empty")
 	}
+	if cfg.WatermarkColumn != "" && cfg.QualifiedWatermarkTable.GetTable() == "" {
+		return errors.New("watermark table is required when a watermark column is configured")
+	}
 	return nil
 }
 
@@ -174,11 +177,13 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl(
 }
 
 // validateTableMappingIdentifiers rejects duplicate sources/destinations, empty table
-// components, and destination pairs whose LegacyDotted renderings collide (e.g.
-// {"a","b.c"} vs {"a.b","c"}) — those would merge in the raw table, which stores the
-// dotted format.
+// components, and source/destination pairs whose LegacyDotted renderings collide (e.g.
+// {"a","b.c"} vs {"a.b","c"}) — colliding destinations would merge in the raw table,
+// which stores the dotted format, and colliding sources would produce duplicate
+// snapshot clone child-workflow IDs.
 func validateTableMappingIdentifiers(tableMappings []*protos.TableMapping) error {
 	sources := make(map[common.QualifiedTable]struct{}, len(tableMappings))
+	sourcesDotted := make(map[string]common.QualifiedTable, len(tableMappings))
 	destinations := make(map[common.QualifiedTable]struct{}, len(tableMappings))
 	destinationsDotted := make(map[string]common.QualifiedTable, len(tableMappings))
 	for _, tm := range tableMappings {
@@ -197,6 +202,11 @@ func validateTableMappingIdentifiers(tableMappings []*protos.TableMapping) error
 			return fmt.Errorf("duplicate source table %s", source)
 		}
 		sources[source] = struct{}{}
+		if other, ok := sourcesDotted[source.LegacyDotted()]; ok {
+			return fmt.Errorf("source tables %s and %s are ambiguous with each other due to dots in names",
+				source, other)
+		}
+		sourcesDotted[source.LegacyDotted()] = source
 		if _, ok := destinations[destination]; ok {
 			return fmt.Errorf("duplicate destination table %s", destination)
 		}

flow/connectors/clickhouse/avro_sync.go

@@ -50,7 +50,7 @@ func (s *ClickHouseAvroSyncMethod) CopyStageToDestination(ctx context.Context, a
 	}
 
 	query := fmt.Sprintf("INSERT INTO %s SELECT * FROM %s",
-		peerdb_clickhouse.QuoteIdentifier(s.config.DestinationTable.GetTable()), stagingTableFunction)
+		peerdb_clickhouse.QuoteIdentifier(internal.QualifiedTableFromProto(s.config.DestinationTable).LegacyDotted()), stagingTableFunction)
 	return s.exec(ctx, query)
 }
 
@@ -61,7 +61,7 @@ func (s *ClickHouseAvroSyncMethod) SyncRecords(
 	flowJobName string,
 	syncBatchID int64,
 ) (int64, error) {
-	dstTableName := internal.QualifiedTableFromProto(s.config.DestinationTable).Table
+	dstTableName := internal.QualifiedTableFromProto(s.config.DestinationTable).LegacyDotted()
 
 	schema, err := stream.Schema()
 	if err != nil {
@@ -101,7 +101,7 @@ func (s *ClickHouseAvroSyncMethod) SyncQRepRecords(
 	stream *model.QRecordStream,
 ) (int64, shared.QRepWarnings, error) {
 	destinationTable := internal.QualifiedTableFromProto(config.DestinationTable)
-	dstTableName := destinationTable.Table
+	dstTableName := destinationTable.LegacyDotted()
 	startTime := time.Now()
 	schema, err := stream.Schema()
 	if err != nil {
@@ -244,7 +244,7 @@ func (s *ClickHouseAvroSyncMethod) pushStagingDataToClickHouseForSnapshot(
 	config *protos.QRepConfig,
 ) error {
 	insertConfig := &insertFromTableFunctionConfig{
-		destinationTable: config.DestinationTable.GetTable(),
+		destinationTable: internal.QualifiedTableFromProto(config.DestinationTable).LegacyDotted(),
 		schema:           schema,
 		columnNameMap:    columnNameAvroFieldMap,
 		excludedColumns:  config.Exclude,
@@ -318,7 +318,8 @@ func (s *ClickHouseAvroSyncMethod) pushStagingDataToClickHouseForSnapshot(
 					slog.Uint64("numParts", numParts),
 					slog.Int("chunkIdx", chunkIdx),
 					slog.Any("error", err))
-				return exceptions.NewClickHouseQRepSyncError(err, config.DestinationTable.GetTable(), s.ClickHouseConnector.Config.Database)
+				return exceptions.NewClickHouseQRepSyncError(
+					err, internal.QualifiedTableFromProto(config.DestinationTable).LegacyDotted(), s.ClickHouseConnector.Config.Database)
 			}
 
 			s.logger.Info("inserted part",

flow/connectors/clickhouse/cdc.go

@@ -261,7 +261,7 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas(
 		var shardTableName string
 		if c.Config.Cluster != "" && (tm == nil || tm.Engine != protos.TableEngine_CH_ENGINE_NULL) {
 			var err error
-			shardTableName, err = c.getDistributedShardTable(ctx, dstTable.Table)
+			shardTableName, err = c.getDistributedShardTable(ctx, dstTable.LegacyDotted())
 			if err != nil {
 				return fmt.Errorf("failed to resolve shard table for %s: %w", dstTable, err)
 			}
@@ -288,7 +288,7 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas(
 
 			if err := c.execWithLogging(ctx,
 				fmt.Sprintf("ALTER TABLE %s%s ADD COLUMN IF NOT EXISTS %s %s",
-					peerdb_clickhouse.QuoteIdentifier(dstTable.Table), onCluster,
+					peerdb_clickhouse.QuoteIdentifier(dstTable.LegacyDotted()), onCluster,
 					peerdb_clickhouse.QuoteIdentifier(addedColumn.Name), clickHouseColType),
 			); err != nil {
 				return fmt.Errorf("failed to add column %s for table %s: %w", addedColumn.Name, dstTable, err)
@@ -321,7 +321,7 @@ func (c *ClickHouseConnector) RenameTables(
 			continue
 		}
 
-		resyncTableExists, err := c.checkIfTableExists(ctx, c.Config.Database, currentTable.Table)
+		resyncTableExists, err := c.checkIfTableExists(ctx, c.Config.Database, currentTable.LegacyDotted())
 		if err != nil {
 			return nil, fmt.Errorf("unable to check if resync table %s exists: %w", currentTable, err)
 		}
@@ -331,7 +331,7 @@ func (c *ClickHouseConnector) RenameTables(
 			continue
 		}
 
-		originalTableExists, err := c.checkIfTableExists(ctx, c.Config.Database, newTable.Table)
+		originalTableExists, err := c.checkIfTableExists(ctx, c.Config.Database, newTable.LegacyDotted())
 		if err != nil {
 			return nil, fmt.Errorf("unable to check if table %s exists: %w", newTable, err)
 		}
@@ -342,11 +342,11 @@ func (c *ClickHouseConnector) RenameTables(
 			c.logger.Info("attempting atomic exchange",
 				slog.String("OldName", currentTable.String()), slog.String("NewName", newTable.String()))
 			if err = c.execWithLogging(ctx,
-				fmt.Sprintf("EXCHANGE TABLES %s and %s%s", peerdb_clickhouse.QuoteIdentifier(newTable.Table),
-					peerdb_clickhouse.QuoteIdentifier(currentTable.Table), onCluster),
+				fmt.Sprintf("EXCHANGE TABLES %s and %s%s", peerdb_clickhouse.QuoteIdentifier(newTable.LegacyDotted()),
+					peerdb_clickhouse.QuoteIdentifier(currentTable.LegacyDotted()), onCluster),
 			); err == nil {
 				if err := c.execWithLogging(ctx,
-					fmt.Sprintf(dropTableSQLWithCHSetting, peerdb_clickhouse.QuoteIdentifier(currentTable.Table), onCluster),
+					fmt.Sprintf(dropTableSQLWithCHSetting, peerdb_clickhouse.QuoteIdentifier(currentTable.LegacyDotted()), onCluster),
 				); err != nil {
 					return nil, fmt.Errorf("unable to drop exchanged table %s: %w", currentTable, err)
 				}
@@ -361,14 +361,14 @@ func (c *ClickHouseConnector) RenameTables(
 		// or err is set (in which case err comes from EXCHANGE TABLES)
 		if !originalTableExists || err != nil {
 			if err := c.execWithLogging(ctx,
-				fmt.Sprintf(dropTableSQLWithCHSetting, peerdb_clickhouse.QuoteIdentifier(newTable.Table), onCluster),
+				fmt.Sprintf(dropTableSQLWithCHSetting, peerdb_clickhouse.QuoteIdentifier(newTable.LegacyDotted()), onCluster),
 			); err != nil {
 				return nil, fmt.Errorf("unable to drop table %s: %w", newTable, err)
 			}
 
 			if err := c.execWithLogging(ctx, fmt.Sprintf("RENAME TABLE %s TO %s%s",
-				peerdb_clickhouse.QuoteIdentifier(currentTable.Table),
-				peerdb_clickhouse.QuoteIdentifier(newTable.Table), onCluster,
+				peerdb_clickhouse.QuoteIdentifier(currentTable.LegacyDotted()),
+				peerdb_clickhouse.QuoteIdentifier(newTable.LegacyDotted()), onCluster,
 			)); err != nil {
 				return nil, fmt.Errorf("unable to rename table %s to %s: %w", currentTable, newTable, err)
 			}

flow/connectors/clickhouse/normalize.go

@@ -52,7 +52,10 @@ func (c *ClickHouseConnector) SetupNormalizedTable(
 	destinationTable common.QualifiedTable,
 	sourceTableSchema *protos.TableSchema,
 ) (bool, error) {
-	tableAlreadyExists, err := c.checkIfTableExists(ctx, c.Config.Database, destinationTable.Table)
+	// LegacyDotted: ClickHouse table names may contain dots; configs persisted before
+	// the QualifiedTable refactor arrive first-dot-split, LegacyDotted reconstructs the
+	// original single-part name (equal to .Table for new configs, namespace == "")
+	tableAlreadyExists, err := c.checkIfTableExists(ctx, c.Config.Database, destinationTable.LegacyDotted())
 	if err != nil {
 		return false, fmt.Errorf("error occurred while checking if destination ClickHouse table exists: %w", err)
 	}
@@ -89,7 +92,7 @@ func (c *ClickHouseConnector) generateCreateTableSQLForNormalizedTable(
 	chVersion *chproto.Version,
 	flags []string,
 ) ([]string, error) {
-	tableIdentifier := destinationTable.Table
+	tableIdentifier := destinationTable.LegacyDotted()
 	var engine string
 	tmEngine := protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE
 

flow/connectors/clickhouse/normalize_query.go

@@ -330,7 +330,7 @@ func (t *NormalizeQueryGenerator) BuildQuery(ctx context.Context) (string, error
 	}
 
 	insertIntoSelectQuery := fmt.Sprintf("INSERT INTO %s %s %s%s",
-		peerdb_clickhouse.QuoteIdentifier(t.TableName.Table), colSelector.String(), selectQuery.String(), chSettings.String())
+		peerdb_clickhouse.QuoteIdentifier(t.TableName.LegacyDotted()), colSelector.String(), selectQuery.String(), chSettings.String())
 
 	t.Query = insertIntoSelectQuery