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

Code before:
                state_table_ids: fragment_desc.state_table_ids.into_u32_array(),
                upstream_fragment_ids: upstreams.iter().map(|id| *id as _).collect(),
                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();
        }

Review comment (refactor):
Since `entry_fragments()` returns an iterator over a `HashSet`, the items are unordered and may contain duplicates after collection. However, `HashSet` guarantees uniqueness, so `dedup()` after sorting is unnecessary. Consider removing the `dedup()` call as it adds no value after collecting from a `HashSet`.
```suggestion

```

Code after:
                    as _,
                state_table_ids: fragment_desc.state_table_ids.into_u32_array(),
                upstream_fragment_ids: upstreams.iter().map(|id| *id as _).collect(),
                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();