Merge pull request #8736 from xudong963/rename

refactor: remove `V2` postfix in cluster

Author: mergify[bot]

src/query/service/src/api/mod.rs

@@ -33,7 +33,7 @@ pub use rpc::MergeExchange;
 pub use rpc::PrecommitBlock;
 pub use rpc::QueryFragmentsPlanPacket;
 pub use rpc::ServerFlightExchange;
-pub use rpc::ShuffleDataExchangeV2;
+pub use rpc::ShuffleDataExchange;
 pub use rpc_service::RpcService;
 
 pub mod http;

src/query/service/src/api/rpc/exchange/data_exchange.rs

@@ -18,36 +18,36 @@ use common_sql::executor::PhysicalScalar;
 pub enum DataExchange {
     Merge(MergeExchange),
     Broadcast(BroadcastExchange),
-    ShuffleDataExchangeV2(ShuffleDataExchangeV2),
+    ShuffleDataExchange(ShuffleDataExchange),
 }
 
 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::ShuffleDataExchangeV2(exchange) => exchange.destination_ids.clone(),
+            DataExchange::ShuffleDataExchange(exchange) => exchange.destination_ids.clone(),
         }
     }
 
     pub fn from_multiple_nodes(&self) -> bool {
         match self {
             DataExchange::Merge(_) => true,
-            DataExchange::ShuffleDataExchangeV2(_) => true,
+            DataExchange::ShuffleDataExchange(_) => true,
             DataExchange::Broadcast(exchange) => exchange.from_multiple_nodes,
         }
     }
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct ShuffleDataExchangeV2 {
+pub struct ShuffleDataExchange {
     pub destination_ids: Vec<String>,
     pub shuffle_keys: Vec<PhysicalScalar>,
 }
 
-impl ShuffleDataExchangeV2 {
+impl ShuffleDataExchange {
     pub fn create(destination_ids: Vec<String>, shuffle_keys: Vec<PhysicalScalar>) -> DataExchange {
-        DataExchange::ShuffleDataExchangeV2(ShuffleDataExchangeV2 {
+        DataExchange::ShuffleDataExchange(ShuffleDataExchange {
             destination_ids,
             shuffle_keys,
         })

src/query/service/src/api/rpc/exchange/exchange_manager.rs

@@ -41,7 +41,7 @@ use crate::api::rpc::exchange::statistics_receiver::StatisticsReceiver;
 use crate::api::rpc::exchange::statistics_sender::StatisticsSender;
 use crate::api::rpc::flight_client::FlightExchange;
 use crate::api::rpc::flight_scatter_broadcast::BroadcastFlightScatter;
-use crate::api::rpc::flight_scatter_hash_v2::HashFlightScatterV2;
+use crate::api::rpc::flight_scatter_hash::HashFlightScatter;
 use crate::api::rpc::Packet;
 use crate::api::DataExchange;
 use crate::api::FlightClient;
@@ -636,14 +636,14 @@ impl FragmentCoordinator {
                     )?)),
                 }))
             }
-            Some(DataExchange::ShuffleDataExchangeV2(exchange)) => {
+            Some(DataExchange::ShuffleDataExchange(exchange)) => {
                 Ok(ExchangeParams::ShuffleExchange(ShuffleExchangeParams {
                     schema: self.payload.schema()?,
                     fragment_id: self.fragment_id,
                     query_id: info.query_id.to_string(),
                     executor_id: info.current_executor.to_string(),
                     destination_ids: exchange.destination_ids.to_owned(),
-                    shuffle_scatter: Arc::new(HashFlightScatterV2::try_create(
+                    shuffle_scatter: Arc::new(HashFlightScatter::try_create(
                         info.query_ctx.try_get_function_context()?,
                         exchange.shuffle_keys.clone(),
                         exchange.destination_ids.len(),
@@ -658,7 +658,7 @@ impl FragmentCoordinator {
             self.initialized = true;
 
             match &self.payload {
-                FragmentPayload::PlanV2(plan) => {
+                FragmentPayload::Plan(plan) => {
                     let pipeline_ctx = QueryContext::create_from(ctx);
                     let pipeline_builder = PipelineBuilderV2::create(pipeline_ctx);
                     self.pipeline_build_res = Some(pipeline_builder.finalize(plan)?);

src/query/service/src/api/rpc/exchange/mod.rs

@@ -26,5 +26,5 @@ mod statistics_sender;
 pub use data_exchange::BroadcastExchange;
 pub use data_exchange::DataExchange;
 pub use data_exchange::MergeExchange;
-pub use data_exchange::ShuffleDataExchangeV2;
+pub use data_exchange::ShuffleDataExchange;
 pub use exchange_manager::DataExchangeManager;

src/query/service/src/api/rpc/flight_scatter_hash_v2.rs renamed to src/query/service/src/api/rpc/flight_scatter_hash.rs

@@ -30,14 +30,14 @@ use crate::sql::evaluator::Evaluator;
 use crate::sql::evaluator::TypedVector;
 
 #[derive(Clone)]
-pub struct HashFlightScatterV2 {
+pub struct HashFlightScatter {
     func_ctx: FunctionContext,
     hash_keys: Vec<EvalNode>,
     hash_functions: Vec<Box<dyn Function>>,
     scatter_size: usize,
 }
 
-impl HashFlightScatterV2 {
+impl HashFlightScatter {
     pub fn try_create(
         func_ctx: FunctionContext,
         scalars: Vec<PhysicalScalar>,
@@ -187,7 +187,7 @@ impl FlightScatter for OneHashKeyFlightScatter {
     }
 }
 
-impl FlightScatter for HashFlightScatterV2 {
+impl FlightScatter for HashFlightScatter {
     fn execute(&self, data_block: &DataBlock, _num: usize) -> Result<Vec<DataBlock>> {
         let hash_keys = self
             .hash_keys

src/query/service/src/api/rpc/mod.rs

@@ -21,7 +21,7 @@ mod flight_actions;
 mod flight_client;
 mod flight_scatter;
 mod flight_scatter_broadcast;
-mod flight_scatter_hash_v2;
+mod flight_scatter_hash;
 mod flight_service;
 mod packets;
 mod request_builder;
@@ -30,7 +30,7 @@ pub use exchange::BroadcastExchange;
 pub use exchange::DataExchange;
 pub use exchange::DataExchangeManager;
 pub use exchange::MergeExchange;
-pub use exchange::ShuffleDataExchangeV2;
+pub use exchange::ShuffleDataExchange;
 pub use flight_client::ClientFlightExchange;
 pub use flight_client::ServerFlightExchange;
 pub use packets::ConnectionInfo;

src/query/service/src/api/rpc/packets/packet_fragment.rs

@@ -26,21 +26,21 @@ use crate::sql::executor::PhysicalPlan;
 #[allow(clippy::large_enum_variant)]
 #[derive(Clone, serde::Serialize, serde::Deserialize)]
 pub enum FragmentPayload {
-    PlanV2(PhysicalPlan),
+    Plan(PhysicalPlan),
 }
 
 impl Debug for FragmentPayload {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         match self {
-            FragmentPayload::PlanV2(plan) => write!(f, "PhysicalPlan({:?})", plan),
+            FragmentPayload::Plan(plan) => write!(f, "PhysicalPlan({:?})", plan),
         }
     }
 }
 
 impl FragmentPayload {
     pub fn schema(&self) -> Result<DataSchemaRef> {
         match self {
-            FragmentPayload::PlanV2(plan) => plan.output_schema(),
+            FragmentPayload::Plan(plan) => plan.output_schema(),
         }
     }
 }

src/query/service/src/interpreters/fragments/v2/fragmenter.rs renamed to src/query/service/src/interpreters/fragments/fragmenter.rs

@@ -18,13 +18,13 @@ use common_catalog::table_context::TableContext;
 use common_exception::Result;
 use common_sql::executor::FragmentKind;
 
-use super::FragmentType;
-use super::PlanFragment;
 use crate::api::BroadcastExchange;
 use crate::api::DataExchange;
 use crate::api::MergeExchange;
-use crate::api::ShuffleDataExchangeV2;
+use crate::api::ShuffleDataExchange;
 use crate::clusters::ClusterHelper;
+use crate::interpreters::fragments::plan_fragment::FragmentType;
+use crate::interpreters::fragments::PlanFragment;
 use crate::sessions::QueryContext;
 use crate::sql::executor::Exchange;
 use crate::sql::executor::ExchangeSink;
@@ -76,7 +76,7 @@ impl Fragmenter {
     ) -> Result<Option<DataExchange>> {
         match plan {
             PhysicalPlan::ExchangeSink(plan) => match plan.kind {
-                FragmentKind::Normal => Ok(Some(ShuffleDataExchangeV2::create(
+                FragmentKind::Normal => Ok(Some(ShuffleDataExchange::create(
                     Self::get_executors(ctx),
                     plan.keys.clone(),
                 ))),

src/query/service/src/interpreters/fragments/mod.rs

@@ -12,12 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod fragmenter;
+mod plan_fragment;
 mod query_fragment_actions;
 mod query_fragment_actions_display;
-mod v2;
 
+pub use fragmenter::Fragmenter;
+pub use plan_fragment::PlanFragment;
 pub use query_fragment_actions::QueryFragmentAction;
 pub use query_fragment_actions::QueryFragmentActions;
 pub use query_fragment_actions::QueryFragmentsActions;
-pub use v2::Fragmenter;
-pub use v2::PlanFragment;

src/query/service/src/interpreters/fragments/v2/plan_fragment.rs renamed to src/query/service/src/interpreters/fragments/plan_fragment.rs

@@ -18,8 +18,8 @@ use common_catalog::plan::DataSourcePlan;
 use common_exception::ErrorCode;
 use common_exception::Result;
 
-use super::Fragmenter;
 use crate::api::DataExchange;
+use crate::interpreters::fragments::Fragmenter;
 use crate::interpreters::QueryFragmentAction;
 use crate::interpreters::QueryFragmentActions;
 use crate::interpreters::QueryFragmentsActions;
@@ -69,7 +69,7 @@ impl PlanFragment {
 
         match &self.fragment_type {
             FragmentType::Root => {
-                let action = QueryFragmentAction::create_v2(
+                let action = QueryFragmentAction::create(
                     Fragmenter::get_local_executor(ctx),
                     self.plan.clone(),
                 );
@@ -87,15 +87,15 @@ impl PlanFragment {
                 {
                     // If this is a intermediate fragment with merge input,
                     // we will only send it to coordinator node.
-                    let action = QueryFragmentAction::create_v2(
+                    let action = QueryFragmentAction::create(
                         Fragmenter::get_local_executor(ctx),
                         self.plan.clone(),
                     );
                     fragment_actions.add_action(action);
                 } else {
                     // Otherwise distribute the fragment to all the executors.
                     for executor in Fragmenter::get_executors(ctx) {
-                        let action = QueryFragmentAction::create_v2(executor, self.plan.clone());
+                        let action = QueryFragmentAction::create(executor, self.plan.clone());
                         fragment_actions.add_action(action);
                     }
                 }
@@ -155,10 +155,8 @@ impl PlanFragment {
             };
             plan = replace_read_source.replace(&plan)?;
 
-            fragment_actions.add_action(QueryFragmentAction::create_v2(
-                executor.clone(),
-                plan.clone(),
-            ));
+            fragment_actions
+                .add_action(QueryFragmentAction::create(executor.clone(), plan.clone()));
         }
 
         Ok(fragment_actions)