leaves12138 commented on code in PR #534:
URL: https://github.com/apache/paimon-rust/pull/534#discussion_r3600840085
##########
crates/paimon/src/table/blob_resolver.rs:
##########
@@ -252,6 +394,187 @@ fn merge_blob_read_requests(mut requests:
Vec<BlobReadRequest>) -> Vec<MergedBlo
mod tests {
use super::*;
+ #[derive(Clone)]
+ struct TrackingFileRead {
+ bytes: Bytes,
+ in_flight: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+ max_in_flight: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+ }
+
+ impl TrackingFileRead {
+ fn new(bytes: Bytes) -> Self {
+ Self {
+ bytes,
+ in_flight:
std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
+ max_in_flight:
std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
+ }
+ }
+
+ fn with_counters(
+ bytes: Bytes,
+ in_flight: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+ max_in_flight: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+ ) -> Self {
+ Self {
+ bytes,
+ in_flight,
+ max_in_flight,
+ }
+ }
+
+ fn max_in_flight(&self) -> usize {
+ self.max_in_flight.load(std::sync::atomic::Ordering::SeqCst)
+ }
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for TrackingFileRead {
+ async fn read(&self, range: std::ops::Range<u64>) ->
crate::Result<Bytes> {
+ let in_flight = self
+ .in_flight
+ .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
+ + 1;
+ self.max_in_flight
+ .fetch_max(in_flight, std::sync::atomic::Ordering::SeqCst);
+ tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+ self.in_flight
+ .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
+ Ok(self.bytes.slice(range.start as usize..range.end as usize))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_blob_range_reads_use_bounded_parallelism() {
+ let reader =
TrackingFileRead::new(Bytes::from_static(b"abcdefghijkl"));
+ let reads = (0..12)
+ .map(|row| MergedBlobRead {
+ start: row,
+ end: row + 1,
+ requests: vec![BlobReadRequest {
+ row: row as usize,
+ offset: row,
+ length: 1,
+ }],
+ })
+ .collect();
+
+ let results = read_merged_blob_ranges(
+ "memory:/blob.bin",
+ std::sync::Arc::new(reader.clone()),
+ reads,
+ BlobReadLimiter::new(),
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(results.len(), 12);
+ assert!(reader.max_in_flight() > 1);
+ assert!(reader.max_in_flight() <= BLOB_DESCRIPTOR_READ_CONCURRENCY);
+ }
+
+ #[tokio::test]
+ async fn test_blob_range_reads_apply_byte_budget_and_preserve_rows() {
+ let reader = TrackingFileRead::new(Bytes::from_static(b"abcdefgh"));
+ let reads = vec![
+ MergedBlobRead {
+ start: 4,
+ end: 8,
+ requests: vec![BlobReadRequest {
+ row: 0,
+ offset: 4,
+ length: 4,
+ }],
+ },
+ MergedBlobRead {
+ start: 0,
+ end: 4,
+ requests: vec![BlobReadRequest {
+ row: 1,
+ offset: 0,
+ length: 4,
+ }],
+ },
+ ];
+
+ let results = read_merged_blob_ranges(
+ "memory:/blob.bin",
+ std::sync::Arc::new(reader.clone()),
+ reads,
+ BlobReadLimiter::with_limits(8, 4, 1),
+ )
+ .await
+ .unwrap();
+
+ let mut by_row = results
+ .into_iter()
+ .map(|result| (result.merged.requests[0].row, result.data))
+ .collect::<Vec<_>>();
+ by_row.sort_by_key(|(row, _)| *row);
+ assert_eq!(by_row[0], (0, Bytes::from_static(b"efgh")));
+ assert_eq!(by_row[1], (1, Bytes::from_static(b"abcd")));
+ assert_eq!(reader.max_in_flight(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_blob_range_reads_overlap_across_uris() {
+ let in_flight =
std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+ let max_in_flight =
std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+ let groups = b"ab"
Review Comment:
The `check` job still fails on Clippy’s `byte_char_slices` lint here. Please
use `*b"ab"` instead of `[b'a', b'b']` so the required Clippy check can
complete.
--
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]
