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

[rustyconover]

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:

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)

[rustyconover]

Hi @westonpace am I missing something for back pressure?