Revert "Revert "provide create job obj in finish_streaming_job""

This reverts commit c71f181.

src/meta/src/controller/catalog.rs

@@ -635,10 +635,15 @@ impl CatalogController {
         let inner = self.inner.write().await;
         let txn = inner.db.begin().await?;
 
-        let creating_jobs: Vec<(ObjectId, ObjectType)> = streaming_job::Entity::find()
+        let creating_job_objs: Vec<PartialObject> = streaming_job::Entity::find()
             .select_only()
             .column(streaming_job::Column::JobId)
-            .column(object::Column::ObjType)
+            .columns([
+                object::Column::Oid,
+                object::Column::ObjType,
+                object::Column::SchemaId,
+                object::Column::DatabaseId,
+            ])
             .join(JoinType::InnerJoin, streaming_job::Relation::Object.def())
             .filter(
                 streaming_job::Column::JobStatus.eq(JobStatus::Initial).or(
@@ -647,13 +652,13 @@ impl CatalogController {
                         .and(streaming_job::Column::CreateType.eq(CreateType::Foreground)),
                 ),
             )
-            .into_tuple()
+            .into_partial_model()
             .all(&txn)
             .await?;
 
         let changed = Self::clean_dirty_sink_downstreams(&txn).await?;
 
-        if creating_jobs.is_empty() {
+        if creating_job_objs.is_empty() {
             if changed {
                 txn.commit().await?;
             }
@@ -666,7 +671,9 @@ impl CatalogController {
         let mut creating_source_ids = vec![];
         let mut creating_sink_ids = vec![];
         let mut creating_job_ids = vec![];
-        for (job_id, job_type) in creating_jobs {
+        for creating_job_obj in creating_job_objs {
+            let job_id = creating_job_obj.oid;
+            let job_type = creating_job_obj.obj_type;
             creating_job_ids.push(job_id);
             match job_type {
                 ObjectType::Table | ObjectType::Index => creating_table_ids.push(job_id),