Repository navigation
Conversation
a73f21a to
2f49a5e
Compare
|
hey @comphead , did you get a chance to take a look again? lmk if something is required / needs to be updated, thanks!! |
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @swanandx for the clear write-up and for a test that reproduces the stall without a network. I ran it with pipeline.rs reverted to the merge base and it stalls with three footer prefetches sitting at 0 bytes consumed, then passes with your change.
| Box::pin( | ||
| tasks | ||
| .map_ok(move |task| task_reader.clone().process(task)) | ||
| .map_ok(move |task| { | ||
| task_reader | ||
| .clone() | ||
| .process(task) | ||
| .try_flatten_stream() | ||
| .boxed() | ||
| }) | ||
| .map_err(|err| { | ||
| Error::new(ErrorKind::Unexpected, "file scan task generate failed") | ||
| .with_source(err) | ||
| }) | ||
| .try_buffer_unordered(concurrency_limit_data_files) | ||
| .try_flatten_unordered(concurrency_limit_data_files), | ||
| ) |
There was a problem hiding this comment.
The description says this hasn't been measured against a remote store. Could you share before and after wall times for a scan against GCS, on a table with many small data files, at a concurrency where main doesn't stall? Please include the concurrency you used, since the gap should depend on it. Small files are where I'd expect opening a file to be a large share of its read time. Opening covers the footer fetch, plus a page index fetch when row selection is on with a predicate or the task has deletes. On main that work for the next files overlaps the reads of the current ones, and with this change it doesn't.
Could the comparison also include a table with positional deletes or deletion vectors? As I read it, delete loading is affected too. process() calls load_deletes before it opens the file (pipeline.rs#L152-L154), and load_deletes spawns the loading onto the IO runtime (caching_delete_file_loader.rs#L207). On main, the first poll of a buffered future starts that task, so delete files for up to concurrency_limit_data_files upcoming data files load while the current ones are read. With this change, a data file's deletes start loading only when it gets a read slot.
If the cost turns out to be large, that would help decide between this change and the spawn plus bounded channel approach you mentioned.
There was a problem hiding this comment.
I measured this on GCS and the first version of the PR was clearly slower, so I have replaced it.
Scan of 64 files of 25 KB, concurrency 8, metadata size hint 8 KiB, and for the deletes case one positional delete file per data file. I ran the variants turn by turn in the same process, 12 rounds each, and compared them round by round, because a plain before/after run on a home connection is all noise.
| no deletes | with deletes | |
|---|---|---|
| main | 8489 ms | 13344 ms |
| first version of this PR | +52.8% | +37.8% |
| new version | +0.7% | +1.1% |
The first version was slower in 12 rounds out of 12 (95% CI [+3348, +5470] ms without deletes). The new version is within [-241, +243] ms of main, so I cannot measure a difference. You were right that removing the prefetch was the wrong trade.
The prefetch is not what causes the stall. The cause is a prefetched open that nobody polls any more: its response sits half read and keeps part of the HTTP/2 connection window, and enough of those stop the server sending on that connection at all. That is also why load_deletes never triggered it, it spawns, so it finishes whether or not anyone polls it.
So the new version leaves the combinators exactly as they are on main and only spawns the open, the same way caching_delete_file_loader.rs already does:
let opening = runtime.io().spawn(task_reader.clone().process(task));The stall test now passes from concurrency 2 to 24, where main stalls from 3 to 10. It is a #[tokio::test], so this holds on a current-thread runtime too.
Two things to be aware of: a panic inside process now comes back as an error through JoinError instead of unwinding, and a scan stream that is dropped early leaves up to concurrency_limit_data_files opens running with their results thrown away, which is what load_deletes already does.
Opens run ahead of the reads. Once try_flatten_unordered holds concurrency_limit_data_files streams it stops polling the opens, and a half-read response keeps part of the HTTP/2 connection window, which stalls every other read on that connection. Spawning the open is what load_deletes already does. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com>
2f49a5e to
9b65ad1
Compare
swanandx
changed the title
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @swanandx for the GCS numbers and for reworking this to keep the prefetch. I reran the harness at concurrency 2 to 24 with pipeline.rs at the merge base and at this head. The merge base stalls at 3 to 10, and this head passes at every value.
| // 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)); |
There was a problem hiding this 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), 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):
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, L108-L112), and clippy.toml disallows tokio::task::spawn in its favor (clippy.toml#L2). LimitStream drops its input as soon as it has enough rows (limit.rs#L715-L729). This reader's stream is the scan stream that datafusion-iceberg's IcebergDataSource::open returns (scan.rs#L182-L219), and the one Comet's IcebergScanExec::execute returns (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):
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). Each one prefetches up to 512 KiB of footer (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). load_deletes does detach its task (caching_delete_file_loader.rs#L207-L257). But that task fills the shared DeleteFilter, and other tasks wait on it (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?
There was a problem hiding this comment.
Thanks for the detailed feedback and suggestions, i totally missed that we must abort on drop.
took help from Claude for this / tests.
Fixed in fa551d9 [ seperate commit for easier review, can amend later if all good ]
A dropped JoinHandle detaches the task, so a cancelled scan left its opens fetching a result nobody reads. Add a pub(crate) abort-on-drop handle and use it for the opens. JoinHandle already maps JoinError, so the open needs no second wrapper. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com>
|
cc @gabotechs @NoahKusaba and anyone else from DataFusion-land who might want to take a look |
This branch has not been deployed
3 participants
Which issue does this PR close?
What changes are included in this PR?
Spawn each data file open on the IO runtime, so the open finishes even when nothing polls it.
The reader opens files ahead of the reads.
try_flatten_unorderedstops polling the opens once it holdsconcurrency_limit_data_filesstreams. Every open it drops leaves a half-read response, and that response holds part of the HTTP/2 connection window. Enough of them stop the server from sending on that connection at all. This is the stall in #3276.load_deletesnever caused it, because it already spawns, seecaching_delete_file_loader.rs.The combinators stay as they are on main, so the prefetch and the concurrency limits do not change. The spawn is the only new behavior.
Two side effects:
processreturns an error throughJoinErrorinstead of unwinding in the caller.concurrency_limit_data_filesopens running and discards their results.load_deletesalready works this way.An earlier version of this PR removed the prefetch. On GCS that cost about half the scan time on small files, so I replaced it. The numbers are in the thread.
Are these changes tested?
Yes.
flow_control_tests.rsscans over one HTTP/2 connection that aStoragemodels, with no network.test_concurrent_scan_does_not_starve_shared_connectionstalls on main and passes with this change. It covers concurrency 3, 4, 10, and 16. Against this harness, every value from 3 to 10 stalls.test_sequential_scan_over_shared_connectionis a control. Concurrency 1 takes the other branch of the reader, so it passes either way. It shows that the harness itself does not stall.AI Disclosure
Claude Code (Claude Opus 5) helped to analyze the stall, to draft the test harness, and to run the GCS measurements. I reviewed every line, and confirmed that the test fails without the fix.