Merge branch 'main' into refactor-mysql

Author: koushiro

core/Cargo.lock

@@ -239,7 +239,7 @@ required-features = ["tests"]
 [dependencies]
 # Required dependencies
 anyhow = { version = "1.0.100", features = ["std"] }
-backon = { version = "1.5", features = ["tokio-sleep"] }
+backon = { version = "1.6", features = ["tokio-sleep"] }
 base64 = "0.22"
 bytes = "1.6"
 futures = { version = "0.3", default-features = false, features = [
@@ -254,7 +254,7 @@ md-5 = "0.10"
 percent-encoding = "2"
 url = "2.5"
 quick-xml = { version = "0.38", features = ["serialize", "overlapped-lists"] }
-reqwest = { version = "0.12.23", features = [
+reqwest = { version = "0.12.24", features = [
   "stream",
 ], default-features = false }
 serde = { version = "1", features = ["derive"] }
@@ -406,7 +406,7 @@ tracing = { version = "0.1", optional = true }
 probe = { version = "0.5.1", optional = true }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
-backon = { version = "1.2", features = ["gloo-timers-sleep"] }
+backon = { version = "1.6", features = ["gloo-timers-sleep"] }
 getrandom = { version = "0.2", features = ["js"] }
 tokio = { version = "1.46", features = ["time"] }
 uuid = { version = "1.18", features = ["serde", "v4", "js"] }

core/Cargo.toml

@@ -26,7 +26,7 @@ version = "0.0.0"
 
 [dependencies]
 criterion = { version = "0.7", features = ["async", "async_tokio"] }
-opendal = { path = "../..", features = ["tests"] }
+opendal = { path = "../..", features = ["blocking", "tests"] }
 rand = "0.8"
 tokio = { version = "1", features = ["full"] }
 uuid = { version = "1", features = ["v4"] }

core/benches/vs_fs/Cargo.toml

@@ -33,11 +33,7 @@ use crate::*;
 /// - Expose `PageLister` as `Accessor::Lister`
 pub trait PageList: Send + Sync + Unpin + 'static {
     /// next_page is used to fetch next page of entries from underlying storage.
-    #[cfg(not(target_arch = "wasm32"))]
     fn next_page(&self, ctx: &mut PageContext) -> impl Future<Output = Result<()>> + MaybeSend;
-    #[cfg(target_arch = "wasm32")]
-    /// next_page is used to fetch next page of entries from underlying storage.
-    fn next_page(&self, ctx: &mut PageContext) -> impl Future<Output = Result<()>>;
 }
 
 /// PageContext is the context passing between `PageList`.

core/src/raw/oio/list/page_list.rs

@@ -17,19 +17,15 @@
 
 use std::fmt::Debug;
 use std::fmt::Formatter;
+use std::sync::Arc;
 
-use tikv_client::Config;
-use tikv_client::RawClient;
 use tokio::sync::OnceCell;
 
-use crate::Builder;
-use crate::Capability;
-use crate::Error;
-use crate::ErrorKind;
-use crate::Scheme;
-use crate::raw::Access;
-use crate::raw::adapters::kv;
-use crate::services::TikvConfig;
+use super::config::TikvConfig;
+use super::core::*;
+use super::deleter::TikvDeleter;
+use super::writer::TikvWriter;
+use crate::raw::*;
 use crate::*;
 
 /// TiKV backend builder
@@ -112,7 +108,7 @@ impl Builder for TikvBuilder {
             )?;
         }
 
-        Ok(TikvBackend::new(Adapter {
+        Ok(TikvBackend::new(TikvCore {
             client: OnceCell::new(),
             endpoints,
             insecure: self.config.insecure,
@@ -124,107 +120,86 @@ impl Builder for TikvBuilder {
 }
 
 /// Backend for TiKV service
-pub type TikvBackend = kv::Backend<Adapter>;
-
-#[derive(Clone)]
-pub struct Adapter {
-    client: OnceCell<RawClient>,
-    endpoints: Vec<String>,
-    insecure: bool,
-    ca_path: Option<String>,
-    cert_path: Option<String>,
-    key_path: Option<String>,
+#[derive(Clone, Debug)]
+pub struct TikvBackend {
+    core: Arc<TikvCore>,
+    root: String,
+    info: Arc<AccessorInfo>,
 }
 
-impl Debug for Adapter {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        let mut ds = f.debug_struct("Adapter");
-
-        ds.field("endpoints", &self.endpoints);
-        ds.finish()
-    }
-}
-
-impl Adapter {
-    async fn get_connection(&self) -> Result<RawClient> {
-        if let Some(client) = self.client.get() {
-            return Ok(client.clone());
+impl TikvBackend {
+    fn new(core: TikvCore) -> Self {
+        let info = AccessorInfo::default();
+        info.set_scheme(Scheme::Tikv.into_static());
+        info.set_name("TiKV");
+        info.set_root("/");
+        info.set_native_capability(Capability {
+            read: true,
+            stat: true,
+            write: true,
+            write_can_empty: true,
+            delete: true,
+            shared: true,
+            ..Default::default()
+        });
+
+        Self {
+            core: Arc::new(core),
+            root: "/".to_string(),
+            info: Arc::new(info),
         }
-        let client = if self.insecure {
-            RawClient::new(self.endpoints.clone())
-                .await
-                .map_err(parse_tikv_config_error)?
-        } else if self.ca_path.is_some() && self.key_path.is_some() && self.cert_path.is_some() {
-            let (ca_path, key_path, cert_path) = (
-                self.ca_path.clone().unwrap(),
-                self.key_path.clone().unwrap(),
-                self.cert_path.clone().unwrap(),
-            );
-            let config = Config::default().with_security(ca_path, cert_path, key_path);
-            RawClient::new_with_config(self.endpoints.clone(), config)
-                .await
-                .map_err(parse_tikv_config_error)?
-        } else {
-            return Err(
-                Error::new(ErrorKind::ConfigInvalid, "invalid configuration")
-                    .with_context("service", Scheme::Tikv)
-                    .with_context("endpoints", format!("{:?}", self.endpoints)),
-            );
-        };
-        self.client.set(client.clone()).ok();
-        Ok(client)
     }
 }
 
-impl kv::Adapter for Adapter {
-    type Scanner = ();
-
-    fn info(&self) -> kv::Info {
-        kv::Info::new(
-            Scheme::Tikv,
-            "TiKV",
-            Capability {
-                read: true,
-                write: true,
-                shared: true,
-                ..Default::default()
-            },
-        )
+impl Access for TikvBackend {
+    type Reader = Buffer;
+    type Writer = TikvWriter;
+    type Lister = ();
+    type Deleter = oio::OneShotDeleter<TikvDeleter>;
+
+    fn info(&self) -> Arc<AccessorInfo> {
+        self.info.clone()
     }
 
-    async fn get(&self, path: &str) -> Result<Option<Buffer>> {
-        let result = self
-            .get_connection()
-            .await?
-            .get(path.to_owned())
-            .await
-            .map_err(parse_tikv_error)?;
-        Ok(result.map(Buffer::from))
+    async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+        let p = build_abs_path(&self.root, path);
+
+        if p == build_abs_path(&self.root, "") {
+            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+        } else {
+            let bs = self.core.get(&p).await?;
+            match bs {
+                Some(bs) => Ok(RpStat::new(
+                    Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64),
+                )),
+                None => Err(Error::new(ErrorKind::NotFound, "kv not found in tikv")),
+            }
+        }
     }
 
-    async fn set(&self, path: &str, value: Buffer) -> Result<()> {
-        self.get_connection()
-            .await?
-            .put(path.to_owned(), value.to_vec())
-            .await
-            .map_err(parse_tikv_error)
+    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+        let p = build_abs_path(&self.root, path);
+        let bs = match self.core.get(&p).await? {
+            Some(bs) => bs,
+            None => return Err(Error::new(ErrorKind::NotFound, "kv not found in tikv")),
+        };
+        Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
     }
 
-    async fn delete(&self, path: &str) -> Result<()> {
-        self.get_connection()
-            .await?
-            .delete(path.to_owned())
-            .await
-            .map_err(parse_tikv_error)
+    async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let p = build_abs_path(&self.root, path);
+        Ok((RpWrite::new(), TikvWriter::new(self.core.clone(), p)))
     }
-}
 
-fn parse_tikv_error(e: tikv_client::Error) -> Error {
-    Error::new(ErrorKind::Unexpected, "error from tikv").set_source(e)
-}
+    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+        Ok((
+            RpDelete::default(),
+            oio::OneShotDeleter::new(TikvDeleter::new(self.core.clone(), self.root.clone())),
+        ))
+    }
 
-fn parse_tikv_config_error(e: tikv_client::Error) -> Error {
-    Error::new(ErrorKind::ConfigInvalid, "invalid configuration")
-        .with_context("service", Scheme::Tikv)
-        .set_source(e)
+    async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
+        let _ = build_abs_path(&self.root, path);
+        Ok((RpList::default(), ()))
+    }
 }

core/src/services/tikv/backend.rs

@@ -18,10 +18,11 @@
 use std::fmt::Debug;
 use std::fmt::Formatter;
 
-use super::backend::TikvBuilder;
 use serde::Deserialize;
 use serde::Serialize;
 
+use super::backend::TikvBuilder;
+
 /// Config for Tikv services support.
 #[derive(Default, Serialize, Deserialize, Clone, PartialEq, Eq)]
 #[serde(default)]
@@ -54,6 +55,7 @@ impl Debug for TikvConfig {
 
 impl crate::Configurator for TikvConfig {
     type Builder = TikvBuilder;
+
     fn from_uri(uri: &crate::types::OperatorUri) -> crate::Result<Self> {
         let map = uri.options().clone();