mbutrovich commented on code in PR #3277:
URL: https://github.com/apache/iceberg-rust/pull/3277#discussion_r4401403011
##########
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:
> Spawning it is a few lines more than the open, because the call needs
`&mut record_batch_stream_builder`. Do you want that in this PR?
I'd like to see it handled here. The bloom filter read is now where the
footer read was before this change. It runs inside the buffered `process`
future
([`pipeline.rs#L578-L584`](https://github.com/apache/iceberg-rust/blob/7775d7e0d779607574ab0a021c59dd92e89a0f11/crates/iceberg/src/arrow/reader/pipeline.rs#L578-L584),
[`L753-L755`](https://github.com/apache/iceberg-rust/blob/7775d7e0d779607574ab0a021c59dd92e89a0f11/crates/iceberg/src/arrow/reader/pipeline.rs#L753-L755)),
so nothing polls it while the flatten stage is full.
The harness can show it once the reads also record when they finish, since
`bytes_read()` counts a read when it goes out. I gave `DeferredRead` a
`finished` counter that it bumps after the yield. Then I scanned one file
written with `set_bloom_filter_enabled(true)` and a bound `id = 2` predicate,
with `with_bloom_filter_enabled(true)`. The test polls the scan until
`bytes_read()` passes the bytes one open reads, stops polling, and spins. At
this head the read left in flight is the bloom filter (`51..98` of a 594-byte
file), and it never finishes: `left: 552, right: 599`.
What do you think about spawning the fetch the same way as the open? The
builder can move into the task and come back out with the result:
```rust
// Spawned for the same reason as the open (#3276).
let bloom_predicate = predicate.clone();
let bloom_candidates = candidate_rgs.to_vec();
let bloom_field_ids = field_id_map.clone();
let mut builder = record_batch_stream_builder;
let (builder, bloom_filtered) = self
.runtime
.io()
.spawn(async move {
let filtered = Self::filter_row_groups_by_bloom_filter(
&bloom_predicate,
&mut builder,
&bloom_candidates,
&bloom_field_ids,
)
.await;
(builder, filtered)
})
.abort_on_drop()
.await?;
record_batch_stream_builder = builder;
let bloom_filtered = bloom_filtered?;
```
This replaces the `filter_row_groups_by_bloom_filter` call at L578-L584. It
clones the predicate, the candidate row groups, and the field id map once per
file, and only when bloom filtering runs.
Could the PR add that test next to the two open tests? One thing to watch is
that it's only meaningful while the bloom filter read is the first read after
the open. With bloom filtering off, the next read is a data page read from the
returned stream, and that read stops when the scan stops being polled, which is
expected. Asserting that the in-flight range is the bloom filter's offset would
keep the test from passing for the wrong reason. A drop test for this spawn
would repeat what `test_dropped_scan_stops_opening_files` already checks for
`abort_on_drop`.
--
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]
