feat: Add typed kv adapter and migrate moka to it

core/benches/ops/utils.rs

@@ -45,6 +45,8 @@ pub fn services() -> Vec<(&'static str, Option<Operator>)> {
         ("fs", service::<services::Fs>()),
         ("s3", service::<services::S3>()),
         ("memory", service::<services::Memory>()),
+        #[cfg(feature = "services-moka")]
+        ("moka", service::<services::Moka>()),
     ]
 }
 

core/src/raw/adapters/kv/mod.rs

@@ -18,12 +18,6 @@
 //! Providing Key Value Adapter for OpenDAL.
 //!
 //! Any services that implement `Adapter` can be used an OpenDAL Service.
-//!
-//! # Notes
-//!
-//! This adapter creates a new storage format which is not stable.
-//!
-//! Any service that built upon this adapter should not be persisted.
 
 mod api;
 pub use api::Adapter;

core/src/raw/adapters/mod.rs

@@ -46,3 +46,4 @@
 //! - [`kv::Adapter`]: Adapter for Key Value Services like in-memory map, `redis`.
 
 pub mod kv;
+pub mod typed_kv;

core/src/raw/adapters/typed_kv/api.rs

@@ -0,0 +1,89 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::{fmt::Debug, mem::size_of};
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::Utc;
+
+use crate::*;
+
+/// Adapter is the typed adapter to underlying kv services.
+///
+/// By implement this trait, any kv service can work as an OpenDAL Service.
+///
+/// # Notes
+///
+/// `typed_kv::Adapter` is the typed version of `kv::Adapter`. It's more
+/// efficient if the uderlying kv service can store data with its type. For
+/// example, we can store `Bytes` along with it's metadata so that we don't
+/// need to serialize/deserialize it when we get it from the service.
+///
+/// Ideally, we should use `typed_kv::Adapter` instead of `kv::Adapter` for
+/// in-memory rust libs like moka and dashmap.
+#[async_trait]
+pub trait Adapter: Send + Sync + Debug + Unpin + 'static {
+    /// Get the scheme and name of current adapter.
+    fn metadata(&self) -> (Scheme, String);
+
+    /// Get a value from adapter.
+    async fn get(&self, path: &str) -> Result<Option<Value>>;
+
+    /// Get a value from adapter.
+    fn blocking_get(&self, path: &str) -> Result<Option<Value>>;
+
+    /// Set a value into adapter.
+    async fn set(&self, path: &str, value: Value) -> Result<()>;
+
+    /// Set a value into adapter.
+    fn blocking_set(&self, path: &str, value: Value) -> Result<()>;
+
+    /// Delete a value from adapter.
+    async fn delete(&self, path: &str) -> Result<()>;
+
+    /// Delete a value from adapter.
+    fn blocking_delete(&self, path: &str) -> Result<()>;
+}
+
+/// Value is the typed value stored in adapter.
+///
+/// It's cheap to clone so that users can read data without extra copy.
+#[derive(Debug, Clone)]
+pub struct Value {
+    /// Metadata of this value.
+    pub metadata: Metadata,
+    /// The correbonding content of this value.
+    pub value: Bytes,
+}
+
+impl Value {
+    /// Create a new dir of value.
+    pub fn new_dir() -> Self {
+        Self {
+            metadata: Metadata::new(EntryMode::DIR)
+                .with_content_length(0)
+                .with_last_modified(Utc::now()),
+            value: Bytes::new(),
+        }
+    }
+
+    /// Size returns the in-memory size of Value.
+    pub fn size(&self) -> usize {
+        size_of::<Metadata>() + self.value.len()
+    }
+}

core/src/raw/adapters/typed_kv/backend.rs

@@ -0,0 +1,282 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes::Bytes;
+
+use super::Adapter;
+use super::Value;
+use crate::ops::*;
+use crate::raw::oio::VectorCursor;
+use crate::raw::*;
+use crate::*;
+
+/// The typed kv backend which implements Accessor for for typed kv adapter.
+#[derive(Debug, Clone)]
+pub struct Backend<S: Adapter> {
+    kv: Arc<S>,
+    root: String,
+}
+
+impl<S> Backend<S>
+where
+    S: Adapter,
+{
+    /// Create a new kv backend.
+    pub fn new(kv: S) -> Self {
+        Self {
+            kv: Arc::new(kv),
+            root: "/".to_string(),
+        }
+    }
+
+    /// Configure root within this backend.
+    pub fn with_root(mut self, root: &str) -> Self {
+        self.root = normalize_root(root);
+        self
+    }
+}
+
+#[async_trait]
+impl<S: Adapter> Accessor for Backend<S> {
+    type Reader = oio::Cursor;
+    type BlockingReader = oio::Cursor;
+    type Writer = KvWriter<S>;
+    type BlockingWriter = KvWriter<S>;
+    type Pager = ();
+    type BlockingPager = ();
+
+    fn info(&self) -> AccessorInfo {
+        let (scheme, name) = self.kv.metadata();
+
+        let mut am = AccessorInfo::default();
+        am.set_scheme(scheme);
+        am.set_name(&name);
+        am.set_root(&self.root);
+
+        let cap = am.capability_mut();
+        cap.read = true;
+        cap.read_can_seek = true;
+        cap.read_can_next = true;
+        cap.read_with_range = true;
+        cap.stat = true;
+
+        cap.write = true;
+        cap.write_with_cache_control = true;
+        cap.write_with_content_disposition = true;
+        cap.write_with_content_type = true;
+        cap.write_without_content_length = true;
+        cap.create_dir = true;
+        cap.delete = true;
+
+        am
+    }
+
+    async fn create_dir(&self, path: &str, _: OpCreate) -> Result<RpCreate> {
+        let p = build_abs_path(&self.root, path);
+        self.kv.set(&p, Value::new_dir()).await?;
+        Ok(RpCreate::default())
+    }
+
+    fn blocking_create_dir(&self, path: &str, _: OpCreate) -> Result<RpCreate> {
+        let p = build_abs_path(&self.root, path);
+        self.kv.blocking_set(&p, Value::new_dir())?;
+
+        Ok(RpCreate::default())
+    }
+
+    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+        let p = build_abs_path(&self.root, path);
+
+        let bs = match self.kv.get(&p).await? {
+            // TODO: we can reuse the metadata in value to build content range.
+            Some(bs) => bs.value,
+            None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+        };
+
+        let bs = self.apply_range(bs, args.range());
+
+        let length = bs.len();
+        Ok((RpRead::new(length as u64), oio::Cursor::from(bs)))
+    }
+
+    fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+        let p = build_abs_path(&self.root, path);
+
+        let bs = match self.kv.blocking_get(&p)? {
+            // TODO: we can reuse the metadata in value to build content range.
+            Some(bs) => bs.value,
+            None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+        };
+
+        let bs = self.apply_range(bs, args.range());
+        Ok((RpRead::new(bs.len() as u64), oio::Cursor::from(bs)))
+    }
+
+    async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let p = build_abs_path(&self.root, path);
+
+        Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p, args)))
+    }
+
+    fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+        let p = build_abs_path(&self.root, path);
+
+        Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p, args)))
+    }
+
+    async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+        let p = build_abs_path(&self.root, path);
+
+        if p.is_empty() || p.ends_with('/') {
+            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+        } else {
+            let bs = self.kv.get(&p).await?;
+            match bs {
+                Some(bs) => Ok(RpStat::new(bs.metadata)),
+                None => Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+            }
+        }
+    }
+
+    fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+        let p = build_abs_path(&self.root, path);
+
+        if p.is_empty() || p.ends_with('/') {
+            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+        } else {
+            let bs = self.kv.blocking_get(&p)?;
+            match bs {
+                Some(bs) => Ok(RpStat::new(bs.metadata)),
+                None => Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+            }
+        }
+    }
+
+    async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+        let p = build_abs_path(&self.root, path);
+
+        self.kv.delete(&p).await?;
+        Ok(RpDelete::default())
+    }
+
+    fn blocking_delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+        let p = build_abs_path(&self.root, path);
+
+        self.kv.blocking_delete(&p)?;
+        Ok(RpDelete::default())
+    }
+}
+
+impl<S> Backend<S>
+where
+    S: Adapter,
+{
+    fn apply_range(&self, mut bs: Bytes, br: BytesRange) -> Bytes {
+        match (br.offset(), br.size()) {
+            (Some(offset), Some(size)) => {
+                let mut bs = bs.split_off(offset as usize);
+                if (size as usize) < bs.len() {
+                    let _ = bs.split_off(size as usize);
+                }
+                bs
+            }
+            (Some(offset), None) => bs.split_off(offset as usize),
+            (None, Some(size)) => bs.split_off(bs.len() - size as usize),
+            (None, None) => bs,
+        }
+    }
+}
+
+pub struct KvWriter<S> {
+    kv: Arc<S>,
+    path: String,
+
+    op: OpWrite,
+    buf: VectorCursor,
+}
+
+impl<S> KvWriter<S> {
+    fn new(kv: Arc<S>, path: String, op: OpWrite) -> Self {
+        KvWriter {
+            kv,
+            path,
+            op,
+            buf: VectorCursor::new(),
+        }
+    }
+
+    fn build(&self) -> Value {
+        let mut metadata = Metadata::new(EntryMode::FILE);
+        if let Some(v) = self.op.cache_control() {
+            metadata.set_cache_control(v);
+        }
+        if let Some(v) = self.op.content_disposition() {
+            metadata.set_content_disposition(v);
+        }
+        if let Some(v) = self.op.content_type() {
+            metadata.set_content_type(v);
+        }
+        if let Some(v) = self.op.content_length() {
+            metadata.set_content_length(v);
+        } else {
+            metadata.set_content_length(self.buf.len() as u64);
+        }
+
+        Value {
+            metadata,
+            value: self.buf.peak_all(),
+        }
+    }
+}
+
+#[async_trait]
+impl<S: Adapter> oio::Write for KvWriter<S> {
+    // TODO: we need to support append in the future.
+    async fn write(&mut self, bs: Bytes) -> Result<()> {
+        self.buf.push(bs);
+
+        Ok(())
+    }
+
+    async fn abort(&mut self) -> Result<()> {
+        self.buf.clear();
+
+        Ok(())
+    }
+
+    async fn close(&mut self) -> Result<()> {
+        self.kv.set(&self.path, self.build()).await?;
+        Ok(())
+    }
+}
+
+impl<S: Adapter> oio::BlockingWrite for KvWriter<S> {
+    fn write(&mut self, bs: Bytes) -> Result<()> {
+        self.buf.push(bs);
+
+        Ok(())
+    }
+
+    fn close(&mut self) -> Result<()> {
+        self.kv.blocking_set(&self.path, self.build())?;
+
+        Ok(())
+    }
+}