ndscoder opened a new issue, #27379:
URL: https://github.com/apache/datafusion/issues/27379
### Describe the bug
A `FULL` or `LEFT` join planned as `HashJoinExec: mode=CollectLeft` whose
right (probe) side yields no row emits every unmatched row of its left (build)
side as a single batch, whatever `datafusion.execution.batch_size` is. With one
probe row, matching or not, the same rows come in batches of `batch_size`.
This matters under a memory pool. A `SortExec` above the join reserves about
twice the batch's size (9.2 MB below, for a batch of about 4.6 MB) before it
holds anything it could spill, so the statement fails with `Resources exhausted
... for ExternalSorter[0] with 0.0 B already allocated`, where the same rows in
batches of 8192 sort and spill within the pool.
Settings that did not change it: `datafusion.execution.batch_size = 1000`
(still one batch of 200000), `enforce_batch_size_in_joins = true`,
`coalesce_batches = false`, `hash_join_buffering_capacity = 0` and `1048576`.
### To Reproduce
With datafusion-python 55.0.0:
```python
import pyarrow as pa
from datafusion import RuntimeEnvBuilder, SessionConfig, SessionContext
rows = 200_000
table = pa.table({"k": pa.array(range(rows), pa.int64()), "v":
pa.array(range(rows), pa.float64())})
def context(pool_bytes=None):
config =
SessionConfig().with_target_partitions(1).set("datafusion.execution.sort_spill_reservation_bytes",
str(2**20))
runtime = RuntimeEnvBuilder().with_disk_manager_os()
if pool_bytes:
runtime = runtime.with_fair_spill_pool(pool_bytes)
ctx = SessionContext(config, runtime)
ctx.register_record_batches("a", [table.to_batches(max_chunksize=8192)])
ctx.register_record_batches("b", [table.to_batches(max_chunksize=8192)])
return ctx
for where in ("k < 0", "k < 1"): # the probe side gives no row, then one row
sql = f"SELECT a.k, a.v, b.v AS w FROM a FULL JOIN (SELECT * FROM b
WHERE {where}) b ON a.k = b.k"
sizes = [batch.to_pyarrow().num_rows for batch in
context().sql(sql).execute_stream()]
print(f"WHERE {where}: {len(sizes)} batches, largest {max(sizes)}")
try:
sorted_rows = sum(b.to_pyarrow().num_rows for b in context(8 *
2**20).sql(sql + " ORDER BY a.v DESC").execute_stream())
print(f" ORDER BY under an 8 MB fair spill pool: sorted
{sorted_rows} rows")
except Exception as error:
print(f" ORDER BY under an 8 MB fair spill pool:
{str(error).splitlines()[-1]}")
```
Output:
```
WHERE k < 0: 1 batches, largest 200000
ORDER BY under an 8 MB fair spill pool: Resources exhausted: Failed to
allocate additional 9.2 MB for ExternalSorter[0] with 0.0 B already allocated
for this reservation - 5.3 MB remain available for the total memory pool:
fair(pool_size: 8.0 MB)
WHERE k < 1: 25 batches, largest 8192
ORDER BY under an 8 MB fair spill pool: sorted 200000 rows
```
The full error starts with `Not enough memory to continue external sort.
Consider increasing the memory limit config ...`. The plan:
```
SortExec: expr=[v@1 DESC], preserve_partitioning=[false]
ProjectionExec: expr=[k@0 as k, v@1 as v, v@2 as w]
HashJoinExec: mode=CollectLeft, join_type=Full, on=[(k@0, k@0)],
projection=[k@0, v@1, v@3]
DataSourceExec: partitions=1, partition_sizes=[25]
FilterExec: k@0 < 0
DataSourceExec: partitions=1, partition_sizes=[25]
```
`LEFT JOIN` gives the same batches (`[200000]` against batches of 8192).
Reading the two tables from Parquet does too; there, with `collect_statistics =
true`, the planner makes the side without rows the build side and the rows come
in batches.
### Expected behavior
The unmatched build rows emitted in batches of `batch_size`, as they are
once the probe side has given a row, so the `ORDER BY` above sorts and spills
within the pool in both cases.
### Additional context
The probe side giving no row is common in practice: a filter on the joined
table that matches nothing, such as a key range one table holds no row of. A
query then fails or succeeds depending on whether one table happens to be empty
for it, with the same memory pool.
### Willingness to contribute
I cannot contribute a fix for this bug at this time
--
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]
