mbutrovich commented on code in PR #3277:
URL: https://github.com/apache/iceberg-rust/pull/3277#discussion_r4270813742
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -91,9 +91,22 @@ impl ArrowReader {
.try_flatten(),
)
} else {
+ let runtime = self.runtime.clone();
Box::pin(
tasks
- .map_ok(move |task| task_reader.clone().process(task))
+ .map_ok(move |task| {
+ // Open the file on the IO runtime. The flatten below
stops polling
+ // the opens once it holds
`concurrency_limit_data_files` streams, and
+ // a half-read response keeps part of the HTTP/2
connection window,
+ // which stalls every other read on that connection.
+ let opening =
runtime.io().spawn(task_reader.clone().process(task));
Review Comment:
What do you think about aborting the spawned opens when the scan stream is
dropped? Our `JoinHandle` has no `Drop` impl
([`runtime/mod.rs#L37-L49`](https://github.com/apache/iceberg-rust/blob/1270e329b517a88bdb8eb641b4a59f45bdc0e872/crates/iceberg/src/runtime/mod.rs#L37-L49)),
so dropping it detaches the task.
DataFusion and Comet cancel a scan by dropping its stream. DataFusion makes
that part of the `ExecutionPlan::execute` contract
([`execution_plan.rs#L539-L547`](https://github.com/apache/datafusion/blob/0d79252f61bd8777990b2749c216ffd07079c30d/datafusion/physical-plan/src/execution_plan.rs#L539-L547)):
> The [`Stream`] that is returned must ensure that any allocated resources
are freed when the stream itself is dropped. This is particularly important for
[`spawn`]ed tasks or threads. Unless care is taken to "abort" such tasks, they
may continue to consume resources even after the plan is dropped, generating
intermediate results that are never used.
DataFusion follows this itself. `SpawnedTask` is a `JoinHandle` that aborts
on drop
([`common.rs#L27-L36`](https://github.com/apache/datafusion/blob/0d79252f61bd8777990b2749c216ffd07079c30d/datafusion/common-runtime/src/common.rs#L27-L36),
[`L108-L112`](https://github.com/apache/datafusion/blob/0d79252f61bd8777990b2749c216ffd07079c30d/datafusion/common-runtime/src/common.rs#L108-L112)),
and `clippy.toml` disallows `tokio::task::spawn` in its favor
([`clippy.toml#L2`](https://github.com/apache/datafusion/blob/0d79252f61bd8777990b2749c216ffd07079c30d/clippy.toml#L2)).
`LimitStream` drops its input as soon as it has enough rows
([`limit.rs#L715-L729`](https://github.com/apache/datafusion/blob/0d79252f61bd8777990b2749c216ffd07079c30d/datafusion/physical-plan/src/limit.rs#L715-L729)).
This reader's stream is the scan stream that datafusion-iceberg's
`IcebergDataSource::open` returns
([`scan.rs#L182-L219`](https://github.com/apache/datafusion-iceberg/blob/b641e5ebe9e92967499ae7e440ba10d
7797712df/crates/datafusion/src/read/scan.rs#L182-L219)), and the one Comet's
`IcebergScanExec::execute` returns
([`iceberg_scan.rs#L223-L234`](https://github.com/apache/datafusion-comet/blob/7d8d9b1544b734c85a93cca2cc9d5380ad70accd/native/core/src/execution/operators/iceberg_scan.rs#L223-L234)).
Comet releases a plan when its Spark task completes, and it relies on the drop
to stop that work
([`jni_api.rs#L1593-L1595`](https://github.com/apache/datafusion-comet/blob/7d8d9b1544b734c85a93cca2cc9d5380ad70accd/native/core/src/execution/jni_api.rs#L1593-L1595)):
> Once this returns the Spark task can end, and Spark hands whatever the
task still holds to other tasks. Dropping the plan aborts the tasks its
operators spawned, and they give back what they hold the next time they yield,
so wait for them.
With this change, every open still buffered in `try_buffer_unordered` keeps
fetching after the drop. With the defaults that is up to
`available_parallelism()` opens
([`mod.rs#L67-L72`](https://github.com/apache/iceberg-rust/blob/1270e329b517a88bdb8eb641b4a59f45bdc0e872/crates/iceberg/src/arrow/reader/mod.rs#L67-L72)).
Each one prefetches up to 512 KiB of footer
([`mod.rs#L35`](https://github.com/apache/iceberg-rust/blob/1270e329b517a88bdb8eb641b4a59f45bdc0e872/crates/iceberg/src/arrow/reader/mod.rs#L35)),
plus the page index when it's loaded.
I checked this with your harness. I scanned 48 files at concurrency 8 with a
64-byte metadata size hint, polled the scan once, and dropped it. On the merge
base, no request goes out after the drop. At this head, 15 more requests go out
in the next 200 ms.
As I read it, aborting wouldn't bring the stall back. The abort fires only
when the handle is dropped, which happens only when the whole scan is dropped.
Dropping the read future resets its stream and returns its window, which
`ResponseFuture`'s `Drop` models
([`flow_control_tests.rs#L239-L245`](https://github.com/apache/iceberg-rust/blob/9b65ad16b7b15f625ac24c8d7c584bb3611234a4/crates/iceberg/src/arrow/reader/flow_control_tests.rs#L239-L245)).
`load_deletes` does detach its task
([`caching_delete_file_loader.rs#L207-L257`](https://github.com/apache/iceberg-rust/blob/1270e329b517a88bdb8eb641b4a59f45bdc0e872/crates/iceberg/src/arrow/caching_delete_file_loader.rs#L207-L257)).
But that task fills the shared `DeleteFilter`, and other tasks wait on it
([`L236-L237`](https://github.com/apache/iceberg-rust/blob/1270e329b517a88bdb8eb641b4a59f45bdc0e872/crates/iceberg/src/arrow/caching_delete_file_loader.rs#L236-L237)).
Nothing uses the result of an open once the scan is gone.
One way is a `pub(crate)` abort-on-drop wrapper around `JoinHandle` in
`runtime/mod.rs`, like DataFusion's `SpawnedTask`, so the reader still goes
through `Runtime`. Could a test in `flow_control_tests.rs` also check that a
dropped scan sends no more requests?
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -91,9 +91,22 @@ impl ArrowReader {
.try_flatten(),
)
} else {
+ let runtime = self.runtime.clone();
Box::pin(
tasks
- .map_ok(move |task| task_reader.clone().process(task))
+ .map_ok(move |task| {
+ // Open the file on the IO runtime. The flatten below
stops polling
+ // the opens once it holds
`concurrency_limit_data_files` streams, and
+ // a half-read response keeps part of the HTTP/2
connection window,
+ // which stalls every other read on that connection.
+ let opening =
runtime.io().spawn(task_reader.clone().process(task));
+ async move {
+ opening.await.map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "opening a
data file failed")
+ .with_source(err)
+ })?
+ }
Review Comment:
Should this be `opening.await?` without the `map_err`? `JoinHandle` already
turns a `JoinError` into `ErrorKind::Unexpected` with the `JoinError` as its
source
([`runtime/mod.rs#L41-L49`](https://github.com/apache/iceberg-rust/blob/1270e329b517a88bdb8eb641b4a59f45bdc0e872/crates/iceberg/src/runtime/mod.rs#L41-L49)),
so this adds a second `Unexpected` layer. The message also reads as if it
labels failed opens. It only wraps a panic or a cancellation, because errors
from `process` come through the inner `Result` unchanged. The other spawn sites
use `.await?` directly, for example
[`scan/mod.rs#L484-L493`](https://github.com/apache/iceberg-rust/blob/1270e329b517a88bdb8eb641b4a59f45bdc0e872/crates/iceberg/src/scan/mod.rs#L484-L493).
```suggestion
async move { opening.await? }
```
--
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]
