chore: deprecate get_reschedule_plan and clean up code

proto/meta.proto

@@ -529,40 +529,9 @@ message TableParallelism {
   }
 }
 
-message GetReschedulePlanRequest {
-  uint64 revision = 1;
-
-  message WorkerChanges {
-    repeated uint32 include_worker_ids = 1;
-    repeated uint32 exclude_worker_ids = 2;
-    optional uint32 target_parallelism = 3;
-    optional uint32 target_parallelism_per_worker = 4;
-  }
-
-  message StableResizePolicy {
-    map<uint32, WorkerChanges> fragment_worker_changes = 1;
-  }
-
-  oneof policy {
-    // The StableResizePolicy will generate a stable ReschedulePlan, without altering the distribution on WorkerId that's not involved.
-    // Note that this "Stable" doesn't refer to the "determinacy" of the algorithm.
-    // Multiple repeated calls may yield different ReschedulePlan results.
-    StableResizePolicy stable_resize_policy = 2;
-  }
-}
-
-message GetReschedulePlanResponse {
-  uint64 revision = 1;
-  // reschedule plan for each fragment
-  map<uint32, Reschedule> reschedules = 2;
-  // todo, refactor needed
-  bool success = 3;
-}
-
 service ScaleService {
   rpc GetClusterInfo(GetClusterInfoRequest) returns (GetClusterInfoResponse);
   rpc Reschedule(RescheduleRequest) returns (RescheduleResponse);
-  rpc GetReschedulePlan(GetReschedulePlanRequest) returns (GetReschedulePlanResponse);
 }
 
 message MembersRequest {}

src/batch/src/executor/join/local_lookup_join.rs

@@ -383,7 +383,7 @@ impl BoxedExecutorBuilder for LocalLookupJoinExecutorBuilder {
         let worker_slot_mapping: HashMap<WorkerSlotId, WorkerNode> = worker_nodes
             .iter()
             .flat_map(|worker| {
-                (0..(worker.parallel_units.len()))
+                (0..(worker.parallelism()))
                     .map(|i| (WorkerSlotId::new(worker.id, i), worker.clone()))
             })
             .collect();

src/batch/src/worker_manager/worker_node_manager.rs

@@ -160,8 +160,7 @@ impl WorkerNodeManager {
             .worker_nodes
             .iter()
             .flat_map(|worker| {
-                (0..worker.parallel_units.len())
-                    .map(move |i| (WorkerSlotId::new(worker.id, i), worker))
+                (0..worker.parallelism()).map(move |i| (WorkerSlotId::new(worker.id, i), worker))
             })
             .collect();
 
@@ -335,10 +334,7 @@ impl WorkerNodeSelector {
         } else {
             self.apply_worker_node_mask(self.manager.list_serving_worker_nodes())
         };
-        worker_nodes
-            .iter()
-            .map(|node| node.parallel_units.len())
-            .sum()
+        worker_nodes.iter().map(|node| node.parallelism()).sum()
     }
 
     pub fn fragment_mapping(&self, fragment_id: FragmentId) -> Result<WorkerSlotMapping> {

src/common/src/vnode_mapping/vnode_placement.rs

@@ -37,7 +37,7 @@ pub fn place_vnode(
         .iter()
         .filter(|w| w.property.as_ref().map_or(false, |p| p.is_serving))
         .sorted_by_key(|w| w.id)
-        .map(|w| (0..w.parallel_units.len()).map(|idx| WorkerSlotId::new(w.id, idx)))
+        .map(|w| (0..w.parallelism()).map(|idx| WorkerSlotId::new(w.id, idx)))
         .collect();
 
     // Set serving parallelism to the minimum of total number of worker slots, specified

src/ctl/src/cmd_impl/meta/reschedule.rs

@@ -20,9 +20,8 @@ use inquire::Confirm;
 use itertools::Itertools;
 use regex::{Match, Regex};
 use risingwave_pb::common::WorkerNode;
-use risingwave_pb::meta::get_reschedule_plan_request::PbPolicy;
 use risingwave_pb::meta::table_fragments::ActorStatus;
-use risingwave_pb::meta::{GetClusterInfoResponse, GetReschedulePlanResponse, Reschedule};
+use risingwave_pb::meta::{GetClusterInfoResponse, Reschedule};
 use serde::{Deserialize, Serialize};
 use thiserror_ext::AsReport;
 
@@ -225,16 +224,6 @@ fn parse_plan(mut plan: String) -> Result<HashMap<u32, Reschedule>, Error> {
     Ok(reschedules)
 }
 
-pub async fn get_reschedule_plan(
-    context: &CtlContext,
-    policy: PbPolicy,
-    revision: u64,
-) -> Result<GetReschedulePlanResponse> {
-    let meta_client = context.meta_client().await?;
-    let response = meta_client.get_reschedule_plan(policy, revision).await?;
-    Ok(response)
-}
-
 pub async fn unregister_workers(
     context: &CtlContext,
     workers: Vec<String>,