Repository: risingwavelabs/risingwave
File: src/meta/src/controller/fragment.rs (Rust)

Code before:
        }

        mapping
    }

    fn format_fragment_parallelism_policy(
        fragment: &fragment::Model,
        job_parallelism: Option<&StreamingParallelism>,
        root_fragments: &[FragmentId],
    ) -> String {
        if fragment.distribution_type == DistributionType::Single {
            return "single".to_owned();
        }

        if let Some(parallelism) = fragment.parallelism.as_ref() {
            return format!(
                "override({})",
                Self::format_streaming_parallelism(parallelism)
            );
        }

        if !root_fragments.is_empty() {
            let mut upstreams = root_fragments.to_vec();
            upstreams.sort_unstable();
            upstreams.dedup();

            return format!("upstream_fragment({upstreams:?})");
        }

        let inherited = job_parallelism
            .map(Self::format_streaming_parallelism)
            .unwrap_or_else(|| "unknown".to_owned());
        format!("inherit({inherited})")
    }

    fn format_streaming_parallelism(parallelism: &StreamingParallelism) -> String {
        match parallelism {
            StreamingParallelism::Adaptive => "adaptive".to_owned(),
            StreamingParallelism::Fixed(n) => format!("fixed({n})"),
            StreamingParallelism::Custom => "custom".to_owned(),
        }
    }

    pub async fn list_sink_actor_mapping(
        &self,
    ) -> MetaResult<HashMap<SinkId, (String, Vec<ActorId>)>> {
        let inner = self.inner.read().await;
        let sink_id_names: Vec<(SinkId, String)> = Sink::find()
            .select_only()
            .columns([sink::Column::SinkId, sink::Column::Name])
            .into_tuple()

Review comment (refactor):
The operations `sort_unstable()` and `dedup()` are redundant here. The `root_fragments` slice is already sorted and deduplicated in `collect_root_fragment_mapping` (lines 1194-1195) before being stored in the HashMap. These operations can be removed to avoid unnecessary computation.
```suggestion
            let upstreams = root_fragments.to_vec();
```

Code after:
            }
        }

        mapping
    }

    fn format_fragment_parallelism_policy(
        fragment: &fragment::Model,
        job_parallelism: Option<&StreamingParallelism>,
        root_fragments: &[FragmentId],
    ) -> String {
        if fragment.distribution_type == DistributionType::Single {
            return "single".to_owned();
        }

        if let Some(parallelism) = fragment.parallelism.as_ref() {
            return format!(
                "override({})",
                Self::format_streaming_parallelism(parallelism)
            );
        }

        if !root_fragments.is_empty() {
            let mut upstreams = root_fragments.to_vec();
            upstreams.sort_unstable();
            upstreams.dedup();

            return format!("upstream_fragment({upstreams:?})");
        }

        let inherited = job_parallelism
            .map(Self::format_streaming_parallelism)
            .unwrap_or_else(|| "unknown".to_owned());
        format!("inherit({inherited})")
    }

    fn format_streaming_parallelism(parallelism: &StreamingParallelism) -> String {
        match parallelism {
            StreamingParallelism::Adaptive => "adaptive".to_owned(),
            StreamingParallelism::Fixed(n) => format!("fixed({n})"),
            StreamingParallelism::Custom => "custom".to_owned(),
        }
    }

    pub async fn list_sink_actor_mapping(
        &self,
    ) -> MetaResult<HashMap<SinkId, (String, Vec<ActorId>)>> {
        let inner = self.inner.read().await;
        let sink_id_names: Vec<(SinkId, String)> = Sink::find()
            .select_only()
            .columns([sink::Column::SinkId, sink::Column::Name])