Repository: risingwavelabs/risingwave
File: src/meta/src/stream/scale.rs (Rust)

Code before:
        tracing::debug!(
            ?job_reschedule_post_updates,
            ?job_parallelism_updates,
            ?no_shuffle_source_fragment_ids,
            ?no_shuffle_target_fragment_ids,
            ?fragment_distribution_map,
            ?actor_location,
            ?table_fragment_id_map,
            ?fragment_actor_id_map,
            "generate_table_resize_plan, after build_index"
        );

        let adaptive_parallelism_strategy = self
            .env
            .system_params_reader()
            .await
            .adaptive_parallelism_strategy();

        let mut target_plan = HashMap::new();

        for (
            table_id,
            JobUpdate {
                filtered_worker_ids,
                parallelism,
            },
        ) in job_parallelism_updates
        {
            let assigner = AssignerBuilder::new(table_id)
                .with_actor_oriented_balancing()
                .build();

            let fragment_map = table_fragment_id_map.remove(&table_id).unwrap();

            let available_worker_slots = workers
                .iter()
                .filter(|(id, _)| filtered_worker_ids.contains(&(**id as WorkerId)))
                .map(|(_, worker)| {
                    (
                        worker.id as WorkerId,
                        NonZeroUsize::new(worker.compute_node_parallelism()).unwrap(),
                    )
                })
                .collect::<BTreeMap<_, _>>();

            for fragment_id in fragment_map {
                // Currently, all of our NO_SHUFFLE relation propagations are only transmitted from upstream to downstream.
                if no_shuffle_target_fragment_ids.contains(&fragment_id) {
                    continue;
                }


Review comment (suggestion):
[nitpick] Since the default `balance_strategy` is now `ActorCounts` (line 660 in assignment.rs), the explicit call to `.with_actor_oriented_balancing()` is redundant. Consider removing it for consistency with other usages in the codebase, or keep it if the explicit documentation of the balancing strategy is intentionally desired.
```suggestion

```

Code after:
            ?no_shuffle_target_fragment_ids,
            ?fragment_distribution_map,
            ?actor_location,
            ?table_fragment_id_map,
            ?fragment_actor_id_map,
            "generate_table_resize_plan, after build_index"
        );

        let adaptive_parallelism_strategy = self
            .env
            .system_params_reader()
            .await
            .adaptive_parallelism_strategy();

        let mut target_plan = HashMap::new();

        for (
            table_id,
            JobUpdate {
                filtered_worker_ids,
                parallelism,
            },
        ) in job_parallelism_updates
        {
            let assigner = AssignerBuilder::new(table_id)
                .with_actor_oriented_balancing()
                .build();

            let fragment_map = table_fragment_id_map.remove(&table_id).unwrap();

            let available_worker_slots = workers
                .iter()
                .filter(|(id, _)| filtered_worker_ids.contains(&(**id as WorkerId)))
                .map(|(_, worker)| {
                    (
                        worker.id as WorkerId,
                        NonZeroUsize::new(worker.compute_node_parallelism()).unwrap(),
                    )
                })
                .collect::<BTreeMap<_, _>>();

            for fragment_id in fragment_map {
                // Currently, all of our NO_SHUFFLE relation propagations are only transmitted from upstream to downstream.
                if no_shuffle_target_fragment_ids.contains(&fragment_id) {
                    continue;
                }

                let mut fragment_slots: BTreeMap<WorkerId, usize> = BTreeMap::new();

                for actor_id in &fragment_actor_id_map[&fragment_id] {
                    let worker_id = actor_location[actor_id];