[python] Reuse streams for coalesced blob reads

[XiaoHongbo-Hope]

Purpose

Coalesced BLOB range reads currently open and close an input stream for every merged span. A batch containing several spans from the same physical file therefore repeats stream-open overhead.

Changes

• Reuse input streams per path through the existing FileIO.new_input_stream API.
• Distribute range tasks over path-aware worker lanes; each lane exclusively owns and reuses one stream.
• Cap each path at 16 lanes and redistribute remaining capacity with capped water-fill allocation.
• Bound active streams globally by the requested parallelism.
• Close a failed lane stream before falling back to an independently opened range.
• Close every lane stream without masking the original read error.
• Read unknown-length ranges with cursor-based access only on their exclusive lane.

Backend-specific HDFS and Jindo stream implementations are unchanged. The direct FormatBlobReader path is also unchanged.

Tests

• Verify 64 same-path ranges use exactly 16 exclusive lane streams with four reads per stream.
• Verify skewed multi-path workloads redistribute capacity after applying the per-path cap.
• Verify a hot path retains 16 lanes beside 63 singleton paths while active streams stay within 64.
• Verify multi-path and fallback reads do not exceed the requested parallelism.
• Verify barrier-based concurrency tests time out and fail explicitly on scheduling regressions.
• Verify known-length and unknown-length reads are safe on an exclusive lane.
• Verify failed streams are closed before fallback.
• Verify all streams are closed and close errors do not mask read errors.
• Relevant tests: 264 passed, 21 skipped, 45 subtests passed.

Read-only benchmark

For 64 sparse ranges totaling 15.5 MiB from one OSS object, with global parallelism=64:

Per-path lane cap	Stream opens	Payload read
64	64	0.564s
32	32	0.622s
16	16	0.886s
8	8	1.180s

All results were byte-identical. The cap of 16 reduces same-object stream-open/HEAD requests by 75% while adding about 0.32s to payload reading. Planning and descriptor reading took about 6.7s in this workload, so the end-to-end overhead was about 5%. The cap is per path; capped water-fill allocation allows multi-path workloads to continue using the requested global parallelism.

[JingsongLi]

Could we avoid introducing new_range_input_stream and keep a single stream API instead?

A cleaner design would be for Jindo’s new_input_stream to return a lazy adapter that:

• opens the native Jindo stream on the first read, seek, or positional read;
• implements the regular file interface (read, seek, tell, and close);
• exposes native pread/read_at;
• declares whether positional reads are safe to execute concurrently.

This follows the capability-based design used by POSIX pread, Hadoop PositionedReadable, Go ReaderAt, and PyArrow NativeFile. It also avoids adding and manually forwarding another FileIO method through Caching, Resolving, RESTToken, and future wrappers.

One important detail is that the range-read path must not go through pyarrow.PythonFile: I verified that PythonFile.read_at() performs seek + read on the wrapped object, even when the wrapped object itself provides read_at. Therefore, the Jindo adapter should be returned directly by new_input_stream so that coalesced reads can invoke native pread, while regular callers can continue using the standard stream methods.

I also noticed two correctness issues:

1. _RangeStreamPool.close() stops at the first close failure after clearing the pool. This can leak all remaining streams and may mask the original read exception. Please attempt to close every stream and then propagate the first close error.
2. _HdfsReaderAdapter.read_at() directly calls hdfs-native’s read_range. That implementation panics when offset + length exceeds the file length, whereas the previous seek + read behavior returned the remaining bytes. Please clamp the requested range to EOF and add an out-of-range test.

[leaves12138]

Reviewed the latest lane-based implementation. Exclusive lane ownership, bounded per-path concurrency, stream switching, fallback cleanup, and error propagation look correct. Targeted tests, randomized multi-path stress, and repeated real OSS reads all passed without data mismatches or crashes.

[JingsongLi]

I reviewed the final lane-based diff specifically for scope and unnecessary changes. The lane ownership model itself looks reasonable, but I think the PR can be narrowed further.

Changes that can be removed directly

1. Remove the remaining supports_concurrent_pread declarations and test attributes.

   The final lane implementation never reads this capability. It was used by the earlier shared-stream pool, but that consumer was removed in commit 730db9a. The declarations in HDFS/Jindo, the Jindo assertion, and the fake-stream attributes in blob_test.py are now dead scaffolding.

2. Restore the original global supports_pread / pread helpers.

   The new _fileno() helper changes pread() from preferring read_at to preferring fileno() + os.pread. Stream reuse does not require this change, and these helpers are also used by the btree, bitmap, vindex, caching, and mosaic readers. This therefore changes behavior outside the BLOB path and should not be bundled into this PR.

3. Remove two redundant tests.

   • test_fetch_bodies_reuses_bounded_streams_for_same_uri duplicates the same-path reuse/cap/close assertions already covered by test_same_path_uses_bounded_exclusive_lanes, while locking the implementation to exactly 16 opens/closes.
   • test_reader_adapter_read_at_is_concurrent exercises generic lane concurrency rather than HDFS adapter behavior and overlaps the lane concurrency tests in blob_test.py.

Changes that should preferably be split into backend-specific PRs

• HDFS read_at/read_range: exclusive lanes can reuse and concurrently execute the existing seek + read implementation on separate handles. Native read_range is an additional positional-read optimization, not a dependency of stream reuse.

• Jindo lazy/direct adapter: the existing PythonFile can already be retained and reused by one lane. Bypassing it to expose native pread is a separate optimization. It also changes the generic PyArrowFileIO.new_input_stream return type for every Jindo caller. In particular, the new JindoInputFile does not implement readinto, while OffsetInputStream.readinto delegates directly to the wrapped stream, so this widens the compatibility surface beyond this feature.

Moving the HDFS/Jindo work out would remove about 210 touched lines from the core PR and leave it focused on lane scheduling and stream lifecycle.

Needs justification

_MAX_RANGE_LANES_PER_PATH = 16 is a resource policy rather than a correctness requirement. The requested parallelism already bounds the total number of active streams, while the hard-coded cap prevents a single path from using requested parallelism above 16. Please either provide benchmark/resource-limit evidence for 16, make it configurable, or rely on the existing parallelism bound.

I would keep the path-aware lane assignment, exclusive cursor access for unknown-length ranges, closing a failed lane before fallback, global stream bounding across paths, and the close-error/read-error handling. Those are meaningful correctness and resource-lifecycle parts of the final design.

[JingsongLi]

Follow-up review of the narrowed lane-based implementation.

[JingsongLi]

Could we document and justify this fixed value? parallelism already bounds global concurrency, while this additionally changes single-path behavior: for example, Daft's default max_concurrency=64 is silently reduced to 16 for one object. The current benchmark uses parallelism 8, so it does not establish why 16 is the right resource/performance trade-off.

[XiaoHongbo-Hope]

Could we document and justify this fixed value? parallelism already bounds global concurrency, while this additionally changes single-path behavior: for example, Daft's default max_concurrency=64 is silently reduced to 16 for one object. The current benchmark uses parallelism 8, so it does not establish why 16 is the right resource/performance trade-off.

Got your point. If there are 64 tasks with parallelism=64 and no per-path cap, the number of stream opens is not reduced, so it cannot solve the QPS issue. That is my concern. So I try to add a max values here, In our same-Blob benchmark, it reduced stream opens by 75% compared with 64, with only about 5% end-to-end latency overhead. Lower values increased latency more noticeably, while higher values saved less open/HEAD QPS, so 16 was chosen as the balance point. Will update PR doc with more testing benchmark.

[JingsongLi]

Could we redistribute lanes after applying the per-path cap? proportional_lanes is calculated against the uncapped task distribution, so a hot path permanently discards the lanes removed by the cap. With parallelism=64 and path task counts [9900, 100], this assigns only 16 + 1 = 17 non-empty lanes even though 32 are available under the cap; [1000, 100, 100, 100] uses only 31 even though all 64 could be used. A capped/water-fill allocation (or another rebalancing pass) would avoid this substantial underutilization.

[JingsongLi]

Please give this barrier a timeout and explicitly assert that it was not broken. If a scheduling regression creates fewer than eight concurrent lanes, the current test will hang CI indefinitely instead of failing with a useful diagnostic.

[JingsongLi]

Could this test assert the expected allocation more precisely? 1 < len(streams) <= 16 would still pass if a regression allocated only two lanes. For 64 tasks with parallelism 64, the current policy should produce exactly 16 streams, with four reads per stream. It would also be valuable to add a skewed multi-path case that catches lanes lost after applying the cap.

[JingsongLi]

[P2] Do not cap reusable path memberships at the worker count

remaining_lanes treats the sum of per-path lane memberships as an exclusive global budget. However, a physical worker lane can serve multiple paths sequentially, so these memberships do not need to sum to at most workers. For example, with parallelism=64, one path containing 10,000 spans, and 63 paths containing one span each, len(path_task_groups) == workers makes remaining_lanes zero and assigns only one lane to the hot path. The other 63 workers become idle after their singleton reads while the hot path stays serial, making the tail close to 16x slower than the per-path cap permits; the previous proportional allocator did give this path 16 lanes. Please preserve capped proportional memberships and only water-fill an under-allocation without shrinking totals above workers, or use dynamic scheduling. A hot-path-plus-63-singletons test should assert 16 hot-path lanes while global active streams remain at most 64.

Retracted at the author request.

[JingsongLi]

[P2] Balance static lanes by estimated I/O cost

The allocation ratios, lane_loads, and this round-robin placement all treat every task as equal, although a merged span can range from a few bytes to 8 MiB and these static lanes cannot steal work. For one path using 16 lanes, 256 non-coalescing spans with an 8 MiB span at indices 0, 16, ..., 240 send all sixteen large reads to lane 0 via index % path_lanes: that lane transfers 128 MiB while the other 15 lanes process only tiny reads. The previous per-task executor did not pin this periodic size pattern to one worker. Please assign an estimated weight to each task, such as span_len plus a fixed request cost, and place each task on the least weighted-load selected lane. A heterogeneous-span regression test should verify that the large reads are distributed across lanes.

[JingsongLi]

[P2] Stop before fallback when closing the failed handle also fails

The exception path calls _close_current(), which clears the stream reference and records a close error, and then opens a fresh fallback handle here. A failed close() does not guarantee that the underlying descriptor or connection was released. With parallelism=1, a single read failure followed by a close failure can therefore leave the original handle active while read_file_range opens a second one, violating the global stream bound and losing the only reference to the leaked handle. The recorded close error also makes the batch fail at the end, so this fallback I/O cannot salvage the result. Please abort the lane before opening another handle when close fails, preserving the read error as primary when applicable, and update test_failed_stream_close_error_is_propagated_after_fallback to assert that no fallback is attempted.