You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/04/16 08:35:58 UTC
(opendal) branch main updated: refactor: kv::adapter should use Buffer (read part) (#4494)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 5b1974b007 refactor: kv::adapter should use Buffer (read part) (#4494)
5b1974b007 is described below
commit 5b1974b007f95aae6b2251d6b3dba80d4fc2d924
Author: tison <wa...@gmail.com>
AuthorDate: Tue Apr 16 16:35:51 2024 +0800
refactor: kv::adapter should use Buffer (read part) (#4494)
refactor: kv::adapter should use Buffer
Signed-off-by: tison <wa...@gmail.com>
---
core/src/raw/adapters/kv/api.rs | 4 ++--
core/src/raw/adapters/kv/backend.rs | 11 ++---------
core/src/raw/oio/read/api.rs | 24 ++++++++++++++++++++++--
core/src/services/atomicserver/backend.rs | 11 ++++-------
core/src/services/cacache/backend.rs | 18 ++++++++----------
core/src/services/cloudflare_kv/backend.rs | 7 ++-----
core/src/services/d1/backend.rs | 2 +-
core/src/services/d1/model.rs | 6 +++---
core/src/services/etcd/backend.rs | 4 ++--
core/src/services/foundationdb/backend.rs | 4 ++--
core/src/services/gridfs/backend.rs | 4 ++--
core/src/services/libsql/backend.rs | 4 ++--
core/src/services/memcached/backend.rs | 6 +++---
core/src/services/mongodb/backend.rs | 7 +++----
core/src/services/mysql/backend.rs | 4 ++--
core/src/services/persy/backend.rs | 6 +++---
core/src/services/postgresql/backend.rs | 4 ++--
core/src/services/redb/backend.rs | 8 ++++----
core/src/services/redis/backend.rs | 17 ++++++-----------
core/src/services/rocksdb/backend.rs | 7 ++++---
core/src/services/sled/backend.rs | 6 +++---
core/src/services/sqlite/backend.rs | 12 ++++++------
core/src/services/surrealdb/backend.rs | 6 +++---
core/src/services/tikv/backend.rs | 8 +++++---
24 files changed, 96 insertions(+), 94 deletions(-)
diff --git a/core/src/raw/adapters/kv/api.rs b/core/src/raw/adapters/kv/api.rs
index beede4c509..a2c3e06912 100644
--- a/core/src/raw/adapters/kv/api.rs
+++ b/core/src/raw/adapters/kv/api.rs
@@ -35,10 +35,10 @@ pub trait Adapter: Send + Sync + Debug + Unpin + 'static {
/// Get a key from service.
///
/// - return `Ok(None)` if this key is not exist.
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>>;
+ async fn get(&self, path: &str) -> Result<Option<Buffer>>;
/// The blocking version of get.
- fn blocking_get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
let _ = path;
Err(Error::new(
diff --git a/core/src/raw/adapters/kv/backend.rs b/core/src/raw/adapters/kv/backend.rs
index 05af29da98..9053930f26 100644
--- a/core/src/raw/adapters/kv/backend.rs
+++ b/core/src/raw/adapters/kv/backend.rs
@@ -62,8 +62,8 @@ where
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<S: Adapter> Accessor for Backend<S> {
- type Reader = Bytes;
- type BlockingReader = Bytes;
+ type Reader = Buffer;
+ type BlockingReader = Buffer;
type Writer = KvWriter<S>;
type BlockingWriter = KvWriter<S>;
type Lister = HierarchyLister<KvLister>;
@@ -94,26 +94,19 @@ impl<S: Adapter> Accessor for Backend<S> {
async fn read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
let p = build_abs_path(&self.root, path);
-
let bs = match self.kv.get(&p).await? {
Some(bs) => bs,
None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
};
-
- let bs = Bytes::from(bs);
-
Ok((RpRead::new(), bs))
}
fn blocking_read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
let p = build_abs_path(&self.root, path);
-
let bs = match self.kv.blocking_get(&p)? {
Some(bs) => bs,
None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
};
-
- let bs = Bytes::from(bs);
Ok((RpRead::new(), bs))
}
diff --git a/core/src/raw/oio/read/api.rs b/core/src/raw/oio/read/api.rs
index c874dd089e..01d2d0c9d3 100644
--- a/core/src/raw/oio/read/api.rs
+++ b/core/src/raw/oio/read/api.rs
@@ -103,7 +103,6 @@ impl Read for () {
}
impl Read for Bytes {
- /// TODO: we can check if the offset is out of range.
async fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
if offset >= self.len() as u64 {
return Ok(Buffer::new());
@@ -114,6 +113,17 @@ impl Read for Bytes {
}
}
+impl Read for Buffer {
+ async fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
+ if offset >= self.len() as u64 {
+ return Ok(Buffer::new());
+ }
+ let offset = offset as usize;
+ let limit = limit.min(self.len() - offset);
+ Ok(self.slice(offset..offset + limit))
+ }
+}
+
/// ReadDyn is the dyn version of [`Read`] make it possible to use as
/// `Box<dyn ReadDyn>`.
pub trait ReadDyn: Unpin + Send + Sync {
@@ -168,7 +178,6 @@ impl BlockingRead for () {
}
impl BlockingRead for Bytes {
- /// TODO: we can check if the offset is out of range.
fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
if offset >= self.len() as u64 {
return Ok(Buffer::new());
@@ -179,6 +188,17 @@ impl BlockingRead for Bytes {
}
}
+impl BlockingRead for Buffer {
+ fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
+ if offset >= self.len() as u64 {
+ return Ok(Buffer::new());
+ }
+ let offset = offset as usize;
+ let limit = limit.min(self.len() - offset);
+ Ok(self.slice(offset..offset + limit))
+ }
+}
+
/// `Box<dyn BlockingRead>` won't implement `BlockingRead` automatically.
/// To make BlockingReader work as expected, we must add this impl.
impl<T: BlockingRead + ?Sized> BlockingRead for Box<T> {
diff --git a/core/src/services/atomicserver/backend.rs b/core/src/services/atomicserver/backend.rs
index 74846dc4dc..2244205336 100644
--- a/core/src/services/atomicserver/backend.rs
+++ b/core/src/services/atomicserver/backend.rs
@@ -24,7 +24,6 @@ use atomic_lib::agents::Agent;
use atomic_lib::client::get_authentication_headers;
use atomic_lib::commit::sign_message;
use bytes::Buf;
-use bytes::Bytes;
use http::header::CONTENT_DISPOSITION;
use http::header::CONTENT_TYPE;
use http::Request;
@@ -356,13 +355,11 @@ impl Adapter {
Ok(req)
}
- pub async fn download_from_url(&self, download_url: &String) -> Result<Bytes> {
+ pub async fn download_from_url(&self, download_url: &String) -> Result<Buffer> {
let req = Request::get(download_url);
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
let resp = self.client.send(req).await?;
- let mut bytes_file = resp.into_body();
-
- Ok(bytes_file.copy_to_bytes(bytes_file.remaining()))
+ Ok(resp.into_body())
}
}
@@ -405,7 +402,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let req = self.atomic_get_object_request(path)?;
let resp = self.client.send(req).await?;
let bytes = resp.into_body();
@@ -424,7 +421,7 @@ impl kv::Adapter for Adapter {
.download_from_url(&query_result.results[0].download_url)
.await?;
- Ok(Some(bytes_file.to_vec()))
+ Ok(Some(bytes_file))
}
async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
diff --git a/core/src/services/cacache/backend.rs b/core/src/services/cacache/backend.rs
index 153a7d853b..7a098d6d9f 100644
--- a/core/src/services/cacache/backend.rs
+++ b/core/src/services/cacache/backend.rs
@@ -109,18 +109,16 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
- Ok(Some(
- cacache::read(&self.datadir, path)
- .await
- .map_err(parse_error)?,
- ))
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
+ let result = cacache::read(&self.datadir, path)
+ .await
+ .map_err(parse_error)?;
+ Ok(Some(Buffer::from(result)))
}
- fn blocking_get(&self, path: &str) -> Result<Option<Vec<u8>>> {
- Ok(Some(
- cacache::read_sync(&self.datadir, path).map_err(parse_error)?,
- ))
+ fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
+ let result = cacache::read_sync(&self.datadir, path).map_err(parse_error)?;
+ Ok(Some(Buffer::from(result)))
}
async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
diff --git a/core/src/services/cloudflare_kv/backend.rs b/core/src/services/cloudflare_kv/backend.rs
index 0dd64d3710..18c63d05cb 100644
--- a/core/src/services/cloudflare_kv/backend.rs
+++ b/core/src/services/cloudflare_kv/backend.rs
@@ -225,7 +225,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let url = format!("{}/values/{}", self.url_prefix, path);
let mut req = Request::get(&url);
req = req.header(header::CONTENT_TYPE, "application/json");
@@ -234,10 +234,7 @@ impl kv::Adapter for Adapter {
let resp = self.client.send(req).await?;
let status = resp.status();
match status {
- StatusCode::OK => {
- let mut body = resp.into_body();
- Ok(Some(body.copy_to_bytes(body.remaining()).to_vec()))
- }
+ StatusCode::OK => Ok(Some(resp.into_body())),
_ => Err(parse_error(resp).await?),
}
}
diff --git a/core/src/services/d1/backend.rs b/core/src/services/d1/backend.rs
index b21603d21c..b1005fb472 100644
--- a/core/src/services/d1/backend.rs
+++ b/core/src/services/d1/backend.rs
@@ -303,7 +303,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let query = format!(
"SELECT {} FROM {} WHERE {} = ? LIMIT 1",
self.value_field, self.table, self.key_field
diff --git a/core/src/services/d1/model.rs b/core/src/services/d1/model.rs
index fee83bb13c..3dc713dc0a 100644
--- a/core/src/services/d1/model.rs
+++ b/core/src/services/d1/model.rs
@@ -23,7 +23,7 @@ use serde::Serialize;
use serde_json::Map;
use serde_json::Value;
-use crate::Error;
+use crate::{Buffer, Error};
/// response data from d1
#[derive(Deserialize, Debug)]
@@ -52,7 +52,7 @@ impl D1Response {
Ok(response)
}
- pub fn get_result(&self, key: &str) -> Option<Vec<u8>> {
+ pub fn get_result(&self, key: &str) -> Option<Buffer> {
if self.result.is_empty() || self.result[0].results.is_empty() {
return None;
}
@@ -67,7 +67,7 @@ impl D1Response {
v.push(n.as_u64().unwrap() as u8);
}
}
- Some(v)
+ Some(Buffer::from(v))
}
_ => None,
}
diff --git a/core/src/services/etcd/backend.rs b/core/src/services/etcd/backend.rs
index 1a4bae6372..ef7e21e6a0 100644
--- a/core/src/services/etcd/backend.rs
+++ b/core/src/services/etcd/backend.rs
@@ -349,11 +349,11 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, key: &str) -> Result<Option<Buffer>> {
let mut client = self.conn().await?;
let resp = client.get(key, None).await.map_err(format_etcd_error)?;
if let Some(kv) = resp.kvs().first() {
- Ok(Some(kv.value().to_vec()))
+ Ok(Some(Buffer::from(kv.value().to_vec())))
} else {
Ok(None)
}
diff --git a/core/src/services/foundationdb/backend.rs b/core/src/services/foundationdb/backend.rs
index 11c475ca17..54b6a48913 100644
--- a/core/src/services/foundationdb/backend.rs
+++ b/core/src/services/foundationdb/backend.rs
@@ -149,12 +149,12 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let transaction = self.db.create_trx().expect("Unable to create transaction");
match transaction.get(path.as_bytes(), false).await {
Ok(slice) => match slice {
- Some(data) => Ok(Some(data.to_vec())),
+ Some(data) => Ok(Some(Buffer::from(data.to_vec()))),
None => Err(Error::new(
ErrorKind::NotFound,
"foundationdb: key not found",
diff --git a/core/src/services/gridfs/backend.rs b/core/src/services/gridfs/backend.rs
index a989242146..9233767347 100644
--- a/core/src/services/gridfs/backend.rs
+++ b/core/src/services/gridfs/backend.rs
@@ -229,7 +229,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let bucket = self.get_bucket().await?;
let filter = doc! { "filename": path };
let options = GridFsFindOptions::builder().limit(Some(1)).build();
@@ -246,7 +246,7 @@ impl kv::Adapter for Adapter {
.download_to_futures_0_3_writer(file_id, &mut destination)
.await
.map_err(parse_mongodb_error)?;
- Ok(Some(destination))
+ Ok(Some(Buffer::from(destination)))
}
None => Ok(None),
}
diff --git a/core/src/services/libsql/backend.rs b/core/src/services/libsql/backend.rs
index aba55e133e..759a0e99dc 100644
--- a/core/src/services/libsql/backend.rs
+++ b/core/src/services/libsql/backend.rs
@@ -349,7 +349,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let query = format!(
"SELECT {} FROM {} WHERE `{}` = ? LIMIT 1",
self.value_field, self.table, self.key_field
@@ -369,7 +369,7 @@ impl kv::Adapter for Adapter {
let val = &rows[0][0];
match val {
Value::Null => Ok(None),
- Value::Blob { value } => Ok(Some(value.to_owned())),
+ Value::Blob { value } => Ok(Some(Buffer::from(value.to_vec()))),
_ => Err(Error::new(ErrorKind::Unexpected, "invalid value type")),
}
}
diff --git a/core/src/services/memcached/backend.rs b/core/src/services/memcached/backend.rs
index d2a5be030d..6521e4e270 100644
--- a/core/src/services/memcached/backend.rs
+++ b/core/src/services/memcached/backend.rs
@@ -230,10 +230,10 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, key: &str) -> Result<Option<Buffer>> {
let mut conn = self.conn().await?;
-
- conn.get(&percent_encode_path(key)).await
+ let result = conn.get(&percent_encode_path(key)).await?;
+ Ok(result.map(Buffer::from))
}
async fn set(&self, key: &str, value: &[u8]) -> Result<()> {
diff --git a/core/src/services/mongodb/backend.rs b/core/src/services/mongodb/backend.rs
index f09edabe85..3847329325 100644
--- a/core/src/services/mongodb/backend.rs
+++ b/core/src/services/mongodb/backend.rs
@@ -263,7 +263,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let collection = self.get_collection().await?;
let filter = doc! {self.key_field.as_str():path};
let result = collection
@@ -274,9 +274,8 @@ impl kv::Adapter for Adapter {
Some(doc) => {
let value = doc
.get_binary_generic(&self.value_field)
- .map_err(parse_bson_error)?
- .to_vec();
- Ok(Some(value))
+ .map_err(parse_bson_error)?;
+ Ok(Some(Buffer::from(value.to_vec())))
}
None => Ok(None),
}
diff --git a/core/src/services/mysql/backend.rs b/core/src/services/mysql/backend.rs
index cdc4dd9598..3826ee8af7 100644
--- a/core/src/services/mysql/backend.rs
+++ b/core/src/services/mysql/backend.rs
@@ -234,7 +234,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let query = format!(
"SELECT `{}` FROM `{}` WHERE `{}` = :path LIMIT 1",
self.value_field, self.table, self.key_field
@@ -255,7 +255,7 @@ impl kv::Adapter for Adapter {
.await
.map_err(parse_mysql_error)?;
match result {
- Some(v) => Ok(Some(v)),
+ Some(v) => Ok(Some(Buffer::from(v))),
None => Ok(None),
}
}
diff --git a/core/src/services/persy/backend.rs b/core/src/services/persy/backend.rs
index 29f63e3a86..f26c76a753 100644
--- a/core/src/services/persy/backend.rs
+++ b/core/src/services/persy/backend.rs
@@ -176,7 +176,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let cloned_self = self.clone();
let cloned_path = path.to_string();
task::spawn_blocking(move || cloned_self.blocking_get(cloned_path.as_str()))
@@ -185,14 +185,14 @@ impl kv::Adapter for Adapter {
.and_then(|inner_result| inner_result)
}
- fn blocking_get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
let mut read_id = self
.persy
.get::<String, persy::PersyId>(&self.index, &path.to_string())
.map_err(parse_error)?;
if let Some(id) = read_id.next() {
let value = self.persy.read(&self.segment, &id).map_err(parse_error)?;
- return Ok(value);
+ return Ok(value.map(Buffer::from));
}
Ok(None)
diff --git a/core/src/services/postgresql/backend.rs b/core/src/services/postgresql/backend.rs
index 7e3aafd01c..76862a5562 100644
--- a/core/src/services/postgresql/backend.rs
+++ b/core/src/services/postgresql/backend.rs
@@ -276,7 +276,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let query = format!(
"SELECT {} FROM {} WHERE {} = $1 LIMIT 1",
self.value_field, self.table, self.key_field
@@ -299,7 +299,7 @@ impl kv::Adapter for Adapter {
return Ok(None);
}
let value: Vec<u8> = rows[0].get(0);
- Ok(Some(value))
+ Ok(Some(Buffer::from(value)))
}
async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
diff --git a/core/src/services/redb/backend.rs b/core/src/services/redb/backend.rs
index ab3c0f6017..e1482493d7 100644
--- a/core/src/services/redb/backend.rs
+++ b/core/src/services/redb/backend.rs
@@ -133,7 +133,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let cloned_self = self.clone();
let cloned_path = path.to_string();
@@ -143,7 +143,7 @@ impl kv::Adapter for Adapter {
.and_then(|inner_result| inner_result)
}
- fn blocking_get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
let read_txn = self.db.begin_read().map_err(parse_transaction_error)?;
let table_define: redb::TableDefinition<&str, &[u8]> =
@@ -157,8 +157,8 @@ impl kv::Adapter for Adapter {
Ok(Some(v)) => Ok(Some(v.value().to_vec())),
Ok(None) => Ok(None),
Err(e) => Err(parse_storage_error(e)),
- };
- result
+ }?;
+ Ok(result.map(Buffer::from))
}
async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
diff --git a/core/src/services/redis/backend.rs b/core/src/services/redis/backend.rs
index 14d647f9ca..f4db2092d5 100644
--- a/core/src/services/redis/backend.rs
+++ b/core/src/services/redis/backend.rs
@@ -389,18 +389,13 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, key: &str) -> Result<Option<Buffer>> {
let conn = self.conn().await?;
- match conn {
- RedisConnection::Normal(mut conn) => {
- let bs = conn.get(key).await.map_err(format_redis_error)?;
- Ok(bs)
- }
- RedisConnection::Cluster(mut conn) => {
- let bs = conn.get(key).await.map_err(format_redis_error)?;
- Ok(bs)
- }
- }
+ let result: Option<bytes::Bytes> = match conn {
+ RedisConnection::Normal(mut conn) => conn.get(key).await.map_err(format_redis_error),
+ RedisConnection::Cluster(mut conn) => conn.get(key).await.map_err(format_redis_error),
+ }?;
+ Ok(result.map(Buffer::from))
}
async fn set(&self, key: &str, value: &[u8]) -> Result<()> {
diff --git a/core/src/services/rocksdb/backend.rs b/core/src/services/rocksdb/backend.rs
index 2787ca79e5..dd1ea3a1e7 100644
--- a/core/src/services/rocksdb/backend.rs
+++ b/core/src/services/rocksdb/backend.rs
@@ -126,7 +126,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let cloned_self = self.clone();
let cloned_path = path.to_string();
@@ -135,8 +135,9 @@ impl kv::Adapter for Adapter {
.map_err(new_task_join_error)?
}
- fn blocking_get(&self, path: &str) -> Result<Option<Vec<u8>>> {
- self.db.get(path).map_err(parse_rocksdb_error)
+ fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
+ let result = self.db.get(path).map_err(parse_rocksdb_error)?;
+ Ok(result.map(Buffer::from))
}
async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
diff --git a/core/src/services/sled/backend.rs b/core/src/services/sled/backend.rs
index 74f845eccb..07ae091e30 100644
--- a/core/src/services/sled/backend.rs
+++ b/core/src/services/sled/backend.rs
@@ -171,7 +171,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let cloned_self = self.clone();
let cloned_path = path.to_string();
@@ -180,12 +180,12 @@ impl kv::Adapter for Adapter {
.map_err(new_task_join_error)?
}
- fn blocking_get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
Ok(self
.tree
.get(path)
.map_err(parse_error)?
- .map(|v| v.to_vec()))
+ .map(|v| Buffer::from(v.to_vec())))
}
async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
diff --git a/core/src/services/sqlite/backend.rs b/core/src/services/sqlite/backend.rs
index a89eeef9a2..f0f82652eb 100644
--- a/core/src/services/sqlite/backend.rs
+++ b/core/src/services/sqlite/backend.rs
@@ -169,14 +169,14 @@ impl Builder for SqliteBuilder {
ErrorKind::ConfigInvalid,
"connection_string is required but not set",
)
- .with_context("service", Scheme::Sqlite))
+ .with_context("service", Scheme::Sqlite));
}
};
let table = match self.config.table.clone() {
Some(v) => v,
None => {
return Err(Error::new(ErrorKind::ConfigInvalid, "table is empty")
- .with_context("service", Scheme::Sqlite))
+ .with_context("service", Scheme::Sqlite));
}
};
let key_field = match self.config.key_field.clone() {
@@ -269,7 +269,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let this = self.clone();
let path = path.to_string();
@@ -278,7 +278,7 @@ impl kv::Adapter for Adapter {
.map_err(new_task_join_error)?
}
- fn blocking_get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
let conn = self.pool.get().map_err(parse_r2d2_error)?;
let query = format!(
@@ -286,9 +286,9 @@ impl kv::Adapter for Adapter {
self.value_field, self.table, self.key_field
);
let mut statement = conn.prepare(&query).map_err(parse_rusqlite_error)?;
- let result = statement.query_row([path], |row| row.get(0));
+ let result: rusqlite::Result<Vec<u8>> = statement.query_row([path], |row| row.get(0));
match result {
- Ok(v) => Ok(Some(v)),
+ Ok(v) => Ok(Some(Buffer::from(v))),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(err) => Err(parse_rusqlite_error(err)),
}
diff --git a/core/src/services/surrealdb/backend.rs b/core/src/services/surrealdb/backend.rs
index f0e362d47d..a2020b6528 100644
--- a/core/src/services/surrealdb/backend.rs
+++ b/core/src/services/surrealdb/backend.rs
@@ -30,11 +30,11 @@ use tokio::sync::OnceCell;
use crate::raw::adapters::kv;
use crate::raw::normalize_root;
use crate::raw::ConfigDeserializer;
-use crate::Builder;
use crate::Capability;
use crate::Error;
use crate::ErrorKind;
use crate::Scheme;
+use crate::{Buffer, Builder};
/// Config for Surrealdb services support.
#[derive(Default, Deserialize)]
@@ -330,7 +330,7 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> crate::Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> crate::Result<Option<Buffer>> {
let query: String = if self.key_field == "id" {
"SELECT type::field($value_field) FROM type::thing($table, $path)".to_string()
} else {
@@ -352,7 +352,7 @@ impl kv::Adapter for Adapter {
.take((0, self.value_field.as_str()))
.map_err(parse_surrealdb_error)?;
- Ok(value)
+ Ok(value.map(Buffer::from))
}
async fn set(&self, path: &str, value: &[u8]) -> crate::Result<()> {
diff --git a/core/src/services/tikv/backend.rs b/core/src/services/tikv/backend.rs
index 987fee60e1..3fa4c573f4 100644
--- a/core/src/services/tikv/backend.rs
+++ b/core/src/services/tikv/backend.rs
@@ -231,12 +231,14 @@ impl kv::Adapter for Adapter {
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
- self.get_connection()
+ 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)
+ .map_err(parse_tikv_error)?;
+ Ok(result.map(Buffer::from))
}
async fn set(&self, path: &str, value: &[u8]) -> Result<()> {