feat: use OceanBase as object store (#887)

## Related Issues
Closes #

## Detailed Changes
OceanBase is a highly reliable, distributed relational database.
Currently, it can be used as a write-ahead log (WAL) for CeresDB. If
OceanBase can be used as an object store for CeresDB, it would decrease
CeresDB's dependencies and improve stability.
And this change set implements the trait `ObjectStore` based on
OceanBase.

## Test Plan
Add new unit tests for the new module.

---------

Co-authored-by: WEI Xikai <[email protected]>

Author: MichaelLeeHZ

Cargo.lock

@@ -5,14 +5,15 @@
 use std::{num::NonZeroUsize, path::Path, pin::Pin, sync::Arc};
 
 use async_trait::async_trait;
-use common_util::{define_result, runtime::Runtime};
+use common_util::define_result;
 use futures::Future;
 use message_queue::kafka::kafka_impl::KafkaImpl;
 use object_store::{
     aliyun,
     disk_cache::DiskCacheStore,
     mem_cache::{MemCache, MemCacheStore},
     metrics::StoreWithMetrics,
+    obkv,
     prefix::StoreWithPrefix,
     LocalFileSystem, ObjectStoreRef,
 };
@@ -110,11 +111,8 @@ pub struct EngineBuilder<'a> {
 
 impl<'a> EngineBuilder<'a> {
     pub async fn build(self) -> Result<TableEngineRef> {
-        let opened_storages = open_storage(
-            self.config.storage.clone(),
-            self.engine_runtimes.io_runtime.clone(),
-        )
-        .await?;
+        let opened_storages =
+            open_storage(self.config.storage.clone(), self.engine_runtimes.clone()).await?;
         let manifest_storages = ManifestStorages {
             wal_manager: self.opened_wals.manifest_wal.clone(),
             oss_storage: opened_storages.default_store().clone(),
@@ -412,7 +410,7 @@ impl ObjectStorePicker for OpenedStorages {
 // ```
 fn open_storage(
     opts: StorageOptions,
-    runtime: Arc<Runtime>,
+    engine_runtimes: Arc<EngineRuntimes>,
 ) -> Pin<Box<dyn Future<Output = Result<OpenedStorages>> + Send>> {
     Box::pin(async move {
         let mut store = match opts.object_store {
@@ -444,6 +442,26 @@ fn open_storage(
                 let store_with_prefix = StoreWithPrefix::new(aliyun_opts.prefix, oss);
                 Arc::new(store_with_prefix.context(OpenObjectStore)?) as _
             }
+            ObjectStoreOptions::Obkv(obkv_opts) => {
+                let obkv_config = obkv_opts.client;
+                let obkv = engine_runtimes
+                    .write_runtime
+                    .spawn_blocking(move || ObkvImpl::new(obkv_config).context(OpenObkv))
+                    .await
+                    .context(RuntimeExec)??;
+
+                let oss: ObjectStoreRef = Arc::new(
+                    obkv::ObkvObjectStore::try_new(
+                        Arc::new(obkv),
+                        obkv_opts.shard_num,
+                        obkv_opts.part_size.0 as usize,
+                        obkv_opts.max_object_size.0 as usize,
+                        obkv_opts.upload_parallelism,
+                    )
+                    .context(OpenObjectStore)?,
+                );
+                Arc::new(StoreWithPrefix::new(obkv_opts.prefix, oss).context(OpenObjectStore)?) as _
+            }
         };
 
         if opts.disk_cache_capacity.as_byte() > 0 {
@@ -464,7 +482,10 @@ fn open_storage(
             ) as _;
         }
 
-        store = Arc::new(StoreWithMetrics::new(store, runtime));
+        store = Arc::new(StoreWithMetrics::new(
+            store,
+            engine_runtimes.io_runtime.clone(),
+        ));
 
         if opts.mem_cache_capacity.as_byte() > 0 {
             let mem_cache = Arc::new(

analytic_engine/src/setup.rs

@@ -4,6 +4,7 @@ use std::time::Duration;
 
 use common_util::config::{ReadableDuration, ReadableSize};
 use serde::{Deserialize, Serialize};
+use table_kv::config::ObkvConfig;
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(default)]
@@ -39,9 +40,11 @@ impl Default for StorageOptions {
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(tag = "type")]
+#[allow(clippy::large_enum_variant)]
 pub enum ObjectStoreOptions {
     Local(LocalOptions),
     Aliyun(AliyunOptions),
+    Obkv(ObkvOptions),
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -83,3 +86,36 @@ impl AliyunOptions {
         ReadableDuration::from(Duration::from_secs(2))
     }
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ObkvOptions {
+    pub prefix: String,
+    #[serde(default = "ObkvOptions::default_shard_num")]
+    pub shard_num: usize,
+    #[serde(default = "ObkvOptions::default_part_size")]
+    pub part_size: ReadableSize,
+    #[serde(default = "ObkvOptions::default_max_object_size")]
+    pub max_object_size: ReadableSize,
+    #[serde(default = "ObkvOptions::default_upload_parallelism")]
+    pub upload_parallelism: usize,
+    /// Obkv client config
+    pub client: ObkvConfig,
+}
+
+impl ObkvOptions {
+    fn default_max_object_size() -> ReadableSize {
+        ReadableSize::gb(1)
+    }
+
+    fn default_part_size() -> ReadableSize {
+        ReadableSize::mb(1)
+    }
+
+    fn default_shard_num() -> usize {
+        512
+    }
+
+    fn default_upload_parallelism() -> usize {
+        8
+    }
+}

analytic_engine/src/storage_options.rs

@@ -25,10 +25,13 @@ lru = { workspace = true }
 prometheus = { workspace = true }
 prometheus-static-metric = { workspace = true }
 prost = { workspace = true }
+rand = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 snafu = { workspace = true }
+table_kv = { workspace = true }
 tokio = { workspace = true }
+twox-hash = "1.6"
 upstream = { package = "object_store", version = "0.5.6", features = [ "aws" ] }
 
 [dev-dependencies]

components/object_store/Cargo.toml

@@ -14,6 +14,7 @@ pub mod disk_cache;
 pub mod mem_cache;
 pub mod metrics;
 pub mod multipart;
+pub mod obkv;
 pub mod prefix;
 
 pub type ObjectStoreRef = Arc<dyn ObjectStore>;