fix(sql-backend): clean up dirty source catalogs for failed table with connectors and recovery (#15110)

Author: yezizp2012

src/meta/model_v2/migration/src/m20230908_072257_init.rs

@@ -519,6 +519,14 @@ impl MigrationTrait for Migration {
                             .to(Connection::Table, Connection::ConnectionId)
                             .to_owned(),
                     )
+                    .foreign_key(
+                        &mut ForeignKey::create()
+                            .name("FK_source_optional_associated_table_id")
+                            .from(Source::Table, Source::OptionalAssociatedTableId)
+                            .to(Object::Table, Object::Oid)
+                            .on_delete(ForeignKeyAction::Cascade)
+                            .to_owned(),
+                    )
                     .to_owned(),
             )
             .await?;

src/meta/src/controller/catalog.rs

@@ -468,6 +468,7 @@ impl CatalogController {
             .clone()
             .into_iter()
             .chain(state_table_ids.clone().into_iter())
+            .chain(associated_source_ids.clone().into_iter())
             .collect();
 
         let res = Object::delete_many()

src/meta/src/controller/streaming_job.rs

@@ -401,13 +401,24 @@ impl CatalogController {
             .all(&txn)
             .await?;
 
+        let associated_source_id: Option<SourceId> = Table::find_by_id(job_id)
+            .select_only()
+            .column(table::Column::OptionalAssociatedSourceId)
+            .filter(table::Column::OptionalAssociatedSourceId.is_not_null())
+            .into_tuple()
+            .one(&txn)
+            .await?;
+
         Object::delete_by_id(job_id).exec(&txn).await?;
         if !internal_table_ids.is_empty() {
             Object::delete_many()
                 .filter(object::Column::Oid.is_in(internal_table_ids))
                 .exec(&txn)
                 .await?;
         }
+        if let Some(source_id) = associated_source_id {
+            Object::delete_by_id(source_id).exec(&txn).await?;
+        }
         txn.commit().await?;
 
         Ok(true)