Fix bucket-merge scheduling and ReadFromCluster filter staleness - #2302
Open
VighneshPath wants to merge 3 commits into
Open
Fix bucket-merge scheduling and ReadFromCluster filter staleness#2302VighneshPath wants to merge 3 commits into
VighneshPath wants to merge 3 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
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: