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

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)

• 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.