feat: add mysql object store backend (#8560)

* feat: add mysql object store backend

Signed-off-by: fys <fengys1996@gmail.com>

* fix: code review

Signed-off-by: fys <fengys1996@gmail.com>

* add feature guard

---------

Signed-off-by: fys <fengys1996@gmail.com>

Cargo.lock

@@ -9295,6 +9295,7 @@ dependencies = [
  "snafu 0.8.6",
  "tempfile",
  "tokio",
+ "toml 0.8.23",
  "uuid",
 ]
 
@@ -9431,6 +9432,7 @@ dependencies = [
  "opendal-service-fs",
  "opendal-service-gcs",
  "opendal-service-http",
+ "opendal-service-mysql",
  "opendal-service-oss",
  "opendal-service-s3",
 ]
@@ -9618,6 +9620,18 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "opendal-service-mysql"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5cb3364a16d32aeb9c9c3232a81648926a5a8efa680f3316e0743ca2d5a37744"
+dependencies = [
+ "mea",
+ "opendal-core",
+ "serde",
+ "sqlx",
+]
+
 [[package]]
 name = "opendal-service-oss"
 version = "0.57.0"

src/cmd/Cargo.toml

@@ -25,6 +25,7 @@ enterprise = ["common-meta/enterprise", "frontend/enterprise", "meta-srv/enterpr
 # Developer-only helper binary gate for `query_perf_fixture`.
 # Kept out of `default` so normal/release builds don't compile them.
 dev-tools = []
+mysql-object-store = ["object-store/mysql-object-store"]
 tokio-console = ["common-telemetry/tokio-console"]
 vector_index = ["mito2/vector_index", "query/vector_index"]
 

src/object-store/Cargo.toml

@@ -8,6 +8,7 @@ license.workspace = true
 workspace = true
 
 [features]
+mysql-object-store = ["opendal/services-mysql"]
 services-memory = ["opendal/services-memory"]
 testing = ["derive_builder"]
 
@@ -49,3 +50,4 @@ object_store_opendal.workspace = true
 rand.workspace = true
 tempfile.workspace = true
 tokio.workspace = true
+toml.workspace = true

src/object-store/src/config.rs

@@ -16,6 +16,8 @@ use std::time::Duration;
 
 use common_base::readable_size::ReadableSize;
 use common_base::secrets::{ExposeSecret, SecretString};
+#[cfg(feature = "mysql-object-store")]
+use opendal::services::Mysql;
 use opendal::services::{Azblob, Gcs, Oss, S3};
 use serde::{Deserialize, Serialize};
 
@@ -32,6 +34,8 @@ pub enum ObjectStoreConfig {
     Oss(OssConfig),
     Azblob(AzblobConfig),
     Gcs(GcsConfig),
+    #[cfg(feature = "mysql-object-store")]
+    Mysql(MysqlConfig),
 }
 
 impl Default for ObjectStoreConfig {
@@ -49,6 +53,8 @@ impl ObjectStoreConfig {
             Self::Oss(_) => "Oss",
             Self::Azblob(_) => "Azblob",
             Self::Gcs(_) => "Gcs",
+            #[cfg(feature = "mysql-object-store")]
+            Self::Mysql(_) => "Mysql",
         }
     }
 
@@ -66,6 +72,8 @@ impl ObjectStoreConfig {
             Self::Oss(oss) => &oss.name,
             Self::Azblob(az) => &az.name,
             Self::Gcs(gcs) => &gcs.name,
+            #[cfg(feature = "mysql-object-store")]
+            Self::Mysql(mysql) => &mysql.name,
         };
 
         if name.trim().is_empty() {
@@ -83,6 +91,8 @@ impl ObjectStoreConfig {
             Self::Oss(oss) => Some(&oss.cache),
             Self::Azblob(az) => Some(&az.cache),
             Self::Gcs(gcs) => Some(&gcs.cache),
+            #[cfg(feature = "mysql-object-store")]
+            Self::Mysql(mysql) => Some(&mysql.cache),
         }
     }
 
@@ -94,6 +104,8 @@ impl ObjectStoreConfig {
             Self::Oss(oss) => Some(&mut oss.cache),
             Self::Azblob(az) => Some(&mut az.cache),
             Self::Gcs(gcs) => Some(&mut gcs.cache),
+            #[cfg(feature = "mysql-object-store")]
+            Self::Mysql(mysql) => Some(&mut mysql.cache),
         }
     }
 }
@@ -276,6 +288,40 @@ impl From<&GcsConnection> for Gcs {
             .endpoint(&connection.endpoint)
     }
 }
+
+#[cfg(feature = "mysql-object-store")]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
+#[serde(default)]
+pub struct MysqlConfig {
+    pub name: String,
+    pub root: String,
+    #[serde(skip_serializing)]
+    pub connection_string: SecretString,
+    pub table: Option<String>,
+    #[serde(flatten)]
+    pub cache: ObjectStorageCacheConfig,
+}
+
+#[cfg(feature = "mysql-object-store")]
+impl From<&MysqlConfig> for Mysql {
+    fn from(config: &MysqlConfig) -> Self {
+        let root = util::normalize_dir(&config.root);
+        let mut builder = Mysql::default()
+            .connection_string(config.connection_string.expose_secret())
+            .root(&root)
+            .key_field("key")
+            .value_field("value");
+
+        if let Some(table) = &config.table {
+            builder = builder.table(table);
+        } else {
+            builder = builder.table("greptime");
+        }
+
+        builder
+    }
+}
+
 /// The http client options to the storage.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(default)]
@@ -364,6 +410,20 @@ mod tests {
         });
         assert_eq!("test", s3_config.config_name());
         assert_eq!("S3", s3_config.provider_name());
+
+        #[cfg(feature = "mysql-object-store")]
+        {
+            let mysql_config = ObjectStoreConfig::Mysql(MysqlConfig::default());
+            assert_eq!("Mysql", mysql_config.config_name());
+            assert_eq!("Mysql", mysql_config.provider_name());
+
+            let mysql_config = ObjectStoreConfig::Mysql(MysqlConfig {
+                name: "test".to_string(),
+                ..Default::default()
+            });
+            assert_eq!("test", mysql_config.config_name());
+            assert_eq!("Mysql", mysql_config.provider_name());
+        }
     }
 
     #[test]
@@ -378,5 +438,41 @@ mod tests {
         assert!(gcs_config.is_object_storage());
         let azblob_config = ObjectStoreConfig::Azblob(AzblobConfig::default());
         assert!(azblob_config.is_object_storage());
+        #[cfg(feature = "mysql-object-store")]
+        {
+            let mysql_config = ObjectStoreConfig::Mysql(MysqlConfig::default());
+            assert!(mysql_config.is_object_storage());
+        }
+    }
+
+    #[cfg(feature = "mysql-object-store")]
+    #[test]
+    fn test_mysql_config_connection_string_serde() {
+        let config: ObjectStoreConfig = toml::from_str(
+            r#"
+type = "Mysql"
+name = "mysql-store"
+root = "/greptimedb"
+connection_string = "mysql://user:password@127.0.0.1:3306/greptime"
+table = "object_store"
+"#,
+        )
+        .unwrap();
+
+        let ObjectStoreConfig::Mysql(mysql_config) = config else {
+            unreachable!()
+        };
+
+        assert_eq!("mysql-store", mysql_config.name);
+        assert_eq!("/greptimedb", mysql_config.root);
+        assert_eq!(
+            "mysql://user:password@127.0.0.1:3306/greptime",
+            mysql_config.connection_string.expose_secret()
+        );
+        assert_eq!(Some("object_store"), mysql_config.table.as_deref());
+
+        let serialized = toml::to_string(&mysql_config).unwrap();
+        assert!(!serialized.contains("connection_string"));
+        assert!(!serialized.contains("password"));
     }
 }

src/object-store/src/factory.rs

@@ -16,9 +16,13 @@ use std::{fs, path};
 
 use common_telemetry::info;
 use opendal::layers::HttpClientLayer;
+#[cfg(feature = "mysql-object-store")]
+use opendal::services::Mysql;
 use opendal::services::{Fs, Gcs, Oss, S3};
 use snafu::prelude::*;
 
+#[cfg(feature = "mysql-object-store")]
+use crate::config::MysqlConfig;
 use crate::config::{AzblobConfig, FileConfig, GcsConfig, ObjectStoreConfig, OssConfig, S3Config};
 use crate::error::{self, Result};
 use crate::services::Azblob;
@@ -36,9 +40,28 @@ pub async fn new_raw_object_store(
         ObjectStoreConfig::Oss(oss_config) => new_oss_object_store(oss_config).await,
         ObjectStoreConfig::Azblob(azblob_config) => new_azblob_object_store(azblob_config).await,
         ObjectStoreConfig::Gcs(gcs_config) => new_gcs_object_store(gcs_config).await,
+        #[cfg(feature = "mysql-object-store")]
+        ObjectStoreConfig::Mysql(mysql_config) => new_mysql_object_store(mysql_config).await,
     }
 }
 
+#[cfg(feature = "mysql-object-store")]
+pub async fn new_mysql_object_store(mysql_config: &MysqlConfig) -> Result<ObjectStore> {
+    let root = util::normalize_dir(&mysql_config.root);
+    info!(
+        "The mysql object storage table is: {}, root is: {}",
+        mysql_config.table.as_deref().unwrap_or("greptime"),
+        root
+    );
+
+    let builder = Mysql::from(mysql_config);
+    let operator = ObjectStore::new(builder)
+        .context(error::InitBackendSnafu)?
+        .finish();
+
+    Ok(operator)
+}
+
 /// A helper function to create a file system object store.
 pub fn new_fs_object_store(data_home: &str, _file_config: &FileConfig) -> Result<ObjectStore> {
     fs::create_dir_all(path::Path::new(&data_home))