From 6b745b20ecb3d9245cd31cee4b38efeb32477e09 Mon Sep 17 00:00:00 2001
From: Hengfei Yang <hengfei.yang@gmail.com>
Date: Mon, 7 Apr 2025 22:55:32 +0800
Subject: [PATCH] fix: enrichment loading on ingester (#6483)

- [x] fix: enrichment loading on ingester #6480
- [x] fix: vrl with grpc search
---
 src/service/enrichment_table/mod.rs          | 5 +++--
 src/service/search/cache/mod.rs              | 5 ++++-
 src/service/search/cluster/flight.rs         | 4 ++--
 src/service/search/super_cluster/follower.rs | 4 ++--
 4 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/src/service/enrichment_table/mod.rs b/src/service/enrichment_table/mod.rs
index 15190f649b..e12201cdc0 100644
--- a/src/service/enrichment_table/mod.rs
+++ b/src/service/enrichment_table/mod.rs
@@ -220,12 +220,13 @@ pub async fn save_enrichment_data(
         .as_ref()
         .clone()
         .with_metadata(HashMap::new());
+    let schema = Arc::new(schema);
     let schema_key = schema.hash_key();
     buf.insert(
         hour_key,
         SchemaRecords {
             schema_key,
-            schema: Arc::new(schema),
+            schema: schema.clone(),
             records,
             records_size,
         },
@@ -253,7 +254,7 @@ pub async fn save_enrichment_data(
         };
 
     // notify update
-    if stream_schema.has_fields {
+    if !schema.fields().is_empty() {
         if let Err(e) = super::db::enrichment_table::notify_update(org_id, stream_name).await {
             log::error!("Error notifying enrichment table {org_id}/{stream_name} update: {e}");
         };
diff --git a/src/service/search/cache/mod.rs b/src/service/search/cache/mod.rs
index e085578527..c8c2b94869 100644
--- a/src/service/search/cache/mod.rs
+++ b/src/service/search/cache/mod.rs
@@ -86,7 +86,10 @@ pub async fn search(
         .query
         .query_fn
         .as_ref()
-        .and_then(|v| base64::decode_url(v).ok());
+        .map(|v| match base64::decode_url(v) {
+            Ok(v) => v,
+            Err(_) => v.to_string(),
+        });
     let action = req
         .query
         .action_id
diff --git a/src/service/search/cluster/flight.rs b/src/service/search/cluster/flight.rs
index 60cd98f3c0..1739617708 100644
--- a/src/service/search/cluster/flight.rs
+++ b/src/service/search/cluster/flight.rs
@@ -140,9 +140,9 @@ pub async fn search(
         .unwrap_or(None);
     let mut nodes = get_online_querier_nodes(trace_id, node_group).await?;
 
-    // local mode, only use local node as querier for the query
+    // local mode, only use local node as querier node
     if req.local_mode.unwrap_or_default() && LOCAL_NODE.is_querier() {
-        nodes = vec![LOCAL_NODE.clone()];
+        nodes.retain(|n| n.is_ingester() || n.name.eq(&LOCAL_NODE.name));
     }
 
     let querier_num = nodes.iter().filter(|node| node.is_querier()).count();
diff --git a/src/service/search/super_cluster/follower.rs b/src/service/search/super_cluster/follower.rs
index 8ea231d91f..f28a8d9ee4 100644
--- a/src/service/search/super_cluster/follower.rs
+++ b/src/service/search/super_cluster/follower.rs
@@ -162,9 +162,9 @@ pub async fn search(
         .unwrap_or(None);
     let mut nodes = get_online_querier_nodes(&trace_id, node_group).await?;
 
-    // local mode, only use local node as querier for the query
+    // local mode, only use local node as querier node
     if req.local_mode.unwrap_or_default() && LOCAL_NODE.is_querier() {
-        nodes = vec![LOCAL_NODE.clone()];
+        nodes.retain(|n| n.is_ingester() || n.name.eq(&LOCAL_NODE.name));
     }
 
     let querier_num = nodes.iter().filter(|node| node.is_querier()).count();