Edit Mirror: don't strip shared destination on table removal (#4414)

Multiple source tables mapping to one destination works by keying on
source. In raw table we only have destination table. Removing one such
source previously cleaned based on destination, breaking surviving
source

cdc_flow: compute exclusivelyRemovedTables, removed mappings whose
destination is not fed by any surviving mapping, & scope cleanup to
those. Publication alteration keeps full RemovedTables set: keyed on
source, so every removed source must leave regardless of shared
destination

RemoveTablesFromRawTable: sort+compact destination names & skip when
none remain or syncBatchID <= normBatchID. The DELETE predicate is
_peerdb_batch_id in (normBatchID, syncBatchID]; an empty range makes it
a no-op, so skipping avoids connecting to destination for nothing

clickhouse: delete in 100 table batches to group mutations

Adds e2e Test_Removal_Shared_Destination covering surviving-source
replication into a shared destination after removal. This also adds
coverage for N:1 mirrors

Author: serprex

flow/activities/flowable.go

@@ -1876,6 +1876,19 @@ func (a *FlowableActivity) RemoveTablesFromRawTable(
 		return a.Alerter.LogFlowError(ctx, cfg.FlowJobName, err)
 	}
 
+	tableNames := make([]string, 0, len(tablesToRemove))
+	for _, table := range tablesToRemove {
+		tableNames = append(tableNames, table.DestinationTableIdentifier)
+	}
+	slices.Sort(tableNames)
+	tableNames = slices.Compact(tableNames)
+	if len(tableNames) == 0 || syncBatchID <= normBatchID {
+		logger.Info("[RemoveTablesFromRawTable] no pending raw rows to remove, skipping",
+			slog.Int64("syncBatchID", syncBatchID), slog.Int64("normalizeBatchID", normBatchID),
+			slog.Int("tables", len(tableNames)))
+		return nil
+	}
+
 	dstConn, dstClose, err := connectors.GetByNameAs[connectors.RawTableConnector](ctx, cfg.Env, a.CatalogPool, cfg.DestinationName)
 	if err != nil {
 		if errors.Is(err, errors.ErrUnsupported) {
@@ -1889,10 +1902,6 @@ func (a *FlowableActivity) RemoveTablesFromRawTable(
 	}
 	defer dstClose(ctx)
 
-	tableNames := make([]string, 0, len(tablesToRemove))
-	for _, table := range tablesToRemove {
-		tableNames = append(tableNames, table.DestinationTableIdentifier)
-	}
 	if err := dstConn.RemoveTableEntriesFromRawTable(ctx, &protos.RemoveTablesFromRawTableInput{
 		FlowJobName:           cfg.FlowJobName,
 		DestinationTableNames: tableNames,

flow/connectors/clickhouse/cdc.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"slices"
 	"strconv"
 	"strings"
 
@@ -410,17 +411,22 @@ func (c *ClickHouseConnector) RemoveTableEntriesFromRawTable(
 		return nil
 	}
 
-	for _, tableName := range req.DestinationTableNames {
+	// chunk to bound statement size, one mutation covers many tables instead of one mutation each
+	for chunk := range slices.Chunk(req.DestinationTableNames, 100) {
+		quoted := make([]string, len(chunk))
+		for i, tableName := range chunk {
+			quoted[i] = peerdb_clickhouse.QuoteLiteral(tableName)
+		}
 		// Better to use lightweight deletes here as the main goal is to not have
 		// rows in the table be visible by the NormalizeRecords' INSERT INTO SELECT queries
-		if err := c.execWithLogging(ctx, fmt.Sprintf("DELETE FROM %s WHERE _peerdb_destination_table_name = %s"+
+		if err := c.execWithLogging(ctx, fmt.Sprintf("DELETE FROM %s WHERE _peerdb_destination_table_name IN (%s)"+
 			" AND _peerdb_batch_id > %d AND _peerdb_batch_id <= %d",
-			c.GetRawTableName(req.FlowJobName), peerdb_clickhouse.QuoteLiteral(tableName), req.NormalizeBatchId, req.SyncBatchId),
+			c.GetRawTableName(req.FlowJobName), strings.Join(quoted, ","), req.NormalizeBatchId, req.SyncBatchId),
 		); err != nil {
-			return fmt.Errorf("unable to remove table %s from raw table: %w", tableName, err)
+			return fmt.Errorf("unable to remove tables from raw table: %w", err)
 		}
 
-		c.logger.Info("successfully removed entries for table from raw table", slog.String("table", tableName))
+		c.logger.Info("successfully removed entries for tables from raw table", slog.Any("tables", chunk))
 	}
 
 	return nil

flow/e2e/clickhouse_test.go

@@ -207,6 +207,98 @@ func (s ClickHouseSuite) Test_Addition_Removal() {
 	RequireEnvCanceled(s.t, env)
 }
 
+// Removing one of several source tables feeding a shared destination must keep
+// that destination's raw rows & catalog schema mapping intact, otherwise the
+// surviving source can no longer normalize into it. Disjoint id ranges keep the
+// two sources' rows distinct under the destination's ReplacingMergeTree.
+func (s ClickHouseSuite) Test_Removal_Shared_Destination() {
+	tc := NewTemporalClient(s.t)
+
+	srcTableA := s.attachSchemaSuffix("test_shared_dst_a")
+	srcTableB := s.attachSchemaSuffix("test_shared_dst_b")
+	dstTableName := "test_shared_dst_target"
+
+	for _, srcTableName := range []string{srcTableA, srcTableB} {
+		require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
+			CREATE TABLE IF NOT EXISTS %s (
+				id INT PRIMARY KEY,
+				"key" TEXT NOT NULL
+			);
+		`, srcTableName)))
+	}
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:      s.attachSuffix("clickhousesharedremoval"),
+		TableNameMapping: map[string]string{srcTableA: dstTableName, srcTableB: dstTableName},
+		Destination:      s.Peer().Name,
+	}
+
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.MaxBatchSize = 1
+
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s (id,"key") VALUES (1,'a1')`, srcTableA)))
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s (id,"key") VALUES (101,'b1')`, srcTableB)))
+	EnvWaitForCount(env, s, "both sources reach shared destination", dstTableName, "id,\"key\"", 2)
+
+	SignalWorkflow(s.t.Context(), env, model.FlowSignal, model.PauseSignal)
+	EnvWaitFor(s.t, env, 3*time.Minute, "pausing for removing shared table", func() bool {
+		return env.GetFlowStatus(s.t) == protos.FlowStatus_STATUS_PAUSED
+	})
+
+	if pgconn, ok := s.source.Connector().(*connpostgres.PostgresConnector); ok {
+		conn := pgconn.Conn()
+		_, err := conn.Exec(s.t.Context(),
+			fmt.Sprintf(`SELECT pg_terminate_backend(pid) FROM pg_stat_activity
+				WHERE query LIKE '%%START_REPLICATION%%' AND query LIKE '%%%s%%' AND backend_type='walsender'`,
+				s.attachSuffix("clickhousesharedremoval")))
+		require.NoError(s.t, err)
+
+		EnvWaitFor(s.t, env, 3*time.Minute, "waiting for replication to stop", func() bool {
+			rows, err := conn.Query(s.t.Context(),
+				fmt.Sprintf(`SELECT pid FROM pg_stat_activity
+					WHERE query LIKE '%%START_REPLICATION%%' AND query LIKE '%%%s%%' AND backend_type='walsender'`,
+					s.attachSuffix("clickhousesharedremoval")))
+			require.NoError(s.t, err)
+			defer rows.Close()
+			return !rows.Next()
+		})
+	}
+
+	runID := EnvGetRunID(s.t, env)
+	SignalWorkflow(s.t.Context(), env, model.CDCDynamicPropertiesSignal, &protos.CDCFlowConfigUpdate{
+		RemovedTables: []*protos.TableMapping{{
+			SourceTableIdentifier:      srcTableA,
+			DestinationTableIdentifier: dstTableName,
+		}},
+	})
+
+	EnvWaitFor(s.t, env, 4*time.Minute, "removing shared source", func() bool {
+		return env.GetFlowStatus(s.t) == protos.FlowStatus_STATUS_RUNNING
+	})
+	EnvWaitFor(s.t, env, time.Minute, "ContinueAsNew", func() bool {
+		return runID != EnvGetRunID(s.t, env)
+	})
+
+	// removed source: must not replicate; surviving source: must still reach shared destination
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s (id,"key") VALUES (2,'a2')`, srcTableA)))
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s (id,"key") VALUES (102,'b2')`, srcTableB)))
+
+	EnvWaitForCount(env, s, "surviving source still reaches shared destination", dstTableName, "id,\"key\"", 3)
+
+	rows, err := s.GetRows(dstTableName, "id")
+	require.NoError(s.t, err)
+	require.Len(s.t, rows.Records, 3, "removed source must not leak into shared destination")
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
 func (s ClickHouseSuite) Test_NullableMirrorSetting() {
 	srcTableName := "test_nullable_mirror"
 	srcFullName := s.attachSchemaSuffix(srcTableName)

flow/workflows/cdc_flow.go

@@ -382,10 +382,24 @@ func processTableRemovals(
 		}
 		logger.Info("tables removed from publication")
 
+		// destinations still fed by remaining mappings keep raw rows & schema mapping,
+		// pending rows still normalize into surviving destination table
+		// relies on TableMappings being trimmed before selector callbacks fire
+		remainingDestinations := make(map[string]struct{}, len(state.SyncFlowOptions.TableMappings))
+		for _, tm := range state.SyncFlowOptions.TableMappings {
+			remainingDestinations[tm.DestinationTableIdentifier] = struct{}{}
+		}
+		exclusivelyRemovedTables := make([]*protos.TableMapping, 0, len(state.FlowConfigUpdate.RemovedTables))
+		for _, tm := range state.FlowConfigUpdate.RemovedTables {
+			if _, shared := remainingDestinations[tm.DestinationTableIdentifier]; !shared {
+				exclusivelyRemovedTables = append(exclusivelyRemovedTables, tm)
+			}
+		}
+
 		rawTableCleanupFuture := workflow.ExecuteActivity(
 			removeTablesCtx,
 			flowable.RemoveTablesFromRawTable,
-			cfg, state.FlowConfigUpdate.RemovedTables)
+			cfg, exclusivelyRemovedTables)
 		removeTablesSelector.AddFuture(rawTableCleanupFuture, func(f workflow.Future) {
 			if err := f.Get(ctx, nil); err != nil {
 				logger.Error("failed to clean up raw table for removed tables", slog.Any("error", err))
@@ -397,7 +411,7 @@ func processTableRemovals(
 			removeTablesFromCatalogFuture := workflow.ExecuteActivity(
 				removeTablesCtx,
 				flowable.RemoveTablesFromCatalog,
-				cfg, state.FlowConfigUpdate.RemovedTables)
+				cfg, exclusivelyRemovedTables)
 			removeTablesSelector.AddFuture(removeTablesFromCatalogFuture, func(f workflow.Future) {
 				if err := f.Get(ctx, nil); err != nil {
 					logger.Error("failed to clean up raw table for removed tables", slog.Any("error", err))

ui/app/mirrors/create/schema.ts

@@ -26,18 +26,7 @@ export const tableMappingSchema = z
       partitionKey: z.string().optional(),
     })
   )
-  .nonempty('At least one table mapping is required')
-  .superRefine((mappingArray, ctx) => {
-    if (
-      mappingArray.map((val) => val.destinationTableIdentifier).length !==
-      new Set(mappingArray).size
-    ) {
-      ctx.addIssue({
-        code: 'custom',
-        message: `Two source tables have been mapped to the same destination table`,
-      });
-    }
-  });
+  .nonempty('At least one table mapping is required');
 
 export const cdcSchema = z.object({
   sourceName: z.string({ error: 'Source peer is required' }).min(1),