@@ -209,6 +209,98 @@ func (s ClickHouseSuite) Test_Addition_Removal() {
209209 RequireEnvCanceled (s .t , env )
210210}
211211
212+ // Removing one of several source tables feeding a shared destination must keep
213+ // that destination's raw rows & catalog schema mapping intact, otherwise the
214+ // surviving source can no longer normalize into it. Disjoint id ranges keep the
215+ // two sources' rows distinct under the destination's ReplacingMergeTree.
216+ func (s ClickHouseSuite ) Test_Removal_Shared_Destination () {
217+ tc := NewTemporalClient (s .t )
218+
219+ srcTableA := s .attachSchemaSuffix ("test_shared_dst_a" )
220+ srcTableB := s .attachSchemaSuffix ("test_shared_dst_b" )
221+ dstTableName := "test_shared_dst_target"
222+
223+ for _ , srcTableName := range []string {srcTableA , srcTableB } {
224+ require .NoError (s .t , s .source .Exec (s .t .Context (), fmt .Sprintf (`
225+ CREATE TABLE IF NOT EXISTS %s (
226+ id INT PRIMARY KEY,
227+ "key" TEXT NOT NULL
228+ );
229+ ` , srcTableName )))
230+ }
231+
232+ connectionGen := FlowConnectionGenerationConfig {
233+ FlowJobName : s .attachSuffix ("clickhousesharedremoval" ),
234+ TableNameMapping : map [string ]string {srcTableA : dstTableName , srcTableB : dstTableName },
235+ Destination : s .Peer ().Name ,
236+ }
237+
238+ flowConnConfig := connectionGen .GenerateFlowConnectionConfigs (s )
239+ flowConnConfig .MaxBatchSize = 1
240+
241+ env := ExecutePeerflow (s .t , tc , flowConnConfig )
242+
243+ SetupCDCFlowStatusQuery (s .t , env , flowConnConfig )
244+ require .NoError (s .t , s .source .Exec (s .t .Context (),
245+ fmt .Sprintf (`INSERT INTO %s (id,"key") VALUES (1,'a1')` , srcTableA )))
246+ require .NoError (s .t , s .source .Exec (s .t .Context (),
247+ fmt .Sprintf (`INSERT INTO %s (id,"key") VALUES (101,'b1')` , srcTableB )))
248+ EnvWaitForCount (env , s , "both sources reach shared destination" , dstTableName , "id,\" key\" " , 2 )
249+
250+ SignalWorkflow (s .t .Context (), env , model .FlowSignal , model .PauseSignal )
251+ EnvWaitFor (s .t , env , 3 * time .Minute , "pausing for removing shared table" , func () bool {
252+ return env .GetFlowStatus (s .t ) == protos .FlowStatus_STATUS_PAUSED
253+ })
254+
255+ if pgconn , ok := s .source .Connector ().(* connpostgres.PostgresConnector ); ok {
256+ conn := pgconn .Conn ()
257+ _ , err := conn .Exec (s .t .Context (),
258+ fmt .Sprintf (`SELECT pg_terminate_backend(pid) FROM pg_stat_activity
259+ WHERE query LIKE '%%START_REPLICATION%%' AND query LIKE '%%%s%%' AND backend_type='walsender'` ,
260+ s .attachSuffix ("clickhousesharedremoval" )))
261+ require .NoError (s .t , err )
262+
263+ EnvWaitFor (s .t , env , 3 * time .Minute , "waiting for replication to stop" , func () bool {
264+ rows , err := conn .Query (s .t .Context (),
265+ fmt .Sprintf (`SELECT pid FROM pg_stat_activity
266+ WHERE query LIKE '%%START_REPLICATION%%' AND query LIKE '%%%s%%' AND backend_type='walsender'` ,
267+ s .attachSuffix ("clickhousesharedremoval" )))
268+ require .NoError (s .t , err )
269+ defer rows .Close ()
270+ return ! rows .Next ()
271+ })
272+ }
273+
274+ runID := EnvGetRunID (s .t , env )
275+ SignalWorkflow (s .t .Context (), env , model .CDCDynamicPropertiesSignal , & protos.CDCFlowConfigUpdate {
276+ RemovedTables : []* protos.TableMapping {{
277+ SourceTableIdentifier : srcTableA ,
278+ DestinationTableIdentifier : dstTableName ,
279+ }},
280+ })
281+
282+ EnvWaitFor (s .t , env , 4 * time .Minute , "removing shared source" , func () bool {
283+ return env .GetFlowStatus (s .t ) == protos .FlowStatus_STATUS_RUNNING
284+ })
285+ EnvWaitFor (s .t , env , time .Minute , "ContinueAsNew" , func () bool {
286+ return runID != EnvGetRunID (s .t , env )
287+ })
288+
289+ // removed source: must not replicate; surviving source: must still reach shared destination
290+ require .NoError (s .t , s .source .Exec (s .t .Context (),
291+ fmt .Sprintf (`INSERT INTO %s (id,"key") VALUES (2,'a2')` , srcTableA )))
292+ require .NoError (s .t , s .source .Exec (s .t .Context (),
293+ fmt .Sprintf (`INSERT INTO %s (id,"key") VALUES (102,'b2')` , srcTableB )))
294+
295+ EnvWaitForCount (env , s , "surviving source still reaches shared destination" , dstTableName , "id,\" key\" " , 3 )
296+
297+ rows , err := s .GetRows (dstTableName , "id" )
298+ require .NoError (s .t , err )
299+ require .Len (s .t , rows .Records , 3 , "removed source must not leak into shared destination" )
300+ env .Cancel (s .t .Context ())
301+ RequireEnvCanceled (s .t , env )
302+ }
303+
212304func (s ClickHouseSuite ) Test_NullableMirrorSetting () {
213305 srcTableName := "test_nullable_mirror"
214306 srcFullName := s .attachSchemaSuffix (srcTableName )
0 commit comments