chore: simplify schema creation from storage (#1368)

remove functions that creates schema from ingestors and queriers separately
reused function `fetch_schema` that fetches all schema files and merges 
the schemas into one

this ensures the schema is always the latest

Author: nikhilsinhaparseable

src/handlers/http/query.rs

@@ -439,12 +439,22 @@ pub async fn create_streams_for_querier() -> Result<(), QueryError> {
     let querier_streams = PARSEABLE.streams.list();
 
     let querier_streams_set: HashSet<_> = querier_streams.into_iter().collect();
-
-    let storage_streams = store.list_streams().await?;
-
-    let missing_streams: Vec<_> = storage_streams
+    // fetch querier streams which have field list blank
+    // now missing streams should be list of streams which are in storage but not in querier
+    // and also have no fields in the schema
+    // this is to ensure that we do not create streams for querier which already exist in querier
+
+    let missing_streams: Vec<_> = store
+        .list_streams()
+        .await?
         .into_iter()
-        .filter(|stream_name| !querier_streams_set.contains(stream_name))
+        .filter(|stream_name| {
+            !querier_streams_set.contains(stream_name)
+                || PARSEABLE
+                    .get_stream(stream_name)
+                    .map(|s| s.get_schema().fields().is_empty())
+                    .unwrap_or(false)
+        })
         .collect();
 
     if missing_streams.is_empty() {

src/migration/mod.rs

@@ -206,7 +206,7 @@ async fn migration_stream(
 ) -> anyhow::Result<Option<LogStreamMetadata>> {
     let mut arrow_schema: Schema = Schema::empty();
 
-    let schema = fetch_or_create_schema(stream, storage).await?;
+    let schema = storage.create_schema_from_storage(stream).await?;
     let stream_metadata = fetch_or_create_stream_metadata(stream, storage).await?;
 
     let mut stream_meta_found = true;
@@ -234,29 +234,6 @@ async fn migration_stream(
     Ok(Some(metadata))
 }
 
-async fn fetch_or_create_schema(
-    stream: &str,
-    storage: &dyn ObjectStorage,
-) -> anyhow::Result<Bytes> {
-    let schema_path = schema_path(stream);
-    if let Ok(schema) = storage.get_object(&schema_path).await {
-        Ok(schema)
-    } else {
-        let querier_schema = storage
-            .create_schema_from_querier(stream)
-            .await
-            .unwrap_or_default();
-        if !querier_schema.is_empty() {
-            Ok(querier_schema)
-        } else {
-            Ok(storage
-                .create_schema_from_ingestor(stream)
-                .await
-                .unwrap_or_default())
-        }
-    }
-}
-
 async fn fetch_or_create_stream_metadata(
     stream: &str,
     storage: &dyn ObjectStorage,

src/parseable/mod.rs

@@ -281,7 +281,7 @@ impl Parseable {
 
         let (stream_metadata_bytes, schema_bytes) = try_join!(
             storage.create_stream_from_ingestor(stream_name),
-            storage.create_schema_from_ingestor(stream_name)
+            storage.create_schema_from_storage(stream_name)
         )?;
 
         let stream_metadata = if stream_metadata_bytes.is_empty() {

src/storage/object_storage.rs

@@ -49,6 +49,7 @@ use crate::catalog::{self, manifest::Manifest, snapshot::Snapshot};
 use crate::correlation::{CorrelationConfig, CorrelationError};
 use crate::event::format::LogSource;
 use crate::event::format::LogSourceEntry;
+use crate::handlers::http::fetch_schema;
 use crate::handlers::http::modal::ingest_server::INGESTOR_EXPECT;
 use crate::handlers::http::modal::ingest_server::INGESTOR_META;
 use crate::handlers::http::users::CORRELATION_DIR;
@@ -652,44 +653,18 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
         Ok(Bytes::new())
     }
 
-    ///create schema from querier schema from storage
-    async fn create_schema_from_querier(
+    ///create schema from storage
+    async fn create_schema_from_storage(
         &self,
         stream_name: &str,
     ) -> Result<Bytes, ObjectStorageError> {
-        let path =
-            RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, SCHEMA_FILE_NAME]);
-        if let Ok(querier_schema_bytes) = self.get_object(&path).await {
-            self.put_object(&schema_path(stream_name), querier_schema_bytes.clone())
-                .await?;
-            return Ok(querier_schema_bytes);
-        }
-        Ok(Bytes::new())
-    }
-
-    ///create schema from ingestor schema from storage
-    async fn create_schema_from_ingestor(
-        &self,
-        stream_name: &str,
-    ) -> Result<Bytes, ObjectStorageError> {
-        let path = RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY]);
-        if let Some(schema_obs) = self
-            .get_objects(
-                Some(&path),
-                Box::new(|file_name| {
-                    file_name.starts_with(".ingestor") && file_name.ends_with("schema")
-                }),
-            )
-            .await
-            .into_iter()
-            .next()
-        {
-            let schema_ob = &schema_obs[0];
-            self.put_object(&schema_path(stream_name), schema_ob.clone())
-                .await?;
-            return Ok(schema_ob.clone());
-        }
-        Ok(Bytes::new())
+        let schema = fetch_schema(stream_name).await?;
+        // convert to bytes
+        let schema = serde_json::to_vec(&schema)?;
+        let schema_bytes = Bytes::from(schema);
+        self.put_object(&schema_path(stream_name), schema_bytes.clone())
+            .await?;
+        Ok(schema_bytes)
     }
 
     async fn get_stream_meta_from_storage(