[data] Download op fusion / removal of interleaved partitioners#56462
Merged
bveeramani merged 4 commits intoray-project:masterfrom Sep 11, 2025
Merged
[data] Download op fusion / removal of interleaved partitioners#56462bveeramani merged 4 commits intoray-project:masterfrom
bveeramani merged 4 commits intoray-project:masterfrom
Conversation
Signed-off-by: Matthew Owen <mowen@anyscale.com>
Signed-off-by: Matthew Owen <mowen@anyscale.com>
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces a valuable optimization for chained download operations by removing intermediate partitioner operators, which should significantly improve performance in those scenarios. The logic to detect chained downloads and defer block splitting to the URIDownloader is well-implemented. I have identified a critical syntax error related to an invalid return statement in a generator function, and a high-severity bug in the block splitting logic that could result in an incorrect number of output blocks. Addressing these issues will ensure the correctness and robustness of this new optimization.
bveeramani
approved these changes
Sep 11, 2025
Member
bveeramani
left a comment
There was a problem hiding this comment.
Gemini comments, but otherwise LGTM
Contributor
|
Please also add a test for multiple chained downloads use case? |
Signed-off-by: Matthew Owen <mowen@anyscale.com>
Signed-off-by: Matthew Owen <mowen@anyscale.com>
ZacAttack
pushed a commit
to ZacAttack/ray
that referenced
this pull request
Sep 24, 2025
…project#56462) ## Why are these changes needed? If we have multiple chained downloads e.g. `ds.with_column("bytes_1", download("uri_1")).with_column("bytes_2", download("uri_2")).with_column("bytes_3", download("uri_3"))`, then we would have an operator structure like `URIPartitioner->URIDownloader->URIPartitioner->URIDownloader->URIPartitioner->URIDownloader`. Each of the `URIPartitioner` operators will be implemented with an ActorPoolMapOperator with concurrency of 1. In these chained downloads, these become bottlenecks and scaling the concurrency of these up will result in additional resource usage that will take resources away from other operators. This solves the problem by deferring some of the partitioning to the `URIDownloader` so we can remove the interleaved partitioners. The result is an operator structure like `URIPartitioner->URIDownloader->URIDownloader->URIDownloader` which delivers much better performance for these cases. ## Related issue number <!-- For example: "Closes ray-project#1234" --> ## Checks - [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR. - [ ] I've run `scripts/format.sh` to lint the changes in this PR. - [ ] I've included any doc changes needed for https://docs.ray.io/en/master/. - [ ] I've added any new APIs to the API Reference. For example, if I added a method in Tune, I've added it in `doc/source/tune/api/` under the corresponding `.rst` file. - [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/ - Testing Strategy - [ ] Unit tests - [ ] Release tests - [ ] This PR is not tested :( --------- Signed-off-by: Matthew Owen <mowen@anyscale.com> Signed-off-by: zac <zac@anyscale.com>
marcostephan
pushed a commit
to marcostephan/ray
that referenced
this pull request
Sep 24, 2025
…project#56462) ## Why are these changes needed? If we have multiple chained downloads e.g. `ds.with_column("bytes_1", download("uri_1")).with_column("bytes_2", download("uri_2")).with_column("bytes_3", download("uri_3"))`, then we would have an operator structure like `URIPartitioner->URIDownloader->URIPartitioner->URIDownloader->URIPartitioner->URIDownloader`. Each of the `URIPartitioner` operators will be implemented with an ActorPoolMapOperator with concurrency of 1. In these chained downloads, these become bottlenecks and scaling the concurrency of these up will result in additional resource usage that will take resources away from other operators. This solves the problem by deferring some of the partitioning to the `URIDownloader` so we can remove the interleaved partitioners. The result is an operator structure like `URIPartitioner->URIDownloader->URIDownloader->URIDownloader` which delivers much better performance for these cases. ## Related issue number <!-- For example: "Closes ray-project#1234" --> ## Checks - [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR. - [ ] I've run `scripts/format.sh` to lint the changes in this PR. - [ ] I've included any doc changes needed for https://docs.ray.io/en/master/. - [ ] I've added any new APIs to the API Reference. For example, if I added a method in Tune, I've added it in `doc/source/tune/api/` under the corresponding `.rst` file. - [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/ - Testing Strategy - [ ] Unit tests - [ ] Release tests - [ ] This PR is not tested :( --------- Signed-off-by: Matthew Owen <mowen@anyscale.com> Signed-off-by: Marco Stephan <marco@magic.dev>
dstrodtman
pushed a commit
to dstrodtman/ray
that referenced
this pull request
Oct 6, 2025
…project#56462) ## Why are these changes needed? If we have multiple chained downloads e.g. `ds.with_column("bytes_1", download("uri_1")).with_column("bytes_2", download("uri_2")).with_column("bytes_3", download("uri_3"))`, then we would have an operator structure like `URIPartitioner->URIDownloader->URIPartitioner->URIDownloader->URIPartitioner->URIDownloader`. Each of the `URIPartitioner` operators will be implemented with an ActorPoolMapOperator with concurrency of 1. In these chained downloads, these become bottlenecks and scaling the concurrency of these up will result in additional resource usage that will take resources away from other operators. This solves the problem by deferring some of the partitioning to the `URIDownloader` so we can remove the interleaved partitioners. The result is an operator structure like `URIPartitioner->URIDownloader->URIDownloader->URIDownloader` which delivers much better performance for these cases. ## Related issue number <!-- For example: "Closes ray-project#1234" --> ## Checks - [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR. - [ ] I've run `scripts/format.sh` to lint the changes in this PR. - [ ] I've included any doc changes needed for https://docs.ray.io/en/master/. - [ ] I've added any new APIs to the API Reference. For example, if I added a method in Tune, I've added it in `doc/source/tune/api/` under the corresponding `.rst` file. - [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/ - Testing Strategy - [ ] Unit tests - [ ] Release tests - [ ] This PR is not tested :( --------- Signed-off-by: Matthew Owen <mowen@anyscale.com> Signed-off-by: Douglas Strodtman <douglas@anyscale.com>
justinyeh1995
pushed a commit
to justinyeh1995/ray
that referenced
this pull request
Oct 20, 2025
…project#56462) ## Why are these changes needed? If we have multiple chained downloads e.g. `ds.with_column("bytes_1", download("uri_1")).with_column("bytes_2", download("uri_2")).with_column("bytes_3", download("uri_3"))`, then we would have an operator structure like `URIPartitioner->URIDownloader->URIPartitioner->URIDownloader->URIPartitioner->URIDownloader`. Each of the `URIPartitioner` operators will be implemented with an ActorPoolMapOperator with concurrency of 1. In these chained downloads, these become bottlenecks and scaling the concurrency of these up will result in additional resource usage that will take resources away from other operators. This solves the problem by deferring some of the partitioning to the `URIDownloader` so we can remove the interleaved partitioners. The result is an operator structure like `URIPartitioner->URIDownloader->URIDownloader->URIDownloader` which delivers much better performance for these cases. ## Related issue number <!-- For example: "Closes ray-project#1234" --> ## Checks - [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR. - [ ] I've run `scripts/format.sh` to lint the changes in this PR. - [ ] I've included any doc changes needed for https://docs.ray.io/en/master/. - [ ] I've added any new APIs to the API Reference. For example, if I added a method in Tune, I've added it in `doc/source/tune/api/` under the corresponding `.rst` file. - [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/ - Testing Strategy - [ ] Unit tests - [ ] Release tests - [ ] This PR is not tested :( --------- Signed-off-by: Matthew Owen <mowen@anyscale.com>
landscapepainter
pushed a commit
to landscapepainter/ray
that referenced
this pull request
Nov 17, 2025
…project#56462) ## Why are these changes needed? If we have multiple chained downloads e.g. `ds.with_column("bytes_1", download("uri_1")).with_column("bytes_2", download("uri_2")).with_column("bytes_3", download("uri_3"))`, then we would have an operator structure like `URIPartitioner->URIDownloader->URIPartitioner->URIDownloader->URIPartitioner->URIDownloader`. Each of the `URIPartitioner` operators will be implemented with an ActorPoolMapOperator with concurrency of 1. In these chained downloads, these become bottlenecks and scaling the concurrency of these up will result in additional resource usage that will take resources away from other operators. This solves the problem by deferring some of the partitioning to the `URIDownloader` so we can remove the interleaved partitioners. The result is an operator structure like `URIPartitioner->URIDownloader->URIDownloader->URIDownloader` which delivers much better performance for these cases. ## Related issue number <!-- For example: "Closes ray-project#1234" --> ## Checks - [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR. - [ ] I've run `scripts/format.sh` to lint the changes in this PR. - [ ] I've included any doc changes needed for https://docs.ray.io/en/master/. - [ ] I've added any new APIs to the API Reference. For example, if I added a method in Tune, I've added it in `doc/source/tune/api/` under the corresponding `.rst` file. - [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/ - Testing Strategy - [ ] Unit tests - [ ] Release tests - [ ] This PR is not tested :( --------- Signed-off-by: Matthew Owen <mowen@anyscale.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why are these changes needed?
If we have multiple chained downloads e.g.
ds.with_column("bytes_1", download("uri_1")).with_column("bytes_2", download("uri_2")).with_column("bytes_3", download("uri_3")), then we would have an operator structure likeURIPartitioner->URIDownloader->URIPartitioner->URIDownloader->URIPartitioner->URIDownloader. Each of theURIPartitioneroperators will be implemented with an ActorPoolMapOperator with concurrency of 1. In these chained downloads, these become bottlenecks and scaling the concurrency of these up will result in additional resource usage that will take resources away from other operators.This solves the problem by deferring some of the partitioning to the
URIDownloaderso we can remove the interleaved partitioners. The result is an operator structure likeURIPartitioner->URIDownloader->URIDownloader->URIDownloaderwhich delivers much better performance for these cases.Related issue number
Checks
git commit -s) in this PR.scripts/format.shto lint the changes in this PR.method in Tune, I've added it in
doc/source/tune/api/under thecorresponding
.rstfile.