fix(clickhouse): drop orphaned shard table after mirror resync exchange (#4490)

## Problem

After a mirror resync on a clustered ClickHouse destination, the old
local shard table (e.g. `itunes_apps_shard`) is left behind as an
orphan.

During resync, `RenameTables` does:
1. `EXCHANGE TABLES itunes_apps AND itunes_apps_resync` — swaps the two
Distributed tables
2. `DROP TABLE itunes_apps_resync` — drops the now-old Distributed shell

But this only handles the Distributed layer. The local shard that the
original `itunes_apps` pointed to (`itunes_apps_shard`) has no
Distributed table referencing it anymore and is never dropped, leaving
it as dead storage.

## Fix

Before the `EXCHANGE`, resolve the shard table name from the original
Distributed table's `engine_full` definition using the existing
`getDistributedShardTable` helper. After the exchange and Distributed
drop succeed, explicitly drop that shard table.

If the shard lookup fails (e.g. non-Distributed engine, single-node
deployment), we log a warning and continue — the swap itself is not
blocked.

## Test plan

- [ ] Run a mirror resync on a clustered ClickHouse destination
- [ ] Confirm `itunes_apps_shard` (or equivalent old shard) is gone
after resync completes
- [ ] Confirm `itunes_apps` Distributed table still points to the new
timestamped shard and data is intact

Author: andreyzhelnin-st

flow/connectors/clickhouse/cdc.go

@@ -331,6 +331,18 @@ func (c *ClickHouseConnector) RenameTables(
 		}
 
 		if originalTableExists {
+			// Before swapping, resolve the shard table that the original Distributed table currently points to.
+			// After the EXCHANGE the Distributed table is dropped, leaving that shard orphaned unless we drop it explicitly.
+			var originalShardTable string
+			if c.Config.Cluster != "" {
+				originalShardTable, err = c.getDistributedShardTable(ctx, renameRequest.NewName)
+				if err != nil {
+					c.logger.Warn("could not resolve shard table for original distributed table, old shard may be left behind",
+						slog.String("table", renameRequest.NewName), slog.Any("error", err))
+					originalShardTable = ""
+				}
+			}
+
 			// target table exists, so we can attempt to swap. In most cases, we will have Atomic engine,
 			// which supports a special query to exchange two tables, allowing dependent (materialized) views and dictionaries on these tables
 			c.logger.Info("attempting atomic exchange",
@@ -344,6 +356,15 @@ func (c *ClickHouseConnector) RenameTables(
 				); err != nil {
 					return nil, fmt.Errorf("unable to drop exchanged table %s: %w", renameRequest.CurrentName, err)
 				}
+				// Drop the original shard table that is now orphaned after the Distributed table was exchanged and dropped.
+				if originalShardTable != "" {
+					if err := c.execWithLogging(ctx,
+						fmt.Sprintf(dropTableSQLWithCHSetting, peerdb_clickhouse.QuoteIdentifier(originalShardTable), onCluster),
+					); err != nil {
+						return nil, fmt.Errorf("unable to drop orphaned shard table %s: %w", originalShardTable, err)
+					}
+					c.logger.Info("dropped orphaned shard table after resync exchange", slog.String("shardTable", originalShardTable))
+				}
 			} else if ex, ok := err.(*clickhouse.Exception); !ok || chproto.Error(ex.Code) != chproto.ErrNotImplemented {
 				// move on to the fallback code if unimplemented, in all other error codes / types return,
 				// since we know/assume exchange would be the sensible action