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 311f324cc fix(services): improve stat implementation (#7857)
311f324cc is described below
commit 311f324cca8d82a16880b93b0c387eb0faa709d0
Author: Erick Guan <[email protected]>
AuthorDate: Mon Jul 6 16:00:06 2026 +0800
fix(services): improve stat implementation (#7857)
* fix: avoid full value reads in kv stat
* harden sql services and improve d1 and gridfs
* Fix linting
---
core/services/d1/src/backend.rs | 7 +--
core/services/d1/src/core.rs | 49 +++++++++++----
core/services/d1/src/model.rs | 60 ++++++++++++++++++
core/services/gridfs/src/backend.rs | 7 +--
core/services/gridfs/src/core.rs | 30 ++++++---
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 | 34 ++++++++++-
core/services/postgresql/src/backend.rs | 7 +--
core/services/postgresql/src/core.rs | 32 +++++++++-
core/services/redis/src/backend.rs | 8 +--
core/services/sqlite/src/backend.rs | 9 +--
core/services/sqlite/src/core.rs | 104 +++++++++++++++++++++++---------
core/services/surrealdb/src/backend.rs | 7 +--
core/services/surrealdb/src/core.rs | 36 +++++++++++
16 files changed, 380 insertions(+), 84 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..9e99f5e59 100644
--- a/core/services/d1/src/core.rs
+++ b/core/services/d1/src/core.rs
@@ -47,6 +47,8 @@ impl Debug for D1Core {
}
}
+const CLOUDFLARE_API_BASE_URL: &str = "https://api.cloudflare.com/client/v4";;
+
impl D1Core {
fn create_d1_query_request(
&self,
@@ -59,11 +61,7 @@ impl D1Core {
"/accounts/{}/d1/database/{}/query",
self.account_id, self.database_id
);
- let url: String = format!(
- "{}{}",
- "https://api.cloudflare.com/client/v4";,
- percent_encode_path(&p)
- );
+ let url: String = format!("{}{}", CLOUDFLARE_API_BASE_URL,
percent_encode_path(&p));
let mut req = Request::post(&url);
if let Some(auth) = &self.authorization {
@@ -84,8 +82,13 @@ impl D1Core {
}
pub async fn get(&self, ctx: &OperationContext, path: &str) ->
Result<Option<Buffer>> {
+ // d1 follows SQLite SQL syntax, we use `"` for identifier quote. A
quoted identifier is case-sensitive and
+ // can contain special characters.
+ // Read more
https://www.sqlite.org/quirks.html#double_quoted_string_literals
+ //
+ // We uses identifier quote for trusted table and field configuration
to ensure correctness.
let query = format!(
- "SELECT {} FROM {} WHERE {} = ? LIMIT 1",
+ r#"SELECT "{}" FROM "{}" WHERE "{}" = ? LIMIT 1"#,
self.value_field, self.table, self.key_field
);
let req =
@@ -104,15 +107,36 @@ impl D1Core {
}
}
+ pub async fn get_length(&self, ctx: &OperationContext, path: &str) ->
Result<Option<usize>> {
+ let query = format!(
+ r#"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;
let value_field = &self.value_field;
let query = format!(
- "INSERT INTO {table} ({key_field}, {value_field}) \
- VALUES (?, ?) \
- ON CONFLICT ({key_field}) \
- DO UPDATE SET {value_field} = EXCLUDED.{value_field}",
+ r#"INSERT INTO "{table}" ("{key_field}", "{value_field}") \
+ VALUES ('?', ?) \
+ ON CONFLICT ("{key_field}") \
+ DO UPDATE SET "{value_field}" = EXCLUDED."{value_field}""#,
);
let params = vec![path.into(), value.to_vec().into()];
@@ -127,7 +151,10 @@ impl D1Core {
}
pub async fn delete(&self, ctx: &OperationContext, path: &str) ->
Result<()> {
- let query = format!("DELETE FROM {} WHERE {} = ?", self.table,
self.key_field);
+ let query = format!(
+ r#"DELETE FROM "{}" WHERE "{}" = ?"#,
+ self.table, self.key_field
+ );
let req =
self.create_d1_query_request(&query, vec![path.into()],
Operation::Delete, "Delete")?;
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/gridfs/src/backend.rs
b/core/services/gridfs/src/backend.rs
index 2a14a879e..88200ca19 100644
--- a/core/services/gridfs/src/backend.rs
+++ b/core/services/gridfs/src/backend.rs
@@ -223,10 +223,9 @@ impl Service for GridfsBackend {
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(len) => Ok(RpStat::new(
+ Metadata::new(EntryMode::FILE).with_content_length(len as
u64),
)),
None => Err(Error::new(ErrorKind::NotFound, "kv not found in
gridfs")),
}
diff --git a/core/services/gridfs/src/core.rs b/core/services/gridfs/src/core.rs
index 3f5f9a9bb..9ea608822 100644
--- a/core/services/gridfs/src/core.rs
+++ b/core/services/gridfs/src/core.rs
@@ -21,7 +21,7 @@ use futures::AsyncReadExt;
use futures::AsyncWriteExt;
use mea::once::OnceCell;
use mongodb::bson::doc;
-use mongodb::gridfs::GridFsBucket;
+use mongodb::gridfs::{FilesCollectionDocument, GridFsBucket};
use mongodb::options::ClientOptions;
use mongodb::options::GridFsBucketOptions;
use opendal_core::raw::*;
@@ -67,10 +67,18 @@ impl GridfsCore {
.await
}
+ async fn get_one_doc(
+ bucket: &GridFsBucket,
+ path: &str,
+ ) -> Result<Option<FilesCollectionDocument>> {
+ let filter = doc! { "filename": path };
+ let doc = bucket.find_one(filter).await.map_err(parse_mongodb_error)?;
+ Ok(doc)
+ }
+
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let bucket = self.get_bucket().await?;
- let filter = doc! { "filename": path };
- let Some(doc) =
bucket.find_one(filter).await.map_err(parse_mongodb_error)? else {
+ let Some(doc) = Self::get_one_doc(bucket, path).await? else {
return Ok(None);
};
@@ -87,12 +95,21 @@ impl GridfsCore {
Ok(Some(Buffer::from(destination)))
}
+ /// Get the byte length of the file.
+ pub async fn get_length(&self, path: &str) -> Result<Option<usize>> {
+ let bucket = self.get_bucket().await?;
+ let Some(doc) = Self::get_one_doc(bucket, path).await? else {
+ return Ok(None);
+ };
+
+ Ok(Some(doc.length as usize))
+ }
+
pub async fn set(&self, path: &str, value: Buffer) -> Result<()> {
let bucket = self.get_bucket().await?;
// delete old file if exists
- let filter = doc! { "filename": path };
- if let Some(doc) =
bucket.find_one(filter).await.map_err(parse_mongodb_error)? {
+ if let Some(doc) = Self::get_one_doc(bucket, path).await? {
let file_id = doc.id;
bucket.delete(file_id).await.map_err(parse_mongodb_error)?;
};
@@ -113,8 +130,7 @@ impl GridfsCore {
pub async fn delete(&self, path: &str) -> Result<()> {
let bucket = self.get_bucket().await?;
- let filter = doc! { "filename": path };
- let Some(doc) =
bucket.find_one(filter).await.map_err(parse_mongodb_error)? else {
+ let Some(doc) = Self::get_one_doc(bucket, path).await? else {
return Ok(());
};
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..961c29807 100644
--- a/core/services/mysql/src/core.rs
+++ b/core/services/mysql/src/core.rs
@@ -45,6 +45,16 @@ impl MysqlCore {
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let pool = self.get_client().await?;
+ // MySQL uses a backtick for identifier quote. An identifier may or
may not be case-sensitive,
+ // depending on database configuration. Only a quoted identifier can
have special characters.
+ // Read more:
+ // https://dev.mysql.com/doc/refman/9.7/en/identifiers.html
+ //
https://dev.mysql.com/doc/refman/9.7/en/identifier-case-sensitivity.html
+ //
+ // We use:
+ // - formatted, quoted identifiers for trusted table and field
configuration
+ // - bind parameters for values to avoid malformed SQL and SQL
injection
+ // to ensure correctness.
let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
"SELECT `{}` FROM `{}` WHERE `{}` = ? LIMIT 1",
self.value_field, self.table, self.key_field
@@ -57,12 +67,34 @@ 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?;
sqlx::query(&format!(
r#"INSERT INTO `{}` (`{}`, `{}`) VALUES (?, ?)
- ON DUPLICATE KEY UPDATE `{}` = VALUES({})"#,
+ ON DUPLICATE KEY UPDATE `{}` = VALUES(`{}`)"#,
self.table, self.key_field, self.value_field, self.value_field,
self.value_field
))
.bind(path)
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..ff6107ab1 100644
--- a/core/services/postgresql/src/core.rs
+++ b/core/services/postgresql/src/core.rs
@@ -45,6 +45,14 @@ impl PostgresqlCore {
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let pool = self.get_client().await?;
+ // PostgreSQL uses `"` for identifier quote. A quoted identifier is
case-sensitive and
+ // can contain special characters.
+ // Read more
https://www.postgresql.org/docs/18/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS
+ //
+ // We use:
+ // - formatted, quoted identifiers for trusted table and field
configuration
+ // - bind parameters for values to avoid malformed SQL and SQL
injection
+ // to ensure correctness.
let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
r#"SELECT "{}" FROM "{}" WHERE "{}" = $1 LIMIT 1"#,
self.value_field, self.table, self.key_field
@@ -57,6 +65,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?;
@@ -82,7 +112,7 @@ impl PostgresqlCore {
let pool = self.get_client().await?;
sqlx::query(&format!(
- "DELETE FROM {} WHERE {} = $1",
+ r#"DELETE FROM "{}" WHERE "{}" = $1"#,
self.table, self.key_field
))
.bind(path)
diff --git a/core/services/redis/src/backend.rs
b/core/services/redis/src/backend.rs
index 57375b9b6..67aa3896d 100644
--- a/core/services/redis/src/backend.rs
+++ b/core/services/redis/src/backend.rs
@@ -340,15 +340,15 @@ 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")),
}
}
}
+
fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) ->
Result<Self::Reader> {
let output: oio::StreamReader<RedisReader> = {
Ok(oio::StreamReader::new(RedisReader::new(
diff --git a/core/services/sqlite/src/backend.rs
b/core/services/sqlite/src/backend.rs
index 5a96dae0e..b32a9164e 100644
--- a/core/services/sqlite/src/backend.rs
+++ b/core/services/sqlite/src/backend.rs
@@ -240,14 +240,7 @@ impl Service for SqliteBackend {
} else {
format!("{}/", p)
};
- let count: i64 = sqlx::query_scalar(&format!(
- "SELECT COUNT(*) FROM `{}` WHERE `{}` LIKE $1 LIMIT 1",
- self.core.table, self.core.key_field
- ))
- .bind(format!("{}%", dir_path))
- .fetch_one(self.core.get_client().await?)
- .await
- .map_err(parse_sqlite_error)?;
+ let count = self.core.count_under(&dir_path).await?;
if count > 0 {
// Directory exists (has children)
diff --git a/core/services/sqlite/src/core.rs b/core/services/sqlite/src/core.rs
index 16ee5f68a..478884595 100644
--- a/core/services/sqlite/src/core.rs
+++ b/core/services/sqlite/src/core.rs
@@ -48,8 +48,17 @@ impl SqliteCore {
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let pool = self.get_client().await?;
+ // SQLite prefers using standard `"` for identifier quotes instead of
+ // MySQL-compatible backtick ` or [] quotes.
+ // Identifier quotes are case-sensitive and allow special characters.
+ // Read more
https://www.sqlite.org/quirks.html#double_quoted_string_literals_are_accepted
+ //
+ // We use:
+ // - formatted, quoted identifiers for trusted table and field
configuration
+ // - bind parameters for values to avoid malformed SQL and SQL
injection
+ // to ensure correctness.
let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
- "SELECT `{}` FROM `{}` WHERE `{}` = $1 LIMIT 1",
+ r#"SELECT "{}" FROM "{}" WHERE "{}" = $1 LIMIT 1"#,
self.value_field, self.table, self.key_field
))
.bind(path)
@@ -64,7 +73,7 @@ impl SqliteCore {
let pool = self.get_client().await?;
let value: Option<i64> = sqlx::query_scalar(&format!(
- "SELECT LENGTH(CAST(`{}` AS BLOB)) FROM `{}` WHERE `{}` = $1 LIMIT
1",
+ r#"SELECT LENGTH(CAST("{}" AS BLOB)) FROM "{}" WHERE "{}" = $1
LIMIT 1"#,
self.value_field, self.table, self.key_field
))
.bind(path)
@@ -82,6 +91,19 @@ impl SqliteCore {
.transpose()
}
+ pub async fn count_under(&self, path: &str) -> Result<i64> {
+ let pool = self.get_client().await?;
+
+ sqlx::query_scalar(&format!(
+ r#"SELECT COUNT(*) FROM "{}" WHERE "{}" LIKE $1 LIMIT 1"#,
+ self.table, self.key_field
+ ))
+ .bind(format!("{}%", path))
+ .fetch_one(pool)
+ .await
+ .map_err(parse_sqlite_error)
+ }
+
pub async fn get_range(
&self,
path: &str,
@@ -89,31 +111,57 @@ impl SqliteCore {
limit: Option<isize>,
) -> Result<Option<(Buffer, u64)>> {
let pool = self.get_client().await?;
- let query = match limit {
- Some(limit) => format!(
- "SELECT SUBSTR(CAST(`{}` AS BLOB), {}, {}), LENGTH(CAST(`{}`
AS BLOB)) FROM `{}` WHERE `{}` = $1 LIMIT 1",
- self.value_field,
- start + 1,
- limit,
- self.value_field,
- self.table,
- self.key_field
- ),
- None => format!(
- "SELECT SUBSTR(CAST(`{}` AS BLOB), {}), LENGTH(CAST(`{}` AS
BLOB)) FROM `{}` WHERE `{}` = $1 LIMIT 1",
- self.value_field,
- start + 1,
- self.value_field,
- self.table,
- self.key_field
- ),
+ if start < 0 || limit.is_some_and(|v| v < 0) {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "sqlite range contains negative value",
+ ));
+ }
+
+ let start = start.checked_add(1).ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "sqlite range start exceeds supported value",
+ )
+ })?;
+ let start: i64 = start.try_into().map_err(|err| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "sqlite range start exceeds supported value",
+ )
+ .set_source(err)
+ })?;
+ let value = match limit {
+ Some(limit) => {
+ let limit: i64 = limit.try_into().map_err(|err| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "sqlite range size exceeds supported value",
+ )
+ .set_source(err)
+ })?;
+ sqlx::query_as(&format!(
+ r#"SELECT SUBSTR(CAST("{}" AS BLOB), $1, $2),
LENGTH(CAST("{}" AS BLOB)) FROM "{}" WHERE "{}" = $3 LIMIT 1"#,
+ self.value_field, self.value_field, self.table,
self.key_field
+ ))
+ .bind(start)
+ .bind(limit)
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ }
+ None => {
+ sqlx::query_as(&format!(
+ r#"SELECT SUBSTR(CAST("{}" AS BLOB), $1), LENGTH(CAST("{}"
AS BLOB)) FROM "{}" WHERE "{}" = $2 LIMIT 1"#,
+ self.value_field, self.value_field, self.table,
self.key_field
+ ))
+ .bind(start)
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ }
};
-
- let value: Option<(Vec<u8>, i64)> = sqlx::query_as(&query)
- .bind(path)
- .fetch_optional(pool)
- .await
- .map_err(parse_sqlite_error)?;
+ let value: Option<(Vec<u8>, i64)> = value.map_err(parse_sqlite_error)?;
Ok(value.map(|(bs, size)| (Buffer::from(bs), size as u64)))
}
@@ -122,7 +170,7 @@ impl SqliteCore {
let pool = self.get_client().await?;
sqlx::query(&format!(
- "INSERT OR REPLACE INTO `{}` (`{}`, `{}`) VALUES ($1, $2)",
+ r#"INSERT OR REPLACE INTO "{}" ("{}", "{}") VALUES ($1, $2)"#,
self.table, self.key_field, self.value_field,
))
.bind(path)
@@ -138,7 +186,7 @@ impl SqliteCore {
let pool = self.get_client().await?;
sqlx::query(&format!(
- "DELETE FROM `{}` WHERE `{}` = $1",
+ r#"DELETE FROM "{}" WHERE "{}" = $1"#,
self.table, self.key_field
))
.bind(path)
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 {} ({}, {}) \
