unlock unlimited code runs for $79, limited time →
NOSSLETTER.techLIVE
back to OSSpedia
ClickHouse/ClickHouse
senior8/10
query-optimizationalgorithmic-optimizationconsistency-models

Fix wrong results with parallel_hash JOIN and read-in-order-through-join

Project: ClickHouse/ClickHouse | Language: C++ | Concepts: query-optimization, algorithmic-optimization, consistency-models | Difficulty: senior
Link: https://github.com/ClickHouse/ClickHouse/pull/109225

The problem

When using the parallel_hash join algorithm in ClickHouse, if the left table's sort order was relied upon by downstream optimizations (like optimize_aggregation_in_order, optimize_distinct_in_order, or LIMIT BY), the join sometimes broke the order, leading to wrong query results. This happened because the join could scatter left-side rows across multiple slots, destroying the original order, but the query planner still assumed the order was preserved and enabled optimizations that require sorted input.

Why it's non-trivial

A naive fix might simply disable the optimization for all joins, but that would unnecessarily degrade performance for join types that do preserve order. The challenge is to accurately detect which join implementations truly preserve the left input order, and only propagate the "sorted" property through those joins. This requires deep understanding of join internals and careful gating of optimization propagation, rather than a blanket disable.

The approach

The PR introduces a new contract: joins must explicitly declare if they preserve the left block order by overriding preservesLeftBlockOrder(). The default is now false (fail-closed), so only joins that are known to preserve order opt in.

1. Change the default in the base interface to fail-closed:

-    /// Whether the join emits left rows in the same order they arrive. HashJoin/DirectJoin/ConcurrentHashJoin
-    /// stream the probe side, so they do. PartialMergeJoin re-sorts left blocks by the join key, so it does not;
-    /// the read-in-order-through-join optimisation in optimizeReadInOrder.cpp must not propagate through such joins.
-    virtual bool preservesLeftBlockOrder() const { return true; }
+    /// Whether the join emits left rows in their original stream order. Read-in-order relies on
+    /// this to keep the left sort property, so the default is fail-closed: a join has to opt in.
+    virtual bool preservesLeftBlockOrder() const { return false; }

Annotation:
The default is now false, so unless a join explicitly says it preserves order, the planner will not propagate the "sorted" property through it. This prevents accidental propagation through joins that might reorder the data.

2. Opt-in for joins that truly preserve order:

For each join implementation that does preserve the left block order, override the method to return true with a comment explaining why.

HashJoin:

+    /// The left side is streamed through once, each row emitted in input order.
+    bool preservesLeftBlockOrder() const override { return true; }

DirectKeyValueJoin:

+    /// Each left row's key is looked up once and the row is emitted in input order.
+    bool preservesLeftBlockOrder() const override { return true; }

ConstantJoin:

+    /// Every result class walks the probe block by ascending row index.
+    bool preservesLeftBlockOrder() const override { return true; }

PasteJoin:

+    /// The left and right blocks are concatenated side by side by row position.
+    bool preservesLeftBlockOrder() const override { return true; }

ConcurrentHashJoin (conditional):

+    /// A single slot passes the left block through unscattered, and so does a shared two-level map.
+    /// With a single-level map (a key materializing to UInt8 or UInt16) the block is scattered
+    /// across the slots and emitted slot by slot, so equal left key values stop being contiguous.
+    bool preservesLeftBlockOrder() const override { return slots == 1 || hash_joins[0]->data->twoLevelMapIsUsed(); }

Annotation:
This logic means order is only preserved if the join is not scattering left rows across slots, i.e., either there's only one slot, or a two-level map is used (which keeps order).

3. Update optimization propagation logic and documentation:

In optimizer comments and logic, clarify that order is only propagated through joins that declare preservesLeftBlockOrder() == true:

-///  2. The join preserves left order but its physicalization inserts a `Sort ... before
-///     JOIN` on the preserved input (`FullSortingMergeJoin`, `full_sorting_merge`; see
+///  2. The join's physicalization inserts a `Sort ... before JOIN` on the preserved input
+///     (`FullSortingMergeJoin`, `full_sorting_merge`; see

Annotation:
The comments and logic in the optimizer are updated to explain the new contract and why it's necessary.

4. Add regression test to prove correctness:

A new test is added that would fail on the old code, but passes with the fix:

-- 8 groups of 8 rows each. The test runner randomizes the settings below, and the ordered read only
-- happens with read-in-order enabled, the join sides unswapped and the join not spilling.
SELECT a, count() FROM l LEFT JOIN r ON l.j = r.j GROUP BY a ORDER BY a
SETTINGS join_algorithm = 'parallel_hash', max_threads = 8, optimize_aggregation_in_order = 1,
         optimize_read_in_order = 1, query_plan_read_in_order = 1,
         query_plan_read_in_order_through_join = 1, query_plan_join_swap_table = 'false',
         max_bytes_before_external_join = 0, max_bytes_ratio_before_external_join = 0,
         enable_parallel_replicas = 0;

Annotation:
This test would previously produce wrong results if the join broke order, but now reliably produces the correct output.

From the review thread

[Reviewer, AI Review]:
Flipping IJoin::preservesLeftBlockOrder() to fail-closed changes every join until it opts back in again. This head adds a good parallel_hash regression, but I still could not find a focused read_in_order_through_join probe for the new ConstantJoin/DirectKeyValueJoin opt-ins: 03711_read_in_order_through_join only covers HashJoin/ConcurrentHashJoin, while the current direct/constant tests only prove algorithm selection or unrelated bugs. Please add one small LEFT JOIN ... ON 1 = 1 probe for ConstantJoin and one dictionary LEFT JOIN probe for DirectKeyValueJoin with pinned read-in-order settings, asserting the left ReadType: InOrder (or another guard-sensitive sorted-consumer result).

[Author, groeneai]:
cc @vdimir @KochetovNicolai — could you review this? It stops the read-in-order-through-join optimization from advertising the left sort order across PartialMergeJoin (which re-scans left key ranges per right block and so does not preserve it), fixing wrong DISTINCT results in release and the sorted-DISTINCT LOGICAL_ERROR in debug (#109216). New IJoin::preservesLeftBlockOrder() gates the propagation; hash/direct joins are unchanged.

Takeaways

  • Fail-closed is safer than fail-open for correctness contracts: When an optimization or property (like sortedness) is only valid for certain implementations, default to "off" and require explicit opt-in. This prevents subtle correctness bugs.
  • Optimization propagation must be guarded by true behavioral contracts: Only propagate properties (like sorted order) through operators that guarantee them, not just those that "usually" do.
  • Tests should target the contract boundary: When changing a contract like order preservation, add regression tests that would fail if the contract is broken, and ensure all opt-in cases are covered.