refactor: ObjectStorage: Send

Author: Devdutt Shenoi

server/src/catalog/mod.rs

@@ -318,7 +318,7 @@ async fn create_manifest(
 }
 
 pub async fn remove_manifest_from_snapshot(
-    storage: Arc<dyn ObjectStorage + Send>,
+    storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
     dates: Vec<String>,
 ) -> Result<Option<String>, ObjectStorageError> {
@@ -343,7 +343,7 @@ pub async fn remove_manifest_from_snapshot(
 }
 
 pub async fn get_first_event(
-    storage: Arc<dyn ObjectStorage + Send>,
+    storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
     dates: Vec<String>,
 ) -> Result<Option<String>, ObjectStorageError> {

server/src/hottier.rs

@@ -289,7 +289,7 @@ impl HotTierManager {
         stream: &str,
         manifest_files_to_download: &mut BTreeMap<String, Vec<String>>,
         parquet_file_size: &mut u64,
-        object_store: Arc<dyn ObjectStorage + Send>,
+        object_store: Arc<dyn ObjectStorage>,
     ) -> Result<(), HotTierError> {
         if manifest_files_to_download.is_empty() {
             return Ok(());

server/src/migration/mod.rs

@@ -297,7 +297,7 @@ pub async fn run_file_migration(config: &Config) -> anyhow::Result<()> {
 }
 
 async fn run_meta_file_migration(
-    object_store: &Arc<dyn ObjectStorage + Send>,
+    object_store: &Arc<dyn ObjectStorage>,
     old_meta_file_path: RelativePathBuf,
 ) -> anyhow::Result<()> {
     // get the list of all meta files
@@ -328,9 +328,7 @@ async fn run_meta_file_migration(
     Ok(())
 }
 
-async fn run_stream_files_migration(
-    object_store: &Arc<dyn ObjectStorage + Send>,
-) -> anyhow::Result<()> {
+async fn run_stream_files_migration(object_store: &Arc<dyn ObjectStorage>) -> anyhow::Result<()> {
     let streams = object_store
         .list_old_streams()
         .await?

server/src/option.rs

@@ -39,7 +39,7 @@ pub static CONFIG: Lazy<Arc<Config>> = Lazy::new(|| Arc::new(Config::new()));
 #[derive(Debug)]
 pub struct Config {
     pub parseable: Cli,
-    storage: Arc<dyn ObjectStorageProvider + Send + Sync>,
+    storage: Arc<dyn ObjectStorageProvider>,
     pub storage_name: &'static str,
 }
 

server/src/storage/azure_blob.rs

@@ -163,7 +163,7 @@ impl ObjectStorageProvider for AzureBlobConfig {
         RuntimeConfig::new().with_object_store_registry(Arc::new(object_store_registry))
     }
 
-    fn get_object_store(&self) -> Arc<dyn super::ObjectStorage + Send> {
+    fn get_object_store(&self) -> Arc<dyn super::ObjectStorage> {
         let azure = self.get_default_builder().build().unwrap();
         // limit objectstore to a concurrent request limit
         let azure = LimitStore::new(azure, super::MAX_OBJECT_STORE_REQUESTS);

server/src/storage/localfs.rs

@@ -67,7 +67,7 @@ impl ObjectStorageProvider for FSConfig {
         RuntimeConfig::new()
     }
 
-    fn get_object_store(&self) -> Arc<dyn ObjectStorage + Send> {
+    fn get_object_store(&self) -> Arc<dyn ObjectStorage> {
         Arc::new(LocalFS::new(self.root.clone()))
     }
 

server/src/storage/object_storage.rs

@@ -58,15 +58,15 @@ use std::{
     time::{Duration, Instant},
 };
 
-pub trait ObjectStorageProvider: StorageMetrics + std::fmt::Debug {
+pub trait ObjectStorageProvider: StorageMetrics + std::fmt::Debug + Send + Sync {
     fn get_datafusion_runtime(&self) -> RuntimeConfig;
-    fn get_object_store(&self) -> Arc<dyn ObjectStorage + Send>;
+    fn get_object_store(&self) -> Arc<dyn ObjectStorage>;
     fn get_endpoint(&self) -> String;
     fn register_store_metrics(&self, handler: &PrometheusMetrics);
 }
 
 #[async_trait]
-pub trait ObjectStorage: Sync + 'static {
+pub trait ObjectStorage: Send + Sync + 'static {
     async fn get_object(&self, path: &RelativePath) -> Result<Bytes, ObjectStorageError>;
     // TODO: make the filter function optional as we may want to get all objects
     async fn get_objects(

server/src/storage/s3.rs

@@ -289,7 +289,7 @@ impl ObjectStorageProvider for S3Config {
         RuntimeConfig::new().with_object_store_registry(Arc::new(object_store_registry))
     }
 
-    fn get_object_store(&self) -> Arc<dyn ObjectStorage + Send> {
+    fn get_object_store(&self) -> Arc<dyn ObjectStorage> {
         let s3 = self.get_default_builder().build().unwrap();
 
         // limit objectstore to a concurrent request limit