chore: simplify schema creation from storage

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/cli.rs

@@ -451,7 +451,7 @@ pub struct Options {
         help = "Object store sync threshold in seconds"
     )]
     pub object_store_sync_threshold: u64,
-    // the oidc scope 
+    // the oidc scope
     #[arg(
         long = "oidc-scope",
         name = "oidc-scope",

src/handlers/http/oidc.rs

@@ -77,7 +77,13 @@ pub async fn login(
     let session_key = extract_session_key_from_req(&req).ok();
     let (session_key, oidc_client) = match (session_key, oidc_client) {
         (None, None) => return Ok(redirect_no_oauth_setup(query.redirect.clone())),
-        (None, Some(client)) => return Ok(redirect_to_oidc(query, client, PARSEABLE.options.scope.to_string().as_str())),
+        (None, Some(client)) => {
+            return Ok(redirect_to_oidc(
+                query,
+                client,
+                PARSEABLE.options.scope.to_string().as_str(),
+            ))
+        }
         (Some(session_key), client) => (session_key, client),
     };
     // try authorize
@@ -113,7 +119,11 @@ pub async fn login(
             } else {
                 Users.remove_session(&key);
                 if let Some(oidc_client) = oidc_client {
-                    redirect_to_oidc(query, oidc_client, PARSEABLE.options.scope.to_string().as_str())
+                    redirect_to_oidc(
+                        query,
+                        oidc_client,
+                        PARSEABLE.options.scope.to_string().as_str(),
+                    )
                 } else {
                     redirect_to_client(query.redirect.as_str(), None)
                 }

src/migration/mod.rs

@@ -184,7 +184,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;
@@ -212,29 +212,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

@@ -273,7 +273,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

@@ -65,6 +65,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::ingest::PostError;
 use crate::handlers::http::modal::ingest_server::INGESTOR_EXPECT;
 use crate::handlers::http::modal::ingest_server::INGESTOR_META;
@@ -671,44 +672,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(