swanandx commented on code in PR #3277:
URL: https://github.com/apache/iceberg-rust/pull/3277#discussion_r4402283603
##########
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:
rebased the PR on main and addressed this, thank you so much for detailed
feedback as always and the insights that i missed. helped a lot
--
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]
