comphead commented on code in PR #3277:
URL: https://github.com/apache/iceberg-rust/pull/3277#discussion_r4318870802
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -144,16 +147,32 @@ impl FileScanTaskReader {
.delete_file_loader
.load_deletes(task.deletes(), task.schema_ref());
- // Open the Parquet file once, loading its metadata
- let (parquet_file_reader, arrow_metadata) =
ArrowReader::open_parquet_file(
- task.data_file_path(),
- &self.file_io,
- task.file_size_in_bytes(),
- parquet_read_options,
- self.scan_metrics.bytes_read_counter(),
- task.key_metadata(),
- )
- .await?;
+ // Open the Parquet file once, loading its metadata. Spawn it: once the
+ // caller holds `concurrency_limit_data_files` streams it stops
polling for
+ // more, and an open left mid-response keeps part of the HTTP/2
connection
+ // window, which stalls every read sharing that connection. Aborts on
drop,
+ // so a dropped scan stops opening files.
+ let path = task.data_file_path().to_string();
+ let file_io = self.file_io.clone();
+ let file_size_in_bytes = task.file_size_in_bytes();
+ let bytes_read = Arc::clone(self.scan_metrics.bytes_read_counter());
+ let key_metadata = task.key_metadata().map(|km| km.to_vec());
+ let opening = self
Review Comment:
`filter_row_groups_by_bloom_filter` still fetches inside the buffered
`process` future when `bloom_filter_enabled` is set with a predicate. A bloom
filter read in flight when the flatten stage fills would sit unpolled the way
the footer did. The description says this could not be made to stall. Is that
because those reads fit in one frame, or because the harness does not cover
them? I did not test it.
For comparison, arrow-rs `SpawnedReader`
(`parquet/src/arrow/async_reader/spawn.rs`) spawns every `AsyncFileReader`
request, which would cover the footer, page index, and bloom filter in one
place. It does not abort on drop, so the abort wrapper here would still be
needed.
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -3795,4 +3814,242 @@ mod tests {
let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(total_rows, 3);
}
+
+ /// A data file must be opened even when nothing polls the scan. The stream
+ /// stops polling the opens once it holds `concurrency_limit_data_files` of
+ /// them, and on a shared HTTP/2 connection an open stopped mid-request
holds
+ /// part of the connection window, which stalls every other read (#3276).
+ mod open_progress {
+ use std::io::Cursor;
+ use std::ops::Range;
+ use std::sync::atomic::{AtomicU64, Ordering};
+
+ use async_trait::async_trait;
+ use bytes::Bytes;
+ use futures::StreamExt;
+ use futures::stream::BoxStream;
+ use serde::{Deserialize, Serialize};
+
+ use super::*;
+ use crate::Result;
+ use crate::io::{
+ FileIOBuilder, FileMetadata, FileRead, FileWrite, InputFile,
MemoryStorage, OutputFile,
+ Storage, StorageConfig, StorageFactory,
+ };
+ use crate::scan::ArrowRecordBatchStream;
+
+ const CONCURRENCY: usize = 2;
+ /// Smaller than a footer, so one open takes more than one request.
+ const METADATA_SIZE_HINT: usize = 64;
+ /// Scheduler turns, far more than a request needs to finish.
+ const YIELDS: usize = 100;
+
+ /// In-memory files that count the requests a scan makes against them.
+ /// Writing does not go through [`CountedRead`], so it is not counted.
+ #[derive(Debug, Clone, Default, Serialize, Deserialize)]
+ struct CountedStorage {
+ #[serde(skip)]
+ files: MemoryStorage,
+ #[serde(skip)]
+ requests: Arc<AtomicU64>,
Review Comment:
`ScanResult::metrics().bytes_read()` already counts each read when it is
sent (`CountingFileRead` adds before the inner read), so it can replace this
counter. Removing `requests` from `CountedStorage` and `CountedRead`, the
`sent` helper, and the `AtomicU64` and `Ordering` imports leaves only the
`yield_now` in `CountedRead::read`. `counted_scan` then returns
`result.metrics().clone()`, and the assertions use `bytes_read()`.
I tried this on the PR head. Removing `abort_on_drop` still fails only
`test_dropped_scan_stops_opening_files`, and inlining the open still fails only
`test_open_proceeds_without_the_scan_being_polled`.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
