fix: get schema for distributed (#1008)

current: querier checks if schema is available in STREAM_INFO
if not, create schema from storage, load in STREAM_INFO and return

STREAM_INFO cannot guarantee the latest state of the schema 
for distributed

fix: merge schema from all ingestors and queriers and then 
load in STREAM_INFO and return

Author: nikhilsinhaparseable

src/handlers/http/logstream.rs

@@ -23,6 +23,7 @@ use super::ingest::create_stream_if_not_exists;
 use super::modal::utils::logstream_utils::{
     create_stream_and_schema_from_storage, create_update_stream,
 };
+use super::query::update_schema_when_distributed;
 use crate::alerts::Alerts;
 use crate::event::format::update_data_type_to_datetime;
 use crate::handlers::STREAM_TYPE_KEY;
@@ -117,23 +118,26 @@ pub async fn detect_schema(body: Bytes) -> Result<impl Responder, StreamError> {
 
 pub async fn schema(req: HttpRequest) -> Result<impl Responder, StreamError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
-    let schema = match STREAM_INFO.schema(&stream_name) {
-        Ok(schema) => schema,
 
-        //if schema not found in memory map
-        //create stream and schema from storage and memory
-        //return from memory map
+    match STREAM_INFO.schema(&stream_name) {
+        Ok(_) => {}
         Err(_) if CONFIG.parseable.mode == Mode::Query => {
-            if create_stream_and_schema_from_storage(&stream_name).await? {
-                STREAM_INFO.schema(&stream_name)?
-            } else {
+            if !create_stream_and_schema_from_storage(&stream_name).await? {
                 return Err(StreamError::StreamNotFound(stream_name.clone()));
             }
         }
-        Err(_) => return Err(StreamError::StreamNotFound(stream_name)),
+        Err(err) => return Err(StreamError::from(err)),
     };
-
-    Ok((web::Json(schema), StatusCode::OK))
+    match update_schema_when_distributed(vec![stream_name.clone()]).await {
+        Ok(_) => {
+            let schema = STREAM_INFO.schema(&stream_name)?;
+            Ok((web::Json(schema), StatusCode::OK))
+        }
+        Err(err) => Err(StreamError::Custom {
+            msg: err.to_string(),
+            status: StatusCode::EXPECTATION_FAILED,
+        }),
+    }
 }
 
 pub async fn get_alert(req: HttpRequest) -> Result<impl Responder, StreamError> {

src/handlers/http/mod.rs

@@ -21,7 +21,7 @@ use arrow_schema::Schema;
 use itertools::Itertools;
 use serde_json::Value;
 
-use crate::option::CONFIG;
+use crate::{option::CONFIG, storage::STREAM_ROOT_DIRECTORY};
 
 use self::{cluster::get_ingestor_info, query::Query};
 
@@ -77,7 +77,7 @@ pub fn base_path_without_preceding_slash() -> String {
 /// An `anyhow::Result` containing the `arrow_schema::Schema` for the specified stream.
 pub async fn fetch_schema(stream_name: &str) -> anyhow::Result<arrow_schema::Schema> {
     let path_prefix =
-        relative_path::RelativePathBuf::from(format!("{}/{}", stream_name, ".stream"));
+        relative_path::RelativePathBuf::from(format!("{}/{}", stream_name, STREAM_ROOT_DIRECTORY));
     let store = CONFIG.storage().get_object_store();
     let res: Vec<Schema> = store
         .get_objects(