Enhancement: Introduce scalable query nodes

server/src/analytics.rs

@@ -224,7 +224,7 @@ async fn fetch_ingestors_metrics(
         // send analytics for ingest servers
 
         // ingestor infos should be valid here, if not some thing is wrong
-        let ingestor_infos = cluster::get_ingestor_info().await.unwrap();
+        let ingestor_infos = cluster::get_ingestor_info_storage().await.unwrap();
 
         for im in ingestor_infos {
             if !check_liveness(&im.domain_name).await {

server/src/catalog.rs

@@ -396,13 +396,12 @@ pub async fn get_first_event(
             }
         }
         Mode::Query => {
-            let ingestor_metadata =
-                handlers::http::cluster::get_ingestor_info()
-                    .await
-                    .map_err(|err| {
-                        log::error!("Fatal: failed to get ingestor info: {:?}", err);
-                        ObjectStorageError::from(err)
-                    })?;
+            let ingestor_metadata = handlers::http::cluster::get_ingestor_info_storage()
+                .await
+                .map_err(|err| {
+                    log::error!("Fatal: failed to get ingestor info: {:?}", err);
+                    ObjectStorageError::from(err)
+                })?;
             let mut ingestors_first_event_at: Vec<String> = Vec::new();
             for ingestor in ingestor_metadata {
                 let url = format!(

server/src/event.rs

@@ -113,6 +113,7 @@ impl Event {
         .map_err(PostError::Event)
     }
 
+    #[allow(unused)]
     pub fn clear(&self, stream_name: &str) {
         STREAM_WRITERS.clear(stream_name);
     }

server/src/event/writer.rs

@@ -202,6 +202,7 @@ impl WriterTable {
         Ok(())
     }
 
+    #[allow(unused)]
     pub fn clear(&self, stream_name: &str) {
         let map = self.write().unwrap();
         if let Some(writer) = map.get(stream_name) {

server/src/handlers/airplane.rs

@@ -32,7 +32,7 @@
 use tonic::transport::{Identity, Server, ServerTlsConfig};
 use tonic_web::GrpcWebLayer;
 
-use crate::handlers::http::cluster::get_ingestor_info;
+use crate::handlers::http::cluster::get_ingestor_info_storage;
 
 use crate::handlers::{CACHE_RESULTS_HEADER_KEY, CACHE_VIEW_HEADER_KEY, USER_ID_HEADER_KEY};
 use crate::metrics::QUERY_EXECUTE_TIME;
@@ -46,8 +46,7 @@
 use crate::query::{TableScanVisitor, QUERY_SESSION};
 use crate::querycache::QueryCacheManager;
 use crate::utils::arrow::flight::{
-    append_temporary_events, get_query_from_ticket, into_flight_data, run_do_get_rpc,
-    send_to_ingester,
+    get_query_from_ticket, into_flight_data, run_do_get_rpc, send_to_ingester,
 };
 use arrow_flight::{
     flight_service_server::FlightService, Action, ActionType, Criteria, Empty, FlightData,
@@ -204,7 +203,9 @@
             .await
             .map_err(|_| Status::internal("Failed to parse query"))?;
 
+        // this is the flow for getting staging data from Ingestors
+        // fix this (it sends double records)
         let event =
             if send_to_ingester(query.start.timestamp_millis(), query.end.timestamp_millis()) {
                 let sql = format!("select * from {}", &stream_name);
                 let start_time = ticket.start_time.clone();
@@ -216,7 +217,7 @@
                 })
                 .to_string();
 
-                let ingester_metadatas = get_ingestor_info()
+                let ingester_metadatas = get_ingestor_info_storage()
                     .await
                     .map_err(|err| Status::failed_precondition(err.to_string()))?;
                 let mut minute_result: Vec<RecordBatch> = vec![];
@@ -226,9 +227,10 @@
                         minute_result.append(&mut batches);
                     }
                 }
-                let mr = minute_result.iter().collect::<Vec<_>>();
-                let event = append_temporary_events(&stream_name, mr).await?;
-                Some(event)
+
+                // log::warn!("minute_result-\n{mr:?}\n");
+                // let event = append_temporary_events(&stream_name, mr).await?;
+                Some(minute_result)
             } else {
                 None
             };
@@ -252,11 +254,15 @@
             Status::permission_denied("User Does not have permission to access this")
         })?;
         let time = Instant::now();
-        let (records, _) = query
+        let (mut records, _) = query
             .execute(stream_name.clone())
             .await
             .map_err(|err| Status::internal(err.to_string()))?;
 
+        // if let Some(event) = event {
+        //     records.extend(event);
+        // }
+
         if let Err(err) = put_results_in_cache(
             cache_results,
             user_id,
@@ -285,10 +291,6 @@
          */
         let out = into_flight_data(records);
 
-        if let Some(event) = event {
-            event.clear(&stream_name);
-        }
-
         let time = time.elapsed().as_secs_f64();
         QUERY_EXECUTE_TIME
             .with_label_values(&[&format!("flight-query-{}", stream_name)])

server/src/handlers/http.rs

@@ -23,7 +23,7 @@ use serde_json::Value;
 
 use crate::option::CONFIG;
 
-use self::{cluster::get_ingestor_info, query::Query};
+use self::{cluster::get_ingestor_info_storage, query::Query};
 
 pub(crate) mod about;
 mod cache;
@@ -99,7 +99,7 @@ pub async fn fetch_schema(stream_name: &str) -> anyhow::Result<arrow_schema::Sch
 pub async fn send_query_request_to_ingestor(query: &Query) -> anyhow::Result<Vec<Value>> {
     // send the query request to the ingestor
     let mut res = vec![];
-    let ima = get_ingestor_info().await?;
+    let ima = get_ingestor_info_storage().await?;
 
     for im in ima.iter() {
         let uri = format!(