You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ko...@apache.org on 2025/10/23 14:33:35 UTC
(opendal) 01/01: refactor: migrate rocksdb service from adapter::kv to impl Access directly
This is an automated email from the ASF dual-hosted git repository.
koushiro pushed a commit to branch refactor-rocksdb
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 081e821c42077acadced1d88eef84e99efdf70f8
Author: koushiro <ko...@gmail.com>
AuthorDate: Thu Oct 23 21:23:57 2025 +0800
refactor: migrate rocksdb service from adapter::kv to impl Access directly
---
core/Cargo.lock | 44 +++----
core/Cargo.toml | 2 +-
core/src/services/rocksdb/backend.rs | 139 +++++++++++++----------
core/src/services/rocksdb/config.rs | 4 +-
core/src/services/rocksdb/core.rs | 74 ++++++++++++
core/src/services/rocksdb/{mod.rs => deleter.rs} | 28 ++++-
core/src/services/rocksdb/docs.md | 9 +-
core/src/services/rocksdb/lister.rs | 59 ++++++++++
core/src/services/rocksdb/mod.rs | 5 +
core/src/services/rocksdb/writer.rs | 59 ++++++++++
10 files changed, 326 insertions(+), 97 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index 123b1d7a5..70139b8c8 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -1182,16 +1182,17 @@ dependencies = [
[[package]]
name = "bindgen"
-version = "0.65.1"
+version = "0.69.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cfdf7b466f9a4903edc73f95d6d2bcd5baf8ae620638762244d3f60143643cc5"
+checksum = "271383c67ccabffb7381723dea0672a673f292304fcb45c01cc648c7a8d58088"
dependencies = [
- "bitflags 1.3.2",
+ "bitflags 2.9.1",
"cexpr",
"clang-sys",
+ "itertools 0.12.1",
"lazy_static",
"lazycell",
- "peeking_take_while",
+ "log",
"prettyplease",
"proc-macro2",
"quote",
@@ -1199,20 +1200,19 @@ dependencies = [
"rustc-hash 1.1.0",
"shlex",
"syn 2.0.101",
+ "which",
]
[[package]]
name = "bindgen"
-version = "0.69.5"
+version = "0.70.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "271383c67ccabffb7381723dea0672a673f292304fcb45c01cc648c7a8d58088"
+checksum = "f49d8fed880d473ea71efb9bf597651e77201bdd4893efe54c9e5d65ae04ce6f"
dependencies = [
"bitflags 2.9.1",
"cexpr",
"clang-sys",
- "itertools 0.12.1",
- "lazy_static",
- "lazycell",
+ "itertools 0.13.0",
"log",
"prettyplease",
"proc-macro2",
@@ -1221,25 +1221,22 @@ dependencies = [
"rustc-hash 1.1.0",
"shlex",
"syn 2.0.101",
- "which",
]
[[package]]
name = "bindgen"
-version = "0.70.1"
+version = "0.72.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f49d8fed880d473ea71efb9bf597651e77201bdd4893efe54c9e5d65ae04ce6f"
+checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895"
dependencies = [
"bitflags 2.9.1",
"cexpr",
"clang-sys",
"itertools 0.13.0",
- "log",
- "prettyplease",
"proc-macro2",
"quote",
"regex",
- "rustc-hash 1.1.0",
+ "rustc-hash 2.1.1",
"shlex",
"syn 2.0.101",
]
@@ -4426,14 +4423,13 @@ dependencies = [
[[package]]
name = "librocksdb-sys"
-version = "0.11.0+8.1.1"
+version = "0.17.3+10.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d3386f101bcb4bd252d8e9d2fb41ec3b0862a15a62b478c355b2982efa469e3e"
+checksum = "cef2a00ee60fe526157c9023edab23943fae1ce2ab6f4abb2a807c1746835de9"
dependencies = [
- "bindgen 0.65.1",
+ "bindgen 0.72.1",
"bzip2-sys",
"cc",
- "glob",
"libc",
"libz-sys",
]
@@ -5772,12 +5768,6 @@ dependencies = [
"sha2",
]
-[[package]]
-name = "peeking_take_while"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099"
-
[[package]]
name = "pem"
version = "3.0.5"
@@ -7070,9 +7060,9 @@ checksum = "4e27ee8bb91ca0adcf0ecb116293afa12d393f9c2b9b9cd54d33e8078fe19839"
[[package]]
name = "rocksdb"
-version = "0.21.0"
+version = "0.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb6f170a4041d50a0ce04b0d2e14916d6ca863ea2e422689a5b694395d299ffe"
+checksum = "ddb7af00d2b17dbd07d82c0063e25411959748ff03e8d4f96134c2ff41fce34f"
dependencies = [
"libc",
"librocksdb-sys",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index c716f8a1e..179dee701 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -328,7 +328,7 @@ redis = { version = "0.32", features = [
"connection-manager",
], optional = true }
# for services-rocksdb
-rocksdb = { version = "0.21.0", default-features = false, optional = true }
+rocksdb = { version = "0.24.0", default-features = false, optional = true }
# for services-sled
sled = { version = "0.34.7", optional = true }
# for services-ftp
diff --git a/core/src/services/rocksdb/backend.rs b/core/src/services/rocksdb/backend.rs
index 624d2abe6..d7cfae8ff 100644
--- a/core/src/services/rocksdb/backend.rs
+++ b/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,104 @@ 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,
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);
+ Ok((RpWrite::new(), RocksdbWriter::new(self.core.clone(), p)))
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(RocksdbDeleter::new(self.core.clone(), self.root.clone())),
+ ))
+ }
+
+ 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)?;
+ let lister = oio::HierarchyLister::new(lister, path, args.recursive());
+ Ok((RpList::default(), lister))
+ }
}
diff --git a/core/src/services/rocksdb/config.rs b/core/src/services/rocksdb/config.rs
index e1fc5d8ec..0f1b05ffe 100644
--- a/core/src/services/rocksdb/config.rs
+++ b/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();
diff --git a/core/src/services/rocksdb/core.rs b/core/src/services/rocksdb/core.rs
new file mode 100644
index 000000000..33e4cf875
--- /dev/null
+++ b/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)
+}
diff --git a/core/src/services/rocksdb/mod.rs b/core/src/services/rocksdb/deleter.rs
similarity index 60%
copy from core/src/services/rocksdb/mod.rs
copy to core/src/services/rocksdb/deleter.rs
index 9a55a345e..a42c62144 100644
--- a/core/src/services/rocksdb/mod.rs
+++ b/core/src/services/rocksdb/deleter.rs
@@ -15,8 +15,28 @@
// specific language governing permissions and limitations
// under the License.
-mod backend;
-pub use backend::RocksdbBuilder as Rocksdb;
+use std::sync::Arc;
-mod config;
-pub use config::RocksdbConfig;
+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(())
+ }
+}
diff --git a/core/src/services/rocksdb/docs.md b/core/src/services/rocksdb/docs.md
index 9bf02eabf..5e64ae0bd 100644
--- a/core/src/services/rocksdb/docs.md
+++ b/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
diff --git a/core/src/services/rocksdb/lister.rs b/core/src/services/rocksdb/lister.rs
new file mode 100644
index 000000000..e5eb0422b
--- /dev/null
+++ b/core/src/services/rocksdb/lister.rs
@@ -0,0 +1,59 @@
+// 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 std::vec::IntoIter;
+
+use super::core::*;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+pub struct RocksdbLister {
+ root: String,
+ iter: IntoIter<String>,
+}
+
+impl RocksdbLister {
+ pub fn new(core: Arc<RocksdbCore>, root: String, path: String) -> Result<Self> {
+ let entries = core.list(&path)?;
+
+ Ok(Self {
+ root,
+ iter: entries.into_iter(),
+ })
+ }
+}
+
+impl oio::List for RocksdbLister {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ if let Some(key) = self.iter.next() {
+ let path = build_rel_path(&self.root, &key);
+
+ // Determine if it's a file or directory based on trailing slash
+ let mode = if key.ends_with('/') {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+ let entry = oio::Entry::new(&path, Metadata::new(mode));
+ return Ok(Some(entry));
+ }
+
+ Ok(None)
+ }
+}
diff --git a/core/src/services/rocksdb/mod.rs b/core/src/services/rocksdb/mod.rs
index 9a55a345e..48c733f1c 100644
--- a/core/src/services/rocksdb/mod.rs
+++ b/core/src/services/rocksdb/mod.rs
@@ -16,6 +16,11 @@
// under the License.
mod backend;
+mod core;
+mod deleter;
+mod lister;
+mod writer;
+
pub use backend::RocksdbBuilder as Rocksdb;
mod config;
diff --git a/core/src/services/rocksdb/writer.rs b/core/src/services/rocksdb/writer.rs
new file mode 100644
index 000000000..40c19e9cf
--- /dev/null
+++ b/core/src/services/rocksdb/writer.rs
@@ -0,0 +1,59 @@
+// 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::*;
+
+pub struct RocksdbWriter {
+ core: Arc<RocksdbCore>,
+ path: String,
+ buffer: oio::QueueBuf,
+}
+
+impl RocksdbWriter {
+ pub fn new(core: Arc<RocksdbCore>, path: String) -> Self {
+ Self {
+ core,
+ path,
+ buffer: oio::QueueBuf::new(),
+ }
+ }
+}
+
+impl oio::Write for RocksdbWriter {
+ async fn write(&mut self, bs: Buffer) -> Result<()> {
+ self.buffer.push(bs);
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<Metadata> {
+ let buf = self.buffer.clone().collect();
+ let length = buf.len() as u64;
+ self.core.set(&self.path, buf)?;
+
+ let meta = Metadata::new(EntryMode::from_path(&self.path)).with_content_length(length);
+ Ok(meta)
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.buffer.clear();
+ Ok(())
+ }
+}