This is an automated email from the ASF dual-hosted git repository. erickguan pushed a commit to branch fix-stat-impl in repository https://gitbox.apache.org/repos/asf/opendal.git
commit eba7a98c8ec0f172577a9abfc483bc37904c5676 Author: Erick Guan <[email protected]> AuthorDate: Sat Jul 4 22:33:47 2026 +0800 fix: avoid full value reads in kv stat --- core/services/d1/src/backend.rs | 7 ++-- core/services/d1/src/core.rs | 21 ++++++++++++ core/services/d1/src/model.rs | 60 +++++++++++++++++++++++++++++++++ core/services/mongodb/src/backend.rs | 7 ++-- core/services/mongodb/src/core.rs | 60 +++++++++++++++++++++++++++++++++ core/services/mysql/src/backend.rs | 7 ++-- core/services/mysql/src/core.rs | 22 ++++++++++++ core/services/postgresql/src/backend.rs | 7 ++-- core/services/postgresql/src/core.rs | 22 ++++++++++++ core/services/redis/src/backend.rs | 7 ++-- core/services/surrealdb/src/backend.rs | 7 ++-- core/services/surrealdb/src/core.rs | 36 ++++++++++++++++++++ 12 files changed, 239 insertions(+), 24 deletions(-) diff --git a/core/services/d1/src/backend.rs b/core/services/d1/src/backend.rs index 7c67dc377..ad8347b17 100644 --- a/core/services/d1/src/backend.rs +++ b/core/services/d1/src/backend.rs @@ -249,10 +249,9 @@ impl Service for D1Backend { if p == build_abs_path(&self.root, "") { Ok(RpStat::new(Metadata::new(EntryMode::DIR))) } else { - let bs = self.core.get(ctx, &p).await?; - match bs { - Some(bs) => Ok(RpStat::new( - Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64), + match self.core.get_length(ctx, &p).await? { + Some(length) => Ok(RpStat::new( + Metadata::new(EntryMode::FILE).with_content_length(length as u64), )), None => Err(Error::new(ErrorKind::NotFound, "kv not found in d1")), } diff --git a/core/services/d1/src/core.rs b/core/services/d1/src/core.rs index 86997991d..96343b4ca 100644 --- a/core/services/d1/src/core.rs +++ b/core/services/d1/src/core.rs @@ -104,6 +104,27 @@ impl D1Core { } } + pub async fn get_length(&self, ctx: &OperationContext, path: &str) -> Result<Option<usize>> { + let query = format!( + "SELECT LENGTH(CAST({} AS BLOB)) AS content_length FROM {} WHERE {} = ? LIMIT 1", + self.value_field, self.table, self.key_field + ); + let req = + self.create_d1_query_request(&query, vec![path.into()], Operation::Stat, "Stat")?; + + let resp = ctx.http_transport().send(req).await?; + let status = resp.status(); + match status { + StatusCode::OK | StatusCode::PARTIAL_CONTENT => { + let body = resp.into_body(); + let bs = body.to_bytes(); + let d1_response = D1Response::parse(&bs)?; + d1_response.get_usize_result("content_length") + } + _ => Err(parse_error(resp)), + } + } + pub async fn set(&self, ctx: &OperationContext, path: &str, value: Buffer) -> Result<()> { let table = &self.table; let key_field = &self.key_field; diff --git a/core/services/d1/src/model.rs b/core/services/d1/src/model.rs index 138341497..4c20c19a1 100644 --- a/core/services/d1/src/model.rs +++ b/core/services/d1/src/model.rs @@ -71,6 +71,38 @@ impl D1Response { _ => None, } } + + pub fn get_usize_result(&self, key: &str) -> Result<Option<usize>, Error> { + if self.result.is_empty() || self.result[0].results.is_empty() { + return Ok(None); + } + let result = &self.result[0].results[0]; + let Some(value) = result.get(key) else { + return Ok(None); + }; + + match value { + Value::Number(n) => { + let value = n.as_u64().ok_or_else(|| { + Error::new( + opendal_core::ErrorKind::Unexpected, + "d1 value length is invalid", + ) + })?; + value.try_into().map(Some).map_err(|err| { + Error::new( + opendal_core::ErrorKind::Unexpected, + "d1 value length is invalid", + ) + .set_source(err) + }) + } + _ => Err(Error::new( + opendal_core::ErrorKind::Unexpected, + "d1 value length is invalid", + )), + } + } } #[derive(Deserialize, Debug)] @@ -121,4 +153,32 @@ mod test { let response: D1Response = serde_json::from_str(data).unwrap(); println!("{:?}", response.result[0].results[0]); } + + #[test] + fn test_get_usize_result() { + let data = r#" + { + "result": [ + { + "results": [ + { + "content_length": 6 + } + ], + "success": true, + "meta": {} + } + ], + "success": true, + "errors": [], + "messages": [] + }"#; + + let response: D1Response = serde_json::from_str(data).unwrap(); + assert_eq!( + response.get_usize_result("content_length").unwrap(), + Some(6) + ); + assert_eq!(response.get_usize_result("missing").unwrap(), None); + } } diff --git a/core/services/mongodb/src/backend.rs b/core/services/mongodb/src/backend.rs index 1fa4b4b4f..f792db5b6 100644 --- a/core/services/mongodb/src/backend.rs +++ b/core/services/mongodb/src/backend.rs @@ -241,10 +241,9 @@ impl Service for MongodbBackend { 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), + match self.core.get_length(&p).await? { + Some(length) => Ok(RpStat::new( + Metadata::new(EntryMode::FILE).with_content_length(length as u64), )), None => Err(Error::new(ErrorKind::NotFound, "kv not found in mongodb")), } diff --git a/core/services/mongodb/src/core.rs b/core/services/mongodb/src/core.rs index 99c9d4a13..b4febd45b 100644 --- a/core/services/mongodb/src/core.rs +++ b/core/services/mongodb/src/core.rs @@ -19,6 +19,7 @@ use std::fmt::Debug; use mea::once::OnceCell; use mongodb::bson::Binary; +use mongodb::bson::Bson; use mongodb::bson::Document; use mongodb::bson::doc; use mongodb::options::ClientOptions; @@ -80,6 +81,32 @@ impl MongodbCore { } } + pub async fn get_length(&self, path: &str) -> Result<Option<usize>> { + let collection = self.get_collection().await?; + let mut cursor = collection + .aggregate(vec![ + doc! { "$match": { self.key_field.as_str(): path } }, + doc! { "$limit": 1 }, + doc! { + "$project": { + "_id": 0, + "content_length": { + "$binarySize": format!("${}", self.value_field) + } + } + }, + ]) + .await + .map_err(parse_mongodb_error)?; + + if !cursor.advance().await.map_err(parse_mongodb_error)? { + return Ok(None); + } + + let doc: Document = cursor.deserialize_current().map_err(parse_mongodb_error)?; + parse_bson_usize(doc.get("content_length")) + } + pub async fn set(&self, path: &str, value: Buffer) -> Result<()> { let collection = self.get_collection().await?; let filter = doc! { self.key_field.as_str(): path }; @@ -111,3 +138,36 @@ fn parse_mongodb_error(err: mongodb::error::Error) -> Error { fn parse_bson_error(err: mongodb::bson::document::ValueAccessError) -> Error { Error::new(ErrorKind::Unexpected, "bson error").set_source(err) } + +fn parse_bson_usize(value: Option<&Bson>) -> Result<Option<usize>> { + let Some(value) = value else { + return Ok(None); + }; + + match value { + Bson::Int32(v) => (*v).try_into().map(Some).map_err(|err| { + Error::new(ErrorKind::Unexpected, "mongodb value length is invalid").set_source(err) + }), + Bson::Int64(v) => (*v).try_into().map(Some).map_err(|err| { + Error::new(ErrorKind::Unexpected, "mongodb value length is invalid").set_source(err) + }), + _ => Err(Error::new( + ErrorKind::Unexpected, + "mongodb value length is invalid", + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_bson_usize() { + assert_eq!(parse_bson_usize(Some(&Bson::Int32(6))).unwrap(), Some(6)); + assert_eq!(parse_bson_usize(Some(&Bson::Int64(6))).unwrap(), Some(6)); + assert_eq!(parse_bson_usize(None).unwrap(), None); + assert!(parse_bson_usize(Some(&Bson::Int32(-1))).is_err()); + assert!(parse_bson_usize(Some(&Bson::String("6".to_string()))).is_err()); + } +} diff --git a/core/services/mysql/src/backend.rs b/core/services/mysql/src/backend.rs index 1dabe6b8f..da2b672f2 100644 --- a/core/services/mysql/src/backend.rs +++ b/core/services/mysql/src/backend.rs @@ -222,10 +222,9 @@ impl Service for MysqlBackend { 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), + match self.core.get_length(&p).await? { + Some(length) => Ok(RpStat::new( + Metadata::new(EntryMode::FILE).with_content_length(length as u64), )), None => Err(Error::new(ErrorKind::NotFound, "kv not found in mysql")), } diff --git a/core/services/mysql/src/core.rs b/core/services/mysql/src/core.rs index f977eba41..807688300 100644 --- a/core/services/mysql/src/core.rs +++ b/core/services/mysql/src/core.rs @@ -57,6 +57,28 @@ impl MysqlCore { Ok(value.map(Buffer::from)) } + pub async fn get_length(&self, path: &str) -> Result<Option<usize>> { + let pool = self.get_client().await?; + + let value: Option<i64> = sqlx::query_scalar(&format!( + "SELECT OCTET_LENGTH(`{}`) FROM `{}` WHERE `{}` = ? LIMIT 1", + self.value_field, self.table, self.key_field + )) + .bind(path) + .fetch_optional(pool) + .await + .map_err(parse_mysql_error)?; + + value + .map(|v| { + v.try_into().map_err(|err| { + Error::new(ErrorKind::Unexpected, "mysql value length is invalid") + .set_source(err) + }) + }) + .transpose() + } + pub async fn set(&self, path: &str, value: Buffer) -> Result<()> { let pool = self.get_client().await?; diff --git a/core/services/postgresql/src/backend.rs b/core/services/postgresql/src/backend.rs index 04f3aa8eb..ebd33a007 100644 --- a/core/services/postgresql/src/backend.rs +++ b/core/services/postgresql/src/backend.rs @@ -214,10 +214,9 @@ impl Service for PostgresqlBackend { 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), + match self.core.get_length(&p).await? { + Some(length) => Ok(RpStat::new( + Metadata::new(EntryMode::FILE).with_content_length(length as u64), )), None => Err(Error::new( ErrorKind::NotFound, diff --git a/core/services/postgresql/src/core.rs b/core/services/postgresql/src/core.rs index 9733b4479..670d7c276 100644 --- a/core/services/postgresql/src/core.rs +++ b/core/services/postgresql/src/core.rs @@ -57,6 +57,28 @@ impl PostgresqlCore { Ok(value.map(Buffer::from)) } + pub async fn get_length(&self, path: &str) -> Result<Option<usize>> { + let pool = self.get_client().await?; + + let value: Option<i64> = sqlx::query_scalar(&format!( + r#"SELECT OCTET_LENGTH("{}")::BIGINT FROM "{}" WHERE "{}" = $1 LIMIT 1"#, + self.value_field, self.table, self.key_field + )) + .bind(path) + .fetch_optional(pool) + .await + .map_err(parse_postgres_error)?; + + value + .map(|v| { + v.try_into().map_err(|err| { + Error::new(ErrorKind::Unexpected, "postgresql value length is invalid") + .set_source(err) + }) + }) + .transpose() + } + pub async fn set(&self, path: &str, value: Buffer) -> Result<()> { let pool = self.get_client().await?; diff --git a/core/services/redis/src/backend.rs b/core/services/redis/src/backend.rs index 57375b9b6..6afba8a8b 100644 --- a/core/services/redis/src/backend.rs +++ b/core/services/redis/src/backend.rs @@ -340,10 +340,9 @@ impl Service for RedisBackend { 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), + match self.core.len(&p).await? { + Some(len) => Ok(RpStat::new( + Metadata::new(EntryMode::FILE).with_content_length(len as u64), )), None => Err(Error::new(ErrorKind::NotFound, "key not found in redis")), } diff --git a/core/services/surrealdb/src/backend.rs b/core/services/surrealdb/src/backend.rs index b6b5e5247..d993f679f 100644 --- a/core/services/surrealdb/src/backend.rs +++ b/core/services/surrealdb/src/backend.rs @@ -268,10 +268,9 @@ impl Service for SurrealdbBackend { 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), + match self.core.get_length(&p).await? { + Some(length) => Ok(RpStat::new( + Metadata::new(EntryMode::FILE).with_content_length(length as u64), )), None => Err(Error::new(ErrorKind::NotFound, "kv not found in surrealdb")), } diff --git a/core/services/surrealdb/src/core.rs b/core/services/surrealdb/src/core.rs index 365f74afb..f92a3ebf2 100644 --- a/core/services/surrealdb/src/core.rs +++ b/core/services/surrealdb/src/core.rs @@ -115,6 +115,42 @@ impl SurrealdbCore { Ok(value.map(Buffer::from)) } + pub async fn get_length(&self, path: &str) -> Result<Option<usize>> { + let query: String = if self.key_field == "id" { + "SELECT bytes::len(type::field($value_field)) AS content_length FROM type::thing($table, $path)" + .to_string() + } else { + format!( + "SELECT bytes::len(type::field($value_field)) AS content_length FROM type::table($table) WHERE {} = $path LIMIT 1", + self.key_field + ) + }; + + let mut result = self + .get_connection() + .await? + .query(query) + .bind(("namespace", "opendal")) + .bind(("path", path.to_string())) + .bind(("table", self.table.to_string())) + .bind(("value_field", self.value_field.to_string())) + .await + .map_err(parse_surrealdb_error)?; + + let value: Option<i64> = result + .take((0, "content_length")) + .map_err(parse_surrealdb_error)?; + + value + .map(|v| { + v.try_into().map_err(|err| { + Error::new(ErrorKind::Unexpected, "surrealdb value length is invalid") + .set_source(err) + }) + }) + .transpose() + } + pub async fn set(&self, path: &str, value: Buffer) -> Result<()> { let query = format!( "INSERT INTO {} ({}, {}) \
