feat: implement process manager and information_schema.process_list

src/catalog/src/error.rs

@@ -277,6 +277,27 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Failed to bump sequence for process manager"))]
+    BumpSequence {
+        source: common_meta::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to start process report task"))]
+    StartReportTask {
+        source: common_runtime::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to report process state"))]
+    ReportProcess {
+        source: common_meta::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 impl Error {
@@ -342,9 +363,11 @@ impl ErrorExt for Error {
             Error::Datafusion { error, .. } => datafusion_status_code::<Self>(error, None),
             Error::ProjectViewColumns { .. } => StatusCode::EngineExecuteQuery,
             Error::TableMetadataManager { source, .. } => source.status_code(),
-            Error::GetViewCache { source, .. } | Error::GetTableCache { source, .. } => {
-                source.status_code()
-            }
+            Error::GetViewCache { source, .. }
+            | Error::GetTableCache { source, .. }
+            | Error::BumpSequence { source, .. }
+            | Error::ReportProcess { source, .. } => source.status_code(),
+            Error::StartReportTask { .. } => StatusCode::Internal,
         }
     }
 

src/catalog/src/kvbackend/manager.rs

@@ -51,6 +51,7 @@ use crate::error::{
 };
 use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
 use crate::kvbackend::TableCacheRef;
+use crate::process_manager::ProcessManager;
 use crate::system_schema::pg_catalog::PGCatalogProvider;
 use crate::system_schema::SystemSchemaProvider;
 use crate::CatalogManager;
@@ -84,6 +85,7 @@ impl KvBackendCatalogManager {
         backend: KvBackendRef,
         cache_registry: LayeredCacheRegistryRef,
         procedure_manager: Option<ProcedureManagerRef>,
+        process_manager: Option<Arc<ProcessManager>>,
     ) -> Arc<Self> {
         Arc::new_cyclic(|me| Self {
             information_extension,
@@ -102,12 +104,14 @@ impl KvBackendCatalogManager {
                     DEFAULT_CATALOG_NAME.to_string(),
                     me.clone(),
                     Arc::new(FlowMetadataManager::new(backend.clone())),
+                    process_manager.clone(),
                 )),
                 pg_catalog_provider: Arc::new(PGCatalogProvider::new(
                     DEFAULT_CATALOG_NAME.to_string(),
                     me.clone(),
                 )),
                 backend,
+                process_manager,
             },
             cache_registry,
             procedure_manager,
@@ -419,6 +423,7 @@ struct SystemCatalog {
     information_schema_provider: Arc<InformationSchemaProvider>,
     pg_catalog_provider: Arc<PGCatalogProvider>,
     backend: KvBackendRef,
+    process_manager: Option<Arc<ProcessManager>>,
 }
 
 impl SystemCatalog {
@@ -486,6 +491,7 @@ impl SystemCatalog {
                         catalog.to_string(),
                         self.catalog_manager.clone(),
                         Arc::new(FlowMetadataManager::new(self.backend.clone())),
+                        self.process_manager.clone(),
                     ))
                 });
             information_schema_provider.table(table_name)

src/catalog/src/lib.rs

@@ -40,6 +40,7 @@ pub mod information_schema {
     pub use crate::system_schema::information_schema::*;
 }
 
+pub mod process_manager;
 pub mod table_source;
 
 #[async_trait::async_trait]

src/catalog/src/memory/manager.rs

@@ -356,6 +356,7 @@ impl MemoryCatalogManager {
             catalog,
             Arc::downgrade(self) as Weak<dyn CatalogManager>,
             Arc::new(FlowMetadataManager::new(Arc::new(MemoryKvBackend::new()))),
+            None, // we don't need ProcessManager on regions server.
         );
         let information_schema = information_schema_provider.tables().clone();