rustyconover opened a new issue, #51047:
URL: https://github.com/apache/arrow/issues/51047
## Title
[C++][Acero] record_batch_reader_source doesn't seem to apply backpressure
## 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)
--
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]
