fix: Add tests for non-pk table cdc works in append-only mode with Merge Tree

flow/connectors/clickhouse/validate.go

@@ -120,6 +120,15 @@ func (c *ClickHouseConnector) ValidateMirrorDestination(
 		); err != nil {
 			return err
 		}
+		if err := chvalidate.ValidateClusterShardingKey(
+			c.Config.Cluster,
+			tableMapping.ShardingKey,
+			tableMapping.SourceTableIdentifier,
+			len(processedSchema.PrimaryKeyColumns) > 0,
+			sortingKeys,
+		); err != nil {
+			return err
+		}
 
 		// if destination table does not exist, we're good
 		if _, ok := chTableColumnsMapping[dstTableName]; !ok {

flow/e2e/clickhouse.go

@@ -379,6 +379,21 @@ func (s ClickHouseSuite) GetRows(table string, cols string) (*model.QRecordBatch
 	return batch, rows.Err()
 }
 
+// CountNonDeletedRows returns the number of rows in table where _peerdb_is_deleted = 0.
+// Unlike GetRows it does not use FINAL, so it works for engines that reject FINAL (e.g. MergeTree).
+func (s ClickHouseSuite) CountNonDeletedRows(table string) (int, error) {
+	ch, err := connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig())
+	if err != nil {
+		return 0, err
+	}
+	defer ch.Close()
+	var count uint64
+	err = ch.QueryRow(s.t.Context(),
+		fmt.Sprintf(`SELECT count() FROM "%s" WHERE _peerdb_is_deleted = 0 SETTINGS use_query_cache = false`, table),
+	).Scan(&count)
+	return int(count), err
+}
+
 func (s ClickHouseSuite) queryRawTable(conn clickhouse.Conn, table string, cols string) (driver.Rows, error) {
 	return conn.Query(
 		s.t.Context(),

flow/e2e/clickhouse_mysql_test.go

@@ -2031,3 +2031,84 @@ func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_Arbitrary_FullTable() {
 	env.Cancel(s.t.Context())
 	RequireEnvCanceled(s.t, env)
 }
+
+func (s ClickHouseSuite) Test_MySQL_NoPrimaryKey_CDC() {
+	if _, ok := s.source.(*MySqlSource); !ok {
+		s.t.Skip("only applies to mysql")
+	}
+
+	var shardingKey string
+	if s.cluster {
+		shardingKey = "rand()"
+	}
+
+	srcTableName := "test_no_pkey_cdc"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	dstTableName := "test_no_pkey_cdc"
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (id INT, val TEXT)`, srcFullName)))
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s VALUES (1, 'a'), (2, 'b')`, srcFullName)))
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName: s.attachSuffix("mysql_no_pkey_cdc"),
+		TableMappings: []*protos.TableMapping{{
+			SourceTableIdentifier:      srcFullName,
+			DestinationTableIdentifier: dstTableName,
+			Engine:                     protos.TableEngine_CH_ENGINE_MERGE_TREE,
+			// rand() is required for cluster mode: Distributed tables with multiple shards
+			// need a sharding expression. No natural key exists on a PK-less table.
+			ShardingKey: shardingKey,
+		}},
+		Destination: s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	// waitForNonDeletedCount polls CH without FINAL (MergeTree rejects FINAL).
+	waitForNonDeletedCount := func(expected int, reason string) {
+		s.t.Helper()
+		EnvWaitFor(s.t, env, 3*time.Minute, reason, func() bool {
+			s.t.Helper()
+			n, err := s.CountNonDeletedRows(dstTableName)
+			if err != nil {
+				s.t.Log(err)
+				return false
+			}
+			return n == expected
+		})
+	}
+
+	// Snapshot: 2 rows expected in CH.
+	waitForNonDeletedCount(2, "waiting on snapshot")
+
+	// CDC INSERT: 3 rows in CH.
+	EnvNoError(s.t, env, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s VALUES (3, 'c')`, srcFullName)))
+	waitForNonDeletedCount(3, "waiting on cdc insert")
+
+	// CDC UPDATE: MergeTree appends the updated row without removing the original.
+	// Source has 3 rows; CH now has 4 non-deleted rows (row-1-original, row-1-updated, row-2, row-3).
+	// The count growing from 3→4 documents the append-only semantics: no dedup without an ordering key.
+	EnvNoError(s.t, env, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`UPDATE %s SET val='updated' WHERE id=1`, srcFullName)))
+	waitForNonDeletedCount(4, "waiting on cdc update")
+
+	// CDC DELETE + subsequent INSERT: the DELETE must not crash the pipeline. A delete-marker row
+	// (is_deleted=1) is appended to CH but the original row-2 (is_deleted=0) stays visible.
+	// The following INSERT (id=4) arriving at count=5 confirms the pipeline survived the DELETE.
+	// count=5: row1_orig, row1_updated, row2_orig (is_deleted=0 retained), row3, row4
+	EnvNoError(s.t, env, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`DELETE FROM %s WHERE id=2`, srcFullName)))
+	EnvNoError(s.t, env, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s VALUES (4, 'd')`, srcFullName)))
+	waitForNonDeletedCount(5, "waiting on cdc after delete")
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}

flow/pkg/clickhouse/validation.go

@@ -303,3 +303,23 @@ func ValidateOrderingKeys(ctx context.Context, logger log.Logger, conn clickhous
 		sourceTable,
 	)
 }
+
+// ValidateClusterShardingKey returns an error when the destination is a multi-shard cluster,
+// the source table has no primary key and no custom ordering, and no explicit sharding_key is
+// provided. In that configuration ClickHouse would reject writes to the Distributed table
+// (error 55: "Method write is not supported by storage Distributed with more than one shard
+// and no sharding key provided").
+func ValidateClusterShardingKey(cluster, shardingKey, sourceTable string, hasPrimaryKeys bool, sortingKeys []string) error {
+	if cluster == "" {
+		return nil
+	}
+	if shardingKey != "" || hasPrimaryKeys || len(sortingKeys) > 0 {
+		return nil
+	}
+	return fmt.Errorf(
+		"table %q has no primary key and no custom ordering columns; "+
+			"an explicit sharding_key is required for cluster deployments "+
+			"(e.g. sharding_key: rand() for random distribution across shards)",
+		sourceTable,
+	)
+}

flow/pkg/clickhouse/validation_test.go

@@ -162,3 +162,55 @@ func TestCheckIfTablesEmptyAndEngine(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateClusterShardingKey(t *testing.T) {
+	tests := []struct {
+		name           string
+		cluster        string
+		shardingKey    string
+		sourceTable    string
+		hasPrimaryKeys bool
+		sortingKeys    []string
+		wantErr        string
+	}{
+		{
+			name:        "non-cluster mode always passes",
+			cluster:     "",
+			sourceTable: "db.no_pk",
+		},
+		{
+			name:        "cluster with explicit sharding key passes",
+			cluster:     "cicluster",
+			shardingKey: "rand()",
+			sourceTable: "db.no_pk",
+		},
+		{
+			name:           "cluster with primary keys passes",
+			cluster:        "cicluster",
+			sourceTable:    "db.has_pk",
+			hasPrimaryKeys: true,
+		},
+		{
+			name:        "cluster with custom sorting columns passes",
+			cluster:     "cicluster",
+			sourceTable: "db.no_pk",
+			sortingKeys: []string{"created_at"},
+		},
+		{
+			name:        "cluster, no pk, no sharding key, no sorting → error",
+			cluster:     "cicluster",
+			sourceTable: "db.no_pk",
+			wantErr:     "sharding_key",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := ValidateClusterShardingKey(tt.cluster, tt.shardingKey, tt.sourceTable, tt.hasPrimaryKeys, tt.sortingKeys)
+			if tt.wantErr != "" {
+				require.ErrorContains(t, err, tt.wantErr)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}