fix: use uuid as upload_id in obkv object store (#970)

## Rationale
There is a possibility for the `upload_id` in the `obkv object store` to
become corrupted after restarting the server. This can result in errors
in the object's data.

## Detailed Changes
* use uuid as `upload_id`

components/object_store/Cargo.toml

@@ -33,6 +33,7 @@ table_kv = { workspace = true }
 tokio = { workspace = true }
 twox-hash = "1.6"
 upstream = { package = "object_store", version = "0.5.6", features = [ "aws" ] }
+uuid = { version = "1.3.3", features = ["v4"] }
 
 [dev-dependencies]
 tempfile = { workspace = true }

components/object_store/src/obkv/meta.rs

@@ -107,10 +107,7 @@ pub struct ObkvObjectMeta {
     /// The paths of multi upload parts.
     #[serde(rename = "parts")]
     pub parts: Vec<String>,
-    /// The version of object, we use the upload_id as version.
-    /// TODO: Since `upload_id` is used by multiple objects, it may become very
-    /// large. Should we assign a version number to each object to avoid
-    /// this issue?
+    /// The version of object, Now we use the upload_id as version.
     #[serde(rename = "version")]
     pub version: String,
 }

components/object_store/src/obkv/mod.rs

@@ -32,6 +32,7 @@ use upstream::{
     path::{Path, DELIMITER},
     Error as StoreError, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Result,
 };
+use uuid::Uuid;
 
 use crate::{
     multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart},
@@ -167,7 +168,6 @@ pub struct ObkvObjectStore<T> {
     /// The manager to manage object store meta, which persist in obkv
     meta_manager: Arc<MetaManager<T>>,
     client: Arc<T>,
-    current_upload_id: AtomicU64,
     /// The size of one object part persited in obkv
     /// It may cause problem to save huge data in one obkv value, so we
     /// need to split data into small parts.
@@ -212,7 +212,6 @@ impl<T: TableKv> ObkvObjectStore<T> {
             shard_manager,
             meta_manager: Arc::new(meta_manager),
             client,
-            current_upload_id: AtomicU64::new(0),
             part_size,
             max_object_size,
             max_upload_concurrency,
@@ -347,13 +346,12 @@ impl<T: TableKv> ObjectStore for ObkvObjectStore<T> {
     ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
         let instant = Instant::now();
 
-        let upload_id = self.current_upload_id.fetch_add(1, Ordering::Relaxed);
-        let multi_part_id = format!("{upload_id}");
+        let upload_id = Uuid::new_v4().to_string();
         let table_name = self.pick_shard_table(location);
 
         let upload = ObkvMultiPartUpload {
             location: location.clone(),
-            upload_id: multi_part_id.clone(),
+            upload_id: upload_id.clone(),
             table_name: table_name.to_string(),
             size: AtomicU64::new(0),
             client: Arc::clone(&self.client),
@@ -367,7 +365,7 @@ impl<T: TableKv> ObjectStore for ObkvObjectStore<T> {
             "ObkvObjectStore put_multipart operation, location:{location}, table_name:{table_name}, cost:{:?}",
             instant.elapsed()
         );
-        Ok((multi_part_id, Box::new(multi_part_upload)))
+        Ok((upload_id, Box::new(multi_part_upload)))
     }
 
     async fn abort_multipart(&self, location: &Path, multipart_id: &MultipartId) -> Result<()> {
@@ -754,6 +752,7 @@ impl<T: TableKv> CloudMultiPartUploadImpl for ObkvMultiPartUpload<T> {
         };
 
         // Save meta info to specify obkv table.
+        // TODO: We should remove the previous object data when update object.
         self.meta_manager
             .save(meta)
             .await