refactor: associate ObjectStore::commit_schema

Author: Devdutt Shenoi

src/handlers/http/query.rs

@@ -45,7 +45,6 @@ use crate::query::{CountsRequest, CountsResponse, Query as LogicalQuery};
 use crate::query::{TableScanVisitor, QUERY_SESSION};
 use crate::rbac::Users;
 use crate::response::QueryResponse;
-use crate::storage::object_storage::commit_schema_to_storage;
 use crate::storage::ObjectStorageError;
 use crate::utils::actix::extract_session_key_from_req;
 use crate::utils::time::{TimeParseError, TimeRange};
@@ -172,7 +171,11 @@ pub async fn update_schema_when_distributed(tables: &Vec<String>) -> Result<(),
         for table in tables {
             if let Ok(new_schema) = fetch_schema(table).await {
                 // commit schema merges the schema internally and updates the schema in storage.
-                commit_schema_to_storage(table, new_schema.clone()).await?;
+                PARSEABLE
+                    .storage
+                    .get_object_store()
+                    .commit_schema(table, new_schema.clone())
+                    .await?;
 
                 commit_schema(table, Arc::new(new_schema))?;
             }

src/storage/object_storage.rs

@@ -780,7 +780,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
             for path in stream.schema_files() {
                 let file = File::open(&path)?;
                 let schema: Schema = serde_json::from_reader(file)?;
-                commit_schema_to_storage(&stream_name, schema).await?;
+                self.commit_schema(&stream_name, schema).await?;
                 if let Err(e) = remove_file(path) {
                     warn!("Failed to remove staged file: {e}");
                 }
@@ -789,16 +789,16 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
 
         Ok(())
     }
-}
 
-pub async fn commit_schema_to_storage(
-    stream_name: &str,
-    schema: Schema,
-) -> Result<(), ObjectStorageError> {
-    let storage = PARSEABLE.storage().get_object_store();
-    let stream_schema = storage.get_schema(stream_name).await?;
-    let new_schema = Schema::try_merge(vec![schema, stream_schema]).unwrap();
-    storage.put_schema(stream_name, &new_schema).await
+    async fn commit_schema(
+        &self,
+        stream_name: &str,
+        schema: Schema,
+    ) -> Result<(), ObjectStorageError> {
+        let stream_schema = self.get_schema(stream_name).await?;
+        let new_schema = Schema::try_merge(vec![schema, stream_schema]).unwrap();
+        self.put_schema(stream_name, &new_schema).await
+    }
 }
 
 #[inline(always)]