[I] UNION ALL silently returns inflated results when the union stage exceeds one wave of executor task slots [datafusion-ballista]

["andygrove (via GitHub)" <gi...@apache.org>]

andygrove opened a new issue, #2067:
URL: https://github.com/apache/datafusion-ballista/issues/2067

   ## Describe the bug
   
   A `UNION ALL` over two file scans returns **massively inflated results** whenever
   the union's stage has more tasks than the executor has concurrent task slots.
   The query does not error — it silently over-counts, so aggregates come back
   several times too large.
   
   ## To Reproduce
   
   SF1 TPC-DS parquet, 1 scheduler + 1 executor, release build:
   
   ```sql
   select round(sum(p),2) from
    (select ws_ext_sales_price p from web_sales
     union all
     select cs_ext_sales_price p from catalog_sales);
   ```
   
   With `--partitions 16` and an executor started `-c 8`:
   
   ```
   expected 5492796521.50   (single-process DataFusion, same data)
   got     42072988115.00   (~8x inflated; varies run to run)
   ```
   
   Summing each table separately is correct, and the same aggregate over a single
   table (no `UNION ALL`) is correct — only the union is wrong.
   
   Note `select count(*)` over the union is **not** a reproducer: it is answered
   from parquet statistics without reading data. The aggregate must touch a column.
   
   ## The rule: it breaks exactly when the stage exceeds one wave of task slots
   
   A `UnionExec` has as many partitions as its children combined, so an N-partition
   union stage runs `2N` tasks. The result is correct **iff those tasks fit in a
   single concurrent wave** on the executor:
   
   Executor `-c 8`:
   
   | `--partitions` | union stage tasks | result |
   |---:|---:|---|
   | 3 | 6 | correct |
   | 4 | 8 | correct |
   | 5 | 10 | **wrong** (12808834840.20) |
   | 6 | 12 | **wrong** (16466853999.55) |
   | 8 | 16 | **wrong** (16466853999.55) |
   
   Same binary, same data, executor `-c 32` — the threshold moves with the slot
   count, which isolates the cause:
   
   | `--partitions` | union stage tasks | result |
   |---:|---:|---|
   | 5 | 10 | correct |
   | 8 | 16 | correct |
   | 16 | 32 | correct |
   | 20 | 40 | **wrong** (20124873158.90) |
   
   So: **correct iff `2 * partitions <= executor slots`.**
   
   ## Suspected cause
   
   `restrict_scan_to_partition` (`ballista/executor/src/execution_engine.rs`) exists
   precisely because DF54's `DataSourceExec` hands file groups out from a shared
   work-queue that is only divided correctly when all partitions of a plan instance
   are polled concurrently (#1907). It pins each task's scan to its own file group —
   but bails out early:
   
   ```rust
   if partition_id >= config.file_groups.len() {
       return None;   // no restriction -> scan left free to drain the whole queue
   }
   ```
   
   Its own doc comment anticipates this case:
   
   > Returns `None` ... for a `partition_id` outside the source's file groups (e.g.
   > when an operator between the scan and the stage output changed the partition
   > count).
   
   A `UnionExec` is exactly such an operator: with `target_partitions=16` each scan
   has 16 file groups but the union stage has 32 partitions, so tasks 16..31 skip
   the restriction and their scans are left unrestricted. An unrestricted scan only
   behaves when the whole stage happens to be polled in one concurrent wave, which
   is precisely the boundary measured above.
   
   Fixing it presumably means mapping the stage's output partition to each scan's
   *local* partition index (a union offsets its right child's partitions by the left
   child's count) rather than using the stage `partition_id` directly, so every
   scan is pinned to the right file group.
   
   ## Impact
   
   - **16 of the 99 TPC-DS queries use `UNION ALL`** (q2, q4, q5, q11, q14, q23,
     q33, q54, q56, q60, q66, q71, q74, q76, q77, q80), so this is broadly reachable
     on any realistic partitions/slots ratio.
   - It accounts for the q2 and q5 failures currently skipped in the TPC-DS
     correctness gate as "non-deterministic (ORDER BY ties)". That label is wrong:
     with a total order imposed and `LIMIT` removed, both still mismatch, so they
     are real wrong results, not tie ambiguity. (q31 and q71 *are* genuinely
     ordering-only — both verify OK once a total order is imposed.)
   - Silent: nothing errors, results are just too large.
   
   ## Expected behavior
   
   `UNION ALL` returns the same result as single-process DataFusion, independent of
   `target_partitions` and executor slot count.
   
   ## Additional context
   
   - Reproduced at SF1 on a 1-scheduler / 1-executor native cluster, `upstream/main`
     (e843c8e8), default static planner, under both `prefer_hash_join=true` and
     `false`.
   - Regression surface introduced with the DataFusion 54 upgrade; closely related to
     #1907, whose fix covers the non-union case (a plain scan verifies correctly at
     any partition/slot ratio) but leaves partitions beyond `file_groups.len()`
     unrestricted.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] UNION ALL silently returns inflated results when the union stage exceeds one wave of executor task slots [datafusion-ballista]

["andygrove (via GitHub)" <gi...@apache.org>]

andygrove commented on issue #2067:
URL: https://github.com/apache/datafusion-ballista/issues/2067#issuecomment-4994615135

   Fixed in #2070.
   
   Root cause: `restrict_scan_to_partition` applied the stage's partition number directly to every scan, but a `UnionExec` concatenates its children's partitions — stage partition `N + k` is the right child's local partition `k`. For partitions past a scan's group count the guard bailed out and left that scan **unrestricted**, free to read its whole table.
   
   The fix walks the plan from the stage root and mirrors `UnionExec::execute`'s own partition arithmetic, pinning each scan to the group its local partition reads. The minimal case is now correct at every partition count tried (4, 5, 6, 8, 16, 20, 32), and q2/q5 verify in their original form and join the gate.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] UNION ALL silently returns inflated results when the union stage exceeds one wave of executor task slots [datafusion-ballista]

["andygrove (via GitHub)" <gi...@apache.org>]

andygrove closed issue #2067: UNION ALL silently returns inflated results when the union stage exceeds one wave of executor task slots
URL: https://github.com/apache/datafusion-ballista/issues/2067


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org