fix(arrow): spawn data file opens on the IO runtime

[swanandx]

Which issue does this PR close?

• Closes Scan stalls on an HTTP/2 object store when many files are read at once #3276

What changes are included in this PR?

Spawn each data file open on the IO runtime, so the open finishes even when nothing polls it.

The reader opens files ahead of the reads. try_flatten_unordered stops polling the opens once it holds concurrency_limit_data_files streams. Every open it drops leaves a half-read response, and that response holds part of the HTTP/2 connection window. Enough of them stop the server from sending on that connection at all. This is the stall in #3276. load_deletes never caused it, because it already spawns, see caching_delete_file_loader.rs.

The combinators stay as they are on main, so the prefetch and the concurrency limits do not change. The spawn is the only new behavior.

Two side effects:

• A panic inside process returns an error through JoinError instead of unwinding in the caller.
• A scan stream that drops early leaves up to concurrency_limit_data_files opens running and discards their results. load_deletes already works this way.

An earlier version of this PR removed the prefetch. On GCS that cost about half the scan time on small files, so I replaced it. The numbers are in the thread.

Are these changes tested?

Yes. flow_control_tests.rs scans over one HTTP/2 connection that a Storage models, with no network.

test_concurrent_scan_does_not_starve_shared_connection stalls on main and passes with this change. It covers concurrency 3, 4, 10, and 16. Against this harness, every value from 3 to 10 stalls.

test_sequential_scan_over_shared_connection is a control. Concurrency 1 takes the other branch of the reader, so it passes either way. It shows that the harness itself does not stall.

AI Disclosure

Claude Code (Claude Opus 5) helped to analyze the stall, to draft the test harness, and to run the GCS measurements. I reviewed every line, and confirmed that the test fails without the fix.

[swanandx]

hey @comphead , did you get a chance to take a look again? lmk if something is required / needs to be updated, thanks!!

[mbutrovich]

Thanks @swanandx for the clear write-up and for a test that reproduces the stall without a network. I ran it with pipeline.rs reverted to the merge base and it stalls with three footer prefetches sitting at 0 bytes consumed, then passes with your change.

[mbutrovich]

The description says this hasn't been measured against a remote store. Could you share before and after wall times for a scan against GCS, on a table with many small data files, at a concurrency where main doesn't stall? Please include the concurrency you used, since the gap should depend on it. Small files are where I'd expect opening a file to be a large share of its read time. Opening covers the footer fetch, plus a page index fetch when row selection is on with a predicate or the task has deletes. On main that work for the next files overlaps the reads of the current ones, and with this change it doesn't.

Could the comparison also include a table with positional deletes or deletion vectors? As I read it, delete loading is affected too. process() calls load_deletes before it opens the file (pipeline.rs#L152-L154), and load_deletes spawns the loading onto the IO runtime (caching_delete_file_loader.rs#L207). On main, the first poll of a buffered future starts that task, so delete files for up to concurrency_limit_data_files upcoming data files load while the current ones are read. With this change, a data file's deletes start loading only when it gets a read slot.

If the cost turns out to be large, that would help decide between this change and the spawn plus bounded channel approach you mentioned.

[swanandx]

I measured this on GCS and the first version of the PR was clearly slower, so I have replaced it.

Scan of 64 files of 25 KB, concurrency 8, metadata size hint 8 KiB, and for the deletes case one positional delete file per data file. I ran the variants turn by turn in the same process, 12 rounds each, and compared them round by round, because a plain before/after run on a home connection is all noise.

	no deletes	with deletes
main	8489 ms	13344 ms
first version of this PR	+52.8%	+37.8%
new version	+0.7%	+1.1%

The first version was slower in 12 rounds out of 12 (95% CI [+3348, +5470] ms without deletes). The new version is within [-241, +243] ms of main, so I cannot measure a difference. You were right that removing the prefetch was the wrong trade.

The prefetch is not what causes the stall. The cause is a prefetched open that nobody polls any more: its response sits half read and keeps part of the HTTP/2 connection window, and enough of those stop the server sending on that connection at all. That is also why load_deletes never triggered it, it spawns, so it finishes whether or not anyone polls it.

So the new version leaves the combinators exactly as they are on main and only spawns the open, the same way caching_delete_file_loader.rs already does:

let opening = runtime.io().spawn(task_reader.clone().process(task));

The stall test now passes from concurrency 2 to 24, where main stalls from 3 to 10. It is a #[tokio::test], so this holds on a current-thread runtime too.

Two things to be aware of: a panic inside process now comes back as an error through JoinError instead of unwinding, and a scan stream that is dropped early leaves up to concurrency_limit_data_files opens running with their results thrown away, which is what load_deletes already does.