This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new 5367ef60a chore(core): remove the inert shuffle-read sort (#2414)
5367ef60a is described below
commit 5367ef60a3426d146f594904614fbd60249083c0
Author: NoahKusaba <[email protected]>
AuthorDate: Thu Sep 10 23:52:42 2026 -0400
chore(core): remove the inert shuffle-read sort (#2414)
The sort grouped partition locations by executor and interleaved them, then
the next line shuffled the result, discarding the ordering entirely. Take
the
locations directly and keep the shuffle. `sorted_by` was ballista-core's
only
`itertools` use, so the dependency goes with it.
---
Cargo.lock | 1 -
ballista/core/Cargo.toml | 1 -
ballista/core/src/execution_plans/shuffle_reader.rs | 17 ++---------------
3 files changed, 2 insertions(+), 17 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 09b3d5163..5ef48d31b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1125,7 +1125,6 @@ dependencies = [
"datafusion-proto-common",
"datafusion-spark",
"futures",
- "itertools 0.15.0",
"log",
"object_store",
"parking_lot",
diff --git a/ballista/core/Cargo.toml b/ballista/core/Cargo.toml
index 17b24c4c8..35c47082e 100644
--- a/ballista/core/Cargo.toml
+++ b/ballista/core/Cargo.toml
@@ -53,7 +53,6 @@ datafusion-proto = { workspace = true }
datafusion-proto-common = { workspace = true }
datafusion-spark = { workspace = true, optional = true, features =
["datafusion"] }
futures = { workspace = true }
-itertools = { workspace = true }
log = { workspace = true }
object_store = { workspace = true, features = ["aws", "http"], optional = true
}
parking_lot = { workspace = true }
diff --git a/ballista/core/src/execution_plans/shuffle_reader.rs
b/ballista/core/src/execution_plans/shuffle_reader.rs
index 7f7e9723c..7cc9b2bc4 100644
--- a/ballista/core/src/execution_plans/shuffle_reader.rs
+++ b/ballista/core/src/execution_plans/shuffle_reader.rs
@@ -50,7 +50,6 @@ use datafusion::physical_plan::{
};
use datafusion::prelude::SessionConfig;
use futures::{Stream, StreamExt, TryStreamExt, ready};
-use itertools::Itertools;
use log::{debug, error, trace};
use rand::prelude::SliceRandom;
use rand::rng;
@@ -389,20 +388,8 @@ impl ExecutionPlan for ShuffleReaderExec {
config.ballista_shuffle_reader_maximum_concurrent_requests(),
config.ballista_grpc_client_max_message_size()
);
- let mut partition_locations = HashMap::new();
- for p in &self.partition[partition] {
- partition_locations
- .entry(p.executor_meta.id.clone())
- .or_insert_with(Vec::new)
- .push(p.clone());
- }
- // Sort partitions for evenly send fetching partition requests to
avoid hot executors within one task
- let mut partition_locations: Vec<PartitionLocation> =
partition_locations
- .into_values()
- .flat_map(|ps| ps.into_iter().enumerate())
- .sorted_by(|(p1_idx, _), (p2_idx, _)| Ord::cmp(p1_idx, p2_idx))
- .map(|(_, p)| p)
- .collect();
+ let mut partition_locations: Vec<PartitionLocation> =
+ self.partition[partition].clone();
// Shuffle partitions for evenly send fetching partition requests to
avoid hot executors within multiple tasks
partition_locations.shuffle(&mut rng());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
