Fix bucket-merge scheduling and ReadFromCluster filter staleness - #2302
Open
VighneshPath wants to merge 7 commits into
Open
Fix bucket-merge scheduling and ReadFromCluster filter staleness#2302VighneshPath wants to merge 7 commits into
VighneshPath wants to merge 7 commits into
Conversation
…atency Aggregator::mergeBlocks dispatched bucket merges in ascending bucket-id order. Real bucket-row-count distributions are skewed, so a large bucket landing anywhere in that order left most merge threads idle early while one thread finished it alone — a straggler tail dominating wall time for wide GROUP BY queries (found via flamegraph/profile-event analysis of IcebergBench's q12_wide_groupby). Sort buckets by row count descending before dispatch (longest-processing- time-first), so the biggest buckets start while every thread is still free to help, and the small ones are left for whoever finishes first. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
ReadFromCluster::createExtension() (backing icebergCluster(), s3Cluster(), and every other IStorageCluster-based cluster table function) was built eagerly from applyFilters(), then frozen via a one-shot guard. But query plan optimizations call applyFilters() more than once as the plan is refined — a later pass (e.g. aggregation-in-order for a GROUP BY matching the partition/sort key) can insert another FilterStep above the source, making filter_actions_dag strictly more complete on a later call. The frozen extension silently kept whichever predicate happened to be known on the very first call, dropping anything discovered afterward. Live testing against Iceberg tables confirmed the effect: for the same WHERE clause, icebergCluster() read 6-13x more rows than the equivalent ice.`ns.table` query specifically on GROUP BY-by-partition-column queries that trigger the extra optimization pass; queries without it were unaffected, since their filter was already complete on the first call. ReadFromObjectStorageStep (the non-cluster object storage read path) already gets this right: it defers building its file iterator to initializePipeline(), which query optimization only ever reaches after every pass has finished, so it always sees the final filter. Make ReadFromCluster follow the same shape: applyFilters() now only updates filter_actions_dag, and createExtension() is called exactly once, from initializePipeline(), using the filter's final state. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
ASTPtr is boost::intrusive_ptr<IAST>, not std::shared_ptr, so std::make_shared<ASTSelectQuery>() didn't convert. And ReadFromCluster's own applyFilters(ActionDAGNodes) override hides the no-arg SourceStepWithFilterBase::applyFilters() by name across inheritance levels when called on the concrete ReadFromCluster type directly, so it needs explicit base-class qualification. Caught by the local unit_tests_dbms build (this test was never built before pushing). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
VighneshPath
force-pushed
the
fix/antalya-26.6/query-plan-aggregation-perf
branch
from
September 2, 2026 08:22
6e2041d to
5671621
Compare
…e tail latency" This reverts commit 4fa22fd.
…rallel replicas Adds an opt-in `object_storage_cluster_bypass_join_wrap` setting (default off, behavior unchanged unless a query sets it) that lets a `JOIN` against a DataLake-catalog-backed table be pushed onto parallel-replica workers together with partial aggregation, instead of pulling all probe-side rows back to the initiator and joining there. Three changes, gated by the setting: - `PlannerJoinTree.cpp`: skip the leftmost-table subquery-wrap guard that normally excludes any multi-table query from an `IStorageCluster`-derived source. - `QueryAnalyzer.cpp` / `TableFunctionsWithClusterAlternativesVisitor.h`: stop `parallel_replicas_for_cluster_engines` from being forcibly reset to `false` for queries containing a `JOIN`. - `IStorageCluster.cpp`: serialize the query sent to remote nodes with `queryNodeToDistributedSelectQuery()` instead of `query_info.query`, so a `JOIN`'s right-hand side referencing a CTE (e.g. `WITH x AS (...) ... JOIN x`) is inlined by body rather than sent as a bare, unresolvable CTE name. Also adds diagnostic `LOG_WARNING` tracing (`DLPR`, `OSC_STAGE`, `CLUSTER_ALT`) in `DatabaseDataLake.cpp`, `StorageObjectStorageCluster.cpp`, and `QueryAnalyzer.cpp` to make the parallel-replicas eligibility decision observable end-to-end. Validated so far on q17 of an internal Iceberg benchmark: correct results (checksum match), JOIN + partial aggregation confirmed pushed onto workers via `system.query_log` `JoinProbeTableRowCount`, ~37% faster hot runtime. The CTE fix (this commit's `IStorageCluster.cpp` change) is not yet re-validated end-to-end against the live cluster. See ICEBERG_JOIN_EXPERIMENT.md for full findings, dead ends, and open items — none of this is production-ready (diagnostic logging left in, no auto-detection of when the bypass is actually safe, setting is a blanket per-query opt-in rather than a narrow condition). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…replacement Redesigns the q21 whole-query dispatch (nested IStorageCluster driver below a CTE, e.g. `transaction_event` -> `txnlog`) on `StorageDistributed`'s `cloneAndReplace()` pattern instead of the previous AST-position/name-search approach: `StorageObjectStorageCluster::buildClusterTableFunctionAST()` builds the driver's `icebergS3Cluster(...)` replacement by reusing the existing, already-proven leftmost-table rewrite machinery; `buildQueryPlanForObjectStorageCluster()` swaps it in via `IQueryTreeNode::cloneAndReplace()` (exact node identity, no AST search, correct for self-joins by construction); `IStorageCluster::readPreparedClusterQuery()` dispatches the already-prepared query directly. Found and fixed four live bugs surfaced by testing against the real cluster: a `query_info.table_expression`/`planner_context` identity mismatch after `cloneAndReplace()`; this dispatch representing a whole remote query rather than a single-table read (no per-table planner-context lookup applies); the leftmost-table JOIN-wrap bypass being unconditional instead of gated on whether the JOIN's other side is safely serializable by the legacy `IStorageCluster::read()` path, which broke `q2`/`q4`/`q13`/`q16` (CTE on the RHS); and `findObjectStorageClusterWholeQueryDriver()` wrongly treating a plain subquery chain with no JOIN at all as a whole-query candidate. Full 23-query IcebergBench suite now passes with correct checksums, and `q17`/`q21` show the intended parallel-replicas JOIN pushdown speedup vs. the published baseline. Not yet resolved: a new hot-latency regression on `q2`/`q4`/`q13` (not `q4`'s sibling `q16`), suspected to be a side effect of the experimental setting also loosening an unrelated, pre-existing analyzer check for tables that don't benefit from cluster distribution (e.g. a window-function CTE that must fully materialize on one node regardless) -- not yet confirmed with log evidence. See ICEBERG_JOIN_EXPERIMENT.md §3.7. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…l-replicas candidate machinery Replaces the prior ad hoc, three-gate `object_storage_cluster_bypass_join_wrap` dispatch (spread across `QueryAnalyzer.cpp`/`PlannerJoinTree.cpp`/ `findParallelReplicasQuery.cpp`) with a single pluggable eligibility predicate, `isObjectStorageClusterDriverEligible()`, reused by both of ClickHouse's own stock parallel-replicas driver-selection mechanisms: - `findQueryForParallelReplicas()`/`findTableForParallelReplicas()` (a driver nested under a CTE, e.g. `q21`), generalized with a `relax_for_object_storage_driver` policy so a JOIN's non-driver branch needing its own finalization doesn't wrongly narrow the candidate the way it must for `MergeTree`. - `allowParallelReplicasForJoinTree()` (a driver as the immediate leftmost table of a JOIN, e.g. `q17`), now parameterized over a pluggable `ParallelReplicasStorageEligibility` instead of being `MergeTree`-only. Both dispatch to one shared execution backend, `buildQueryPlanForObjectStorageCluster()`, mirroring `StorageDistributed::buildQueryTreeDistributed()`'s exact-`QueryTree`-node replacement pattern. Also fixes a `StorageDummy`-erases-type bug (the disposable dummy-plan-walk used to evaluate a candidate's shape substitutes every table with a generic `StorageDummy`, which silently failed the object-storage eligibility check for a driver reachable only through a CTE), and adds a `!select_query_options.is_subquery` capability boundary to the leftmost-driver mechanism: whole-query dispatch is only proven correct when the JOIN's owning query is the true top level, not nested under further derived-table wrapping (found live via IcebergBench `q16`, whose JOIN sits four subquery layers below a Looker-style window-function pivot). Reverts `QueryAnalyzer.cpp`/`TableFunctionsWithClusterAlternativesVisitor.h` to stock: cluster dispatch is now injected explicitly, per selected driver, at execution time, so no query-scope-wide analyzer flag is needed anymore. See ICEBERG_JOIN_EXPERIMENT.md §12 for the full investigation, including four other live bugs found and fixed along the way (unbounded `only_analyze` recursion, an `isDistributed()` misuse, and two gaps in the branch-role-aware candidate-narrowing walk). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
|
Partially crossed with #2249 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Found via profiling ClickHouse against StarRocks on the IcebergBench cross-engine benchmark. Two independent performance fixes, bundled on one branch as separate commits (unrelated subsystems, found in the same investigation):
Aggregator::mergeBlocks: two-level aggregation bucket merges were dispatched in ascending bucket-id order. Real bucket-row-count distributions are skewed, so a large bucket landing anywhere in that order left most merge threads idle early while one thread finished it alone — a straggler tail dominating wall time for wideGROUP BYqueries. Now dispatches buckets largest-first (longest-processing-time-first scheduling), so the biggest bucket starts while every thread is still free to help, instead of stranding one thread alone with it at the end.ReadFromCluster(backingicebergCluster(),s3Cluster(), and every otherIStorageCluster-based cluster table function): its task/file list was built eagerly from the firstapplyFilters()call and frozen via a one-shot guard. But query plan optimization callsapplyFilters()more than once as the plan is refined — a later pass (e.g. aggregation-in-order for aGROUP BYmatching the partition/sort key) can insert anotherFilterStepabove the source, making the filter strictly more complete on a later call. The frozen extension silently dropped anything discovered after the first call. Measured live against Iceberg tables:icebergCluster()read 6-13x more rows than the equivalentice.\ns.table`query for the identicalWHEREclause, specifically on queries whose plan needed an extra optimization pass.ReadFromObjectStorageStep(the non-cluster object storage path) already gets this right — it defers the equivalent step toinitializePipeline(), which only ever runs after every optimization pass has finished. This change makesReadFromCluster` follow the same shape.Both come with unit tests (
gtest_aggregator_bucket_merge_order.cpp,gtest_read_from_cluster_predicate_pushdown.cpp) that fail against the pre-fix code and pass against the fix.Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
Improved two-level aggregation merge scheduling to avoid thread-idle tails under skewed bucket sizes, and fixed cluster table functions (
icebergCluster(),s3Cluster(), etc.) reading significantly more data than necessary when query optimization refines filter conditions across multiple passes (e.g.GROUP BYon a partitioned column).Documentation entry for user-facing changes
Not applicable — internal implementation fixes, no new settings or syntax; user-visible effect is fewer files/rows read and less merge-phase tail latency.
CI/CD Options
Exclude tests:
Regression jobs to run: