diff --git a/Cargo.lock b/Cargo.lock index 159f8242fa..9e61b00d77 100644 --- a/Cargo.lock +++ b/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" diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml index b06da6e517..8002ab742f 100644 --- a/src/cmd/Cargo.toml +++ b/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"] diff --git a/src/object-store/Cargo.toml b/src/object-store/Cargo.toml index 9ca68bb780..18b246ec8e 100644 --- a/src/object-store/Cargo.toml +++ b/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 diff --git a/src/object-store/src/config.rs b/src/object-store/src/config.rs index 5b408a9b34..1308db92fa 100644 --- a/src/object-store/src/config.rs +++ b/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, + #[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")); } } diff --git a/src/object-store/src/factory.rs b/src/object-store/src/factory.rs index 2ac38e00ed..005817d269 100644 --- a/src/object-store/src/factory.rs +++ b/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 { + 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 { fs::create_dir_all(path::Path::new(&data_home))