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

@@ -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}");
         };

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

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();

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();