fix: not set trigger querier

Author: fengys1996

src/cmd/src/frontend.rs

@@ -469,7 +469,7 @@ impl StartCommand {
         );
         let heartbeat_task = Some(heartbeat_task);
 
-        let instance = FrontendBuilder::new(
+        let builder = FrontendBuilder::new(
             opts.clone(),
             cached_meta_backend.clone(),
             layered_cache_registry.clone(),
@@ -479,10 +479,19 @@ impl StartCommand {
             process_manager,
         )
         .with_plugin(plugins.clone())
-        .with_local_cache_invalidator(layered_cache_registry)
-        .try_build()
-        .await
-        .context(error::StartFrontendSnafu)?;
+        .with_local_cache_invalidator(layered_cache_registry);
+
+        #[cfg(feature = "enterprise")]
+        let builder = if let Some(factory) = extension.trigger_querier_factory {
+            builder.with_trigger_querier(factory)
+        } else {
+            builder
+        };
+
+        let instance = builder
+            .try_build()
+            .await
+            .context(error::StartFrontendSnafu)?;
         let instance = Arc::new(instance);
 
         let export_metrics_task = ExportMetricsTask::try_new(&opts.export_metrics, Some(&plugins))

src/frontend/src/instance/builder.rs

@@ -32,6 +32,8 @@ use operator::flow::FlowServiceOperator;
 use operator::insert::Inserter;
 use operator::procedure::ProcedureServiceOperator;
 use operator::request::Requester;
+#[cfg(feature = "enterprise")]
+use operator::statement::TriggerQuerierFactoryRef;
 use operator::statement::{StatementExecutor, StatementExecutorRef};
 use operator::table::TableMutationOperator;
 use partition::manager::PartitionRuleManager;
@@ -58,6 +60,8 @@ pub struct FrontendBuilder {
     plugins: Option<Plugins>,
     procedure_executor: ProcedureExecutorRef,
     process_manager: ProcessManagerRef,
+    #[cfg(feature = "enterprise")]
+    trigger_querier_factory: Option<TriggerQuerierFactoryRef>,
 }
 
 impl FrontendBuilder {
@@ -81,6 +85,8 @@ impl FrontendBuilder {
             plugins: None,
             procedure_executor,
             process_manager,
+            #[cfg(feature = "enterprise")]
+            trigger_querier_factory: None,
         }
     }
 
@@ -98,6 +104,14 @@ impl FrontendBuilder {
         }
     }
 
+    #[cfg(feature = "enterprise")]
+    pub fn with_trigger_querier(self, trigger_querier: TriggerQuerierFactoryRef) -> Self {
+        Self {
+            trigger_querier_factory: Some(trigger_querier),
+            ..self
+        }
+    }
+
     pub async fn try_build(self) -> Result<Instance> {
         let kv_backend = self.kv_backend;
         let node_manager = self.node_manager;
@@ -188,12 +202,10 @@ impl FrontendBuilder {
         );
 
         #[cfg(feature = "enterprise")]
-        let statement_executor =
-            if let Some(factory) = plugins.get::<operator::statement::TriggerQuerierFactoryRef>() {
-                statement_executor.with_trigger_querier(factory.create(kv_backend.clone()))
-            } else {
-                statement_executor
-            };
+        if let Some(factory) = self.trigger_querier_factory.as_ref() {
+            let trigger_querier = factory.create(kv_backend.clone());
+            statement_executor = statement_executor.with_trigger_querier(trigger_querier);
+        }
 
         let statement_executor = Arc::new(statement_executor);