refactor: Parseable::fetch_schema

Author: Devdutt Shenoi

src/handlers/http/mod.rs

@@ -18,12 +18,10 @@
 
 use actix_cors::Cors;
 use actix_web::Responder;
-use arrow_schema::Schema;
 use http::StatusCode;
-use itertools::Itertools;
 use serde_json::Value;
 
-use crate::{parseable::PARSEABLE, storage::STREAM_ROOT_DIRECTORY, HTTP_CLIENT};
+use crate::{parseable::PARSEABLE, HTTP_CLIENT};
 
 use self::{cluster::get_ingestor_info, query::Query};
 
@@ -75,34 +73,6 @@ pub fn base_path_without_preceding_slash() -> String {
     format!("{API_BASE_PATH}/{API_VERSION}")
 }
 
-/// Fetches the schema for the specified stream.
-///
-/// # Arguments
-///
-/// * `stream_name` - The name of the stream to fetch the schema for.
-///
-/// # Returns
-///
-/// 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_ROOT_DIRECTORY));
-    let store = PARSEABLE.storage.get_object_store();
-    let res: Vec<Schema> = store
-        .get_objects(
-            Some(&path_prefix),
-            Box::new(|file_name: String| file_name.contains(".schema")),
-        )
-        .await?
-        .iter()
-        // we should be able to unwrap as we know the data is valid schema
-        .map(|byte_obj| serde_json::from_slice(byte_obj).expect("data is valid json"))
-        .collect_vec();
-
-    let new_schema = Schema::try_merge(res)?;
-    Ok(new_schema)
-}
-
 /// unused for now, might need it later
 #[allow(unused)]
 pub async fn send_query_request_to_ingestor(query: &Query) -> anyhow::Result<Vec<Value>> {

src/handlers/http/query.rs

@@ -33,10 +33,8 @@ use std::sync::Arc;
 use std::time::Instant;
 use tracing::error;
 
-use crate::event::error::EventError;
-use crate::handlers::http::fetch_schema;
-
 use crate::event::commit_schema;
+use crate::event::error::EventError;
 use crate::metrics::QUERY_EXECUTE_TIME;
 use crate::option::Mode;
 use crate::parseable::PARSEABLE;
@@ -169,7 +167,7 @@ pub async fn get_counts(
 pub async fn update_schema_when_distributed(tables: &Vec<String>) -> Result<(), EventError> {
     if PARSEABLE.options.mode == Mode::Query {
         for table in tables {
-            if let Ok(new_schema) = fetch_schema(table).await {
+            if let Ok(new_schema) = PARSEABLE.fetch_schema(table).await {
                 // commit schema merges the schema internally and updates the schema in storage.
                 PARSEABLE
                     .storage

src/parseable/mod.rs

@@ -25,7 +25,9 @@ use bytes::Bytes;
 use chrono::Utc;
 use clap::{error::ErrorKind, Parser};
 use http::{header::CONTENT_TYPE, HeaderName, HeaderValue, StatusCode};
+use itertools::Itertools;
 use once_cell::sync::Lazy;
+use relative_path::RelativePathBuf;
 pub use staging::StagingError;
 use streams::StreamRef;
 pub use streams::{StreamNotFound, Streams};
@@ -50,7 +52,7 @@ use crate::{
     static_schema::{convert_static_schema_to_arrow_schema, StaticSchema},
     storage::{
         object_storage::parseable_json_path, ObjectStorageError, ObjectStorageProvider,
-        ObjectStoreFormat, Owner, Permisssion, StreamType,
+        ObjectStoreFormat, Owner, Permisssion, StreamType, STREAM_ROOT_DIRECTORY,
     },
     validator,
 };
@@ -765,6 +767,33 @@ impl Parseable {
 
         Some(first_event_at.to_string())
     }
+
+    /// Fetches the schema for the specified stream.
+    ///
+    /// # Arguments
+    ///
+    /// * `stream_name` - The name of the stream to fetch the schema for.
+    ///
+    /// # Returns
+    ///
+    /// An `anyhow::Result` containing the `arrow_schema::Schema` for the specified stream.
+    pub async fn fetch_schema(&self, stream_name: &str) -> anyhow::Result<Schema> {
+        let path_prefix = RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY]);
+        let store = self.storage.get_object_store();
+        let res: Vec<Schema> = store
+            .get_objects(
+                Some(&path_prefix),
+                Box::new(|file_name: String| file_name.contains(".schema")),
+            )
+            .await?
+            .iter()
+            // we should be able to unwrap as we know the data is valid schema
+            .map(|byte_obj| serde_json::from_slice(byte_obj).expect("data is valid json"))
+            .collect_vec();
+
+        let new_schema = Schema::try_merge(res)?;
+        Ok(new_schema)
+    }
 }
 
 pub fn validate_static_schema(