Distributed execution returns incorrect results for several TPC-DS queries (static planner)

[andygrove]

Describe the bug

Under the default (static) planner, Ballista returns results for several TPC-DS queries that differ from single-process DataFusion executing the same SQL over the same data. The differences are stable across repeated runs (not query non-determinism), and single-process DataFusion returns the same answer under both prefer_hash_join=true and prefer_hash_join=false, so this is not a join-strategy artifact — it points to a distributed-execution correctness problem.

Surfaced by a TPC-DS correctness cross-check (SF1) that runs each query on a Ballista cluster and diffs the result against a single-process DataFusion oracle over the same Parquet data (the mechanism added by the TPC-H --verify gate, extended to TPC-DS).

Confirmed cases (SF1, target_partitions=16, prefer_hash_join=false)

Query	Shape	Single-process DataFusion	Ballista (distributed)
q38	count(*) over INTERSECT of 3 channel customer sets	104	0
q87	count(*) over EXCEPT	46753	47011
q78	… ORDER BY … LIMIT 100	100 rows	73 rows (27 dropped)
q4	year-over-year CTE ratios, LIMIT 100	8 rows	0 rows

• Stable, not ties: q38/q78/q87 reproduce identically across repeated Ballista runs, and single-process DataFusion is stable too. (Queries whose results genuinely vary run-to-run due to LIMIT/ORDER BY ties are a separate matter and are excluded from this report.)
• Not a join-config artifact: single-process DataFusion returns the correct answer under both prefer_hash_join=true and false at target_partitions=16 (verified with datafusion-cli). Only distributed vs single-process differs.

The INTERSECT/EXCEPT (q38, q87) and LIMIT row-drop (q78) symptoms suggest the set-operation and limit lowering across shuffle stages as the area to look at.

To Reproduce

Generate SF1 TPC-DS Parquet, start a scheduler + executor, then for the distributed result:

tpcds --host localhost --port 50050 --path <data> --partitions 16 --verify \
  --query 38 -c datafusion.optimizer.prefer_hash_join=false
# -> Query 38 VERIFY MISMATCH: expected 104, got 0

Single-process baseline (both join modes give 104):

datafusion-cli:
  SET datafusion.execution.target_partitions = 16;
  SET datafusion.optimizer.prefer_hash_join = false;   -- and = true
  <register the 24 TPC-DS tables as external parquet>
  <q38 body>   -- -> 104

Expected behavior

Distributed execution returns the same results as single-process DataFusion for the same query and data.

Additional context

• Reproduced at SF1 on a 1-scheduler / 1-executor native cluster (8 slots, 16 partitions), base apache/main @ f45996c.
• Related: Some benchmark queries fail or produce incorrect results #330 (older, general "some benchmark queries produce incorrect results"); Left/full outer join incorrect for CollectLeft / broadcast #1055 (CollectLeft/broadcast outer-join correctness — likely distinct, as this reproduces on the static planner with prefer_hash_join=false, i.e. sort-merge joins, no broadcast).
• The TPC-DS correctness harness that surfaced this is proposed in the PR for Add TPC-DS workflow #1845.

[andygrove]

Update: this issue turned out to bundle two unrelated root causes, one per pair of queries.

q38 / q87 (set operations) — datafusion-proto's logical JoinNode decode rebuilt the join via LogicalPlanBuilder, whose join_with_expr_keys / join_using hardcode NullEqualsNothing, so the encoded null_equality was dropped. Ballista ships the client's logical plan to the scheduler, so INTERSECT / EXCEPT (which carry NullEqualsNull) lost NULL-matching: NULL stopped matching itself, so INTERSECT dropped NULL rows and EXCEPT retained them. Fixed upstream by apache/datafusion#22104, backported to branch-54 as apache/datafusion#22785 — not in the released 54.0.0 tag. PR #2064 (draft until a 54.x patch release is out).

q4 / q78 (joins) — a separate bug, #2065: broadcast promotion rewrites a SortMergeJoinExec into a HashJoinExec(CollectLeft), which does not preserve input order, so a parent SortMergeJoinExec merges unsorted input and silently drops rows. Fixed in #2066.

Note the q78 description above ("distributed LIMIT drops rows") is a red herring — the wrong answer does not depend on ORDER BY or LIMIT and reproduces with neither. Wrapping the query in count(*) changes the plan enough to hide the bug, which is what made it look limit-related.

With both PRs the TPC-DS correctness gate goes from 86 to 90 queries.