ScatterExchangeStep::createSinkAndSourcePair rejected any source with more
than one bucket with a LOGICAL_ERROR. This is reachable from SQL: aggregating
over a Merge table whose children are Distributed tables crashed the server
with 'ScatterExchangeStep should have one source shard, got 8' when
make_distributed_plan=1 (issue ClickHouse#107946).

StorageMerge::ReadFromMerge builds its children at a post-FetchColumns stage
and converts each child plan to a distributed plan on its own
(ReadFromMerge::buildPipeline -> QueryPlan::buildQueryPipeline ->
convertToDistributed). The per-child plan therefore carries a ScatterExchange
above an already-bucketed distributed read, so the scatter sees several source
buckets (distributed_plan_default_reader_bucket_count, 8 by default) and the
assertion aborts the server.

Scatter is documented as a special case of Shuffle whose source bucket count is
1, and it already reuses ShuffleSend/ShuffleReceive as its sink/source. Neither
step requires a single source bucket: ShuffleSend runs per source bucket and
ShuffleReceive reads every source bucket for its destination bucket. So when the
source happens to expose several buckets, the same pair correctly repartitions
all source buckets into the requested result buckets, i.e. behaves as a plain
shuffle - which is exactly what optimizeExchanges already does when it rewrites a
Scatter over a Gather into a ShuffleExchange. Drop the assertion and let the
multi-bucket source flow through.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…on test

The CI test profile (tests/config/users.d/limits.yaml) sets max_rows_to_group_by
to 10G, and make_distributed_plan rejects aggregation with a non-zero limit
(Code 344, SUPPORT_IS_DISABLED). That fired before the query could reach the
ScatterExchangeStep path, so the Fast test for this regression failed without
exercising the fix.

clickhouse-test also randomizes prefer_localhost_replica. With it set to 0 the
Distributed engine serializes the per-child plan to the localhost replica, where
the experimental BlocksMarshalling step is not deserializable (Code 47,
UNKNOWN_IDENTIFIER). That path is unrelated to the crash, which happens earlier
while building the per-child plan.

Pin both settings so the query deterministically reaches the multi-bucket
ScatterExchangeStep. EXPLAIN distributed=1 confirms the per-child plan still
carries a ShuffleExchange over a bucketed ReadFromMergeTree (the step that threw
'should have one source shard, got 8' before the fix), so crash-path coverage is
preserved. 20/20 runs pass under full CI randomization.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Pin distributed_plan_max_rows_to_broadcast = 0 and
distributed_plan_default_reader_bucket_count = 3 so the distributed read is
neither broadcast nor single-bucket, which guarantees the per-child scatter sees
a multi-bucket source and the removed guard is actually exercised (verified: the
test fails with the pre-fix binary and passes with the fix). Add a non-empty
assertion through the same Merge over Distributed topology, filtering on the real
underlying table names that _table exposes, so the test proves rows survive the
multi-bucket scatter, not only that the no-row reproducer no longer throws.

Reword 'crashed the server' / 'survive' / 'abort' to 'raised the LOGICAL_ERROR
exception' / 'throw', since release builds throw an exception rather than crash.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>