…eption

When `ParallelFormattingOutputFormat` had an exception set (via
`http_write_exception_in_output_format`), the inner formatter's error
recovery path wrote the complete document (exception + closing delimiter).
However, `hasDeferredStatistics()` still returned the cached `true` from
construction time, causing `writeDeferredStatisticsAndFinalize` to append
a duplicate statistics section and closing delimiter — producing malformed
JSON/XML output.

Fix: check `exception_message.empty()` in `hasDeferredStatistics`, matching
the behavior of the inner formatters (`JSONRowOutputFormat`, `XMLRowOutputFormat`).

Co-Authored-By: Claude Opus 4.6 <[email protected]>

In `ClientBase::onProgress`, progress is forwarded to `output_format->onProgress`.
However, `output_format` is initialized lazily in `initOutputFormat` (called from
`onData`). If Progress packets arrive from the server before the first Data block,
progress is silently dropped.

This is especially common with parallel replicas on small tables where all reading
can complete before any data blocks are sent.

Fix: buffer progress in `pending_output_format_progress` when `output_format` is
null, and replay it into the format once it is created.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

…tisticsAndFinalize`

The temporary formatter created to write deferred statistics has an
internal write buffer chain (e.g. UTF8 validation wrapper) between it
and the real output buffer `out`. After `writeDeferredStatisticsAndFinalize`
writes the statistics JSON, only the formatter's immediate buffer was
flushed (`ostr->next()`), but not the intermediate buffers in the chain.
When the formatter went out of scope, `~WriteBuffer` called `cancel` on
the unfinalized intermediate buffers, discarding the statistics data.

Add explicit `flush`, `finalizeBuffers`, and `out.next()` calls to
ensure all data reaches the output buffer before the formatter is
destroyed.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

Root cause: with parallel replicas and LIMIT on fast queries (< 100ms),
replicas send Data blocks before Progress packets (because
`interactive_delay` hasn't elapsed). The coordinator's `RemoteSource`
receives Data via the read loop, then `LimitTransform` cancels the
pipeline. `RemoteSource::onCancel` called `cancel` which just sends
Cancel without draining remaining packets. `RemoteQueryExecutor::finish`
then skipped draining because `was_cancelled` was already true. Progress
packets from replicas were never processed, so `rows_read` stayed 0.

Two fixes:
1. `RemoteSource::onCancel`: call `finish` instead of `cancel` to drain
   remaining packets from the connection, ensuring Progress packets sent
   by replicas after data blocks are still processed.
2. `PipelineExecutor::finalizeExecution`: collect progress from ALL
   nodes (not just finished ones) when the pipeline is cancelled. This
   is safe because all execution threads have stopped by this point.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

…lickHouse/ClickHouse into fix-json-xml-statistics-race-condition

# Conflicts:
#	src/Processors/Sources/RemoteSource.cpp

…tics-race-condition

…->cancel`

When `PipelineExecutor::cancel` is called (e.g., from
`PullingAsyncPipelineExecutor::cancel`), there is a race condition
between `graph->cancel()` running in the main thread and
`finalizeExecution()` running in the background thread.

`graph->cancel()` synchronously calls `onCancel()` on all processors.
For `RemoteSource`, `onCancel()` drains remaining packets from
parallel replicas, which may include Progress packets that get
accumulated in `ISource::read_progress`.

However, `finalizeExecution()` (which collects progress from all
processors via `getReadProgress()`) may have already run in the
background thread before the drain completed. In that case, the
progress from draining is never forwarded to the progress callback,
and the server never sends it to the client.

Fix: after `graph->cancel()` returns, iterate over all processors
and collect any remaining progress. Since `getReadProgress()` uses
`std::swap` under a mutex, this is safe even if called concurrently
with `finalizeExecution()` - only one caller gets the progress.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

`SYSTEM STOP MERGES` now also prevents merge selection (not just execution).
The `test_lost_part_same_replica` and `test_lost_part_other_replica` tests
relied on merge entries being created while merges were stopped.

Fix by using `SYSTEM STOP REPLICATION QUEUES` instead, which stops queue
execution but still allows the merge selecting task to create entries.
For `test_lost_part_other_replica`, switch to `SYSTEM STOP MERGES` during
the recovery phase to allow fetches but block merge execution.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

…ecutor`

Both `cancel` and `finalizeExecution` were iterating over graph nodes
and calling `ReadProgressCallback::onProgress` concurrently, which is
not thread-safe. This caused TSan failures ("Server died") in tests
like `02133_issue_32458` and `03289_explain_syntax_statistics`.

The race occurs when `cancel` is called from an external thread
(e.g. `CompletedPipelineExecutor`'s timeout callback) while
`finalizeExecution` runs in the executor's background thread after
`pool->wait` returns.

Fix: Remove duplicate progress collection from `cancel` and add
`cancel_mutex` to synchronize `graph->cancel` (which triggers
`RemoteSource::onCancel` draining) with progress collection in
`finalizeExecution`. This ensures `finalizeExecution` waits for
all `onCancel` calls to complete before reading accumulated progress.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

…tics-race-condition

… in pulling path

After merging master, `sendSelectProfileEvents` now has a `TSA_REQUIRES(callback_mutex)`
annotation. Wrap the calls in the pulling pipeline path with the mutex lock, matching
the pattern already used in the completed pipeline path.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

…rmat::addChunk`

The save/restore logic was doubling `rows_read` values in JSON/XML output.

Root cause: `Progress::operator=(&&)` copies atomic values without clearing
the source (since `std::atomic` cannot be truly moved). After
`unit.statistics = std::move(statistics)`, `statistics.progress` already
retains its original values. The save/restore was then adding the same
values again, exactly doubling every `rows_read` and `bytes_read`.

The fix removes the unnecessary save/restore. The `Progress` move semantics
naturally preserve values in the source, so `writeDeferredStatisticsAndFinalize`
(called later from `IOutputFormat::finalize` or `completeDeferredStatistics`)
sees the correct accumulated progress, including any additional progress from
parallel replicas connection draining added via `onProgress` after the move.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

With parallel replicas and `LIMIT`, there is a race between progress
reporting and pipeline finalization: the pipeline finishes before
`graph->cancel()` drains remaining progress from remote connections.
Under sanitizers the timing shifts enough that `rows_read` can be 0
on rare occasions. Add retries (up to 3 attempts) to handle this
inherent flakiness.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

The two-phase finalization mechanism in the pipeline executor guarantees
that statistics (rows_read, bytes_read) are written after all progress
has been collected from parallel replica connection draining. Retries
are no longer needed and their presence incorrectly documents a
remaining race condition.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

The `cancel_mutex` was from an earlier iteration of the fix that
synchronized `cancel` with `finalizeExecution`. The current approach
does not need it: `graph->cancel` runs synchronously in `cancel`, and
all executor threads are joined before `finalizeExecution` runs, so
there is no concurrent access to processor state.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

…tics-race-condition

…n statistics

When a pipeline completes normally (e.g. LIMIT satisfied with parallel
replicas), `PullingAsyncPipelineExecutor::cancel` skips calling
`executor->cancel()` because the pipeline is already finished. This means
`graph->cancel()` is never called, so `RemoteSource::onCancel` does not
run. While connection draining does happen via `onUpdatePorts` during
execution, there is a race where progress accumulated during that
concurrent drain may not be collected by `executeJob`'s `getReadProgress`
call, leaving it stranded in `ISource::read_progress`.

Fix: call `graph->cancel()` in `finalizeExecution` for non-cancelled
pipelines, after all worker threads have been joined but before
collecting remaining progress. This ensures `RemoteSource::onCancel`
runs (triggering `query_executor->finish` to drain remaining packets),
and the subsequent progress collection loop picks up all accumulated
progress.

The "Pipeline stuck" check is moved before the `graph->cancel()` call
since cancel changes processor states.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Address review feedback: in `finalizeExecution`, use `graph->cancel(false)` instead
of `graph->cancel()` when the pipeline completed normally (not cancelled). This
limits cancellation to source processors only (e.g. `RemoteSource`), which is all
that's needed to drain remaining progress packets from parallel replica connections.
Non-source processors should not be cancelled on a successful completion path —
their `onCancel` handlers are not designed as post-success cleanup hooks.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

…n master

The bugfix validation requires the test to fail on master (to prove the bug exists)
and pass on the PR branch. Since the race condition is non-deterministic, a single
run on master may not trigger it. Run each format check 10 times — on master, at
least one iteration is very likely to see `rows_read=0`, while the fix guarantees
all iterations pass deterministically on the PR branch.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>