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

Code before:
            .filter(
                streaming_job::Column::JobStatus
                    .eq(JobStatus::Created)
                    .or(table::Column::TableType.eq(TableType::MaterializedView)),
            )
            .all(&self.db)
            .await?;

        let job_statuses: HashMap<JobId, JobStatus> = StreamingJob::find()
            .select_only()
            .column(streaming_job::Column::JobId)
            .column(streaming_job::Column::JobStatus)
            .filter(
                streaming_job::Column::JobStatus
                    .eq(JobStatus::Created)
                    .or(streaming_job::Column::CreateType.eq(CreateType::Background)),
            )
            .into_tuple::<(JobId, JobStatus)>()
            .all(&self.db)
            .await?
            .into_iter()
            .collect();

        let sink_ids: HashSet<SinkId> = Sink::find()
            .select_only()
            .column(sink::Column::SinkId)
            .into_tuple::<SinkId>()
            .all(&self.db)
            .await?
            .into_iter()
            .collect();

        let job_ids: HashSet<JobId> = table_objs
            .iter()
            .map(|(t, _)| t.table_id.as_job_id())
            .chain(job_statuses.keys().cloned())
            .chain(sink_ids.into_iter().map(|s| s.as_job_id()))
            .collect();

        let internal_table_objs = Table::find()
            .find_also_related(Object)
            .filter(
                table::Column::TableType
                    .eq(TableType::Internal)
                    .and(table::Column::BelongsToJobId.is_in(job_ids)),
            )
            .all(&self.db)
            .await?;

        Ok(table_objs
            .into_iter()

Review comment (suggestion):
I think we should also add creating sink catalogs in `list_sinks`.

Code after:
            .filter(
                streaming_job::Column::JobStatus.eq(JobStatus::Created).or(
                    table::Column::TableType
                        .eq(TableType::MaterializedView)
                        .or(streaming_job::Column::CreateType.eq(CreateType::Background)),
                ),
            )
            .all(&self.db)
            .await?;

        let job_statuses: HashMap<JobId, JobStatus> = StreamingJob::find()
            .select_only()
            .column(streaming_job::Column::JobId)
            .column(streaming_job::Column::JobStatus)
            .filter(
                streaming_job::Column::JobStatus
                    .eq(JobStatus::Created)
                    .or(streaming_job::Column::CreateType.eq(CreateType::Background)),
            )
            .into_tuple::<(JobId, JobStatus)>()
            .all(&self.db)
            .await?
            .into_iter()
            .collect();

        let sink_ids: HashSet<SinkId> = Sink::find()
            .select_only()
            .column(sink::Column::SinkId)
            .into_tuple::<SinkId>()
            .all(&self.db)
            .await?
            .into_iter()
            .collect();

        let job_ids: HashSet<JobId> = table_objs
            .iter()
            .map(|(t, _)| t.table_id.as_job_id())
            .chain(job_statuses.keys().cloned())
            .chain(sink_ids.into_iter().map(|s| s.as_job_id()))
            .collect();

        let internal_table_objs = Table::find()
            .find_also_related(Object)
            .filter(
                table::Column::TableType
                    .eq(TableType::Internal)
                    .and(table::Column::BelongsToJobId.is_in(job_ids)),
            )
            .all(&self.db)
            .await?;