feat: add flownode_peer_cache to Inserter

GreptimeTeam · Jul 2, 2024 · cc3b9f1 · cc3b9f1
1 parent 9cb55cd
commit cc3b9f1
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 1 deletion.
diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs
@@ -127,11 +127,13 @@ impl FrontendBuilder {
                 .context(error::CacheRequiredSnafu {
                     name: TABLE_FLOWNODE_SET_CACHE_NAME,
                 })?;
+        let flownode_peer_cache = self.layered_cache_registry.get();
         let inserter = Arc::new(Inserter::new(
             self.catalog_manager.clone(),
             partition_manager.clone(),
             node_manager.clone(),
             table_flownode_cache,
+            flownode_peer_cache,
         ));
         let deleter = Arc::new(Deleter::new(
             self.catalog_manager.clone(),

diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs
@@ -25,7 +25,7 @@ use catalog::CatalogManagerRef;
 use client::{OutputData, OutputMeta};
 use common_catalog::consts::default_engine;
 use common_grpc_expr::util::{extract_new_columns, ColumnExpr};
-use common_meta::cache::TableFlownodeSetCacheRef;
+use common_meta::cache::{FlownodePeerCacheRef, TableFlownodeSetCacheRef};
 use common_meta::node_manager::{AffectedRows, NodeManagerRef};
 use common_meta::peer::Peer;
 use common_query::prelude::{GREPTIME_TIMESTAMP, GREPTIME_VALUE};
@@ -62,6 +62,9 @@ pub struct Inserter {
     partition_manager: PartitionRuleManagerRef,
     node_manager: NodeManagerRef,
     table_flownode_set_cache: TableFlownodeSetCacheRef,
+    /// Only available in cluster mode, it not necessary to query peer info in standalone mode.
+    #[allow(dead_code)]
+    flownode_peer_cache: Option<FlownodePeerCacheRef>,
 }
 
 pub type InserterRef = Arc<Inserter>;
@@ -88,12 +91,14 @@ impl Inserter {
         partition_manager: PartitionRuleManagerRef,
         node_manager: NodeManagerRef,
         table_flownode_set_cache: TableFlownodeSetCacheRef,
+        flownode_peer_cache: Option<FlownodePeerCacheRef>,
     ) -> Self {
         Self {
             catalog_manager,
             partition_manager,
             node_manager,
             table_flownode_set_cache,
+            flownode_peer_cache,
         }
     }