Fix RIGHT JOIN with parallel_replicas_min_number_of_rows_per_replica

Project: ClickHouse/ClickHouse | Language: C++ | Concepts: parallelism, query-optimization, error-handling, testing-strategy | Difficulty: senior
Link: https://github.com/ClickHouse/ClickHouse/pull/111332

The problem

When running a RIGHT JOIN query with the setting parallel_replicas_min_number_of_rows_per_replica > 0, ClickHouse would fail with errors like NOT_FOUND_COLUMN_IN_BLOCK or THERE_IS_NO_COLUMN. This happened when projecting columns from the left table. In some cases, the query would even return silently incorrect results. The issue only affected RIGHT JOIN (and related joins) with parallel replicas enabled; other join types and settings worked as expected.

Why it's non-trivial

A naive fix might try to simply adjust which columns are requested or patch the error at the surface, but the root cause is deeper: the query planner was reusing analysis results (like which columns and parts to read) from the leftmost table scan for the right-side scan in a RIGHT JOIN. This is incorrect because, in a RIGHT JOIN, the parallelized scan is actually on the right table, not the left. The fix needs to ensure that analysis is only reused when it is valid for the table being scanned, including in self-join cases.

The approach

1. Track whether the scan is reached via a RIGHT JOIN right-branch

The function that finds the reading steps in the query plan is updated to optionally track if the scan is reached by descending the right branch of a RIGHT JOIN:

-std::vector<QueryPlan::Node *> findReadingSteps(QueryPlan::Node * root, bool allow_view_over_mergetree);
+std::vector<QueryPlan::Node *> findReadingSteps(QueryPlan::Node * root, bool allow_view_over_mergetree, bool * right_branch_selected = nullptr);

This allows the planner to know if the scan is on the "wrong side" for analysis reuse.

2. Propagate the right-branch flag during plan traversal

When traversing the plan, the function sets the flag if it descends the right child of a RIGHT JOIN:

if ((join && join->getJoin()->getTableJoin().kind() == JoinKind::Right)
    || (join_logical && join_logical->getJoinOperator().kind == JoinKind::Right))
+{
+    if (right_branch_selected)
+        *right_branch_selected = true;
    node = node->children.at(1);
+}
else
    node = node->children.at(0);

This ensures that the planner is aware of the context in which each scan node is reached.

3. Only reuse analysis if not on a RIGHT JOIN right-branch

In the main function that creates the local plan for parallel replicas, the code now checks the flag before reusing the pre-analyzed result:

-bool right_branch_selected = false;
-auto reading_nodes = findReadingSteps(query_plan->getRootNode(), allow_view_over_mergetree, &right_branch_selected);
...
-/// For the first reading step, reuse the pre-analyzed result if available.
-ReadFromMergeTree::AnalysisResultPtr analyzed_result_ptr;
-if (analyzed_read_from_merge_tree.get())
+ReadFromMergeTree::AnalysisResultPtr analyzed_result_ptr;
+if (!right_branch_selected && analyzed_read_from_merge_tree.get())
{
    auto * analyzed_merge_tree = typeid_cast<ReadFromMergeTree *>(analyzed_read_from_merge_tree.get());
    if (analyzed_merge_tree)
+    {
+        /// Best-effort bug catcher, not an invariant: the analysis is built by the caller from a
+        /// different plan, and a storage id cannot tell two occurrences of one table apart.
+        const auto * reused_by = typeid_cast<const ReadFromMergeTree *>(reading_nodes.front()->step.get());
+        chassert(reused_by && reused_by->getStorageID() == analyzed_merge_tree->getStorageID(),
+                 "pre-analyzed result belongs to a different table than the scan that reuses it");
        analyzed_result_ptr = analyzed_merge_tree->getAnalyzedResult();
+    }
}

This means the analysis is only reused when it is safe to do so (i.e., not for a scan reached via a RIGHT JOIN right-branch).

4. Regression test

A new regression test is added to ensure that the bug is fixed and does not reappear. The test checks correct results for RIGHT JOINs, including edge cases like self-joins and unmatched rows, and verifies that parallel replicas are actually used:

-- Projecting a left column under `RIGHT JOIN` (issue's Error 10 case).
SELECT r.ver, (l.a + 2) FROM tl AS l RIGHT JOIN tr AS r USING (k) ORDER BY r.ver LIMIT 5;

-- Unmatched right rows: the left side must come back as defaults, not as wrong values.
SELECT r.k, r.ver, (l.a + 2) FROM tl AS l RIGHT JOIN tr AS r USING (k) ORDER BY r.k DESC LIMIT 5;

-- Self `RIGHT JOIN`: both sides share the same storage; the analysis reuse must still be skipped
-- for the right-branch scan.
SELECT r.k, (l.a + 2) FROM tl AS l RIGHT JOIN tl AS r USING (k) ORDER BY r.k LIMIT 5;

From the review thread

| a | Deterministic repro? | Yes. SELECT r.ver, (l.a + 2) FROM tl AS l RIGHT JOIN tr AS r USING (k) with parallel_replicas_min_number_of_rows_per_replica=100 on a 3-replica cluster fails every run with Code: 10 NOT_FOUND_COLUMN_IN_BLOCK: Column a not found in table tr. |

| c | Fix matches root cause? | Yes. Reuse the pre-analyzed result only when the parallelized scan was NOT reached via a RIGHT-JOIN right-branch descent; otherwise pass nullptr so the scan analyzes itself (the same path the working min_rows=0 case takes). Directly addresses the mismatched-analysis reuse. |

Takeaways

Never reuse analysis or metadata for a scan unless you are certain it applies to the specific table and context being scanned. Join direction and plan traversal can change which table is being parallelized.
When fixing query-planner bugs, add regression tests that cover both the error case and the silent-wrong-result case, including edge cases like self-joins and unmatched rows.
Testing for parallel execution features should assert that the intended execution path is actually taken (e.g., by checking the query plan), not just that the result is correct.