Add object_storage_cluster_join_mode='distributed' for whole-query … - #2383
Open
VighneshPath wants to merge 4 commits into
Open
VighneshPath wants to merge 4 commits into
VighneshPath wants to merge 4 commits into
Conversation
…JOIN dispatch When a JOIN's driving table is a `DataLake` catalog table distributed via `object_storage_cluster`, send the whole query to that table's cluster and merge the partial aggregates on the initiator, instead of reading every table back to the initiator and joining there. The JOIN and any `GROUP BY` then run on every node of the cluster rather than on one. `findDistributedObjectStorageCandidate` decides eligibility. It walks down the left side of the join tree to find the driving table, passing through a subquery only when that subquery does not itself aggregate, deduplicate, sort or limit -- each worker runs it against its own slice of the driver, so anything that finalizes across rows would turn a partial result into a final one. Every other table reachable in the query must resolve through a `DataLake` catalog, carry no row policy, and be readable by the current user. `buildDistributedObjectStorageQueryPlan` then replaces the driver with an explicit `*Cluster()` table function, keyed on the exact query tree node, and reads the result back at `WithMergeableState` through a single `ReadFromCluster` step, so the planner's ordinary finalization applies on top. It follows `buildQueryPlanForParallelReplicas` step for step, including the position-based conversion from the rewritten query's header back to the original's. Only the driving table is partitioned across the cluster; every other table in the query is read and recomputed in full on each node. That cost is not estimated when deciding to dispatch. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
Reuse mechanisms that already exist rather than adding parallel ones: `DatabaseCatalog::isDatalakeCatalog` in place of a per-storage marker, `StorageObjectStorageCluster::getClusterName` in place of a second worker-localization path in `DatabaseDataLake`, and `SourceStepWithFilterBase::applyFilters` in place of clearing fields on the `SelectQueryInfo` handed to `ReadFromCluster`. `DatabaseDataLake.cpp` is no longer touched by this feature at all. The setting's documentation described a search for the highest eligible enclosing query, and a fallback for the level an ineligible table appears at. Neither exists: dispatch is attempted only for the outermost `SELECT` of an initial query, and is all-or-nothing. It now says so, and states that only the driving table is partitioned while the rest is recomputed in full per node. `allWorkerLocalReferencesAreSafe` is renamed to `allWorkerLocalTableReferencesAreSafe`, because it proves nothing about ordinary functions: `dictGet`, a user-defined function or `hostName` move from the initiator to the workers unexamined, as they do for `Distributed`. Tests cover both shapes against a real `DataLake` catalog -- the driver as the JOIN's leftmost table, and the driver behind a subquery with the aggregation on the enclosing query and a nested JOIN on the right -- in derived-table and CTE spellings, comparing results against `object_storage_cluster_join_mode='allow'`. They assert that the whole query reaches a worker, that only the driver becomes a cluster function, and that no partner table fans out again from a worker. A local `Memory` JOIN partner must fall back to ordinary planning. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
…cluster-distributed-join Signed-off-by: VighneshPath <pathrikarvighnesh@gmail.com>
VighneshPath
force-pushed
the
feature/antalya-26.6/object-storage-cluster-distributed-join
branch
from
September 16, 2026 10:06
037f342 to
10f70a2
Compare
…cluster-distributed-join
Member
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
2 participants
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.
…JOIN dispatch
Introduces whole-query dispatch for
JOINs involving aDataLake-catalog table backed byobject_storage_cluster:findDistributedObjectStorageCandidate()walks the left spine from the root query to find such a driver table where every other table reachable in the sameJOIN/subquery tree also safely resolves through aDataLakecatalog, andbuildDistributedObjectStorageQueryPlan()rewrites the driver's table expression into a cluster table function viaIStorageCluster::buildClusterTableFunctionAST()(usingQueryTree::cloneAndReplace(), not in-place mutation) and dispatches the whole enclosing query to the driver's cluster nodes.IStorageCluster::readPreparedClusterQuery()andReadFromClustergained anis_whole_query_dispatchmode: driver-side filter/task-iterator pruning is suppressed (safe fallback, since the shipped query already carries its ownWHERE), andquery_info'stable_expression/planner_context-- which describe a single table for an ordinary per-table read -- are reset beforeReadFromClusteris constructed, since they otherwise get consulted byReadFromCluster::applyFilters()as if they described the whole dispatched query, corrupting filter pushdown and, if the resulting exception path is ever hit, dereferencing a nulltable_expression.Verified against a live Iceberg/MinIO-backed cluster with q17- and q21-shaped
JOINqueries: correct, disjoint fan-out results across cluster nodes, filter pushdown intoReadFromClustersurvives real query plan optimization, no crash.Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
...
Documentation entry for user-facing changes
...
CI/CD Options
Exclude tests:
Regression jobs to run: