shyjsarah commented on code in PR #720:
URL: https://github.com/apache/paimon-rust/pull/720#discussion_r3794928129
##########
crates/paimon/src/spec/core_options.rs:
##########
@@ -728,6 +731,35 @@ impl<'a> CoreOptions<'a> {
Ok(value as usize)
}
+ /// Maximum number of concurrent range reads shared by Vindex readers in
one
+ /// search operation. This is independent of
[`Self::global_index_thread_num`].
Review Comment:
**[Minor] Please update the existing option documentation and migration
notes for this new independent limit.**
The current `global_index_thread_num` Rust docs still say it limits
global-index/PK-vector I/O and that `1` gives strict sequential execution;
`docs/src/sql.md` also documents only `global-index.thread-num`. Those
statements are no longer true for Vindex range reads when this option is absent
(the new default is 32). Please narrow the old option's documented scope,
document `global-index.range-read-thread-num` and its default, and call out the
changed upgrade behavior for tables with a non-default
`global-index.thread-num`.
##########
crates/paimon/src/vindex/range_reader.rs:
##########
@@ -319,17 +344,28 @@ impl VindexFileReader {
});
}
- for batch in merged.chunks(RANGE_READ_CONCURRENCY) {
- let ranges: Vec<_> = batch.iter().map(|merged|
merged.range.clone()).collect();
- let fetched = self.fetch_range_batch(&ranges)?;
- for (merged_range, data) in batch.iter().zip(fetched) {
- for &request_index in &merged_range.request_indices {
- let request = &mut requests[request_index];
- let start = (request.pos - merged_range.range.start) as
usize;
- request
- .buf
- .copy_from_slice(&data[start..start +
request.buf.len()]);
- }
+ if let Some(stats) = &self.stats {
+ stats
+ .read_many_merged_ranges
+ .fetch_add(merged.len() as u64, Ordering::Relaxed);
+ }
+ if let Some(stats) = &self.stats {
+ stats.read_many_chunks.fetch_add(1, Ordering::Relaxed);
+ stats
+ .read_many_chunk_sizes
+ .lock()
+ .unwrap()
+ .push(merged.len());
+ }
+ let ranges: Vec<_> = merged.iter().map(|merged|
merged.range.clone()).collect();
+ let fetched = self.fetch_range_batch(&ranges)?;
Review Comment:
**[Major] Please keep the refill pipeline bounded instead of queueing every
merged range at once.**
`try_join_all` creates/polls a future for every range, so the semaphore
limits only active `FileRead` calls—not queued permit waiters or completed
`Bytes` retained until the slowest read finishes. Since Tokio's semaphore is
FIFO, one large `pread` can also enqueue all of its ranges ahead of other
Vindex readers sharing this limiter. This changes the old chunk barrier into
O(R) queued state and can cause cross-reader head-of-line blocking.
Could this use a rolling
`FuturesUnordered`/`buffer_unordered(max_range_read_concurrency)` window,
carrying the original index to restore result order? That preserves immediate
refill while keeping local waiters bounded. A regression test with two cloned
readers sharing permits would also help verify that a large batch cannot
monopolize the queue.
--
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]
