feat(cubestore): Add ExtendedRemoteFs interface and make S3 use priority queue based implementation

Author: srh

rust/cubestore/Cargo.lock

@@ -52,6 +52,7 @@ chrono-tz = "0.8.2"
 lazy_static = "1.4.0"
 mockall = "0.8.1"
 async-std = "0.99"
+async-stream = "0.3.6"
 itertools = "0.11.0"
 bigdecimal = { version = "0.2.0", features = ["serde"] }
 # Right now, it's not possible to use the 0.33 release because it has bugs

rust/cubestore/cubestore/Cargo.toml

@@ -28,7 +28,7 @@ use crate::remotefs::gcs::GCSRemoteFs;
 use crate::remotefs::minio::MINIORemoteFs;
 use crate::remotefs::queue::QueueRemoteFs;
 use crate::remotefs::s3::S3RemoteFs;
-use crate::remotefs::{LocalDirRemoteFs, RemoteFs};
+use crate::remotefs::{ExtendedRemoteFs, LocalDirRemoteFs, RemoteFs};
 use crate::scheduler::SchedulerImpl;
 use crate::sql::cache::SqlResultCache;
 use crate::sql::{SqlService, SqlServiceImpl};
@@ -1907,7 +1907,8 @@ impl Config {
             self.injector
                 .register("cachestore_fs", async move |i| {
                     // TODO metastore works with non queue remote fs as it requires loops to be started prior to load_from_remote call
-                    let original_remote_fs = i.get_service("original_remote_fs").await;
+                    let original_remote_fs: Arc<dyn ExtendedRemoteFs> =
+                        i.get_service("original_remote_fs").await;
                     let arc: Arc<dyn DIService> = BaseRocksStoreFs::new_for_cachestore(
                         original_remote_fs,
                         i.get_service_typed().await,
@@ -1982,7 +1983,8 @@ impl Config {
             self.injector
                 .register("metastore_fs", async move |i| {
                     // TODO metastore works with non queue remote fs as it requires loops to be started prior to load_from_remote call
-                    let original_remote_fs = i.get_service("original_remote_fs").await;
+                    let original_remote_fs: Arc<dyn ExtendedRemoteFs> =
+                        i.get_service("original_remote_fs").await;
                     let arc: Arc<dyn DIService> = BaseRocksStoreFs::new_for_metastore(
                         original_remote_fs,
                         i.get_service_typed().await,

rust/cubestore/cubestore/src/config/mod.rs

@@ -6546,8 +6546,9 @@ mod tests {
 
     #[tokio::test]
     async fn delete_old_snapshots() {
+        let metastore_snapshots_lifetime_secs = 1;
         let config = Config::test("delete_old_snapshots").update_config(|mut obj| {
-            obj.metastore_snapshots_lifetime = 1;
+            obj.metastore_snapshots_lifetime = metastore_snapshots_lifetime_secs;
             obj.minimum_metastore_snapshots_count = 2;
             obj
         });
@@ -6616,14 +6617,22 @@ mod tests {
                     .await
                     .unwrap();
 
-            assert_eq!(uploaded3.len(), 3);
+            assert_eq!(
+                uploaded3.len(),
+                3,
+                "uploaded3 keys: {}",
+                uploaded3.keys().join(", ")
+            );
 
             meta_store
                 .create_schema("foo4".to_string(), false)
                 .await
                 .unwrap();
 
-            tokio::time::sleep(Duration::from_millis(1100)).await;
+            tokio::time::sleep(Duration::from_millis(
+                metastore_snapshots_lifetime_secs * 1000 + 100,
+            ))
+            .await;
             meta_store.upload_check_point().await.unwrap();
 
             let uploaded4 =