JingsongLi commented on PR #8807:
URL: https://github.com/apache/paimon/pull/8807#issuecomment-5178609743
Thanks for the follow-up changes. I re-reviewed the latest head (`1ca2bdb`).
The adaptive default is a meaningful improvement for small local reads, and the
normal `_ReadAheadBudget` acquire/resize/release paths look internally
consistent. However, I found several remaining issues that I think should be
addressed before merging:
1. **Reservation-estimation exceptions can hang the reader.**
`_initial_batch_reservation()` is called before the `try` block in
`read_split`. If reading `split.file_size` / `row_count` or parsing
`read.batch-size` raises, the worker exits without calling `publish_error()`
and without putting an end marker into that split's queue. The consumer then
polls the empty queue forever. I reproduced this by making `read_batch_size()`
raise: both workers exited, but the consumer was still alive after 500 ms and
received no exception.
Please move reservation calculation into the worker's
exception-propagation region, or calculate and validate all initial
reservations synchronously before starting the threads.
2. **The byte budget is not a hard bound on decoded/in-flight memory.**
A worker first acquires an estimated reservation, then fully executes
`next(batch_iterator)`, and only afterwards resizes the reservation from
`batch.nbytes`. If the estimate is low, many workers can decode large batches
concurrently and then block in `resize` while already retaining those batches.
For example, with an 8 MiB budget, four 1 MiB reservations, and four actual 16
MiB batches, the Arrow allocator reached 64 MiB before the budget could
throttle anything.
This is especially relevant for high compression ratios, wide/string
columns, and the row-reader conversion path. To provide a real memory bound,
the implementation needs either a safe cold-decode limit/reservation or
byte-targeted batch production. Otherwise, this should be documented as a soft
accounting budget rather than a memory cap.
3. **The initial reservation can also substantially overestimate projected
or sliced reads.**
The estimator uses whole-split encoded file size and is not
projection-aware. In addition, `IndexedSplit` and `SlicedSplit` expose a
selected `row_count` but still delegate `file_size` to the full underlying
split. A narrow projection or small row range from a large file can therefore
reserve the full 64 MiB budget even when its decoded output is only a few KiB,
effectively serializing cold reads.
Please make the estimate projection/range-aware where possible and apply
a fair per-worker cap (for example, a fraction of the global budget) until an
actual batch size has been observed.
4. **The remote adaptive threshold should consider split/request count, not
only total bytes.**
With the current `ceil(total_file_size / 64 MiB)` policy, any remote
workload totaling at most 64 MiB is forced to one worker. That is often exactly
where parallelism is useful for hiding per-request RTT. In a synthetic workload
with twelve 1 MiB S3-style splits and 5 ms I/O latency per split, the adaptive
default took about 86 ms, while explicit parallelism 12 took about 8.9 ms.
I suggest including split count/request latency in the remote heuristic,
e.g. giving multiple remote splits a modest concurrency floor, then capping it
by the byte budget.
5. **Locality should be derived from the effective `FileIO`, not only table
options and URI scheme.**
Catalog-level local-cache options are not necessarily present in the
table schema options, and a FUSE-backed local `FileIO` may still read an
`oss://` path. Both cases can be classified as remote by the current logic.
Inspecting the runtime `FileIO`/cache wrapper capability would make the
storage-type adaptation more reliable.
Two cleanup issues from the previous revision also remain:
- Closing the public reader cannot interrupt a worker already blocked in
`next(batch_iterator)` / storage I/O. The close path returns after the short
join deadline, while the worker remains alive and continues holding a global
semaphore slot.
- `Thread.start()` is still outside the main cleanup `try/finally`; a start
failure leaks all semaphore slots reserved for that pipeline.
For verification, the 59 parallel-reader tests and the targeted
BLOB/multimodal reader tests passed, and the PR checks are green. The
small-local-read behavior is clearly better, but the exception propagation and
true memory-bound issues remain correctness/safety blockers in my view.
--
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]
