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

Code before:
                actor_dispatchers_map
                    .entry(actor_id as _)
                    .or_default()
                    .push(dispatchers);
            }
        }
        Ok(actor_dispatchers_map)
    }

    pub async fn get_fragment_downstream_relations(
        &self,
        fragment_ids: Vec<FragmentId>,
    ) -> MetaResult<FragmentDownstreamRelation> {
        let inner = self.inner.read().await;
        let mut stream = FragmentRelation::find()
            .filter(fragment_relation::Column::SourceFragmentId.is_in(fragment_ids))
            .stream(&inner.db)
            .await?;
        let mut relations = FragmentDownstreamRelation::new();
        while let Some(relation) = stream.try_next().await? {
            relations
                .entry(relation.source_fragment_id as _)
                .or_default()
                .push(DownstreamFragmentRelation {
                    downstream_fragment_id: relation.target_fragment_id as _,
                    dispatcher_type: relation.dispatcher_type,
                    dist_key_indices: relation.dist_key_indices.into_u32_array(),
                    output_mapping: PbDispatchOutputMapping {
                        indices: relation.output_indices.into_u32_array(),
                        types: relation
                            .output_type_mapping
                            .unwrap_or_default()
                            .to_protobuf(),
                    },
                });
        }
        Ok(relations)
    }

    pub async fn get_job_fragment_backfill_scan_type(
        &self,
        job_id: JobId,
    ) -> MetaResult<HashMap<crate::model::FragmentId, PbStreamScanType>> {
        let inner = self.inner.read().await;
        let fragments: Vec<_> = FragmentModel::find()
            .filter(fragment::Column::JobId.eq(job_id))
            .all(&inner.db)
            .await?;

        let mut result = HashMap::new();


Review comment (suggestion):
The new public method `get_fragment_downstream_relations_in_txn` lacks rustdoc documentation. Since this is a public API that accepts a transaction handle for use in transactional contexts, it should include documentation explaining:
- The purpose of the transactional variant
- Parameters (especially the generic `txn` parameter and `fragment_ids`)
- Return value semantics
- Example usage or when to use this variant vs the non-transactional one

Code after:
                actor_dispatchers_map
                    .entry(actor_id as _)
                    .or_default()
                    .push(dispatchers);
            }
        }
        Ok(actor_dispatchers_map)
    }

    pub async fn get_fragment_downstream_relations(
        &self,
        fragment_ids: Vec<FragmentId>,
    ) -> MetaResult<FragmentDownstreamRelation> {
        let inner = self.inner.read().await;
        self.get_fragment_downstream_relations_in_txn(&inner.db, fragment_ids)
            .await
    }

    pub async fn get_fragment_downstream_relations_in_txn<C>(
        &self,
        txn: &C,
        fragment_ids: Vec<FragmentId>,
    ) -> MetaResult<FragmentDownstreamRelation>
    where
        C: ConnectionTrait + StreamTrait + Send,
    {
        let mut stream = FragmentRelation::find()
            .filter(fragment_relation::Column::SourceFragmentId.is_in(fragment_ids))
            .stream(txn)
            .await?;
        let mut relations = FragmentDownstreamRelation::new();
        while let Some(relation) = stream.try_next().await? {
            relations
                .entry(relation.source_fragment_id as _)
                .or_default()
                .push(DownstreamFragmentRelation {
                    downstream_fragment_id: relation.target_fragment_id as _,
                    dispatcher_type: relation.dispatcher_type,
                    dist_key_indices: relation.dist_key_indices.into_u32_array(),
                    output_mapping: PbDispatchOutputMapping {
                        indices: relation.output_indices.into_u32_array(),
                        types: relation
                            .output_type_mapping
                            .unwrap_or_default()
                            .to_protobuf(),
                    },
                });
        }
        Ok(relations)
    }