chore: cleanup some workaround for background siit (#23749)

Author: yezizp2012

src/frontend/src/handler/cancel_job.rs

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use pgwire::pg_response::StatementType;
+use itertools::Itertools;
+use pgwire::pg_response::{PgResponse, StatementType};
 use risingwave_common::types::Fields;
+use risingwave_pb::id::JobId;
 use risingwave_pb::meta::cancel_creating_jobs_request::{CreatingJobIds, PbJobs};
 use risingwave_sqlparser::ast::JobIdents;
 
@@ -27,46 +29,16 @@ pub(super) async fn handle_cancel(
     jobs: JobIdents,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
-    let job_ids = jobs.0;
-    let mut filtered_job_ids = vec![];
-    let mut notices = vec![];
-    {
-        let catalog_reader = session.env().catalog_reader().read_guard();
-        for job_id in job_ids {
-            let database_catalog = catalog_reader.get_database_by_name(&session.database())?;
-            let sink_catalog = database_catalog
-                .iter_schemas()
-                .find_map(|schema| schema.get_sink_by_id(job_id.into()));
-            if let Some(sink_catalog) = sink_catalog {
-                if sink_catalog.is_created() {
-                    continue; // Skip already created sinks
-                } else if sink_catalog.target_table.is_some() {
-                    notices.push(format!(
-                        "Please use `DROP SINK {}` to cancel sink into table job.",
-                        sink_catalog.name
-                    ));
-                    continue;
-                }
-            }
-            filtered_job_ids.push(job_id.into());
-        }
-    }
+    let job_ids = jobs.0.into_iter().map(JobId::from).collect_vec();
 
-    let mut response_builder = RwPgResponse::builder(StatementType::CANCEL_COMMAND);
-    for notice in notices {
-        response_builder = response_builder.notice(notice);
-    }
-
-    let canceled_jobs = if !filtered_job_ids.is_empty() {
+    let canceled_jobs = if !job_ids.is_empty() {
         // Wrap in async block to convert RpcError to RwError
         execute_with_long_running_notification(
             async {
                 session
                     .env()
                     .meta_client()
-                    .cancel_creating_jobs(PbJobs::Ids(CreatingJobIds {
-                        job_ids: filtered_job_ids,
-                    }))
+                    .cancel_creating_jobs(PbJobs::Ids(CreatingJobIds { job_ids }))
                     .await
                     .map_err(Into::into)
             },
@@ -80,7 +52,9 @@ pub(super) async fn handle_cancel(
     let rows = canceled_jobs
         .into_iter()
         .map(|id| CancelRow { id: id.to_string() });
-    Ok(response_builder.rows(rows).into())
+    Ok(PgResponse::builder(StatementType::CANCEL_COMMAND)
+        .rows(rows)
+        .into())
 }
 
 #[derive(Fields)]

src/meta/src/controller/catalog/mod.rs

@@ -1166,16 +1166,6 @@ impl CatalogControllerInner {
         }
     }
 
-    pub(crate) fn notify_cancelled(&mut self, database_id: DatabaseId, job_id: JobId) {
-        if let Some(creating_tables) = self.creating_table_finish_notifier.get_mut(&database_id)
-            && let Some(tx_list) = creating_tables.remove(&job_id)
-        {
-            for tx in tx_list {
-                let _ = tx.send(Err("Cancelled".to_owned()));
-            }
-        }
-    }
-
     pub async fn list_time_travel_table_ids(&self) -> MetaResult<Vec<TableId>> {
         let table_ids: Vec<TableId> = Table::find()
             .select_only()

src/meta/src/manager/metadata.rs

@@ -732,9 +732,4 @@ impl MetadataManager {
         let mut mgr = self.catalog_controller.get_inner_write_guard().await;
         mgr.notify_finish_failed(database_id, err);
     }
-
-    pub(crate) async fn notify_cancelled(&self, database_id: DatabaseId, job_id: JobId) {
-        let mut mgr = self.catalog_controller.get_inner_write_guard().await;
-        mgr.notify_cancelled(database_id, job_id);
-    }
 }

src/meta/src/stream/stream_manager.rs

@@ -177,11 +177,6 @@ impl CreatingStreamingJobInfo {
 
         Ok((receivers, background_job_ids))
     }
-
-    async fn check_job_exists(&self, job_id: JobId) -> bool {
-        let jobs = self.streaming_jobs.lock().await;
-        jobs.contains_key(&job_id)
-    }
 }
 
 type CreatingStreamingJobInfoRef = Arc<CreatingStreamingJobInfo>;
@@ -594,23 +589,10 @@ impl GlobalStreamManager {
         database_id: DatabaseId,
         removed_actors: Vec<ActorId>,
         streaming_job_ids: Vec<JobId>,
-        state_table_ids: Vec<risingwave_meta_model::TableId>,
+        state_table_ids: Vec<TableId>,
         fragment_ids: HashSet<FragmentId>,
         dropped_sink_fragment_by_targets: HashMap<FragmentId, Vec<FragmentId>>,
     ) {
-        // TODO(august): This is a workaround for canceling SITT via drop, remove it after refactoring SITT.
-        for &job_id in &streaming_job_ids {
-            if self.creating_job_info.check_job_exists(job_id).await {
-                tracing::info!(
-                    ?job_id,
-                    "streaming job is creating, cancel it with drop directly"
-                );
-                self.metadata_manager
-                    .notify_cancelled(database_id, job_id)
-                    .await;
-            }
-        }
-
         if !removed_actors.is_empty()
             || !streaming_job_ids.is_empty()
             || !state_table_ids.is_empty()
@@ -669,6 +651,13 @@ impl GlobalStreamManager {
         // we can directly cancel them by running the barrier command.
         let futures = background_job_ids.into_iter().map(|id| async move {
             let fragment = self.metadata_manager.get_job_fragments_by_id(id).await?;
+            if fragment.is_created() {
+                tracing::warn!(
+                    "streaming job {} is already created, ignore cancel request",
+                    id
+                );
+                return Ok(None);
+            }
             if fragment.is_created() {
                 Err(MetaError::invalid_parameter(format!(
                     "streaming job {} is already created",
@@ -694,15 +683,15 @@ impl GlobalStreamManager {
                     .await?;
             }
 
-            tracing::info!(?id, "cancelled recovered streaming job");
-            Ok(id)
+            tracing::info!(?id, "cancelled background streaming job");
+            Ok(Some(id))
         });
         let cancelled_recovered_ids = join_all(futures)
             .await
             .into_iter()
             .collect::<MetaResult<Vec<_>>>()?;
 
-        cancelled_ids.extend(cancelled_recovered_ids);
+        cancelled_ids.extend(cancelled_recovered_ids.into_iter().flatten());
         Ok(cancelled_ids)
     }