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

Code before:
                    .cloned()
                    .ok_or(anyhow::anyhow!(
                        "sink fragment not found for sink id {}",
                        pb_sink.id
                    ))?;
            let upstream_info = build_upstream_sink_info(
                pb_sink,
                sink_fragment_id,
                target_table,
                target_fragment_id,
            )?;
            upstream_sink_infos.push(upstream_info);
        }

        Ok(upstream_sink_infos)
    }

    pub async fn get_mview_fragment_by_id(&self, job_id: JobId) -> MetaResult<FragmentId> {
        let inner = self.inner.read().await;
        let txn = inner.db.begin().await?;

        let mview_fragment: Vec<FragmentId> = FragmentModel::find()
            .select_only()
            .column(fragment::Column::FragmentId)
            .filter(
                fragment::Column::JobId
                    .eq(job_id)
                    .and(FragmentTypeMask::intersects(FragmentTypeFlag::Mview)),
            )
            .into_tuple()
            .all(&txn)
            .await?;

        if mview_fragment.len() != 1 {
            return Err(anyhow::anyhow!(
                "expected exactly one mview fragment for job {}, found {}",
                job_id,
                mview_fragment.len()
            )
            .into());
        }

        Ok(mview_fragment.into_iter().next().unwrap())
    }

    pub async fn has_table_been_migrated(&self, table_id: TableId) -> MetaResult<bool> {
        let inner = self.inner.read().await;
        let txn = inner.db.begin().await?;
        has_table_been_migrated(&txn, table_id).await
    }


Review comment (suggestion):
The new public method `get_all_upstream_sink_infos_in_txn` lacks rustdoc documentation. This public API should be documented with:
- The purpose of the transactional variant
- Parameters (transaction handle requirements, `target_table`, and `target_fragment_id`)
- Return value description
- Usage context and when to prefer this over the non-transactional version

Code after:
            }
        }
        Ok(source_fragment_ids)
    }

    pub async fn get_all_upstream_sink_infos(
        &self,
        target_table: &PbTable,
        target_fragment_id: FragmentId,
    ) -> MetaResult<Vec<UpstreamSinkInfo>> {
        let inner = self.inner.read().await;
        let txn = inner.db.begin().await?;

        self.get_all_upstream_sink_infos_in_txn(&txn, target_table, target_fragment_id)
            .await
    }

    pub async fn get_all_upstream_sink_infos_in_txn<C>(
        &self,
        txn: &C,
        target_table: &PbTable,
        target_fragment_id: FragmentId,
    ) -> MetaResult<Vec<UpstreamSinkInfo>>
    where
        C: ConnectionTrait,
    {
        let incoming_sinks = self
            .get_table_incoming_sinks_in_txn(txn, target_table.id)
            .await?;

        let sink_ids = incoming_sinks.iter().map(|s| s.id).collect_vec();
        let sink_fragment_ids = get_sink_fragment_by_ids(txn, sink_ids).await?;

        let mut upstream_sink_infos = Vec::with_capacity(incoming_sinks.len());
        for pb_sink in &incoming_sinks {
            let sink_fragment_id =
                sink_fragment_ids
                    .get(&pb_sink.id)
                    .cloned()
                    .ok_or(anyhow::anyhow!(
                        "sink fragment not found for sink id {}",
                        pb_sink.id
                    ))?;
            let upstream_info = build_upstream_sink_info(
                pb_sink,
                sink_fragment_id,
                target_table,
                target_fragment_id,
            )?;
            upstream_sink_infos.push(upstream_info);
        }