commit 18fa146828f2031400317c05b68e7412ae5b8a0c
author Junfan Zhang <zuston@apache.org> Sun Feb 04 16:25:04 2024 +0800
committer GitHub <noreply@github.com> Sun Feb 04 16:25:04 2024 +0800
tree d3b51ff28afcf028295197ed4485368ead970ad9
parent acb2f3f9dcfd9c7362ecaf2012a6d9cc8a427322

[#1407] fix(rust): dont panic when no available local disks (#1504)

### What changes were proposed in this pull request?

fix(rust): dont panic when no available local disks

### Why are the changes needed?

make spill data fallback to hdfs when no available local disks

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

online tests

rust/experimental/server/src/store/localfile.rs

diff --git a/rust/experimental/server/src/store/localfile.rs b/rust/experimental/server/src/store/localfile.rs
index 8afdb7b..8f81e5c 100644
--- a/rust/experimental/server/src/store/localfile.rs
+++ b/rust/experimental/server/src/store/localfile.rs
@@ -40,6 +40,7 @@
 use log::{debug, error, warn};
 
 use crate::runtime::manager::RuntimeManager;
+use dashmap::mapref::entry::Entry;
 use std::sync::atomic::{AtomicI64, Ordering};
 use std::sync::Arc;
 
@@ -187,14 +188,16 @@
             LocalFileStore::gen_relative_path_for_partition(&uid);
 
         let mut parent_dir_is_created = false;
-        let locked_obj = self
-            .partition_locks
-            .entry(data_file_path.clone())
-            .or_insert_with(|| {
+        let locked_obj = match self.partition_locks.entry(data_file_path.clone()) {
+            Entry::Vacant(e) => {
                 parent_dir_is_created = true;
-                Arc::new(LockedObj::from(self.select_disk(&uid).unwrap()))
-            })
-            .clone();
+                let disk = self.select_disk(&uid)?;
+                let locked_obj = Arc::new(LockedObj::from(disk));
+                let obj = e.insert_entry(locked_obj.clone());
+                obj.get().clone()
+            }
+            Entry::Occupied(v) => v.get().clone(),
+        };
 
         let local_disk = &locked_obj.disk;
         let mut next_offset = locked_obj.pointer.load(Ordering::SeqCst);