chore(cubestore): Deleay in remove local files (#6796)

Author: waralexrom

rust/cubestore/cubestore/src/metastore/mod.rs

@@ -20,6 +20,7 @@ pub use rocks_store::*;
 pub use rocks_table::*;
 
 use crate::cluster::node_name_by_partition;
+use crate::metastore::partition::partition_file_name;
 use async_trait::async_trait;
 use log::info;
 use rocksdb::{BlockBasedOptions, Cache, Env, MergeOperands, Options, DB};
@@ -3327,8 +3328,9 @@ impl MetaStore for RocksMetaStore {
 
             for p in PartitionRocksTable::new(db.clone()).table_scan(db.snapshot)? {
                 let p = p?;
-                if let Some(filename) = p.row.get_full_name(p.id) {
-                    filenames.push(filename);
+                if p.row.active || p.row.main_table_row_count == 0 {
+                    //maint_table_row_count == 0 means that partition is just created
+                    filenames.push(partition_file_name(p.id, p.row.suffix()));
                 }
             }
             Ok(filenames)

rust/cubestore/cubestore/src/remotefs/cleanup.rs

@@ -113,6 +113,7 @@ impl RemoteFsCleanup {
         let cleanup_local_files_delay =
             Duration::from_secs(self.config.local_files_cleanup_delay_secs());
 
+        let mut files_to_remove: HashSet<String> = HashSet::new();
         loop {
             // Do the cleanup every now and then.
             tokio::select! {
@@ -122,6 +123,40 @@ impl RemoteFsCleanup {
                 }
             }
 
+            //We delete files on the next iteration after building the file list in order to give time for requests that may use these files to complete
+            if !files_to_remove.is_empty() {
+                log::debug!(
+                    "Cleaning up {} files that were removed remotely",
+                    files_to_remove.len()
+                );
+                log::trace!("The files being removed are {:?}", files_to_remove);
+                //Double check that files don't exists in metastore
+                let files_from_metastore = match self.metastore.get_all_filenames().await {
+                    Err(e) => {
+                        log::error!("could not get the list of files from metastore: {}", e);
+                        continue;
+                    }
+                    Ok(f) => f,
+                };
+
+                // Only keep the files we want to remove in `local_files`.
+                for f in files_from_metastore {
+                    files_to_remove.remove(&f);
+                }
+
+                let local_dir_copy = local_dir.clone();
+                let mut files_to_remove_to_move = HashSet::new();
+                std::mem::swap(&mut files_to_remove, &mut files_to_remove_to_move);
+                cube_ext::spawn_blocking(move || {
+                    for f in files_to_remove_to_move {
+                        let _ = std::fs::remove_file(Path::new(&local_dir_copy).join(f));
+                    }
+                })
+                .await
+                .unwrap();
+            }
+            files_to_remove.clear();
+
             let local_dir_copy = local_dir.clone();
             let res_local_files =
                 cube_ext::spawn_blocking(move || -> Result<HashSet<String>, std::io::Error> {
@@ -213,21 +248,8 @@ impl RemoteFsCleanup {
             }
 
             if !local_files.is_empty() {
-                log::debug!(
-                    "Cleaning up {} files that were removed remotely",
-                    local_files.len()
-                );
-                log::trace!("The files being removed are {:?}", local_files);
+                files_to_remove = local_files;
             }
-
-            let local_dir_copy = local_dir.clone();
-            cube_ext::spawn_blocking(move || {
-                for f in local_files {
-                    let _ = std::fs::remove_file(Path::new(&local_dir_copy).join(f));
-                }
-            })
-            .await
-            .unwrap();
         }
     }
 }
@@ -237,6 +259,17 @@ mod test {
     use crate::config::Config;
     use futures_timer::Delay;
 
+    fn is_root_partition(name: &str) -> bool {
+        name.starts_with("1-") && !name.ends_with(".chunk.parquet")
+    }
+
+    fn remove_root_paritition(names: Vec<String>) -> Vec<String> {
+        names
+            .into_iter()
+            .filter(|name| !is_root_partition(name))
+            .collect::<Vec<_>>()
+    }
+
     #[tokio::test]
     async fn queue_cleanup_local_files() {
         Config::test("cleanup_local_files")
@@ -262,7 +295,7 @@ mod test {
                     .exec_query("INSERT INTO test.tst (a, b) VALUES (20, 20), (40 , 40)")
                     .await
                     .unwrap();
-                let files = meta_store.get_all_filenames().await.unwrap();
+                let files = remove_root_paritition(meta_store.get_all_filenames().await.unwrap());
                 assert_eq!(files.len(), 2);
                 for f in files.iter() {
                     let path = remote_fs.local_file(&f).await.unwrap();
@@ -276,12 +309,15 @@ mod test {
                     .await
                     .unwrap();
 
-                assert_eq!(meta_store.get_all_filenames().await.unwrap().len(), 1);
+                assert_eq!(
+                    remove_root_paritition(meta_store.get_all_filenames().await.unwrap()).len(),
+                    1
+                );
                 for f in files.iter() {
                     let path = remote_fs.local_file(&f).await.unwrap();
                     assert!(Path::new(&path).exists());
                 }
-                Delay::new(Duration::from_millis(3000)).await; // TODO logger init conflict
+                Delay::new(Duration::from_millis(4000)).await; // TODO logger init conflict
 
                 let path = remote_fs.local_file(&files[0]).await.unwrap();
                 assert!(!Path::new(&path).exists());