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
The following commit(s) were added to refs/heads/fix-stat-impl by this push:
new 31a54d610 WIP
31a54d610 is described below
commit 31a54d61046fa6307efcde54e91bc4cf5ea37acb
Author: Erick Guan <[email protected]>
AuthorDate: Sun Jul 5 11:25:21 2026 +0800
WIP
---
core/services/mysql/src/core.rs | 87 ++++++++++++++----------
core/services/postgresql/src/core.rs | 69 +++++++++++--------
core/services/redis/src/backend.rs | 1 +
core/services/sqlite/src/backend.rs | 9 +--
core/services/sqlite/src/core.rs | 128 +++++++++++++++++++++--------------
5 files changed, 172 insertions(+), 122 deletions(-)
diff --git a/core/services/mysql/src/core.rs b/core/services/mysql/src/core.rs
index 807688300..00f3564a8 100644
--- a/core/services/mysql/src/core.rs
+++ b/core/services/mysql/src/core.rs
@@ -45,14 +45,27 @@ impl MysqlCore {
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let pool = self.get_client().await?;
- let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
- "SELECT `{}` FROM `{}` WHERE `{}` = ? LIMIT 1",
- self.value_field, self.table, self.key_field
- ))
- .bind(path)
- .fetch_optional(pool)
- .await
- .map_err(parse_mysql_error)?;
+ // 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
+ //
+ // When ANSI_QUOTES SQL mode is enabled, a string literal must be
quoted as well.
+ //
+ // We use:
+ // - parameterized query to avoid SQL injection
+ // - quoted identifiers and string literal quote
+ // to ensure correctness.
+ let value: Option<Vec<u8>> =
+ sqlx::query_scalar("SELECT `?` FROM `?` WHERE `?` = '?' LIMIT 1")
+ .bind(self.value_field.clone())
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ .map_err(parse_mysql_error)?;
Ok(value.map(Buffer::from))
}
@@ -60,14 +73,15 @@ impl MysqlCore {
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)?;
+ let value: Option<i64> =
+ sqlx::query_scalar("SELECT OCTET_LENGTH(`?`) FROM `?` WHERE `?` =
'?' LIMIT 1")
+ .bind(self.value_field.clone())
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ .map_err(parse_mysql_error)?;
value
.map(|v| {
@@ -82,12 +96,16 @@ impl MysqlCore {
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({})"#,
- self.table, self.key_field, self.value_field, self.value_field,
self.value_field
- ))
+ sqlx::query(
+ "INSERT INTO `?` (`?`, `?`) VALUES ('?', ?)
+ ON DUPLICATE KEY UPDATE `?` = VALUES(`?`)",
+ )
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(self.value_field.clone())
.bind(path)
+ .bind(self.value_field.clone())
+ .bind(self.value_field.clone())
.bind(value.to_vec())
.execute(pool)
.await
@@ -99,14 +117,13 @@ impl MysqlCore {
pub async fn delete(&self, path: &str) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(&format!(
- "DELETE FROM `{}` WHERE `{}` = ?",
- self.table, self.key_field
- ))
- .bind(path)
- .execute(pool)
- .await
- .map_err(parse_mysql_error)?;
+ sqlx::query("DELETE FROM `?` WHERE `?` = '?'")
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(path)
+ .execute(pool)
+ .await
+ .map_err(parse_mysql_error)?;
Ok(())
}
@@ -114,15 +131,13 @@ impl MysqlCore {
pub async fn list(&self, path: &str) -> Result<Vec<String>> {
let pool = self.get_client().await?;
- let mut sql = format!(
- "SELECT `{}` FROM `{}` WHERE `{}` LIKE ?",
- self.key_field, self.table, self.key_field
- );
- sql.push_str(&format!(" ORDER BY `{}`", self.key_field));
-
let escaped = escape_like(path);
- sqlx::query_scalar(&sql)
+ sqlx::query_scalar("SELECT `?` FROM `?` WHERE `?` LIKE '?' ORDER BY
`?`")
+ .bind(self.key_field.clone())
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
.bind(format!("{escaped}%"))
+ .bind(self.key_field.clone())
.fetch_all(pool)
.await
.map_err(parse_mysql_error)
diff --git a/core/services/postgresql/src/core.rs
b/core/services/postgresql/src/core.rs
index 670d7c276..7eae9415b 100644
--- a/core/services/postgresql/src/core.rs
+++ b/core/services/postgresql/src/core.rs
@@ -45,14 +45,24 @@ impl PostgresqlCore {
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let pool = self.get_client().await?;
- let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
- r#"SELECT "{}" FROM "{}" WHERE "{}" = $1 LIMIT 1"#,
- self.value_field, self.table, self.key_field
- ))
- .bind(path)
- .fetch_optional(pool)
- .await
- .map_err(parse_postgres_error)?;
+ // PostgreSQL uses `"` for identifier quote. An quoted identifier is
case-sensitive and
+ // can contain special characters.
+ // PostresSQL uses `'` for string literal quote.
+ // Read more
https://www.postgresql.org/docs/18/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS
+ //
+ // We use:
+ // - parameterized query to avoid SQL injection
+ // - quoted identifiers and string literal quote
+ // to ensure correctness.
+ let value: Option<Vec<u8>> =
+ sqlx::query_scalar(r#"SELECT "$1" FROM "$2" WHERE "$3" = '$4'
LIMIT 1"#)
+ .bind(self.value_field.clone())
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ .map_err(parse_postgres_error)?;
Ok(value.map(Buffer::from))
}
@@ -60,10 +70,12 @@ impl PostgresqlCore {
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
- ))
+ let value: Option<i64> = sqlx::query_scalar(
+ r#"SELECT OCTET_LENGTH("$1")::BIGINT FROM "$2" WHERE "$3" = '$4'
LIMIT 1"#,
+ )
+ .bind(self.value_field.clone())
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
.bind(path)
.fetch_optional(pool)
.await
@@ -82,15 +94,15 @@ impl PostgresqlCore {
pub async fn set(&self, path: &str, value: Buffer) -> Result<()> {
let pool = self.get_client().await?;
- let table = &self.table;
- let key_field = &self.key_field;
- let value_field = &self.value_field;
- sqlx::query(&format!(
- r#"INSERT INTO "{table}" ("{key_field}", "{value_field}")
- VALUES ($1, $2)
- ON CONFLICT ("{key_field}")
- DO UPDATE SET "{value_field}" = EXCLUDED."{value_field}""#,
- ))
+ sqlx::query(
+ r#"INSERT INTO "$1" ("$2", "$3")
+ VALUES ('$4', $5)
+ ON CONFLICT ("$2")
+ DO UPDATE SET "$3" = EXCLUDED."$3""#,
+ )
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(self.value_field.clone())
.bind(path)
.bind(value.to_vec())
.execute(pool)
@@ -103,14 +115,13 @@ impl PostgresqlCore {
pub async fn delete(&self, path: &str) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(&format!(
- "DELETE FROM {} WHERE {} = $1",
- self.table, self.key_field
- ))
- .bind(path)
- .execute(pool)
- .await
- .map_err(parse_postgres_error)?;
+ sqlx::query(r#"DELETE FROM "$1" WHERE "$2" = '$3'"#)
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(path)
+ .execute(pool)
+ .await
+ .map_err(parse_postgres_error)?;
Ok(())
}
diff --git a/core/services/redis/src/backend.rs
b/core/services/redis/src/backend.rs
index 6afba8a8b..67aa3896d 100644
--- a/core/services/redis/src/backend.rs
+++ b/core/services/redis/src/backend.rs
@@ -348,6 +348,7 @@ impl Service for RedisBackend {
}
}
}
+
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..4bd52b5bf 100644
--- a/core/services/sqlite/src/core.rs
+++ b/core/services/sqlite/src/core.rs
@@ -34,6 +34,12 @@ pub struct SqliteCore {
pub value_field: String,
}
+#[derive(Debug)]
+struct SqliteBlobSized {
+ bs: Vec<u8>,
+ size: i64,
+}
+
impl SqliteCore {
pub async fn get_client(&self) -> Result<&SqlitePool> {
self.pool
@@ -48,14 +54,24 @@ impl SqliteCore {
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let pool = self.get_client().await?;
- let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
- "SELECT `{}` FROM `{}` WHERE `{}` = $1 LIMIT 1",
- self.value_field, self.table, self.key_field
- ))
- .bind(path)
- .fetch_optional(pool)
- .await
- .map_err(parse_sqlite_error)?;
+ // SQLite prefer using standard `"` for identifier quotes, `'` for
string literal 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:
+ // - parameterized query to avoid SQL injection
+ // - quoted identifiers and string literal quote
+ // to ensure correctness.
+ let value: Option<Vec<u8>> =
+ sqlx::query_scalar(r#"SELECT "$3" FROM "$1" WHERE "$2" = "$4"
LIMIT 1"#)
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(self.value_field.clone())
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ .map_err(parse_sqlite_error)?;
Ok(value.map(Buffer::from))
}
@@ -63,10 +79,12 @@ impl SqliteCore {
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 LENGTH(CAST(`{}` AS BLOB)) FROM `{}` WHERE `{}` = $1 LIMIT
1",
- self.value_field, self.table, self.key_field
- ))
+ let value: Option<i64> = sqlx::query_scalar(
+ r#"SELECT LENGTH(CAST("$3" AS BLOB)) FROM "$1" WHERE "$2" = '$4'
LIMIT 1"#,
+ )
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(self.value_field.clone())
.bind(path)
.fetch_optional(pool)
.await
@@ -82,6 +100,18 @@ impl SqliteCore {
.transpose()
}
+ pub async fn count_under(&self, path: &str) -> Result<i64> {
+ let pool = self.get_client().await?;
+
+ sqlx::query_scalar(r#"SELECT COUNT(*) FROM "$1" WHERE "$2" LIKE "$3"
LIMIT 1"#)
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(format!("{}%", path))
+ .fetch_one(pool)
+ .await
+ .map_err(parse_sqlite_error)
+ }
+
pub async fn get_range(
&self,
path: &str,
@@ -90,46 +120,47 @@ impl SqliteCore {
) -> 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
- ),
+ Some(limit) => sqlx::query_as(
+ r#"SELECT SUBSTR(CAST("$3" AS BLOB), $4, $5), LENGTH(CAST("$3"
AS BLOB)) FROM "$1" WHERE "$2" = '$6' LIMIT 1"#,
+ )
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(self.value_field.clone())
+ .bind(start+1)
+ .bind(limit)
+ .bind(path)
+ ,
+ None => sqlx::query_as(
+ r#"SELECT SUBSTR(CAST("$3" AS BLOB), $4), LENGTH(CAST("$3" AS
BLOB)) FROM "$1" WHERE "$2" = '$5' LIMIT 1"#,
+ )
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(self.value_field.clone())
+ .bind(start+1)
+ .bind(path)
+ ,
};
- let value: Option<(Vec<u8>, i64)> = sqlx::query_as(&query)
- .bind(path)
+ let value: Option<SqliteBlobSized> = query
.fetch_optional(pool)
.await
.map_err(parse_sqlite_error)?;
- Ok(value.map(|(bs, size)| (Buffer::from(bs), size as u64)))
+ Ok(value.map(|s| (Buffer::from(s.bs), s.size as u64)))
}
pub async fn set(&self, path: &str, value: Buffer) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(&format!(
- "INSERT OR REPLACE INTO `{}` (`{}`, `{}`) VALUES ($1, $2)",
- self.table, self.key_field, self.value_field,
- ))
- .bind(path)
- .bind(value.to_vec())
- .execute(pool)
- .await
- .map_err(parse_sqlite_error)?;
+ sqlx::query(r#"INSERT OR REPLACE INTO "$1" ("$2", "$3") VALUES ('$4',
$5)"#)
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(self.value_field.clone())
+ .bind(path)
+ .bind(value.to_vec())
+ .execute(pool)
+ .await
+ .map_err(parse_sqlite_error)?;
Ok(())
}
@@ -137,14 +168,13 @@ impl SqliteCore {
pub async fn delete(&self, path: &str) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(&format!(
- "DELETE FROM `{}` WHERE `{}` = $1",
- self.table, self.key_field
- ))
- .bind(path)
- .execute(pool)
- .await
- .map_err(parse_sqlite_error)?;
+ sqlx::query(r#"DELETE FROM "$1" WHERE "$2" = '$3'"#)
+ .bind(self.table.clone())
+ .bind(self.key_field.clone())
+ .bind(path)
+ .execute(pool)
+ .await
+ .map_err(parse_sqlite_error)?;
Ok(())
}
