Fix pipeline-stuck logical error when a scatter shard finishes early

[groeneai]

Changelog category (leave one):

• Bug Fix (user-visible misbehavior in an official stable release)

Changelog entry (a user-readable short description of the changes that goes into CHANGELOG.md):

Fixed a Pipeline stuck logical error in queries that scatter data by partition, such as a window function with PARTITION BY or a join with join_algorithm = 'parallel_full_sorting_merge', when one shard's downstream finished while a block was only partially distributed.

Description

ScatterByPartitionTransform::prepare treated a finished output port as a reason to keep
waiting. In the mid-block distribution branch it asked !was_output_processed[i] && canPush(), and canPush is IS_NEEDED && !HAS_DATA, so a finished port is never
canPush.

That state is legitimate: an INNER join whose one side is exhausted is done, and
IMergingTransformBase::prepare then closes its inputs, finishing one shard's chain. If a block
is at that moment only partially distributed, the finished shard stays unprocessed and unpushable
while the other shards are already processed for that block, so prepare returns PortFull
forever, nothing is schedulable, and the executor throws at
PipelineExecutor::finalizeExecution.

work already skipped finished outputs. The bug was that prepare did not agree with it.

This adds the missing isFinished clause and returns Ready when the only unprocessed outputs
are finished, mirroring CopyTransform::prepareGenerate, which has the identical fan-out
accounting and already handles this. Both parts are needed: with the exclusion alone, can_push
is still false and prepare still returns PortFull. Fixing the transform covers both
producers, SortingStep::scatterByPartitionIfNeeded and ShuffleSendStep::updatePipeline.

Found by the AST fuzzer (STID 3833-2f20) on four unrelated PRs, e.g.
AST fuzzer (amd_debug, targeted).
The defect predates parallel_full_sorting_merge (#109005): the same accounting is present
unchanged in 25.8 and in every 26.x release branch, and a pre-#109005 commit hit the same wedge
through the window PARTITION BY path. The existing "it can deadlock" comment in
optimizeJoinByShards.cpp describes two scatters in circular wait, a different mechanism: here
a single scatter starves on one finished shard.

Deterministic repro: 10/10 abort before the fix, 10/10 pass after, and the new test aborts on an
unpatched binary. 50/50 randomized runs green.

Related: #106251 (same wedge class in the sibling BufferedShardByHashTransform)

[groeneai]

Internal second-model review (2 rounds, click to expand)

An independent reviewer and a second model reviewed this change before it was opened. Round 1
returned one substantive finding, which sent the change back for a fix; round 2 returned none.

#	Finding	Verdict	Resolution
❌	The regression test could go vacuous. It asserted only count() = 0, and because the right key is number % NULL that holds whether or not ScatterByPartitionTransform is in the pipeline. optimizeJoinByShards scatters only when both pre-join sorts are Type::Full, and 04500_parallel_full_sorting_merge_join_sorted_inputs asserts that a sorted subquery on both sides is not scattered. This test's left side is such a subquery and is scattered only because the count() projection plus the NULL key keep applyOrder from producing a FinishSorting - an incidental planner outcome, so one optimizer change could have silently stopped covering the bug.	AGREE	Fixed. The test now pins the scatter with an EXPLAIN PIPELINE assertion, in the style the parallel_full_sorting_merge tests already use, over the same query text and settings as the result query. The count of 2 is measured, and stable under the legacy analyzer, query_plan_join_shard_by_pk_ranges, query_plan_convert_join_to_in, optimize_sorting_by_input_stream_properties and max_threads = 8. The pin is not itself vacuous: the same EXPLAIN under full_sorting_merge, which must not scatter, counts 0.
⚠️	The added comment wrote work() and prepare() where the repo convention is to name a function without parentheses.	AGREE	Fixed in the comment and in the commit message. Comment-only: with comments stripped, the code is token-identical.
💡	The 200000-row fixture with max_block_size = 16 is larger than the sibling tests' numbers(4000).	DISAGREE	The shape is load-bearing: it is what leaves a block only partially distributed when one shard's merge join finishes, and it is byte-identical to the arm proven to abort 10/10 on unpatched master. Measured cost is ~3.5 s per run over 50 randomized runs (174 s total), so shrinking it would risk the wedge and save nothing.

Verified independently rather than taken from the change's own notes: the reproducer hits the
same wedge as CI, not merely a hang (ScatterByPartitionTransform in PortFull with one
Finished and three NeedData outputs, mirroring the CI digraph); the test still aborts on an
unpatched binary after gaining the EXPLAIN assertion, with the assertion printing first so it
cannot mask the arm; and both clauses of the fix are independently load-bearing (reverting only the
isFinished clause fails 10/10, and the clause without the pending-output flag fails 6/6).
Termination was checked rather than assumed, since a Ready that made no progress would busy-loop
and be worse than the original error.

Session id: cron:clickhouse-review-slot-49:20260803-210900

[groeneai]

Pre-PR validation gate (click to expand)

#	Question	Answer
a	Deterministic repro?	Yes, 10/10 on unpatched master: clickhouse local --queries-file repro.sql, where the query is SELECT count() FROM (SELECT number AS k FROM numbers(200000) ORDER BY k ASC) AS l INNER JOIN (SELECT number % NULL AS k FROM numbers(100) ORDER BY k) AS r ON l.k = r.k SETTINGS join_algorithm='parallel_full_sorting_merge', max_threads=4, max_block_size=16. The local digraph shows the same wedge as CI: the scatter in PortFull with one Finished and three NeedData outputs.
b	Root cause explained?	Yes. ScatterByPartitionTransform::prepare required !was_output_processed[i] && canPush(), and canPush is IS_NEEDED && !HAS_DATA, so a finished port never satisfies it. An INNER join with an exhausted side finishes and IMergingTransformBase::prepare closes its inputs, finishing one shard's chain. With a block only partially distributed, that shard is unprocessed and unpushable forever while the others are already processed, so prepare returns PortFull forever and the executor throws at PipelineExecutor::finalizeExecution.
c	Fix matches root cause?	Yes. work already skipped finished outputs; the fix makes prepare agree, at the point where the accounting is wrong. No widened bound, no no-random-* tag, no planner-level guard on the symptom.
d	Test intent preserved / new tests added?	Yes. No existing test modified. Added 04714_scatter_by_partition_finished_output, which asserts the query result (stable in every scheduling order) and, before it, pins that the plan really scatters, via an EXPLAIN PIPELINE count of ScatterByPartitionTransform in the style the parallel_full_sorting_merge tests already use. Both halves were checked against going blind: the test file aborts on an unpatched binary, and the same EXPLAIN under full_sorting_merge, which must not scatter, counts 0 and so fails the assertion. The measured count is stable under the legacy analyzer, query_plan_join_shard_by_pk_ranges, query_plan_convert_join_to_in, optimize_sorting_by_input_stream_properties and max_threads = 8, so no no-random-settings tag is needed.
e	Both directions demonstrated?	Yes, with per-arm Build IDs. Unpatched cd03879e...: 10/10 abort. Patched 63bd9403...: 10/10 pass, result 0, cross-checked equal against full_sorting_merge and hash. Mutation arms: reverting only the isFinished clause fails 10/10; the isFinished clause without the pending-output flag fails 6/6, so both parts are load-bearing. The abort arm was re-run after the test gained its EXPLAIN PIPELINE assertion: still RC=134 with Pipeline stuck at PipelineExecutor::finalizeExecution, and the digraph still shows the scatter in PortFull with one Finished and three NeedData outputs.
f	Fix is general across code paths?	Yes. Fixing the transform covers both producers of these scatters, SortingStep::scatterByPartitionIfNeeded (window PARTITION BY and join sharding) and ShuffleSendStep::updatePipeline, with one change. The sibling branches in the same function (the all-finished early out, the input-finished branch, work) were re-read and are already correct. The num_streams > 1 path, where a ResizeProcessor merges per-partition ports, was tested on both arms: no wedge there and no regression. BufferedShardByHashTransform is a different wedge and is deliberately not folded in here.
g	Fix generalizes across inputs (params/datatypes/wrappers)?	N/A by construction, and verified. The change touches only output-port state accounting, with no hashing and no key typing, so Nullable / LowCardinality / LC(Nullable) / Array / Map / Const are not carriers. The relevant dimensions are shard and stream counts, exercised at output_size = 4 and at num_streams 1 and above 1.
h	Backward compatible?	Yes. No setting, no format, no serialization change, so nothing is needed in SettingsChangesHistory.cpp. Purely an internal scheduling fix.
i	Invariants and contracts preserved?	Yes. The invariant is that a finished output is never awaited and counts as satisfied, which is what work and CopyTransform::prepareGenerate already assume. PortFull is still returned in exactly the case that justifies it, an output that is unprocessed, unfinished and not yet pushable. Termination was checked rather than assumed, since a Ready making no progress would be a busy loop and worse than the original error: work skips finished outputs without clearing all_outputs_processed, so the pass completes the block and the next prepare pulls fresh input. Every one of the 60+ local runs returned a row within about 2 to 15 seconds and none hung. No lock, allocation or error path added.

Session id: cron:clickhouse-impl-slot-47:20260803-193500

[groeneai]

cc @vdimir @novikd, could you review this? ScatterByPartitionTransform::prepare counted a finished output port as a reason to keep waiting, so when one shard's downstream finished while a block was only partially distributed the transform returned PortFull forever and the executor threw Pipeline stuck; work already skipped finished outputs, and this makes prepare agree with it.

[clickhouse-gh]

Workflow [PR], commit [8513892]

Summary: ❌

• Performance Comparison: Performance dashboard

job_name	test_name	status	info	comment
AST fuzzer (amd_debug, targeted)		FAIL
	Logical error: ColumnBLOB should be converted to a regular column before usage (STID: 3059-3663)	FAIL	cidb, issue
AST fuzzer (amd_debug, targeted, old_compatibility)		FAIL
	Logical error: Cannot serialize FutureSetFromSubquery with no query plan (STID: 2678-3775)	FAIL	cidb

---

AI Review

Summary

This PR fixes a real scheduler wedge in ScatterByPartitionTransform by treating finished outputs as satisfied in prepare, matching work's existing behavior. I traced the prepare/work state machine through the join, window, and shuffle producers, checked the current PR discussion and CI, and did not find a remaining correctness or coverage gap that warrants a new review comment.

Final Verdict

✅ No new findings.

LLVM Coverage Report

Metric	Baseline	Current	Δ
Lines	86.50%	86.50%	+0.00%
Functions	91.90%	91.90%	+0.00%
Branches	78.70%	78.70%	+0.00%

Changed lines: Changed C/C++ lines covered: 12/12 (100.00%) · Uncovered code

Full report · Diff report

[groeneai]

PR-caused CI fix: 04714 runtime, pushed as 8513892

Stateless tests (amd_asan_ubsan, flaky check) failed on 7d2384fb4722d16: 2 of the 50 runs of
the test this PR adds crossed the 180 s per-run limit (183.0 s and 186.6 s).

It was not two unlucky randomizations. All 50 asan_ubsan runs sit in a tight 143-166 s band, and
the debug arm peaked at 166.0 s, one draw short of the same failure. The runner's --max_threads 1
was not the cause either, since the query pins max_threads = 4 at query level.

The left side's row count was the cost driver and is not what creates the deadlock. Reduced
numbers(200000) to numbers(4000), which is also the size the sibling
parallel_full_sorting_merge tests use. Under a 50-way contention proxy pinned to 8 cores the
median run drops 29.7 s -> 8.6 s and the slowest 31.4 s -> 9.7 s; the 50-run randomized harness run
drops 174 s -> 82 s.

The test is not weakened. On a binary with only the transform fix reverted, the committed 4000-row
file still aborts 20/20 with Logical error: 'Pipeline stuck and the same wedge
(ScatterByPartitionTransform_19 (PortFull) feeding one Finished and three NeedData outputs).
The deadlock reproduces 10/10 at every size from 200000 down to 2000, and 30/30 at 4000. The
EXPLAIN PIPELINE liveness assertion still counts 2 scatters at the smaller size and still counts
0 under full_sorting_merge, so it has not gone vacuous.

I did not use the long tag: it would exempt the runtime check but also cut the flaky-check repeat
count from 50 to 5, which removes most of the coverage on a brand-new test.

parallel_full_sorting_merge family: 16 pass, 0 fail. No source change in this push.

Session id: cron:clickhouse-maint-slot-20:20260804-002400

[groeneai]

CI finish ledger — 8513892

Every failure below has an owner: a fixing PR (ours or external), or a full-effort fix task
whose fixing-PR link will be posted here when it opens. Only CH Inc sync is exempt.

Check / test	Reason	Owner / fixing PR
AST fuzzer (amd_debug, targeted) / Logical error: ColumnBLOB should be converted to a regular column before usage	trunk bug, not this diff	#111997 (ours, open)
AST fuzzer (amd_debug, targeted, old_compatibility) / Logical error: Cannot serialize FutureSetFromSubquery with no query plan	trunk bug, not this diff	#112849 (external, open)

Neither failure is reachable from this diff, which touches only ScatterByPartitionTransform.cpp
and its own test. Both fuzzer queries fail inside ReadFromMerge:

• SELECT count(__table1._table) FROM merge('default', '^m107946$') ... GROUP BY __table1._table
  reaches a consumer with an unconverted ColumnBLOB. ReadFromMerge::createPlanForTable is one of
  the five carriers named in Do not marshal blocks of a local shard plan in a distributed query #111997, which fixes the BlocksMarshallingStep gate itself
  (Closes #111945). 122 rows across 81 pull requests and 10 master rows in the last 14 days.
• SELECT DISTINCT intDiv(i, materialize(257)) FROM merge(currentUser(), toFixedString('^t0$', 4)) QUALIFY exists((SELECT -20, _table LIMIT 274)) serializes a FutureSetFromSubquery whose source
  plan the outer plan already moved. That is the shape Do not apply parallel replicas to the child plans of a Merge table #112849 describes as the one the AST fuzzer
  keeps hitting (Closes #112848). 42 rows across 19 pull requests in the last 14 days.

Both fixing PRs are open and neither is an ancestor of this branch, so this run predates both fixes.

Session id: cron:our-pr-ci-monitor:20260804-070000