GROUP BY returns the same key in multiple rows when the left side of a JOIN is SELECT DISTINCT (parallel_hash + optimize_distinct_in_order #114352

Company or project name

No response

Describe what's wrong

A GROUP BY k on top of a JOIN whose left side is a SELECT DISTINCT subquery intermittently returns the same group key in two separate rows, with the aggregate values split between them. The sum across the duplicate rows always equals the correct single-row value, so no data is lost or duplicated, the group is emitted in more than one piece.

The same query text, same data, same settings returns 3 rows most of the time and 4 rows occasionally (~4–10% under server defaults). system.query_log confirms identical Settings and identical query across both outcomes, so the plan is identical between a correct and an incorrect run.

Turning off optimize_distinct_in_order eliminates it completely. Replacing parallel_hash with any other join algorithm eliminates it completely. Rewriting SELECT DISTINCT a,b,c as SELECT a,b,c GROUP BY a,b,c (semantically identical, verified byte-identical output) also eliminates it completely.

Reproducer with sample data attached , see "How to reproduce" below for the full CREATE TABLE + load + query script.

Does it reproduce on the most recent release?

Yes

How to reproduce

Reproduced on ClickHouse Cloud 26.2.1.558. Have not independently verified against the latest OSS release, happy to if a maintainer confirms it would help.

• ClickHouse server version: 26.2.1.558 (ClickHouse Cloud)
• Interface: observed via both the native protocol (Go driver) and HTTP
• Non-default settings: NONE required, reproduces under pure server defaults. Forcing join_algorithm='parallel_hash', max_threads=4 raises the failure rate from ~5–25% to ~85–95%.

Sample data attached below (dim_part1..6.tsv.gz, facts.tsv.gz — load order matters, see load step below).

CREATE DATABASE IF NOT EXISTS repro;

CREATE TABLE repro.dim
(
    sync_version            UInt64,
    is_deleted              Bool,
    tenant_id               UInt64,
    grp                     LowCardinality(String),  -- the GROUP BY key
    k3                      UInt64,                  -- sort-key filler
    k2                      Int64,                   -- join key
    k4                      UInt64,                  -- sort-key filler
    k5                      Date,                    -- sort-key filler
    k1                      String                   -- join key
)
ENGINE = ReplacingMergeTree(sync_version)
ORDER BY (tenant_id, grp, k3, k2, k4, k5)
SETTINGS max_bytes_to_merge_at_max_space_in_pool = 1;  -- keep the loaded parts distinct

CREATE TABLE repro.facts
(
    tenant_id  UInt64,
    d          Date,
    input_id   UInt64,
    load_id    UUID,
    grp        LowCardinality(String),
    k2         Int64,
    k1         String,
    k6         String,  -- sort-key filler
    k7         String,  -- sort-key filler
    measure    UInt64
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(d)
ORDER BY (tenant_id, d, input_id, load_id, grp, k2, k1, k6, k7);

dim_part1.tsv.gz
dim_part2.tsv.gz
dim_part3.tsv.gz
dim_part4.tsv.gz
dim_part5.tsv.gz
dim_part6.tsv.gz
facts.tsv.gz

Expected behavior

GROUP BY should return each group key exactly once. Instead, under the conditions above, one key is intermittently split across two rows whose values sum to the correct total.

Error message and/or stacktrace

None, no exception is raised. The query completes successfully but returns incorrect (split) results.

Related issues and pull requests

Related: #109216

Additional context

Environment

• ClickHouse Cloud, version 26.2.1.558 (official build)
• join_algorithm = direct,parallel_hash,hash (server default)
• max_threads = auto(N); observed at both 3 and 6 as the service scaled
• optimize_distinct_in_order = 1 (default)
• distributed_group_by_no_merge = 0 (default)
• enable_parallel_replicas = 0
• Single-node query (not distributed)

Reproducing query (structure)

Schema shape (names genericized):

-- left/dimension table
CREATE TABLE dim
(
    sync_version UInt64,
    is_deleted   Bool,
    tenant_id    UInt64,
    grp          LowCardinality(String),   -- the GROUP BY key
    k1           String,                   -- join key
    k2           Int64                     -- join key
)
ENGINE = ReplacingMergeTree(sync_version)
ORDER BY (tenant_id, grp, k1, k2, ...);    -- NOTE: grp is a sort-key prefix column

-- right/fact table
CREATE TABLE facts
(
    tenant_id UInt64,
    d         Date,
    k1        String,
    k2        Int64,
    measure   UInt64
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(d)
ORDER BY (tenant_id, d, ..., k1, k2, ...);

Query:

SELECT
    m.grp,
    sum(a.measure) AS cnt
FROM (
    SELECT DISTINCT grp, k1, k2
    FROM dim FINAL
    WHERE tenant_id = ? AND is_deleted = 0
) AS m
INNER JOIN (
    SELECT k1, k2, measure
    FROM facts
    WHERE tenant_id = ? AND d BETWEEN ? AND ?
) AS a USING (k1, k2)
GROUP BY m.grp

Relevant data characteristics: the left side yields ~580k distinct rows across 3 values of grp, heavily skewed — one group holds ~97% of the rows, the other two are small. Only the large group ever splits; the small ones never do. The dim table has multiple active parts (EXPLAIN PIPELINE shows a 9-way ReplacingSorted/MergeTreeSelect(ReadPoolInOrder, InOrder) fan-in).

Note: this simplified shape is illustrative only and did not reproduce standalone in our testing, see "How to reproduce" above for the exact schema/data that does.

Observed result

Correct (majority of runs):

grp_small_1     828
grp_small_2  173923
grp_large    224701

Incorrect (same query, same settings, ~4–10% of runs):

grp_large       254      <-- same key
grp_small_1     828
grp_small_2  173923
grp_large    224447      <-- same key

254 + 224447 = 224701 — exactly the correct value. The split point differs on every occurrence (observed 15341/209360, 21190/203511, 28030/196671, …), while the total is always identical.

Evidence that the plan is identical between correct and incorrect runs

40 executions of the identical query under defaults, then:

SELECT result_rows, uniqExact(Settings) AS distinct_setting_sets, uniqExact(query) AS distinct_queries
FROM system.query_log
WHERE query LIKE '%<marker>%' AND type = 'QueryFinish'
GROUP BY result_rows

result_rows  distinct_setting_sets  distinct_queries
3            1                      1
4            1                      1

Same query text, same settings on both sides — so this is probably a runtime race, not a plan/semantics difference.

EXPLAIN PLAN sorting=1 — the only difference between broken and fixed

optimize_distinct_in_order = 1 (broken):

Aggregating
  Expression
    Join (JOIN FillRightFirst)
      Expression (Left Pre Join Actions)
          Distinct (DISTINCT)
              Distinct (Preliminary DISTINCT)
              Sorting: __table2.grp ASC                        <-- only in the broken plan
                  ReadFromMergeTree (dim)
                  Sorting: tenant_id ASC, grp ASC              <-- only in the broken plan

optimize_distinct_in_order = 0 (correct): byte-identical plan minus the two Sorting: lines.
EXPLAIN PIPELINE differs in exactly one transform:

broken:  DistinctSortedStreamTransform × 4
correct: DistinctTransform × 4

Everything else — Resize 4 → 1, the final single-stream DistinctTransform, Resize 1 → 4, JoiningTransform × 4, AggregatingTransform × 4 — is identical.

Note that in both plans the Distinct (DISTINCT) step is fully below the join, i.e. deduplication completes on a single stream before any join work happens.

Controls (each figure is an independent measurement on the same data)

optimize_read_in_order = 0 and optimize_aggregation_in_order = 0 do not help (15/16 and 14/16 still incorrect), and neither removes the Sorting: annotation from the plan — the annotation and the bug appear and disappear together across every variation tested.

The DISTINCT subquery on its own is always correct: run standalone 50 times under pure defaults it returned the same 580865 rows every time, with zero duplicate (grp, k1, k2) tuples — including for tuples that genuinely have multiple physical rows in the table. SELECT DISTINCT a,b,c and SELECT a,b,c GROUP BY a,b,c produce identical row counts and identical sum(cityHash64(...)) checksums.

Oddity: CAST vs toString

Wrapping the group key in toString(grp) keeps the bug (16/20, 20/20). Wrapping it in CAST(grp AS String) removes it entirely (0/20 forced, 0/30 defaults), even though:

• both produce identical output (same row count, same checksum),
• both keep the Sorting: annotation in EXPLAIN PLAN (toString(...) ASC vs CAST(..., 'String') ASC),
• and both produce a byte-identical EXPLAIN PIPELINE, including DistinctSortedStreamTransform × 4.

We could not explain this from EXPLAIN output alone; mentioning it in case it narrows down where the two paths diverge internally.

Why we believe this is a bug rather than expected behaviour

SELECT k, agg(...) ... GROUP BY k returning the same k in more than one row contradicts the definition of GROUP BY, independently of what the subquery below it does. The only setting we found that legitimately permits partial, unmerged group output — distributed_group_by_no_merge — is 0 (default), and this is not a distributed query.

Related

Likely the same family as #109216 (partial_merge + optimize_distinct_in_order → wrong DISTINCT results, sort-order property claimed across a join that does not preserve it). Same control setting, same "assume a stream property the join does not actually preserve" shape, different join algorithm and different direction (there DISTINCT above the join, here below it).

Minimal reproducible example

Attached: dim_part1..6.tsv.gz + facts.tsv.gz and the schema/load/query script below.

Schema

CREATE DATABASE IF NOT EXISTS repro;

-- Column order matches the TSV files exactly -- FORMAT TSV maps positionally,
-- not by name, so do not reorder these.
CREATE TABLE repro.dim
(
    sync_version            UInt64,
    is_deleted              Bool,
    tenant_id               UInt64,
    grp                     LowCardinality(String),  -- the GROUP BY key
    k3                      UInt64,                  -- sort-key filler
    k2                      Int64,                   -- join key
    k4                      UInt64,                  -- sort-key filler
    k5                      Date,                    -- sort-key filler
    k1                      String                   -- join key
)
ENGINE = ReplacingMergeTree(sync_version)
ORDER BY (tenant_id, grp, k3, k2, k4, k5)
SETTINGS max_bytes_to_merge_at_max_space_in_pool = 1;  -- keep the loaded parts distinct, see below

CREATE TABLE repro.facts
(
    tenant_id  UInt64,
    d          Date,
    input_id   UInt64,
    load_id    UUID,
    grp        LowCardinality(String),
    k2         Int64,
    k1         String,
    k6         String,  -- sort-key filler
    k7         String,  -- sort-key filler
    measure    UInt64
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(d)
ORDER BY (tenant_id, d, input_id, load_id, grp, k2, k1, k6, k7);

Load

Load the six dim files in order — each becomes exactly one part, and the resulting part-size skew (four ~560–590k-row parts, two tiny ones) is load-bearing:

for f in dim_part1.tsv.gz dim_part2.tsv.gz dim_part3.tsv.gz \
         dim_part4.tsv.gz dim_part5.tsv.gz dim_part6.tsv.gz; do
  gzip -dc "$f" | clickhouse-client --query "INSERT INTO repro.dim FORMAT TSV"
done
gzip -dc facts.tsv.gz | clickhouse-client --query "INSERT INTO repro.facts FORMAT TSV"

Row counts after loading: dim = 2,288,189 (collapses to 581,668 distinct (grp,k1,k2) triples under FINAL+DISTINCT), facts = 450,338 rows / 225,473 distinct (k1,k2) pairs.

Query, run repeatedly

SELECT m.grp, sum(a.measure) AS cnt
FROM (
    SELECT DISTINCT grp, k1, k2
    FROM repro.dim FINAL
    WHERE tenant_id = 1 AND is_deleted = 0
) AS m
INNER JOIN (
    SELECT k1, k2, measure
    FROM repro.facts
    WHERE tenant_id = 1 AND d BETWEEN '2026-08-08' AND '2026-08-09'
) AS a USING (k1, k2)
GROUP BY m.grp

Under pure server defaults (nothing set), this split 5/20, then 1/20 in two independent 20-run samples on a freshly loaded service — no dependency on the originating instance. Forcing join_algorithm='parallel_hash', max_threads=4 raised it to 17/20 and 19/20 in two samples on that same fresh service.

Correct:

code_2    266225
code_3    182985
code_5      1128

Incorrect (same query, same data, same settings — code_2 appears twice, split):

code_2      5133
code_2    261092
code_3    182985
code_5      1128

5133 + 261092 = 266225 — the correct total, matching the sum-preservation pattern from the original observation above.

We're happy to run any candidate diagnostic query against this same reloaded, anonymized dataset and report back.