A FULL or LEFT hash join whose probe side gives no row emits its unmatched build rows as one batch, ignoring batch_size

Aperta
  • Qualcosa non va?
#27,379 1 commento 0 reazioni 1 assegnatario Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@lakshaydevoo-arch ci sta già lavorando.

Dal 12/11/2026.

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

bug
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:

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

Lingua principale
Rust
Stelle
9.6k
Fork
2.6k
Merge medio
3g 4h
PR unite (30g)
345

Preparare l'ambiente

Apri in Codespaces

Avvia il container di sviluppo del progetto nel browser, con il tuo account GitHub.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/datafusion

Tutte le issue di apache/datafusion

Issue simili

Altre issue su Rust

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.