bug fix for schema sync at server restart

Author: nikhilsinhaparseable

server/src/handlers/http/logstream.rs

@@ -116,7 +116,19 @@ 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 = STREAM_INFO.schema(&stream_name)?;
+    let schema = if let Ok(schema) = STREAM_INFO.schema(&stream_name) {
+        schema
+    } else if CONFIG.parseable.mode == Mode::Query {
+        let stream_found = create_stream_and_schema_from_storage(&stream_name).await?;
+        if !stream_found {
+            return Err(StreamError::StreamNotFound(stream_name.clone()));
+        } else {
+            STREAM_INFO.schema(&stream_name)?
+        }
+    } else {
+        return Err(StreamError::StreamNotFound(stream_name));
+    };
+
     Ok((web::Json(schema), StatusCode::OK))
 }
 

server/src/metadata.rs

@@ -32,7 +32,6 @@ use crate::metrics::{
     EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE, LIFETIME_EVENTS_INGESTED,
     LIFETIME_EVENTS_INGESTED_SIZE,
 };
-use crate::option::{Mode, CONFIG};
 use crate::storage::retention::Retention;
 use crate::storage::{LogStream, ObjectStorage, ObjectStoreFormat, StorageDir, StreamType};
 use crate::utils::arrow::MergedRecordReader;
@@ -463,9 +462,7 @@ pub async fn load_stream_metadata_on_server_start(
     }
     let schema =
         update_data_type_time_partition(storage, stream_name, schema, meta.clone()).await?;
-    if CONFIG.parseable.mode == Mode::Ingest {
-        storage.put_schema(stream_name, &schema).await?;
-    }
+    storage.put_schema(stream_name, &schema).await?;
     //load stats from storage
     let stats = meta.stats;
     fetch_stats_from_storage(stream_name, stats).await;

server/src/migration.rs

@@ -28,9 +28,9 @@ use crate::{
     metadata::load_stream_metadata_on_server_start,
     option::{validation::human_size_to_bytes, Config, Mode, CONFIG},
     storage::{
-        object_storage::{parseable_json_path, stream_json_path},
+        object_storage::{parseable_json_path, schema_path, stream_json_path},
         ObjectStorage, ObjectStorageError, PARSEABLE_METADATA_FILE_NAME, PARSEABLE_ROOT_DIRECTORY,
-        SCHEMA_FILE_NAME, STREAM_ROOT_DIRECTORY,
+        STREAM_ROOT_DIRECTORY,
     },
 };
 use arrow_schema::Schema;
@@ -166,15 +166,22 @@ async fn migration_hot_tier(stream: &str) -> anyhow::Result<()> {
 
 async fn migration_stream(stream: &str, storage: &dyn ObjectStorage) -> anyhow::Result<()> {
     let mut arrow_schema: Schema = Schema::empty();
-    let query_schema_path =
-        RelativePathBuf::from_iter([stream, STREAM_ROOT_DIRECTORY, SCHEMA_FILE_NAME]);
-    let schema = if let Ok(schema) = storage.get_object(&query_schema_path).await {
+    let schema_path = schema_path(stream);
+    let schema = if let Ok(schema) = storage.get_object(&schema_path).await {
         schema
     } else {
-        storage
-            .create_schema_from_ingestor(stream)
+        let querier_schema = storage
+            .create_schema_from_querier(stream)
             .await
-            .unwrap_or_default()
+            .unwrap_or_default();
+        if !querier_schema.is_empty() {
+            querier_schema
+        } else {
+            storage
+                .create_schema_from_ingestor(stream)
+                .await
+                .unwrap_or_default()
+        }
     };
 
     let path = stream_json_path(stream);
@@ -222,7 +229,7 @@ async fn migration_stream(stream: &str, storage: &dyn ObjectStorage) -> anyhow::
                 let schema = serde_json::from_slice(&schema).ok();
                 arrow_schema = schema_migration::v1_v4(schema)?;
                 storage
-                    .put_object(&query_schema_path, to_bytes(&arrow_schema))
+                    .put_object(&schema_path, to_bytes(&arrow_schema))
                     .await?;
             }
             Some("v2") => {
@@ -236,7 +243,7 @@ async fn migration_stream(stream: &str, storage: &dyn ObjectStorage) -> anyhow::
                 let schema = serde_json::from_slice(&schema)?;
                 arrow_schema = schema_migration::v2_v4(schema)?;
                 storage
-                    .put_object(&query_schema_path, to_bytes(&arrow_schema))
+                    .put_object(&schema_path, to_bytes(&arrow_schema))
                     .await?;
             }
             Some("v3") => {

server/src/storage/object_storage.rs

@@ -493,6 +493,20 @@ pub trait ObjectStorage: Sync + 'static {
         Ok(Bytes::new())
     }
 
+    async fn create_schema_from_querier(
+        &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())
+    }
+
     async fn create_schema_from_ingestor(
         &self,
         stream_name: &str,