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

Abierto
  • ¿Algo no está bien?
#27,379 1 comentario 0 reacciones 1 asignado Ver en GitHub

Los mantenedores suelen responder en 1 día

@lakshaydevoo-arch ya está trabajando en esto.

Desde el 12/11/2026.

Evaluación

Este issue todavía no se ha evaluado.

Descripción

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

Lenguaje dominante
Rust
Estrellas
9.6k
Forks
2.6k
Merge medio
3 d 4 h
PR fusionados (30 d)
345

Preparar el entorno

Abrir en Codespaces

Inicia el contenedor de desarrollo del proyecto en tu navegador, con tu propia cuenta de GitHub.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/datafusion

Todos los issues de apache/datafusion

Issues similares

Más issues de Rust

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.