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

Code before:
    pub async fn get_table_incoming_sinks(&self, table_id: TableId) -> MetaResult<Vec<PbSink>> {
        let inner = self.inner.read().await;
        let sink_objs = Sink::find()
            .find_also_related(Object)
            .filter(sink::Column::TargetTable.eq(table_id))
            .all(&inner.db)
            .await?;
        Ok(sink_objs
            .into_iter()
            .map(|(sink, obj)| ObjectModel(sink, obj.unwrap()).into())
            .collect())
    }

    pub async fn get_sink_by_id(&self, sink_id: SinkId) -> MetaResult<Option<PbSink>> {
        let inner = self.inner.read().await;
        let sink_objs = Sink::find_by_id(sink_id)
            .find_also_related(Object)
            .one(&inner.db)
            .await?;
        Ok(sink_objs.map(|(sink, obj)| ObjectModel(sink, obj.unwrap()).into()))
    }

    pub async fn get_sink_auto_refresh_schema_from(
        &self,
        table_id: TableId,
    ) -> MetaResult<Vec<PbSink>> {
        let inner = self.inner.read().await;
        let sink_objs = Sink::find()
            .find_also_related(Object)
            .filter(sink::Column::AutoRefreshSchemaFromTable.eq(table_id))
            .all(&inner.db)
            .await?;
        Ok(sink_objs
            .into_iter()
            .map(|(sink, obj)| ObjectModel(sink, obj.unwrap()).into())
            .collect())
    }

    pub async fn get_sink_state_table_ids(&self, sink_id: SinkId) -> MetaResult<Vec<TableId>> {
        let inner = self.inner.read().await;
        let tables: Vec<I32Array> = Fragment::find()
            .select_only()
            .column(fragment::Column::StateTableIds)
            .filter(fragment::Column::JobId.eq(sink_id))
            .into_tuple()
            .all(&inner.db)
            .await?;
        Ok(tables
            .into_iter()
            .flat_map(|ids| {
                ids.into_inner()

Review comment (suggestion):
The new public method `get_table_incoming_sinks_in_txn` lacks rustdoc documentation. This public API should be documented with:
- The purpose of the transactional variant
- Parameter descriptions (transaction handle and `table_id`)
- Return value description
- Usage guidance on when to use this transactional variant

Code after:
    }

    pub async fn get_table_columns(&self, id: TableId) -> MetaResult<Vec<ColumnCatalog>> {
        let inner = self.inner.read().await;
        Ok(get_table_columns(&inner.db, id)
            .await?
            .to_protobuf()
            .into_iter()
            .map(|col| col.into())
            .collect())
    }

    pub async fn get_table_incoming_sinks(&self, table_id: TableId) -> MetaResult<Vec<PbSink>> {
        let inner = self.inner.read().await;
        self.get_table_incoming_sinks_in_txn(&inner.db, table_id)
            .await
    }

    pub async fn get_table_incoming_sinks_in_txn<C>(
        &self,
        txn: &C,
        table_id: TableId,
    ) -> MetaResult<Vec<PbSink>>
    where
        C: ConnectionTrait,
    {
        let sink_objs = Sink::find()
            .find_also_related(Object)
            .filter(sink::Column::TargetTable.eq(table_id))
            .all(txn)
            .await?;
        Ok(sink_objs
            .into_iter()
            .map(|(sink, obj)| ObjectModel(sink, obj.unwrap()).into())
            .collect())
    }

    pub async fn get_sink_by_id(&self, sink_id: SinkId) -> MetaResult<Option<PbSink>> {
        let inner = self.inner.read().await;
        let sink_objs = Sink::find_by_id(sink_id)
            .find_also_related(Object)
            .one(&inner.db)
            .await?;
        Ok(sink_objs.map(|(sink, obj)| ObjectModel(sink, obj.unwrap()).into()))
    }

    pub async fn get_sink_auto_refresh_schema_from(
        &self,
        table_id: TableId,
    ) -> MetaResult<Vec<PbSink>> {
        let inner = self.inner.read().await;