Skip to content

Fix distributed global right join - #2315

Open
ianton-ru wants to merge 8 commits into
antalya-26.6from
bugfix/antalya-26.6/distributed-global-right-join-test
Open

ianton-ru wants to merge 8 commits into
antalya-26.6from
bugfix/antalya-26.6/distributed-global-right-join-test

Conversation

@ianton-ru

Copy link
Copy Markdown

Solved #1664

Changelog category (leave one):

  • Bug Fix (user-visible misbehavior in an official stable release)

Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):

Fix distributed global right join

Documentation entry for user-facing changes

Swap left and right tables for distributed right join.
Fix for object_storage_cluster_join_mode='global' - it was broken after frontports and executed as local.

CI/CD Options

Exclude tests:

  • Fast test
  • Integration Tests
  • Stateless tests
  • Stateful tests
  • Unit tests
  • Performance tests
  • Aarch64 tests
  • All with ASAN
  • All with TSAN
  • All with MSAN
  • All with UBSAN
  • All with Coverage
  • All Regression
  • Disable CI Cache

Regression jobs to run:

  • Fast suites (mostly <1h)
  • Aggregate Functions (2h)
  • Alter (1.5h)
  • Benchmark (30m)
  • CAS (content-addressed storage; Antalya only)
  • ClickHouse Keeper (1h)
  • Iceberg (2h)
  • LDAP (1h)
  • OAuth (5m)
  • Parquet (1.5h)
  • RBAC (1.5h)
  • SSL Server (1h)
  • S3 (2h)
  • S3 Export (2h)
  • Swarms (30m)
  • Tiered Storage (2h)

ianton-ru and others added 3 commits September 4, 2026 13:30
The expected result matches initiator-side join semantics; the test currently fails because shards emit extra unmatched right-side rows.

Related: #1664
Co-authored-by: Cursor <cursoragent@cursor.com>
Broadcast the original left table and keep the preserved right side sharded, so unmatched right rows are not duplicated on every shard.

Related: #1664
Co-authored-by: Cursor <cursoragent@cursor.com>
After ClickHouse#94748 the planner always wrapped the left cluster source, so
object_storage_cluster_join_mode='global' never saw the JOIN and ran
on the initiator. Keep wrapping for allow/local and for RIGHT/FULL to
avoid duplicated unmatched rows.

Co-authored-by: Cursor <cursoragent@cursor.com>
@github-actions

github-actions Bot commented Sep 4, 2026 •

Copy link
Copy Markdown

Workflow [PR], commit [46c5fb6]

@ianton-ru

Copy link
Copy Markdown
Author

@blau-ai

@blau-ai

blau-ai commented Sep 7, 2026

Copy link
Copy Markdown
Collaborator

CI triage for #2315 (@ianton-ru)

Verdict: 0 of the failures are caused by this PR. All red checks are pre-existing flakiness / infra / runner-environment issues in subsystems your change never touches. This PR only edits src/Planner/PlannerJoinTree.cpp plus two tests, and its own new tests pass (05058_distributed_global_right_join is green in every stateless run that executed it; nothing from test_s3_cluster appears in the failure set). Re-running the failed jobs should clear them.

Head SHA: 5ca17e5.

Per-failure breakdown

1. Source upload — infra (not PR)
Checkout submodules step failed: fatal: could not read Username for 'https://github.com' while cloning contrib/* (llvm-project, aws, …), retried 3× and aborted (exit code 123). Pure network/credential failure on the runner, unrelated to source. → Re-run.

2. Integration tests (amd_asan_ubsan, db disk, old analyzer, 4/8) — pre-existing/infra (not PR)
All 12 failures are test_storage_iceberg_with_spark/test_schema_inference.py::test_schema_inference[...] (every s3/azure/local parametrization). Iceberg + Spark schema inference has nothing to do with analyzer JOIN planning; these are heavy external-dependency (Spark/MinIO/Azure) tests that fail as a block when the Spark side is unhealthy. → Re-run.

3. Integration tests (amd_asan_ubsan, db disk, old analyzer, 7/8) — pre-existing/infra (not PR)
test_database_iceberg_lakekeeper_catalog/test.py::test_auth_token_profile_events — Iceberg/Lakekeeper REST catalog auth, again unrelated to the planner. → Re-run.

4. Stateless tests (amd_binary, cas s3 storage, parallel) — flaky storage layer (not PR)
02265_column_ttl: Code: 210 ... did not prove it still holds the manifest it offered for part ... by relink ... (NETWORK_ERROR) from DataPartsExchange::Fetcher::relinkPartToDisk. This is the CAS/shared-storage part-exchange path (Antalya feature), not the query planner. The harness self-diagnosis even labels it flaky. → Re-run.

5. Stateless tests (amd_asan_ubsan, cas s3 storage, parallel, 2/2) — environment / flaky (not PR)

  • 04105_limit_by_into_sort_explain: result diff is only stream counts (MergingSortedTransform 12 → 1 vs 6 → 1, 24→12, etc.). That is the runner's CPU/max_threads count, not query semantics — an environment-sensitive EXPLAIN reference, independent of the JOIN change.
  • 04033_tpc_ds_q60: Code: 241 ... memory limit exceeded (14.55 GiB > 14.40 GiB). Passed on rerun (0/3 reruns failed) — memory pressure. Flaky.
    → Re-run.

6. Stateless tests (amd_asan_ubsan, cas s3 storage, parallel, 1/2) — server death, collateral (not PR)
Server died; the 6 UNKNOWN tests (00051_min_max_array, 02122_parallel_formatting_JSONStrings, …) are just the ones that never ran after the server went down. Same CAS-s3-storage config as #4 above; a source-only planner change cannot make the server die at this scale, and the affected tests are unrelated. → Re-run.

7. Stress test (amd_asan_ubsan) — infra (not PR)
Cannot start clickhouse-server → Check failed. The server never started, so no stress workload ran. Builds all passed and Stress test (amd_debug) is green, so this is a startup/environment flake, not the diff. → Re-run.

Bottom line

Nothing here needs a code change. The PR-relevant signal is all green: builds pass on every arch, Fast test / Stateless tests (amd_debug, parallel) / distributed-plan variants pass with 0 failures, and the regression suites (swarms, s3_export, parquet, oauth) are green. The failures are the usual Iceberg/Spark, CAS-s3-storage, thread-count-EXPLAIN, memory-limit, and submodule-checkout flakes. I'd just re-run the seven failed jobs.

(No fix pushed — there's nothing PR-caused to fix. Happy to re-check after a re-run if any of these persist.)

— @blau-ai

@ianton-ru ianton-ru changed the title Bugfix/antalya 26.6/distributed global right join test Fix distributed global right join Sep 7, 2026
Port of the review fixes made for the `master` pull request
ClickHouse#118611

`tryRewriteGlobalRightJoinAsLeftJoin` only recognised a narrow shape of the
query it was meant to fix, so a `GLOBAL RIGHT JOIN` whose left table is a
sharded `Distributed` table still emitted the rows preserved from the
broadcast side once per shard in these cases:

* `USING` and `NATURAL` joins were skipped, because only `ON` was accepted.
  A `USING` key stores its sides positionally, so the key pairs are now
  swapped together with the table expressions, and a key that does not hold
  a plain column per side leaves the query alone.

* The rewrite required the right side to be a sharded `Distributed` table
  too. Only the left table fans the query out across shards, so the gate is
  now on the left table alone, and a local table, a one shard `Distributed`
  table, a subquery or a table function on the right is handled as well.
  `Any`, `Semi` and `Anti` join over to the swapped sides the same way `All`
  does and are accepted; `Asof` and `RightAny` do not mirror and stay out.

* Only the topmost join was examined. Join trees are left deep, so the walk
  now descends to the join that reads the leftmost table.

A `RIGHT` or `FULL` join that survives the rewrite, such as the one in
`(A INNER C) RIGHT B`, cannot be swapped without moving a join into the right
table expression. Those now read the leftmost table through a subquery, which
keeps the join on the initiator, the only remaining way to emit the preserved
rows once.

`getDistributedStorageFromTableExpression` also missed `remote` and
`clusterAllReplicas` on the left, because it only looked at `TableNode`.

Co-authored-by: Cursor <cursoragent@cursor.com>
ianton-ru and others added 2 commits September 9, 2026 13:09
…istributed tables.

An `Alias`, `Buffer`, `MaterializedView` or `StorageProxy` in front of a sharded `Distributed` table still fans the query out across shards, so unmatched right rows would otherwise be duplicated once per shard.

Co-authored-by: Cursor <cursoragent@cursor.com>
@ianton-ru
ianton-ru marked this pull request as ready for review September 10, 2026 08:02
@k-morozov

Copy link
Copy Markdown

Did you consider another approach? Keep the RIGHT JOIN and still send the right table to shards, but add a row number to it. Each shard returns the set of row numbers that found a match, and the initiator merges these sets and adds the rows without a match itself. With this, big GLOBAL RIGHT JOIN small sends only the small table over the network, and FULL JOIN can also run on shards. I understand it is harder (the shard has to collect row numbers before WHERE/GROUP BY), so I wonder why you chose the current approach.

@Selfeer

Selfeer commented Sep 29, 2026

Copy link
Copy Markdown
Collaborator

Medium: PREWHERE on a multi-shard GLOBAL RIGHT JOIN is left on the table after the sides are swapped, and the query then fails with ILLEGAL_PREWHERE.

tryRewriteGlobalRightJoinAsLeftJoin swaps the join inputs and turns the join into a LEFT JOIN without looking at PREWHERE. The original left table, which owned that filter, becomes the broadcast side. buildQueryTreeForShard materializes that side as a Memory temporary table, and Memory does not support PREWHERE, so a query such as FROM dist_left GLOBAL RIGHT JOIN dist_right PREWHERE dist_left.v != '' throws Code: 182. The same query works as a GLOBAL LEFT JOIN, as a non-global RIGHT JOIN, and when the left table has a single shard, because those paths do not swap. distributed_product_mode = 'global' reaches the same swap. A GLOBAL FULL JOIN, which is not swapped and is instead read through the initiator subquery, also drops PREWHERE: the wrap does not copy it onto that read. Upstream fixed both in ClickHouse#119780 after the change this pull request ports.

ianton-ru and others added 2 commits October 1, 2026 16:24
After the join is rewritten as a LEFT JOIN, a `PREWHERE` bound to the original left table followed it into a `Memory` table and shards threw `ILLEGAL_PREWHERE`. Move the predicate into the broadcast subquery when its columns belong to that table, and skip the swap when `PREWHERE` has no column so the initiator wrapper keeps the filter. When wrapping a remote table that owns `PREWHERE`, move the full predicate onto the wrap and drop the initiator actions so nested `GLOBAL IN` sets are not materialized twice.

Port of ClickHouse#119780

Co-authored-by: Cursor <cursoragent@cursor.com>
@ianton-ru

Copy link
Copy Markdown
Author

@Selfeer Yes, thanks. The same was found in PR for upstream,(ClickHouse#118611), fix (ClickHouse#119780) ported here.

@ianton-ru

Copy link
Copy Markdown
Author

@k-morozov Good idea, looks too global, introduce additional information in responses from nodes. Need to think.

/// `Alias`, `MaterializedView`, `Buffer` and `StorageProxy` (for example `lazy_load_tables`)
/// forward `read` to a nested storage. If that nested storage is `Distributed`, the join still
/// fans out across shards, so look through the wrappers before deciding.
for (size_t i = 0; storage && i < 16; ++i)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Surprised that '16' passed through AI review.
Funny, but Ok.

{
const auto distributed_product_mode = settings[Setting::distributed_product_mode];
return join_node.getLocality() == JoinLocality::Global
|| distributed_product_mode == DistributedProductMode::GLOBAL

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure that I get this.
getLocality returns JoinLocality::Local, but we believe that it is actually Global? Why?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't catch.
getLocality is compared here with JoinLocality::Global, not with JoinLocality::Local.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Even if getLocality is JoinLocality::Local, we can consider the join global based on other conditions. Why?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Settings change behaviour. distributed_product_mode=global and prefer_global_in_and_join=true are exactly to use global variants, when from query semantic itself local is returned.
https://clickhouse.com/docs/reference/settings/session-settings/distributed#distributed_product_mode
https://clickhouse.com/docs/reference/settings/session-settings/prefer#prefer_global_in_and_join

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

5 participants