andygrove opened a new issue, #2656:
URL: https://github.com/apache/datafusion-ballista/issues/2656
**Describe the bug**
TPC-H q5 at SF1000 cannot complete: executors are OOM-killed in the stage
that sort-merge joins `orders` and `lineitem` on the order key. Executor memory
climbs steadily (~1 GB/s) to the pod limit while the DataFusion memory pool
reservation stays flat at ~9 GB, so the memory pool never sees the growth and
nothing spills to relieve it. Each lost executor makes AQE roll back and re-run
the upstream stages, which then hit the same wall, so the job never finishes.
A jemalloc heap profile of one executor shows the untracked memory is remote
shuffle blocks that the shuffle reader has fully decoded into memory:
| Path | Heap at 43 GB peak |
|---|---|
| `shuffle_reader::send_fetch_partitions` → `fetch_partition_buffered` →
`try_collect` → IPC decode | **36.4 GiB (86.5%)** |
| Local shuffle read (`MultiStreamPartitionStream`, streamed) | 2.1 GiB |
| SMJ `advance_buffered_group` | 1.7 GiB |
| Other | 1.8 GiB |
Between a dump taken just after the sort spilled (19 GB) and the peak (43
GB), 96% of the net growth is on the remote-fetch path.
`fetch_partition_buffered` collects a whole remote partition into a
`Vec<RecordBatch>` before yielding anything (added in #1951 so a mid-body
transport failure can be retried by refetching). Three things combine to make
this unbounded in practice:
1. **Whole blocks are materialized.** The `max_bytes_in_flight` governor
only decides *when* a fetch may start. A block larger than the budget takes the
whole budget and is still decoded in full. Here the `lineitem` shuffle blocks
are ~1 GB each decoded.
2. **The budget is per reader partition, not per executor.** The semaphores
are created inside `send_fetch_partitions`, so each output partition of each
`ShuffleReaderExec` gets its own 48 MiB budget. This stage runs 8 partitions ×
2 shuffle readers = 16 independent budgets per executor.
3. **None of it is charged to the memory pool.** The consumer (`SortExec`
feeding the SMJ) is spilling and consumes slowly, while decoded blocks it has
not yet pulled sit outside the pool.
**To Reproduce**
- `apache/main` at b89a4e465
- 2 executors (8 vCPU, 56 GiB limit, `--memory-pool-size=48GB`), 1
scheduler, TPC-H SF1000 Parquet on local disk
- `tpch benchmark ballista --query 5 --path .../sf1000 --format parquet
--partitions 16 --iterations 1`
A reduced query with the same join shape whose join output feeds an
aggregate (`select count(*), sum(l_extendedprice * (1 - l_discount)) from
customer, orders, lineitem where c_custkey = o_custkey and l_orderkey =
o_orderkey and o_orderdate >= date '1994-01-01' and o_orderdate < date
'1995-01-01'`) fails the same way, which rules out the sort shuffle writer. q3,
which has the same `orders` ⋈ `lineitem` SMJ but a filtered `lineitem` side
about half the size, completes.
Stage 4 of q5 as resolved by AQE:
```
SortShuffleWriterExec: partitioning=Hash([l_suppkey@1, c_nationkey@0], 16)
ProjectionExec: ...
SortMergeJoinExec: join_type=Inner, on=[(o_orderkey@1, l_orderkey@0)]
SortExec: expr=[o_orderkey@1 ASC], preserve_partitioning=[true]
ShuffleReaderExec: upstream_stage: 2, partitioning:
Hash([o_orderkey@1], 16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
ShuffleReaderExec: upstream_stage: 3, partitioning:
Hash([l_orderkey@0], 16)
```
Executor memory log during the stage (allocated vs pool reserved):
```
08:35:15 alloc=15839 MiB reserved=15094 MiB
08:35:25 alloc=13096 MiB reserved= 7399 MiB <- sort spills
08:35:45 alloc=18957 MiB reserved= 5072 MiB
08:35:55 alloc=26459 MiB reserved= 8341 MiB
08:36:05 alloc=34226 MiB reserved= 9333 MiB
08:36:16 alloc=43054 MiB reserved= 9515 MiB
08:36:28 OOMKilled
```
**Expected behavior**
The shuffle reader should never hold an entire remote partition in memory.
Reduce-side memory should be bounded by the configured fetch budget (or be
charged to the memory pool so the consumer spills), independent of partition
size and of the number of reader partitions per executor.
**Additional context**
#1952 tracks the disk-spill enhancement for the same buffering. Possible
fixes:
- Stream remote blocks through instead of collecting them, holding the
governor permits for the life of the stream. Retry a fetch only if it fails
before the first batch is yielded; a mid-body failure fails the task so the
scheduler re-runs it (Spark's `FetchFailed` behavior).
- Spill blocks above a threshold to local disk (#1952), keeping whole-block
retry.
- Either way, make the in-flight byte budget per executor rather than per
reader partition.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
