JeonDaehong commented on code in PR #3587:
URL: https://github.com/apache/iceberg-rust/pull/3587#discussion_r4430843885
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -262,13 +272,20 @@ impl CachingDeleteFileLoader {
basic_delete_file_loader: BasicDeleteFileLoader,
del_filter: DeleteFilter,
schema: SchemaRef,
+ io: crate::runtime::RuntimeHandle,
) -> Result<DeleteFileContext> {
match task.file_type() {
DataContentType::PositionDeletes => {
// A V3 deletion vector arrives as a PositionDeletes entry
whose deletes live in
// a Puffin blob, not in a positional-delete parquet file.
if task.file_format() == DataFileFormat::Puffin {
- return Self::load_deletion_vector(task,
basic_delete_file_loader).await;
+ let task = task.clone();
+ return io
+ .spawn(async move {
Review Comment:
`delete_loading_requires_cpu_runtime` uses `setup()`, which only has parquet
delete files, so nothing checks that this DV read actually runs on io. a
split-runtime test with a puffin DV and a thread name check like the one in
`runtime_file_read.rs` would cover it.
##########
crates/iceberg/src/arrow/delete_file_loader.rs:
##########
@@ -78,6 +80,7 @@ impl BasicDeleteFileLoader {
let (parquet_file_reader, arrow_metadata) =
ArrowReader::open_parquet_file(
data_file_path,
&self.file_io,
+ &self.runtime,
Review Comment:
heads up, #3433 (mine) adds a `metadata()` call in `parquet_to_batch_stream`
when the delete file size is unknown. with this change that call would run on
the cpu runtime. whichever of the two lands second should route it through io
too, I can do it on my side if this goes first.
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -815,14 +818,14 @@ impl ArrowReader {
pub(crate) async fn open_parquet_file(
data_file_path: &str,
file_io: &FileIO,
+ runtime: &Runtime,
file_size_in_bytes: u64,
parquet_read_options: ParquetReadOptions,
bytes_read: &Arc<AtomicU64>,
key_metadata: Option<&[u8]>,
) -> Result<(ArrowFileReader, ArrowReaderMetadata)> {
- let parquet_file = file_io.new_input(data_file_path)?;
- let counting_reader =
- CountingFileRead::new(parquet_file.reader().await?,
Arc::clone(bytes_read));
+ let reader = RuntimeFileRead::open(file_io, data_file_path,
runtime.io()).await?;
Review Comment:
since `open_parquet_file` is shared, data file reads also go through
`RuntimeFileRead` with this PR, not only after #3588. page reads made while the
scan stream is polled already get dispatched to io here. fine by me, maybe
worth a line in the description.
--
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]
