refactor(query): rename exchange hash to node to node hash (#18948)

zhang2014 · web-flow · commit d11e97f837a7 · 2025-11-11T16:26:07.000+08:00
diff --git a/src/query/service/src/physical_plans/physical_exchange.rs b/src/query/service/src/physical_plans/physical_exchange.rs
@@ -100,7 +100,7 @@ impl PhysicalPlanBuilder {
         mut required: ColumnSet,
     ) -> Result<PhysicalPlan> {
         // 1. Prune unused Columns.
-        if let databend_common_sql::plans::Exchange::Hash(exprs) = exchange {
+        if let databend_common_sql::plans::Exchange::NodeToNodeHash(exprs) = exchange {
             for expr in exprs {
                 required.extend(expr.used_columns());
             }
@@ -112,7 +112,7 @@ impl PhysicalPlanBuilder {
         let mut keys = vec![];
         let mut allow_adjust_parallelism = true;
         let kind = match exchange {
-            databend_common_sql::plans::Exchange::Hash(scalars) => {
+            databend_common_sql::plans::Exchange::NodeToNodeHash(scalars) => {
                 for scalar in scalars {
                     let expr = scalar
                         .type_check(input_schema.as_ref())?
diff --git a/src/query/service/src/physical_plans/physical_hash_join.rs b/src/query/service/src/physical_plans/physical_hash_join.rs
@@ -1198,7 +1198,7 @@ impl PhysicalPlanBuilder {
         let build_side_data_distribution = s_expr.build_side_child().get_data_distribution()?;
         let broadcast_id = if build_side_data_distribution
             .as_ref()
-            .is_some_and(|e| matches!(e, databend_common_sql::plans::Exchange::Hash(_)))
+            .is_some_and(|e| matches!(e, databend_common_sql::plans::Exchange::NodeToNodeHash(_)))
         {
             Some(self.ctx.get_next_broadcast_id())
         } else {
diff --git a/src/query/service/src/physical_plans/physical_join_filter.rs b/src/query/service/src/physical_plans/physical_join_filter.rs
@@ -191,12 +191,12 @@ impl JoinRuntimeFilter {
 
             let enable_min_max_runtime_filter = build_side_data_distribution
                 .as_ref()
-                .is_none_or(|e| matches!(e, Exchange::Broadcast | Exchange::Hash(_)))
+                .is_none_or(|e| matches!(e, Exchange::Broadcast | Exchange::NodeToNodeHash(_)))
                 && Self::is_type_supported_for_min_max_filter(&data_type);
 
             let enable_inlist_runtime_filter = build_side_data_distribution
                 .as_ref()
-                .is_none_or(|e| matches!(e, Exchange::Broadcast | Exchange::Hash(_)));
+                .is_none_or(|e| matches!(e, Exchange::Broadcast | Exchange::NodeToNodeHash(_)));
 
             // Create and add the runtime filter
             let runtime_filter = PhysicalRuntimeFilter {
diff --git a/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_exchange_injector.rs b/src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_exchange_injector.rs
@@ -239,7 +239,7 @@ impl ExchangeInjector for AggregateInjector {
         match exchange {
             DataExchange::Merge(_) => unreachable!(),
             DataExchange::Broadcast(_) => unreachable!(),
-            DataExchange::ShuffleDataExchange(exchange) => {
+            DataExchange::NodeToNodeExchange(exchange) => {
                 Ok(Arc::new(Box::new(HashTableHashScatter {
                     buckets: exchange.destination_ids.len(),
                 })))
diff --git a/src/query/service/src/pipelines/processors/transforms/sort/sort_exchange_injector.rs b/src/query/service/src/pipelines/processors/transforms/sort/sort_exchange_injector.rs
@@ -40,7 +40,7 @@ impl ExchangeInjector for SortInjector {
     ) -> Result<Arc<Box<dyn FlightScatter>>> {
         match exchange {
             DataExchange::Merge(_) | DataExchange::Broadcast(_) => unreachable!(),
-            DataExchange::ShuffleDataExchange(exchange) => {
+            DataExchange::NodeToNodeExchange(exchange) => {
                 Ok(Arc::new(Box::new(SortBoundScatter {
                     partitions: exchange.destination_ids.len(),
                 })))
diff --git a/src/query/service/src/schedulers/fragments/fragmenter.rs b/src/query/service/src/schedulers/fragments/fragmenter.rs
@@ -48,7 +48,7 @@ use crate::schedulers::PlanFragment;
 use crate::servers::flight::v1::exchange::BroadcastExchange;
 use crate::servers::flight::v1::exchange::DataExchange;
 use crate::servers::flight::v1::exchange::MergeExchange;
-use crate::servers::flight::v1::exchange::ShuffleDataExchange;
+use crate::servers::flight::v1::exchange::NodeToNodeExchange;
 use crate::sessions::QueryContext;
 
 /// Visitor to split a `PhysicalPlan` into fragments.
@@ -217,7 +217,7 @@ impl FragmentDeriveHandle {
 
         Ok(match exchange_sink.kind {
             FragmentKind::Init => None,
-            FragmentKind::Normal => Some(DataExchange::ShuffleDataExchange(ShuffleDataExchange {
+            FragmentKind::Normal => Some(DataExchange::NodeToNodeExchange(NodeToNodeExchange {
                 destination_ids: get_executors(cluster),
                 shuffle_keys: exchange_sink.keys.clone(),
                 allow_adjust_parallelism: exchange_sink.allow_adjust_parallelism,
diff --git a/src/query/service/src/schedulers/fragments/query_fragment_actions_display.rs b/src/query/service/src/schedulers/fragments/query_fragment_actions_display.rs
@@ -71,7 +71,7 @@ impl Display for QueryFragmentActionsWrap<'_> {
             match data_exchange {
                 DataExchange::Merge(_) => writeln!(f, "  DataExchange: Merge")?,
                 DataExchange::Broadcast(_) => writeln!(f, "  DataExchange: Broadcast")?,
-                DataExchange::ShuffleDataExchange(_) => writeln!(f, "  DataExchange: Shuffle")?,
+                DataExchange::NodeToNodeExchange(_) => writeln!(f, "  DataExchange: Shuffle")?,
             }
         }
 
diff --git a/src/query/service/src/servers/flight/v1/exchange/data_exchange.rs b/src/query/service/src/servers/flight/v1/exchange/data_exchange.rs
@@ -18,21 +18,21 @@ use databend_common_expression::RemoteExpr;
 pub enum DataExchange {
     Merge(MergeExchange),
     Broadcast(BroadcastExchange),
-    ShuffleDataExchange(ShuffleDataExchange),
+    NodeToNodeExchange(NodeToNodeExchange),
 }
 
 impl DataExchange {
     pub fn get_destinations(&self) -> Vec<String> {
         match self {
             DataExchange::Merge(exchange) => vec![exchange.destination_id.clone()],
             DataExchange::Broadcast(exchange) => exchange.destination_ids.clone(),
-            DataExchange::ShuffleDataExchange(exchange) => exchange.destination_ids.clone(),
+            DataExchange::NodeToNodeExchange(exchange) => exchange.destination_ids.clone(),
         }
     }
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct ShuffleDataExchange {
+pub struct NodeToNodeExchange {
     pub destination_ids: Vec<String>,
     pub shuffle_keys: Vec<RemoteExpr>,
     pub allow_adjust_parallelism: bool,
diff --git a/src/query/service/src/servers/flight/v1/exchange/exchange_injector.rs b/src/query/service/src/servers/flight/v1/exchange/exchange_injector.rs
@@ -86,7 +86,7 @@ impl ExchangeInjector for DefaultExchangeInjector {
             DataExchange::Broadcast(exchange) => Box::new(BroadcastFlightScatter::try_create(
                 exchange.destination_ids.len(),
             )?),
-            DataExchange::ShuffleDataExchange(exchange) => {
+            DataExchange::NodeToNodeExchange(exchange) => {
                 let local_id = &ctx.get_cluster().local_id;
                 let local_pos = exchange
                     .destination_ids
diff --git a/src/query/service/src/servers/flight/v1/exchange/exchange_manager.rs b/src/query/service/src/servers/flight/v1/exchange/exchange_manager.rs
@@ -1050,7 +1050,7 @@ impl FragmentCoordinator {
                     allow_adjust_parallelism: true,
                 },
             ))),
-            DataExchange::ShuffleDataExchange(exchange) => Ok(Some(
+            DataExchange::NodeToNodeExchange(exchange) => Ok(Some(
                 ExchangeParams::ShuffleExchange(ShuffleExchangeParams {
                     exchange_injector: exchange_injector.clone(),
                     schema: self.physical_plan.output_schema()?,
diff --git a/src/query/service/src/servers/flight/v1/exchange/mod.rs b/src/query/service/src/servers/flight/v1/exchange/mod.rs
@@ -32,7 +32,7 @@ pub mod serde;
 pub use data_exchange::BroadcastExchange;
 pub use data_exchange::DataExchange;
 pub use data_exchange::MergeExchange;
-pub use data_exchange::ShuffleDataExchange;
+pub use data_exchange::NodeToNodeExchange;
 pub use exchange_injector::DefaultExchangeInjector;
 pub use exchange_injector::ExchangeInjector;
 pub use exchange_manager::DataExchangeManager;
diff --git a/src/query/sql/src/planner/format/display_rel_operator.rs b/src/query/sql/src/planner/format/display_rel_operator.rs
@@ -409,14 +409,14 @@ fn limit_to_format_tree<I: IdHumanizer>(_: &I, op: &Limit) -> FormatTreeNode {
 
 fn exchange_to_format_tree<I: IdHumanizer>(id_humanizer: &I, op: &Exchange) -> FormatTreeNode {
     let payload = match op {
-        Exchange::Hash(_) => "Exchange(Hash)",
         Exchange::Broadcast => "Exchange(Broadcast)",
         Exchange::Merge => "Exchange(Merge)",
         Exchange::MergeSort => "Exchange(MergeSort)",
+        Exchange::NodeToNodeHash(_) => "Exchange(Hash)",
     };
 
     match op {
-        Exchange::Hash(keys) => {
+        Exchange::NodeToNodeHash(keys) => {
             FormatTreeNode::with_children(payload.to_string(), vec![FormatTreeNode::new(format!(
                 "Exchange(Hash): keys: [{}]",
                 keys.iter()
diff --git a/src/query/sql/src/planner/optimizer/ir/format.rs b/src/query/sql/src/planner/optimizer/ir/format.rs
@@ -56,7 +56,7 @@ fn display_rel_op(rel_op: &RelOperator) -> String {
         RelOperator::UnionAll(_) => "UnionAll".to_string(),
         RelOperator::Exchange(op) => {
             format!("Exchange: ({})", match op {
-                Exchange::Hash(scalars) => format!(
+                Exchange::NodeToNodeHash(scalars) => format!(
                     "Hash({})",
                     scalars
                         .iter()
diff --git a/src/query/sql/src/planner/optimizer/ir/property/enforcer.rs b/src/query/sql/src/planner/optimizer/ir/property/enforcer.rs
@@ -69,7 +69,9 @@ impl Enforcer for DistributionEnforcer {
         match &self.0 {
             Distribution::Serial => Ok(Exchange::Merge.into()),
             Distribution::Broadcast => Ok(Exchange::Broadcast.into()),
-            Distribution::Hash(hash_keys) => Ok(Exchange::Hash(hash_keys.clone()).into()),
+            Distribution::NodeToNodeHash(hash_keys) => {
+                Ok(Exchange::NodeToNodeHash(hash_keys.clone()).into())
+            }
             Distribution::Random | Distribution::Any => Err(ErrorCode::Internal(
                 "Cannot enforce random or any distribution",
             )),
@@ -120,8 +122,10 @@ impl PropertyEnforcer {
         if let RelOperator::Join(_) = &plan {
             let (probe_required_property, build_required_property) =
                 required_properties.split_at_mut(1);
-            if let Distribution::Hash(probe_keys) = &mut probe_required_property[0].distribution
-                && let Distribution::Hash(build_keys) = &mut build_required_property[0].distribution
+            if let Distribution::NodeToNodeHash(probe_keys) =
+                &mut probe_required_property[0].distribution
+                && let Distribution::NodeToNodeHash(build_keys) =
+                    &mut build_required_property[0].distribution
             {
                 let cast_rules = &BUILTIN_FUNCTIONS.get_auto_cast_rules("eq");
                 for (probe_key, build_key) in probe_keys.iter_mut().zip(build_keys.iter_mut()) {
diff --git a/src/query/sql/src/planner/optimizer/ir/property/property.rs b/src/query/sql/src/planner/optimizer/ir/property/property.rs
@@ -94,7 +94,7 @@ pub enum Distribution {
     Random,
     Serial,
     Broadcast,
-    Hash(Vec<ScalarExpr>),
+    NodeToNodeHash(Vec<ScalarExpr>),
 }
 
 impl Default for Distribution {
@@ -113,11 +113,12 @@ impl Distribution {
             | (Distribution::Random, _)
             | (Distribution::Serial, Distribution::Serial)
             | (Distribution::Broadcast, Distribution::Broadcast)
-            | (Distribution::Hash(_), Distribution::Broadcast) => true,
+            | (Distribution::NodeToNodeHash(_), Distribution::Broadcast) => true,
 
-            (Distribution::Hash(ref keys), Distribution::Hash(ref other_keys)) => {
-                keys == other_keys
-            }
+            (
+                Distribution::NodeToNodeHash(ref keys),
+                Distribution::NodeToNodeHash(ref other_keys),
+            ) => keys == other_keys,
             _ => false,
         }
     }
@@ -130,7 +131,7 @@ impl Display for Distribution {
             Distribution::Random => write!(f, "Random"),
             Distribution::Serial => write!(f, "Serial"),
             Distribution::Broadcast => write!(f, "Broadcast"),
-            Distribution::Hash(ref keys) => write!(
+            Distribution::NodeToNodeHash(ref keys) => write!(
                 f,
                 "Hash({})",
                 keys.iter()
diff --git a/src/query/sql/src/planner/optimizer/optimizers/cascades/cost/model.rs b/src/query/sql/src/planner/optimizer/optimizers/cascades/cost/model.rs
@@ -159,7 +159,7 @@ impl DefaultCostModel {
         let exchange: Exchange = (*m_expr.plan.clone()).clone().try_into()?;
         let group = memo.group(m_expr.group_index)?;
         let cost = match exchange {
-            Exchange::Hash(_) => {
+            Exchange::NodeToNodeHash(_) => {
                 group.stat_info.cardinality * self.network_per_row
                     + group.stat_info.cardinality * self.compute_per_row
             }
diff --git a/src/query/sql/src/planner/optimizer/optimizers/distributed/distributed_merge.rs b/src/query/sql/src/planner/optimizer/optimizers/distributed/distributed_merge.rs
@@ -59,11 +59,15 @@ impl BroadcastToShuffleOptimizer {
 
         let new_join_children = vec![
             Arc::new(SExpr::create_unary(
-                Arc::new(RelOperator::Exchange(Exchange::Hash(left_conditions))),
+                Arc::new(RelOperator::Exchange(Exchange::NodeToNodeHash(
+                    left_conditions,
+                ))),
                 Arc::new(left_exchange_input.clone()),
             )),
             Arc::new(SExpr::create_unary(
-                Arc::new(RelOperator::Exchange(Exchange::Hash(right_conditions))),
+                Arc::new(RelOperator::Exchange(Exchange::NodeToNodeHash(
+                    right_conditions,
+                ))),
                 Arc::new(right_exchange_input.clone()),
             )),
         ];
diff --git a/src/query/sql/src/planner/optimizer/optimizers/distributed/sort_and_limit.rs b/src/query/sql/src/planner/optimizer/optimizers/distributed/sort_and_limit.rs
@@ -109,7 +109,7 @@ impl SortAndLimitPushDownOptimizer {
         match exchange_sexpr.plan() {
             RelOperator::Exchange(exchange) => match exchange {
                 // this is window shuffle sort
-                Exchange::Hash(_) => return Ok(s_expr.clone()),
+                Exchange::NodeToNodeHash(_) => return Ok(s_expr.clone()),
                 Exchange::Merge | Exchange::MergeSort => {}
                 Exchange::Broadcast => unreachable!(),
             },
diff --git a/src/query/sql/src/planner/plans/aggregate.rs b/src/query/sql/src/planner/plans/aggregate.rs
@@ -113,8 +113,12 @@ impl Aggregate {
         }
 
         match group_by_shuffle_mode.as_str() {
-            "before_partial" => Ok(Distribution::Hash(self.get_distribution_keys(true)?)),
-            "before_merge" => Ok(Distribution::Hash(self.get_distribution_keys(false)?)),
+            "before_partial" => Ok(Distribution::NodeToNodeHash(
+                self.get_distribution_keys(true)?,
+            )),
+            "before_merge" => Ok(Distribution::NodeToNodeHash(
+                self.get_distribution_keys(false)?,
+            )),
             value => Err(ErrorCode::Internal(format!(
                 "Bad settings value group_by_shuffle_mode = {:?}",
                 value
diff --git a/src/query/sql/src/planner/plans/exchange.rs b/src/query/sql/src/planner/plans/exchange.rs
@@ -27,10 +27,10 @@ use crate::plans::ScalarExpr;
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub enum Exchange {
-    Hash(Vec<ScalarExpr>),
     Broadcast,
     Merge,
     MergeSort, // For distributed sort
+    NodeToNodeHash(Vec<ScalarExpr>),
 }
 
 impl Operator for Exchange {
@@ -39,7 +39,7 @@ impl Operator for Exchange {
     }
 
     fn scalar_expr_iter(&self) -> Box<dyn Iterator<Item = &ScalarExpr> + '_> {
-        if let Exchange::Hash(hash_keys) = self {
+        if let Exchange::NodeToNodeHash(hash_keys) = self {
             Box::new(hash_keys.iter())
         } else {
             Box::new(std::iter::empty())
@@ -53,7 +53,9 @@ impl Operator for Exchange {
     fn derive_physical_prop(&self, _rel_expr: &RelExpr) -> Result<PhysicalProperty> {
         Ok(PhysicalProperty {
             distribution: match self {
-                Exchange::Hash(hash_keys) => Distribution::Hash(hash_keys.clone()),
+                Exchange::NodeToNodeHash(hash_keys) => {
+                    Distribution::NodeToNodeHash(hash_keys.clone())
+                }
                 Exchange::Broadcast => Distribution::Broadcast,
                 Exchange::Merge | Exchange::MergeSort => Distribution::Serial,
             },
diff --git a/src/query/sql/src/planner/plans/join.rs b/src/query/sql/src/planner/plans/join.rs
@@ -640,9 +640,11 @@ impl Operator for Join {
             // Although the build side is also Hash, it is more efficient to
             // utilize the distribution on the probe side.
             // As soon as we support subset property, we can pass through both sides.
-            (Distribution::Hash(_), Distribution::Hash(_)) => Ok(PhysicalProperty {
-                distribution: probe_prop.distribution.clone(),
-            }),
+            (Distribution::NodeToNodeHash(_), Distribution::NodeToNodeHash(_)) => {
+                Ok(PhysicalProperty {
+                    distribution: probe_prop.distribution.clone(),
+                })
+            }
 
             // Otherwise use random distribution.
             _ => Ok(PhysicalProperty {
@@ -727,14 +729,14 @@ impl Operator for Join {
                 .iter()
                 .map(|condition| condition.left.clone())
                 .collect();
-            required.distribution = Distribution::Hash(left_conditions);
+            required.distribution = Distribution::NodeToNodeHash(left_conditions);
         } else {
             let right_conditions = self
                 .equi_conditions
                 .iter()
                 .map(|condition| condition.right.clone())
                 .collect();
-            required.distribution = Distribution::Hash(right_conditions);
+            required.distribution = Distribution::NodeToNodeHash(right_conditions);
         }
 
         Ok(required)
@@ -766,7 +768,7 @@ impl Operator for Join {
                         distribution: Distribution::Broadcast,
                     },
                     RequiredProperty {
-                        distribution: Distribution::Hash(conditions),
+                        distribution: Distribution::NodeToNodeHash(conditions),
                     },
                 ]);
             } else {
@@ -779,7 +781,7 @@ impl Operator for Join {
 
                 children_required.push(vec![
                     RequiredProperty {
-                        distribution: Distribution::Hash(conditions),
+                        distribution: Distribution::NodeToNodeHash(conditions),
                     },
                     RequiredProperty {
                         distribution: Distribution::Broadcast,
@@ -795,10 +797,10 @@ impl Operator for Join {
             children_required.extend(self.equi_conditions.iter().map(|condition| {
                 vec![
                     RequiredProperty {
-                        distribution: Distribution::Hash(vec![condition.left.clone()]),
+                        distribution: Distribution::NodeToNodeHash(vec![condition.left.clone()]),
                     },
                     RequiredProperty {
-                        distribution: Distribution::Hash(vec![condition.right.clone()]),
+                        distribution: Distribution::NodeToNodeHash(vec![condition.right.clone()]),
                     },
                 ]
             }));
diff --git a/src/query/sql/src/planner/plans/sort.rs b/src/query/sql/src/planner/plans/sort.rs
@@ -111,7 +111,7 @@ impl Operator for Sort {
             .map(|s| s.scalar.clone())
             .collect();
         Ok(PhysicalProperty {
-            distribution: Distribution::Hash(partition_by),
+            distribution: Distribution::NodeToNodeHash(partition_by),
         })
     }
 
@@ -140,7 +140,7 @@ impl Operator for Sort {
             .iter()
             .map(|s| s.scalar.clone())
             .collect();
-        required.distribution = Distribution::Hash(partition_by);
+        required.distribution = Distribution::NodeToNodeHash(partition_by);
 
         Ok(required)
     }
@@ -170,7 +170,7 @@ impl Operator for Sort {
             .map(|s| s.scalar.clone())
             .collect();
 
-        required.distribution = Distribution::Hash(partition_by);
+        required.distribution = Distribution::NodeToNodeHash(partition_by);
         Ok(vec![vec![required]])
     }
 

Original file line number	Diff line number	Diff line change
`@@ -1198,7 +1198,7 @@ impl PhysicalPlanBuilder {`
`1198`	`1198`	`let build_side_data_distribution = s_expr.build_side_child().get_data_distribution()?;`
`1199`	`1199`	`let broadcast_id = if build_side_data_distribution`
`1200`	`1200`	`.as_ref()`
`1201`		`- .is_some_and(\|e\| matches!(e, databend_common_sql::plans::Exchange::Hash(_)))`
	`1201`	`+ .is_some_and(\|e\| matches!(e, databend_common_sql::plans::Exchange::NodeToNodeHash(_)))`
`1202`	`1202`	`{`
`1203`	`1203`	`Some(self.ctx.get_next_broadcast_id())`
`1204`	`1204`	`} else {`
Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ impl Display for QueryFragmentActionsWrap<'_> {`
`71`	`71`	`match data_exchange {`
`72`	`72`	`DataExchange::Merge(_) => writeln!(f, " DataExchange: Merge")?,`
`73`	`73`	`DataExchange::Broadcast(_) => writeln!(f, " DataExchange: Broadcast")?,`
`74`		`- DataExchange::ShuffleDataExchange(_) => writeln!(f, " DataExchange: Shuffle")?,`
	`74`	`+ DataExchange::NodeToNodeExchange(_) => writeln!(f, " DataExchange: Shuffle")?,`
`75`	`75`	`}`
`76`	`76`	`}`
`77`	`77`
Original file line number	Diff line number	Diff line change
`@@ -18,21 +18,21 @@ use databend_common_expression::RemoteExpr;`
`18`	`18`	`pub enum DataExchange {`
`19`	`19`	`Merge(MergeExchange),`
`20`	`20`	`Broadcast(BroadcastExchange),`
`21`		`- ShuffleDataExchange(ShuffleDataExchange),`
	`21`	`+ NodeToNodeExchange(NodeToNodeExchange),`
`22`	`22`	`}`
`23`	`23`
`24`	`24`	`impl DataExchange {`
`25`	`25`	`pub fn get_destinations(&self) -> Vec<String> {`
`26`	`26`	`match self {`
`27`	`27`	`DataExchange::Merge(exchange) => vec![exchange.destination_id.clone()],`
`28`	`28`	`DataExchange::Broadcast(exchange) => exchange.destination_ids.clone(),`
`29`		`- DataExchange::ShuffleDataExchange(exchange) => exchange.destination_ids.clone(),`
	`29`	`+ DataExchange::NodeToNodeExchange(exchange) => exchange.destination_ids.clone(),`
`30`	`30`	`}`
`31`	`31`	`}`
`32`	`32`	`}`
`33`	`33`
`34`	`34`	`#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]`
`35`		`-pub struct ShuffleDataExchange {`
	`35`	`+pub struct NodeToNodeExchange {`
`36`	`36`	`pub destination_ids: Vec<String>,`
`37`	`37`	`pub shuffle_keys: Vec<RemoteExpr>,`
`38`	`38`	`pub allow_adjust_parallelism: bool,`