apache / apache/arrow

[C++][Acero] record_batch_reader_source doesn't seem to apply backpressure

Open
#51,047 1 comment 0 reactions 0 assignees View on GitHub
Dominant language
C++
Stars
17.1k
Forks
4.3k
Avg merge
3d 13h
Merged PRs (30d)
88

Description

## Body

Hi, and first off — thank you for #50801/#50802! `RecordBatchReaderSourceNodeOptions` is exactly what I needed for a project bridging a generator-backed data source into Acero, and it's working great.

While using it, I noticed the source seems to drain a `RecordBatchReader` at full speed regardless of how quickly (or whether at all) the downstream consumer is actually asking for batches. I'm not sure if this is expected given the backpressure work still in progress in #47383, or a separate gap specific to this node — flagging it in case the repro is useful data either way, no urgency on my end.

### Repro

Pure pyarrow, no other dependencies:

```python
import time
import pyarrow as pa
import pyarrow.acero as ac

N = 200
DELAY = 0.05 # simulate a producer that takes real time per batch
table = pa.table({"a": list(range(N))})
batches = table.to_batches(max_chunksize=1)
pulled_count = [0]

def gen():
for i, batch in enumerate(batches):
time.sleep(DELAY)
pulled_count[0] = i + 1
yield batch

reader = pa.RecordBatchReader.from_batches(table.schema, gen())
decl = ac.Declaration("record_batch_reader_source", ac.RecordBatchReaderSourceNodeOptions(reader))
out_reader = decl.to_reader()

t0 = time.monotonic()
out_reader.read_next_batch() # one read to start the plan
print("after first read:", pulled_count[0])

# Never ask for another batch -- just watch what happens anyway.
for _ in range(10):
time.sleep(0.3)
print(f"{time.monotonic() - t0:.2f}s: pulled_count={pulled_count[0]}/{N}")
```

### Observed

```
after first read: pulled_count=1
0.36s: pulled_count=6/200
0.67s: pulled_count=12/200
0.97s: pulled_count=18/200
...
9.18s: pulled_count=171/200
```

`pulled_count` climbs steadily the entire time with no plateau, even though `read_next_batch()` is only ever called once. It reaches 200/200 well before anything downstream has asked for more than the first batch.

### Expected (maybe?)

Given `SourceNode`'s general backpressure machinery (`BackpressureOptions`, the `backpressure_future_` in the read loop), I'd have expected the read loop to eventually pause once enough unconsumed batches have piled up — though I understand `BackpressureOptions` isn't wired up by `Declaration.to_table()`/`.to_reader()` at all today (`pause_if_above` defaults to 0 / disabled), so maybe this is simply "working as currently designed, nothing to configure it with yet." Genuinely not sure whether this belongs under #47383's umbrella or is worth its own tracking — happy to help however's useful, and thanks again for the node, it's a real improvement for exactly the kind of custom-source bridging I'm doing.

### Versions

- pyarrow: `26.0.0.dev175` (scientific-python-nightly-wheels)
- Platform: macOS arm64 (also reproduced on Linux aarch64)

Contributor guide

Open the contributing guide

Research direction

Trace record_batch_reader_source through RecordBatchReaderSourceNodeOptions and SourceNode's read loop, focusing on the backpressure_future_ and BackpressureOptions behavior described in #47383. Reproduce the supplied pyarrow script and compare it with Declaration.to_reader(); done means the pulling behavior is covered by a focused regression test or clearly established as current behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
cpp, python
Domain
data-engineering, stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.