feat: enhanced plugins mechanism

Cargo.toml

@@ -58,6 +58,7 @@ members = [
     "src/partition",
     "src/pipeline",
     "src/plugins",
+    "src/plugins-framework",
     "src/promql",
     "src/puffin",
     "src/query",
@@ -300,6 +301,7 @@ operator = { path = "src/operator" }
 partition = { path = "src/partition" }
 pipeline = { path = "src/pipeline" }
 plugins = { path = "src/plugins" }
+plugins-framework = { path = "src/plugins-framework" }
 promql = { path = "src/promql" }
 puffin = { path = "src/puffin" }
 query = { path = "src/query" }

src/catalog/Cargo.toml

@@ -5,7 +5,6 @@ edition.workspace = true
 license.workspace = true
 
 [features]
-enterprise = []
 testing = []
 
 [lints]

src/catalog/src/kvbackend/builder.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use common_catalog::consts::DEFAULT_CATALOG_NAME;
@@ -23,9 +24,9 @@ use common_procedure::ProcedureManagerRef;
 use moka::sync::Cache;
 use partition::manager::PartitionRuleManager;
 
-#[cfg(feature = "enterprise")]
-use crate::information_schema::InformationSchemaTableFactoryRef;
-use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
+use crate::information_schema::{
+    InformationExtensionRef, InformationSchemaProvider, InformationSchemaTableFactoryRef,
+};
 use crate::kvbackend::KvBackendCatalogManager;
 use crate::kvbackend::manager::{CATALOG_CACHE_MAX_CAPACITY, SystemCatalog};
 use crate::process_manager::ProcessManagerRef;
@@ -38,9 +39,7 @@ pub struct KvBackendCatalogManagerBuilder {
     cache_registry: LayeredCacheRegistryRef,
     procedure_manager: Option<ProcedureManagerRef>,
     process_manager: Option<ProcessManagerRef>,
-    #[cfg(feature = "enterprise")]
-    extra_information_table_factories:
-        std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
+    extra_information_table_factories: HashMap<String, InformationSchemaTableFactoryRef>,
 }
 
 impl KvBackendCatalogManagerBuilder {
@@ -55,8 +54,7 @@ impl KvBackendCatalogManagerBuilder {
             cache_registry,
             procedure_manager: None,
             process_manager: None,
-            #[cfg(feature = "enterprise")]
-            extra_information_table_factories: std::collections::HashMap::new(),
+            extra_information_table_factories: HashMap::new(),
         }
     }
 
@@ -71,10 +69,9 @@ impl KvBackendCatalogManagerBuilder {
     }
 
     /// Sets the extra information tables.
-    #[cfg(feature = "enterprise")]
     pub fn with_extra_information_table_factories(
         mut self,
-        factories: std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
+        factories: HashMap<String, InformationSchemaTableFactoryRef>,
     ) -> Self {
         self.extra_information_table_factories = factories;
         self
@@ -87,7 +84,6 @@ impl KvBackendCatalogManagerBuilder {
             cache_registry,
             procedure_manager,
             process_manager,
-            #[cfg(feature = "enterprise")]
             extra_information_table_factories,
         } = self;
         Arc::new_cyclic(|me| KvBackendCatalogManager {
@@ -111,7 +107,6 @@ impl KvBackendCatalogManagerBuilder {
                         process_manager.clone(),
                         backend.clone(),
                     );
-                    #[cfg(feature = "enterprise")]
                     let provider = provider
                         .with_extra_table_factories(extra_information_table_factories.clone());
                     Arc::new(provider)
@@ -123,7 +118,6 @@ impl KvBackendCatalogManagerBuilder {
                 numbers_table_provider: NumbersTableProvider,
                 backend,
                 process_manager,
-                #[cfg(feature = "enterprise")]
                 extra_information_table_factories,
             },
             cache_registry,

src/catalog/src/kvbackend/manager.rs

@@ -53,9 +53,9 @@ use crate::error::{
     CacheNotFoundSnafu, GetTableCacheSnafu, InvalidTableInfoInCatalogSnafu, ListCatalogsSnafu,
     ListSchemasSnafu, ListTablesSnafu, Result, TableMetadataManagerSnafu,
 };
-#[cfg(feature = "enterprise")]
-use crate::information_schema::InformationSchemaTableFactoryRef;
-use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
+use crate::information_schema::{
+    InformationExtensionRef, InformationSchemaProvider, InformationSchemaTableFactoryRef,
+};
 use crate::kvbackend::TableCacheRef;
 use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::SystemSchemaProvider;
@@ -557,7 +557,6 @@ pub(super) struct SystemCatalog {
     pub(super) numbers_table_provider: NumbersTableProvider,
     pub(super) backend: KvBackendRef,
     pub(super) process_manager: Option<ProcessManagerRef>,
-    #[cfg(feature = "enterprise")]
     pub(super) extra_information_table_factories:
         std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
 }
@@ -628,7 +627,6 @@ impl SystemCatalog {
                         self.process_manager.clone(),
                         self.backend.clone(),
                     );
-                    #[cfg(feature = "enterprise")]
                     let provider = provider
                         .with_extra_table_factories(self.extra_information_table_factories.clone());
                     Arc::new(provider)

src/catalog/src/system_schema/information_schema.rs

@@ -117,7 +117,6 @@ macro_rules! setup_memory_table {
     };
 }
 
-#[cfg(feature = "enterprise")]
 pub struct MakeInformationTableRequest {
     pub catalog_name: String,
     pub catalog_manager: Weak<dyn CatalogManager>,
@@ -128,12 +127,10 @@ pub struct MakeInformationTableRequest {
 ///
 /// This trait allows for extensibility of the information schema by providing
 /// a way to dynamically create custom information schema tables.
-#[cfg(feature = "enterprise")]
 pub trait InformationSchemaTableFactory {
     fn make_information_table(&self, req: MakeInformationTableRequest) -> SystemTableRef;
 }
 
-#[cfg(feature = "enterprise")]
 pub type InformationSchemaTableFactoryRef = Arc<dyn InformationSchemaTableFactory + Send + Sync>;
 
 /// The `information_schema` tables info provider.
@@ -143,9 +140,7 @@ pub struct InformationSchemaProvider {
     process_manager: Option<ProcessManagerRef>,
     flow_metadata_manager: Arc<FlowMetadataManager>,
     tables: HashMap<String, TableRef>,
-    #[allow(dead_code)]
     kv_backend: KvBackendRef,
-    #[cfg(feature = "enterprise")]
     extra_table_factories: HashMap<String, InformationSchemaTableFactoryRef>,
 }
 
@@ -166,7 +161,6 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
     }
 
     fn system_table(&self, name: &str) -> Option<SystemTableRef> {
-        #[cfg(feature = "enterprise")]
         if let Some(factory) = self.extra_table_factories.get(name) {
             let req = MakeInformationTableRequest {
                 catalog_name: self.catalog_name.clone(),
@@ -281,7 +275,6 @@ impl InformationSchemaProvider {
             process_manager,
             tables: HashMap::new(),
             kv_backend,
-            #[cfg(feature = "enterprise")]
             extra_table_factories: HashMap::new(),
         };
 
@@ -290,7 +283,6 @@ impl InformationSchemaProvider {
         provider
     }
 
-    #[cfg(feature = "enterprise")]
     pub(crate) fn with_extra_table_factories(
         mut self,
         factories: HashMap<String, InformationSchemaTableFactoryRef>,
@@ -358,7 +350,6 @@ impl InformationSchemaProvider {
         if let Some(process_list) = self.build_table(PROCESS_LIST) {
             tables.insert(PROCESS_LIST.to_string(), process_list);
         }
-        #[cfg(feature = "enterprise")]
         for name in self.extra_table_factories.keys() {
             tables.insert(name.clone(), self.build_table(name).expect(name));
         }

src/cmd/Cargo.toml

@@ -16,7 +16,7 @@ default = [
     "meta-srv/pg_kvbackend",
     "meta-srv/mysql_kvbackend",
 ]
-enterprise = ["common-meta/enterprise", "frontend/enterprise", "meta-srv/enterprise", "catalog/enterprise"]
+enterprise = ["common-meta/enterprise", "frontend/enterprise", "meta-srv/enterprise"]
 tokio-console = ["common-telemetry/tokio-console"]
 
 [lints]
@@ -71,6 +71,7 @@ nu-ansi-term = "0.46"
 object-store.workspace = true
 parquet = { workspace = true, features = ["object_store"] }
 plugins.workspace = true
+plugins-framework.workspace = true
 prometheus.workspace = true
 prost.workspace = true
 query.workspace = true

src/cmd/src/bin/greptime.rs

@@ -17,10 +17,14 @@
 use clap::{Parser, Subcommand};
 use cmd::datanode::builder::InstanceBuilder;
 use cmd::error::{InitTlsProviderSnafu, Result};
-use cmd::options::GlobalOptions;
+use cmd::options::{EmptyOptions, GlobalOptions};
 use cmd::{App, cli, datanode, flownode, frontend, metasrv, standalone};
 use common_base::Plugins;
 use common_version::{verbose_version, version};
+use meta_srv::bootstrap::plugins::DefaultMetaPluginFactory;
+use plugins_framework::flownode::DefaultFlownodePluginFactory;
+use plugins_framework::frontend::DefaultFrontendPluginFactory;
+use plugins_framework::standalone::DefaultStandalonePluginFactory;
 use servers::install_ring_crypto_provider;
 
 #[derive(Parser)]
@@ -108,30 +112,31 @@ async fn start(cli: Command) -> Result<()> {
                 let opts = start.load_options(&cli.global_options)?;
                 let plugins = Plugins::new();
                 let builder = InstanceBuilder::try_new_with_init(opts, plugins).await?;
-                cmd.build_with(builder).await?.run().await
+                cmd.build_with::<EmptyOptions>(builder).await?.run().await
             }
             datanode::SubCommand::Objbench(ref bench) => bench.run().await,
         },
         SubCommand::Flownode(cmd) => {
-            cmd.build(cmd.load_options(&cli.global_options)?)
+            let opts = cmd.load_options::<EmptyOptions>(&cli.global_options)?;
+            cmd.build(opts, DefaultFlownodePluginFactory)
                 .await?
                 .run()
                 .await
         }
         SubCommand::Frontend(cmd) => {
-            cmd.build(cmd.load_options(&cli.global_options)?)
+            let opts = cmd.load_options::<EmptyOptions>(&cli.global_options)?;
+            cmd.build(opts, DefaultFrontendPluginFactory)
                 .await?
                 .run()
                 .await
         }
         SubCommand::Metasrv(cmd) => {
-            cmd.build(cmd.load_options(&cli.global_options)?)
-                .await?
-                .run()
-                .await
+            let opts = cmd.load_options::<EmptyOptions>(&cli.global_options)?;
+            cmd.build(opts, DefaultMetaPluginFactory).await?.run().await
         }
         SubCommand::Standalone(cmd) => {
-            cmd.build(cmd.load_options(&cli.global_options)?)
+            let opts = cmd.load_options::<EmptyOptions>(&cli.global_options)?;
+            cmd.build(opts, DefaultStandalonePluginFactory)
                 .await?
                 .run()
                 .await

src/cmd/src/cli.rs

@@ -106,9 +106,10 @@ mod tests {
     use client::{Client, Database};
     use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
     use common_telemetry::logging::LoggingOptions;
+    use plugins_framework::standalone::DefaultStandalonePluginFactory;
 
     use crate::error::Result as CmdResult;
-    use crate::options::GlobalOptions;
+    use crate::options::{EmptyOptions, GlobalOptions};
     use crate::{App, cli, standalone};
 
     #[tokio::test(flavor = "multi_thread")]
@@ -122,8 +123,12 @@ mod tests {
             &*output_dir.path().to_string_lossy(),
         ]);
 
-        let standalone_opts = standalone.load_options(&GlobalOptions::default()).unwrap();
-        let mut instance = standalone.build(standalone_opts).await?;
+        let standalone_opts = standalone
+            .load_options::<EmptyOptions>(&GlobalOptions::default())
+            .unwrap();
+        let mut instance = standalone
+            .build(standalone_opts, DefaultStandalonePluginFactory)
+            .await?;
         instance.start().await?;
 
         let client = Client::with_urls(["127.0.0.1:4001"]);