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

Code before:

            result.push((fragment, upstreams));
        }
        Ok(result)
    }

    fn format_fragment_parallelism_policy(
        fragment: &fragment::Model,
        job_parallelism: Option<&StreamingParallelism>,
        noshuffle_upstreams: &[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 !noshuffle_upstreams.is_empty() {
            let mut upstreams = noshuffle_upstreams.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])

Review comment (performance):
The code creates a copy of the slice, sorts it, and deduplicates it. Since the input is already a slice that needs to be copied anyway for mutation, consider using `sort_unstable()` on the original vector if possible, or combine these operations more efficiently. However, if noshuffle_upstreams can contain duplicates from the database query, you should add deduplication logic earlier in the pipeline (lines 1134-1138) to avoid redundant data processing here.
```suggestion
            // Deduplicate and sort in one pass using BTreeSet.
            let upstreams: Vec<_> = noshuffle_upstreams.iter().cloned().collect::<BTreeSet<_>>().into_iter().collect();
```

Code after:
                fragment_type_mask: fragment_desc.fragment_type_mask as _,
                parallelism: parallelism as _,
                vnode_count: fragment_desc.vnode_count as _,
                node: Some(fragment_desc.stream_node.to_protobuf()),
                parallelism_policy,
            };

            result.push((fragment, upstreams));
        }
        Ok(result)
    }

    /// Build a fragment-to-root lookup for all reported root fragment ensembles.
    fn collect_root_fragment_mapping(
        ensembles: Vec<NoShuffleEnsemble>,
    ) -> HashMap<FragmentId, Vec<FragmentId>> {
        let mut mapping = HashMap::new();

        for ensemble in ensembles {
            let mut roots: Vec<_> = ensemble.entry_fragments().collect();
            roots.sort_unstable();
            roots.dedup();

            if roots.is_empty() {
                continue;
            }

            let root_set: HashSet<_> = roots.iter().copied().collect();

            for fragment_id in ensemble.component_fragments() {
                if root_set.contains(&fragment_id) {
                    mapping.insert(fragment_id, Vec::new());
                } else {
                    mapping.insert(fragment_id, roots.clone());
                }
            }
        }

        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() {