Repository navigation
Add object_storage_cluster_join_mode='distributed' for whole-query JOIN dispatch - #2832
Conversation
…JOIN dispatch With the new mode, a `JOIN` query whose driving table is a `DataLake` catalog table with a cluster is sent whole to that cluster. Each node runs the `JOIN` and any `GROUP BY` over its share of the driver's files, and the initiator only merges the partial aggregate states. - findDistributedObjectStorageCandidate walks the left side of the query, through `INNER ALL`/`LEFT` joins and subqueries that do not combine rows, to the driver. Every other table must be a catalog table with no row policy and `SELECT` access. The decision is made once per query, next to parallel replicas, and kept in `GlobalPlannerContext`. - buildDistributedObjectStorageQueryPlan serializes the query and replaces every table in it with the table function its own cluster read sends: the driver's `*Cluster` function reads the files the initiator hands out, every other table's plain function reads all of them. A worker resolves no catalog table and runs the query as it would any cluster table function; an Iceberg table's function names the metadata file resolved on the initiator. A driver referenced twice is not dispatched. - `ReadFromClusterQuery` sends the query over the same transport as `ReadFromCluster`, whose pipe building is now shared, and feeds it the driver's files pruned by the driver's own filter. - An ordinary cluster read reached under the new mode behaves exactly like `allow`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- Shorten the comments to the reason each step exists. - `DistributedObjectStorageCandidate` keeps only the dispatch boundary and the driver's `TableNode`; its storage and filter key are derived from it. - `findDriverOnLeftSpine` returns `std::optional` instead of an `unusable` flag. - Rename `isSafeDataLakeLeaf` to `isSafeCatalogTable`. - Move the driver filter extraction into `buildDriverFilter`. - Replace the `prepareForWholeQueryDispatch` hook with lazy initialization in `StorageObjectStorageCluster::getTaskIteratorExtension`, limited to data lake configurations, and drop the test of the hook. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- `QueryNode::getJoinTree` and `JoinNode::getLeftTableExpression` now return a reference that asserts the node exists; use `getJoinTreeNode` and `getLeftTableExpressionNode`, which can still be null. - Do not dispatch when `build_logical_plan` is set: that plan is serialized for a shard, and `ReadFromClusterQuery` has no serialized form. Plan-based parallel replicas step aside for the same reason. - Strip initiator-only settings such as `limit` from the context sent to workers, as `ReadFromCluster::updateSettings` now does. A worker applying them would cut its partial result before the initiator merges it. - The analyzer now checks table access while resolving identifiers, so the access tests analyze as a user with grants and check the candidate as one without. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- Decline dispatch when a nested subquery or CTE has its own `SETTINGS` clause. The eligibility checks read the outer query's settings, so a nested `additional_table_filters` was not rejected, and a nested `object_storage_cluster` was ignored when choosing the driver's cluster. - Reuse `hasAggregateFunctionNodes` and `hasWindowFunctionNodes` instead of a local copy of the same traversal. - Rename `ReadFromClusterQuery::updateSettings` to `createRemoteContext`: it creates a new context rather than updating one. Build it with `Context::createCopy` and `setSetting`. Related: Altinity#2383 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…cluster-distributed-join
|
@codex review |
Codex Review Summary: ✅ CompletedThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
|
Codex Review: Something went wrong. Try again later by commenting “@codex review”. ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
…cluster-distributed-join
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 46dd813705
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
|
||
| /// Initiator-only settings such as `limit` would cut each worker's partial result, as in `ReadFromCluster`. | ||
| Settings settings = remote_context->getSettingsCopy(); | ||
| ClusterProxy::stripInitiatorOnlySettings(settings); |
There was a problem hiding this comment.
When this mode is enabled during a rolling upgrade, settings still marks object_storage_cluster_join_mode=distributed as changed after this strip, so RemoteQueryExecutor sends that value in the settings packet. A worker from before this commit already knows the setting name but cannot parse the new enum value, causing every otherwise eligible dispatched JOIN to fail before execution; mirror ReadFromCluster::updateSettings and change it to allow in this remote context.
Useful? React with 👍 / 👎.
| return false; | ||
|
|
||
| const auto & storage = static_cast<const IStorageCluster &>(*table_node.getStorage()); | ||
| return !storage.getClusterName(context).empty(); |
There was a problem hiding this comment.
When object_storage_cluster_fallback_to_local_if_empty=1 and the configured object-storage cluster is missing or has no nodes, getClusterName still returns its nonempty configured name, so this accepts a whole-query candidate. The custom path subsequently uses the throwing getClusterImpl in readPreparedClusterQuery instead of resolveClusterRead, causing a qualifying JOIN to fail although the setting promises an ordinary local read; resolve the cluster or decline dispatch before returning the candidate.
Useful? React with 👍 / 👎.
Port of #2383 to
antalya-26.8, with that PR's review comments addressed.With
object_storage_cluster_join_mode = 'distributed', aJOINquery whose driving table is aDataLakecatalog table withobject_storage_clusterset is sent whole to that cluster. Each node runs theJOINand anyGROUP BYover its share of the driver's files, and the initiator only merges the partial aggregate states. Before this, theJOINandGROUP BYalways ran on the initiator.findDistributedObjectStorageCandidatewalks the left side of the query, throughINNER ALL/LEFTjoins and through subqueries that do not combine rows, to the driver. Every other table must be a catalog table with no row policy and withSELECTaccess. Otherwise the whole query is planned as before.buildDistributedObjectStorageQueryPlansends the query with every table replaced by the table function its own cluster read would send: the driver as its*Clusterfunction, which reads the files the initiator hands out, and every other table as its plain function, which each node reads in full.ReadFromClusterQuerysends the query over the same transport asReadFromCluster. It hands out the driver's files, pruned by the driver's own filter.Commits:
antalya-26.6.antalya-26.6.antalya-26.8:limitfrom the context sent to workers, asReadFromClusternow does.allowmode, a declined dispatch takes theallowpath. 8000object_storage_cluster_join_mode='distributed'for whole-query … #2383:hasAggregateFunctionNodes/hasWindowFunctionNodes.ReadFromClusterQuery::updateSettingstocreateRemoteContextand build it withcreateCopy+setSetting.SETTINGSclause. A nestedadditional_table_filtersorobject_storage_clusterwas invisible to the eligibility checks.IMPORTANTdriver settings, theunusableflag andreturn leftwere already gone after the redesign.The query condition cache fix from #2383 is not ported:
antalya-26.8already keys object storage entries by full path and ETag.Related: #2383
Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
Add
object_storage_cluster_join_mode = 'distributed', which sends a wholeJOINquery overDataLakecatalog tables to the cluster of its driving table, so that theJOINandGROUP BYrun on the cluster's nodes instead of on the initiator.Documentation entry for user-facing changes
The
distributedvalue is described in the documentation of theobject_storage_cluster_join_modesetting.CI/CD Options
Exclude tests:
Regression jobs to run:
🤖 Generated with Claude Code