fix: self cr

Author: fengys1996

src/cmd/src/extension/common.rs

@@ -12,56 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
 use std::result::Result;
 use std::sync::Arc;
 
-use catalog::CatalogManagerRef;
-use catalog::information_schema::InformationSchemaTableFactoryRef;
 use common_error::ext::BoxedError;
-use common_meta::FlownodeId;
-use common_meta::kv_backend::KvBackendRef;
-use flow::FrontendClient;
 use servers::grpc::builder::GrpcServerBuilder;
 
-/// Provides additional information schema table factories beyond the built-in
-/// ones.
-///
-/// These are typically provided by enterprise or optional modules, such as:
-/// - `information_schema.triggers`
-/// - `information_schema.alerts`
-#[async_trait::async_trait]
-pub trait InformationSchemaTableFactories: Send + Sync {
-    async fn create_factories(
-        &self,
-        ctx: TableFactoryContext,
-    ) -> Result<HashMap<String, InformationSchemaTableFactoryRef>, BoxedError>;
-}
-
-pub type InformationSchemaTableFactoriesRef = Arc<dyn InformationSchemaTableFactories>;
-
-/// Context for information schema table factory providers.
-pub struct TableFactoryContext {
-    pub fe_client: Arc<FrontendClient>,
-}
-
 /// Allows extending the gRPC server with additional services (e.g., enterprise
 /// features).
 #[async_trait::async_trait]
 pub trait GrpcExtension: Send + Sync {
-    async fn extend_grpc_services(
-        &self,
-        builder: &mut GrpcServerBuilder,
-        ctx: GrpcExtensionContext,
-    ) -> Result<(), BoxedError>;
-}
-
-/// Context provided to gRPC service extensions during server construction.
-pub struct GrpcExtensionContext {
-    pub kv_backend: KvBackendRef,
-    pub fe_client: Arc<FrontendClient>,
-    pub flownode_id: FlownodeId,
-    pub catalog_manager: CatalogManagerRef,
+    async fn extend_grpc_services(&self, builder: &mut GrpcServerBuilder)
+    -> Result<(), BoxedError>;
 }
 
 pub type GrpcExtensionRef = Arc<dyn GrpcExtension>;

src/cmd/src/extension/flownode.rs

@@ -12,10 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
+use catalog::CatalogManagerRef;
 use common_error::ext::BoxedError;
+use common_meta::FlownodeId;
+use common_meta::kv_backend::KvBackendRef;
+use flow::FrontendClient;
 
 use crate::extension::common::GrpcExtensionRef;
 
+/// The extension point for flownode instance.
 #[derive(Default)]
 pub struct Extension {
     pub grpc: Option<GrpcExtensionRef>,
@@ -29,12 +36,19 @@ pub trait ExtensionFactory: Send + Sync {
     ) -> impl Future<Output = Result<Extension, BoxedError>> + Send;
 }
 
-pub struct ExtensionContext {}
+/// Context provided to ExtensionFactory during extension creation.
+pub struct ExtensionContext {
+    pub kv_backend: KvBackendRef,
+    pub fe_client: Arc<FrontendClient>,
+    pub flownode_id: FlownodeId,
+    pub catalog_manager: CatalogManagerRef,
+}
 
+/// Default no-op implementation of ExtensionFactory.
 pub struct DefaultExtensionFactory;
 
 impl ExtensionFactory for DefaultExtensionFactory {
-    async fn create(&self, _ctx: ExtensionContext) -> Result<Extension, BoxedError> {
+    async fn create(&self, _: ExtensionContext) -> Result<Extension, BoxedError> {
         Ok(Extension::default())
     }
 }

src/cmd/src/extension/frontend.rs

@@ -21,6 +21,7 @@ use meta_client::MetaClientRef;
 #[cfg(feature = "enterprise")]
 use operator::statement::TriggerQuerierRef;
 
+/// The extension point for frontend instance.
 #[derive(Default)]
 pub struct Extension {
     pub info_schema_factories: Option<HashMap<String, InformationSchemaTableFactoryRef>>,
@@ -36,15 +37,17 @@ pub trait ExtensionFactory: Send + Sync {
     ) -> impl Future<Output = Result<Extension, BoxedError>> + Send;
 }
 
+/// Context provided to ExtensionFactory during extension creation.
 pub struct ExtensionContext {
     pub kv_backend: KvBackendRef,
     pub meta_client: MetaClientRef,
 }
 
+/// Default no-op implementation of ExtensionFactory.
 pub struct DefaultExtensionFactory;
 
 impl ExtensionFactory for DefaultExtensionFactory {
-    async fn create(&self, _ctx: ExtensionContext) -> Result<Extension, BoxedError> {
+    async fn create(&self, _: ExtensionContext) -> Result<Extension, BoxedError> {
         Ok(Extension::default())
     }
 }

src/cmd/src/extension/standalone.rs

@@ -54,18 +54,22 @@ pub struct ExtensionContext {
 /// These are typically provided by enterprise or optional modules, such as:
 /// - `information_schema.triggers`
 /// - `information_schema.alerts`
+///
+/// It is separated from the [`ExtensionFactory`] to avoid circular dependencies,
+/// since the [`CatalogManagerRef`] of [`ExtensionContext`] is dependent on the
+/// [`InformationSchemaTableFactories`].
 #[async_trait::async_trait]
 pub trait InformationSchemaTableFactories: Send + Sync {
     async fn create_factories(
         &self,
-        ctx: TableFactoryContext,
+        ctx: InfoTableFactoryContext,
     ) -> Result<HashMap<String, InformationSchemaTableFactoryRef>, BoxedError>;
 }
 
 pub type InformationSchemaTableFactoriesRef = Arc<dyn InformationSchemaTableFactories>;
 
 /// Context for information schema table factory providers.
-pub struct TableFactoryContext {
+pub struct InfoTableFactoryContext {
     pub fe_client: Arc<FrontendClient>,
 }
 
@@ -76,7 +80,7 @@ pub struct DefaultExtensionFactory;
 impl InformationSchemaTableFactories for DefaultExtensionFactory {
     async fn create_factories(
         &self,
-        _ctx: TableFactoryContext,
+        _ctx: InfoTableFactoryContext,
     ) -> Result<HashMap<String, InformationSchemaTableFactoryRef>, BoxedError> {
         Ok(HashMap::new())
     }

src/cmd/src/flownode.rs

@@ -47,7 +47,6 @@ use crate::error::{
     BuildCacheRegistrySnafu, InitMetadataSnafu, LoadLayeredConfigSnafu, MetaClientInitSnafu,
     MissingConfigSnafu, OtherSnafu, Result, ShutdownFlownodeSnafu, StartFlownodeSnafu,
 };
-use crate::extension::common::GrpcExtensionContext;
 use crate::extension::flownode::{ExtensionContext, ExtensionFactory};
 use crate::options::{GlobalOptions, GreptimeOptions};
 use crate::{App, create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};
@@ -403,21 +402,22 @@ impl StartCommand {
 
         let mut flownode = flownode_builder.build().await.context(StartFlownodeSnafu)?;
 
-        let mut builder =
-            FlownodeServiceBuilder::grpc_server_builder(&opts, flownode.flownode_server());
+        let context = ExtensionContext {
+            kv_backend: cached_meta_backend.clone(),
+            fe_client: frontend_client.clone(),
+            flownode_id: member_id,
+            catalog_manager: catalog_manager.clone(),
+        };
         let extension = extension_factory
-            .create(ExtensionContext {})
+            .create(context)
             .await
             .context(OtherSnafu)?;
+
+        let mut builder =
+            FlownodeServiceBuilder::grpc_server_builder(&opts, flownode.flownode_server());
         if let Some(grpc_extension) = extension.grpc {
-            let ctx = GrpcExtensionContext {
-                kv_backend: cached_meta_backend.clone(),
-                fe_client: frontend_client.clone(),
-                flownode_id: member_id,
-                catalog_manager: catalog_manager.clone(),
-            };
             grpc_extension
-                .extend_grpc_services(&mut builder, ctx)
+                .extend_grpc_services(&mut builder)
                 .await
                 .context(OtherSnafu)?
         }

src/cmd/src/standalone.rs

@@ -66,7 +66,7 @@ use standalone::options::StandaloneOptions;
 use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{OtherSnafu, Result, StartFlownodeSnafu};
-use crate::extension::standalone::{ExtensionContext, ExtensionFactory, TableFactoryContext};
+use crate::extension::standalone::{ExtensionContext, ExtensionFactory, InfoTableFactoryContext};
 use crate::options::{GlobalOptions, GreptimeOptions};
 use crate::{App, create_resource_limit_metrics, error, log_versions, maybe_activate_heap_profile};
 
@@ -423,7 +423,7 @@ impl StartCommand {
         let frontend_client = Arc::new(frontend_client);
 
         let info_schema_table_factories = extension_factory
-            .create_factories(TableFactoryContext {
+            .create_factories(InfoTableFactoryContext {
                 fe_client: frontend_client.clone(),
             })
             .await
@@ -522,7 +522,7 @@ impl StartCommand {
             .context(error::InitDdlManagerSnafu)?;
         #[cfg(feature = "enterprise")]
         let ddl_manager = if let Some(trigger_ddl_manager) = extension.trigger_ddl_manager {
-            ddl_manager.with_trigger_ddl_manager_opt(Some(trigger_ddl_manager))
+            ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
         } else {
             ddl_manager
         };

src/common/meta/src/ddl_manager.rs

@@ -148,11 +148,8 @@ impl DdlManager {
     }
 
     #[cfg(feature = "enterprise")]
-    pub fn with_trigger_ddl_manager_opt(
-        mut self,
-        trigger_ddl_manager: Option<TriggerDdlManagerRef>,
-    ) -> Self {
-        self.trigger_ddl_manager = trigger_ddl_manager;
+    pub fn with_trigger_ddl_manager(mut self, trigger_ddl_manager: TriggerDdlManagerRef) -> Self {
+        self.trigger_ddl_manager = Some(trigger_ddl_manager);
         self
     }
 

src/meta-srv/src/metasrv/builder.rs

@@ -414,7 +414,7 @@ impl MetasrvBuilder {
         #[cfg(feature = "enterprise")]
         let ddl_manager =
             if let Some(trigger_ddl_manager) = extension.and_then(|e| e.trigger_ddl_manager) {
-                ddl_manager.with_trigger_ddl_manager_opt(Some(trigger_ddl_manager))
+                ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
             } else {
                 ddl_manager
             };