Fix input() read twice when the parallel-replicas row estimate builds an IN set - #121283
groeneai wants to merge 1 commit into
Conversation
Sorry, something went wrong.
|
Internal second-model review: adjudication log (click to expand)
Four rounds by an independent model (engine: codex) plus my own cold read of the code each round:
Severity: ❌ blocker / ⚠️ major / 💡 nit. DISAGREE verdicts carry recorded evidence and are Also measured and deliberately not raised as separate work: join runtime filters do not trip the new Session id: cron:clickhouse-review-slot-9:20260921-062200 |
Sorry, something went wrong.
Pre-PR validation gate (click to expand)
Session id: cron:clickhouse-impl-slot-40:20260921-035600 |
Sorry, something went wrong.
|
Workflow [PR], commit [946f7d1] Summary: ⏳
AI ReviewSummaryThis PR fixes the planner-side double-read of input() by skipping the parallel-replicas row estimate when the read filter still carries an unbuilt set, and the src/ change itself looks coherent. The remaining problem is in the new regression test: it is not safe to rerun in the same database, so a failed first attempt can make the retry fail with TABLE_ALREADY_EXISTS instead of re-exercising the original bug. Findings⚠️ Majors
Tests
Final VerdictRequest changes. |
Sorry, something went wrong.
| PR_DIRECT="$PR, automatic_parallel_replicas_mode = 0" | ||
| PR_AUTO="$PR, automatic_parallel_replicas_mode = 2" | ||
|
|
||
| $CLICKHOUSE_CLIENT -m -q " |
There was a problem hiding this comment.
This regression test already assumes the harness may rerun it in the same database (QUERY_ID is regenerated for exactly that reason a few lines below), but the four tables are created unconditionally and never dropped. If the first attempt fails after these CREATE TABLEs, the retry will stop with TABLE_ALREADY_EXISTS before it reaches the input() path you are trying to reproduce. Please make the setup idempotent here, e.g. DROP TABLE IF EXISTS before the CREATE TABLEs (and preferably clean them up at the end too).
Sorry, something went wrong.
Build profile diff (arm_release)Comparing 946f7d194 with master de0734db6 (stripped binary size, per-symbol sizes and ThinLTO time; compile times per translation unit against the most recent warmup build that recompiled it). ✅ No significant changes. Binary sizesprograms/clickhouse-stripped: smaller than the master baseline by the known offset between the two builds, so the difference is not shown. A delta that differs from the offset by more than 50% of it is shown, in either direction. The official master build is compiled with -g and a pull request build is not, and XRay counts debug instructions towards its instrumentation threshold, so master instruments thousands of functions more and its binary is ~0.4% larger no matter what the pull request does. Compile time of recompiled translation units7 translation units recompiled, 19 s compile time in total, 7 of them have a recent master baseline. |
Sorry, something went wrong.
An INSERT SELECT whose filter references input() failed with INVALID_USAGE_OF_INPUT when parallel replicas were enabled together with a non-zero parallel_replicas_min_number_of_rows_per_replica. With that setting, collectFiltersForAnalysis runs in the Planner constructor over a throwaway planner that owns its own PreparedSets, so the filter DAG handed to the real read carries a ColumnSet for a FutureSetFromSubquery that planner will never build. The advisory replica-count estimate then called selectRangesToRead() purely to read selected_rows; index analysis reached KeyCondition::tryPrepareSetIndexForIn, which built that set and so executed the IN subquery. For input(), a one-shot stream from the client, that was the first read, and the query's own set then read it a second time. Gating the estimate is sufficient for this read because nothing else builds the escaped set on a ReadFromMergeTree step: applyFilters rebuilds filter_actions_dag from the real plan's pushed-down nodes and overwrites query_info.filter_actions_dag with it, and every other selectRangesToRead() caller runs after that. The estimate now declines when the filter DAG references a set that is not built yet, reusing QueryPlanOptimizations::dagContainsNonReadySet, which convertAnyJoinToSemiOrAntiJoin and Optimizations/Utils already use to make a plan-time decision decline rather than force a build. Both routes that reach this call site are covered: the direct planner path and the automatic-parallel-replicas probe plan. One consequence beyond the bug: the estimate no longer prunes with an IN (subquery) set, so it can use more replicas, never fewer for the same data, and, where index analysis would have built that set, it no longer runs the subquery an extra time while planning. The guard is coarser on purpose: it declines for any unbuilt set, while KeyCondition builds only key-column ones. Rejected alternatives. Disabling parallel replicas for any tree that reads input(): GLOBAL IN, a GLOBAL JOIN's materialized side, a reused MATERIALIZED CTE and every plain join are all materialized on the initiator, work today, and would lose the route; the exact predicate is only available after buildQueryTreeForShard, which executes the GLOBAL subqueries and so cannot be evaluated speculatively. Overriding parallel_replicas_min_number_of_rows_per_replica for such trees: silently changes a user setting and still leaves every other query executing its IN subquery twice. Buffering the client stream in StorageInput: a memory regression, and it guards the symptom rather than the cause. Running the estimate with KeyCondition's require_ready_sets instead of declining it: both entry points this call site can take pass the step's own `indexes` member by reference (ReadFromMergeTree.cpp:2735, :2763), so an estimate analysed that way memoizes a set-less KeyCondition on the step; the whole body of ReadFromMergeTree::applyFilters is then guarded by `if (!indexes)` and does nothing, and the executed read inherits that weaker analysis, losing IN-set primary-key pruning at read time. Declining leaves `indexes` empty, so applyFilters runs in full and the executed read analyses against the query's own set. Not addressed here, all of it outside the estimate this change gates. A plain IN (subquery) is re-executed on every follower, since parallel_replicas_allow_in_with_subquery defaults to 1, and a follower has no client input stream, so it throws INVALID_USAGE_OF_INPUT there; the initiator usually answers from its local plan first and discards that error, which makes the client seeing it a race. GLOBAL IN evaluates the subquery on the initiator and ships a temporary table, so the test uses that spelling for the two cases that read from a replica; without this change they still fail on the double read described above. With parallel_replicas_mode = custom_key_sampling a follower can still be asked to read input() and fails with "Input stream is not initialized", identically without this change. And collectFiltersForAnalysis collects the same unbuilt-set DAG for IStorageCluster irrespective of this setting (Planner.cpp:284), where ReadFromCluster::applyFilters falls back to query_info.filter_actions_dag and the task iterators build sets from it through VirtualColumnUtils::buildSetsForDAG. I traced that last one by reading and have not run it. Related: ClickHouse#71028 (comment) Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Flaky check on 946f7d1: the test's own input() cases were racy, fixed in 659c1b212 of 250 flaky-check runs failed, always on the first input() case, with Code: 477 ... Input stream is not initialized raised on a follower (127.0.0.2:9000) rather than on the initiator. Cause, measured on this branch's binary: a plain IN (subquery) is re-executed on every follower, and a follower has no client input stream. Twenty runs of each direct case left a follower ExceptionBeforeStart with code 477 in system.query_log on 7/20 (HTTP) and 16/20 (native), while the initiator answered from its local plan and the client saw nothing. Whether that error reaches the client is a race; with parallel_replicas_local_plan = 0 it is deterministic, on this branch and on the parent commit alike. That is a follower-side branch this change does not reach, so the test must not depend on it. Fix: the two cases that read from a replica now use GLOBAL IN, which evaluates the subquery on the initiator and ships a temporary table, so no follower is asked to read input(). The statistics-only case keeps a plain IN, because that mode contacts no replica at all. The row and route oracles and the .reference file are unchanged, and each input() case still fails on the parent commit through the same estimate. 500 runs of the fixed test, eight at a time, 250 with the CI settings blob forced and 250 with default randomization: no failures, and zero follower 477 rows across all of them. No src/ line changed, so the src/ diff is byte-identical to 946f7d1. The changelog entry and the "Not fixed here" paragraph are now scoped to what this change repairs. |
Sorry, something went wrong.
Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes into CHANGELOG.md):
Fixed the parallel-replicas row estimate running an IN (subquery) while planning, which made an INSERT ... SELECT whose filter reads input() fail with INVALID_USAGE_OF_INPUT when parallel_replicas_min_number_of_rows_per_replica was not zero.
Description
Related: #71028 (comment)
Not introduced by #71028. The same test failed with 477 on v25.11.1.1 in the CI database; that PR's randomizer only makes it visible.
Root cause. With that setting a throwaway planner with its own PreparedSets collects the filters in the Planner constructor, so the real read's DAG carries a ColumnSet nobody will build. The replica-count estimate at PlannerJoinTree.cpp:2669 calls selectRangesToRead() just to read selected_rows; index analysis then reaches KeyCondition::tryPrepareSetIndexForIn, builds that set and runs the IN subquery: read one. The query's own set reads input() again.
The fix. That estimate declines when the filter DAG references a set that is not built yet, reusing QueryPlanOptimizations::dagContainsNonReadySet. For a ReadFromMergeTree step it is the only index analysis before filter push-down: every other selectRangesToRead() caller runs after applyFilters populated the step's indexes. Cost: such a query no longer refines its replica count, so it can use more replicas, never fewer.
Not fixed here. A plain IN (subquery) is re-executed on every follower, which has no client input stream and throws there; the initiator usually answers from its local plan first and discards that error, so whether the client sees it is a race. GLOBAL IN evaluates the subquery on the initiator, so that is the spelling the test's transport cases use. With parallel_replicas_mode = custom_key_sampling a follower can still be asked to read input(). And collectFiltersForAnalysis hands the same unbuilt-set DAG to cluster storages irrespective of this setting, whose task iterators build sets from it (traced by reading only).
Validation. New 05237_parallel_replicas_input_table_function.sh covers both client transports and the automatic route, asserting per case that the route was really taken; each input() case fails on the base commit with 477. 500 randomized runs are stable and leave no follower INVALID_USAGE_OF_INPUT behind; the regression batch shows no reference movement.