refactor: migrate rocksdb service from adapter::kv to impl Access directly (#6732)

* refactor: migrate rocksdb service from adapter::kv to impl Access directly

* revert rocksdb: 0.24.0 => 0.21.0

* fix tests

Author: koushiro

core/src/services/rocksdb/backend.rs

@@ -15,16 +15,16 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::fmt::Debug;
-use std::fmt::Formatter;
 use std::sync::Arc;
 
 use rocksdb::DB;
 
-use crate::Result;
-use crate::raw::adapters::kv;
+use super::config::RocksdbConfig;
+use super::core::*;
+use super::deleter::RocksdbDeleter;
+use super::lister::RocksdbLister;
+use super::writer::RocksdbWriter;
 use crate::raw::*;
-use crate::services::RocksdbConfig;
 use crate::*;
 
 /// RocksDB service support.
@@ -70,83 +70,106 @@ impl Builder for RocksdbBuilder {
                 .set_source(e)
         })?;
 
-        let root = normalize_root(
-            self.config
-                .root
-                .clone()
-                .unwrap_or_else(|| "/".to_string())
-                .as_str(),
-        );
+        let root = normalize_root(&self.config.root.unwrap_or_default());
 
-        Ok(RocksdbBackend::new(Adapter { db: Arc::new(db) }).with_normalized_root(root))
+        Ok(RocksdbBackend::new(RocksdbCore { db: Arc::new(db) }).with_normalized_root(root))
     }
 }
 
 /// Backend for rocksdb services.
-pub type RocksdbBackend = kv::Backend<Adapter>;
-
-#[derive(Clone)]
-pub struct Adapter {
-    db: Arc<DB>,
+#[derive(Clone, Debug)]
+pub struct RocksdbBackend {
+    core: Arc<RocksdbCore>,
+    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("path", &self.db.path());
-        ds.finish()
-    }
-}
-
-impl kv::Adapter for Adapter {
-    type Scanner = kv::Scanner;
-
-    fn info(&self) -> kv::Info {
-        kv::Info::new(
-            Scheme::Rocksdb,
-            &self.db.path().to_string_lossy(),
-            Capability {
+impl RocksdbBackend {
+    pub fn new(core: RocksdbCore) -> Self {
+        let info = AccessorInfo::default();
+        info.set_scheme(Scheme::Rocksdb.into_static())
+            .set_name(&core.db.path().to_string_lossy())
+            .set_root("/")
+            .set_native_capability(Capability {
                 read: true,
+                stat: true,
                 write: true,
+                write_can_empty: true,
+                delete: true,
                 list: true,
+                list_with_recursive: true,
                 shared: false,
                 ..Default::default()
-            },
-        )
-    }
+            });
 
-    async fn get(&self, path: &str) -> Result<Option<Buffer>> {
-        let result = self.db.get(path).map_err(parse_rocksdb_error)?;
-        Ok(result.map(Buffer::from))
+        Self {
+            core: Arc::new(core),
+            root: "/".to_string(),
+            info: Arc::new(info),
+        }
     }
 
-    async fn set(&self, path: &str, value: Buffer) -> Result<()> {
-        self.db
-            .put(path, value.to_vec())
-            .map_err(parse_rocksdb_error)
+    fn with_normalized_root(mut self, root: String) -> Self {
+        self.info.set_root(&root);
+        self.root = root;
+        self
     }
+}
+
+impl Access for RocksdbBackend {
+    type Reader = Buffer;
+    type Writer = RocksdbWriter;
+    type Lister = oio::HierarchyLister<RocksdbLister>;
+    type Deleter = oio::OneShotDeleter<RocksdbDeleter>;
 
-    async fn delete(&self, path: &str) -> Result<()> {
-        self.db.delete(path).map_err(parse_rocksdb_error)
+    fn info(&self) -> Arc<AccessorInfo> {
+        self.info.clone()
     }
 
-    async fn scan(&self, path: &str) -> Result<Self::Scanner> {
-        let it = self.db.prefix_iterator(path).map(|r| r.map(|(k, _)| k));
-        let mut res = Vec::default();
+    async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+        let p = build_abs_path(&self.root, path);
 
-        for key in it {
-            let key = key.map_err(parse_rocksdb_error)?;
-            let key = String::from_utf8_lossy(&key);
-            if !key.starts_with(path) {
-                break;
+        if p == build_abs_path(&self.root, "") {
+            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+        } else {
+            let bs = self.core.get(&p)?;
+            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 rocksdb")),
             }
-            res.push(key.to_string());
         }
+    }
 
-        Ok(Box::new(kv::ScanStdIter::new(res.into_iter().map(Ok))))
+    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)? {
+            Some(bs) => bs,
+            None => {
+                return Err(Error::new(ErrorKind::NotFound, "kv not found in rocksdb"));
+            }
+        };
+        Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
     }
-}
 
-fn parse_rocksdb_error(e: rocksdb::Error) -> Error {
-    Error::new(ErrorKind::Unexpected, "got rocksdb error").set_source(e)
+    async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let p = build_abs_path(&self.root, path);
+        let writer = RocksdbWriter::new(self.core.clone(), p);
+        Ok((RpWrite::new(), writer))
+    }
+
+    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+        let deleter = RocksdbDeleter::new(self.core.clone(), self.root.clone());
+        Ok((RpDelete::default(), oio::OneShotDeleter::new(deleter)))
+    }
+
+    async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+        let p = build_abs_path(&self.root, path);
+        let lister = RocksdbLister::new(self.core.clone(), self.root.clone(), p)?;
+        Ok((
+            RpList::default(),
+            oio::HierarchyLister::new(lister, path, args.recursive()),
+        ))
+    }
 }

core/src/services/rocksdb/config.rs

@@ -17,10 +17,11 @@
 
 use std::fmt::Debug;
 
-use super::backend::RocksdbBuilder;
 use serde::Deserialize;
 use serde::Serialize;
 
+use super::backend::RocksdbBuilder;
+
 /// Config for Rocksdb Service.
 #[derive(Default, Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 #[serde(default)]
@@ -36,6 +37,7 @@ pub struct RocksdbConfig {
 
 impl crate::Configurator for RocksdbConfig {
     type Builder = RocksdbBuilder;
+
     fn from_uri(uri: &crate::types::OperatorUri) -> crate::Result<Self> {
         let mut map = uri.options().clone();
 

core/src/services/rocksdb/core.rs

@@ -0,0 +1,74 @@
+// 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;
+use std::fmt::Formatter;
+use std::sync::Arc;
+
+use rocksdb::DB;
+
+use crate::*;
+
+#[derive(Clone)]
+pub struct RocksdbCore {
+    pub db: Arc<DB>,
+}
+
+impl Debug for RocksdbCore {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let mut ds = f.debug_struct("RocksdbCore");
+        ds.field("path", &self.db.path());
+        ds.finish()
+    }
+}
+
+impl RocksdbCore {
+    pub fn get(&self, path: &str) -> Result<Option<Buffer>> {
+        let result = self.db.get(path).map_err(parse_rocksdb_error)?;
+        Ok(result.map(Buffer::from))
+    }
+
+    pub fn set(&self, path: &str, value: Buffer) -> Result<()> {
+        self.db
+            .put(path, value.to_vec())
+            .map_err(parse_rocksdb_error)
+    }
+
+    pub fn delete(&self, path: &str) -> Result<()> {
+        self.db.delete(path).map_err(parse_rocksdb_error)
+    }
+
+    pub fn list(&self, path: &str) -> Result<Vec<String>> {
+        let it = self.db.prefix_iterator(path).map(|r| r.map(|(k, _)| k));
+        let mut res = Vec::default();
+
+        for key in it {
+            let key = key.map_err(parse_rocksdb_error)?;
+            let key = String::from_utf8_lossy(&key);
+            if !key.starts_with(path) {
+                break;
+            }
+            res.push(key.to_string());
+        }
+
+        Ok(res)
+    }
+}
+
+fn parse_rocksdb_error(e: rocksdb::Error) -> Error {
+    Error::new(ErrorKind::Unexpected, "got rocksdb error").set_source(e)
+}

core/src/services/rocksdb/deleter.rs

@@ -0,0 +1,42 @@
+// 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 super::core::*;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+pub struct RocksdbDeleter {
+    core: Arc<RocksdbCore>,
+    root: String,
+}
+
+impl RocksdbDeleter {
+    pub fn new(core: Arc<RocksdbCore>, root: String) -> Self {
+        Self { core, root }
+    }
+}
+
+impl oio::OneShotDelete for RocksdbDeleter {
+    async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+        let p = build_abs_path(&self.root, &path);
+        self.core.delete(&p)?;
+        Ok(())
+    }
+}

core/src/services/rocksdb/docs.md

@@ -2,16 +2,15 @@
 
 This service can be used to:
 
+- [ ] create_dir
 - [x] stat
 - [x] read
 - [x] write
-- [x] create_dir
 - [x] delete
-- [x] copy
-- [x] rename
-- [ ] ~~list~~
+- [ ] copy
+- [ ] rename
+- [x] list
 - [ ] ~~presign~~
-- [x] blocking
 
 ## Note