[core] Unify the object storage User-Agent format

crates/paimon/Cargo.toml

@@ -129,6 +129,8 @@ parquet = { workspace = true, features = ["async", "zstd", "lz4", "snap"] }
 orc-rust = "0.8.0"
 async-stream = "0.3.6"
 reqwest = { version = "0.12", features = ["json", "gzip", "deflate"] }
+# Already in the tree via `reqwest`; direct dep for the object storage User-Agent transport.
+http = "1"
 # DLF authentication dependencies
 base64 = "0.22"
 hex = "0.4"

crates/paimon/src/io/mod.rs

@@ -31,14 +31,23 @@ pub use storage::*;
     feature = "storage-oss",
     feature = "storage-s3"
 ))]
-fn with_http_transport(op: opendal::Operator) -> opendal::Operator {
+fn with_http_transport(op: opendal::Operator, user_agent: &str) -> opendal::Operator {
     use opendal::{HttpTransporter, OperationContext};
-    use opendal_http_transport_reqwest::ReqwestTransport;
 
-    let transport = HttpTransporter::new(ReqwestTransport::default());
+    let transport = HttpTransporter::new(user_agent::UserAgentTransport::new(user_agent));
     op.with_context(OperationContext::new().with_http_transport(transport))
 }
 
+#[cfg(any(
+    feature = "storage-azdls",
+    feature = "storage-cos",
+    feature = "storage-gcs",
+    feature = "storage-obs",
+    feature = "storage-oss",
+    feature = "storage-s3"
+))]
+mod user_agent;
+
 #[cfg(any(
     feature = "storage-s3",
     feature = "storage-cos",

crates/paimon/src/io/storage.rs

@@ -83,26 +83,31 @@ pub enum Storage {
     #[cfg(feature = "storage-s3")]
     S3 {
         config: Box<S3Config>,
+        user_agent: String,
         operators: Mutex<HashMap<String, Operator>>,
     },
     #[cfg(feature = "storage-cos")]
     Cos {
         config: Box<CosConfig>,
+        user_agent: String,
         operators: Mutex<HashMap<String, Operator>>,
     },
     #[cfg(feature = "storage-azdls")]
     Azdls {
         config: Box<AzdlsStorageConfig>,
+        user_agent: String,
         operators: Mutex<HashMap<String, Operator>>,
     },
     #[cfg(feature = "storage-obs")]
     Obs {
         config: Box<ObsConfig>,
+        user_agent: String,
         operators: Mutex<HashMap<String, Operator>>,
     },
     #[cfg(feature = "storage-gcs")]
     Gcs {
         config: Box<GcsConfig>,
+        user_agent: String,
         operators: Mutex<HashMap<String, Operator>>,
     },
     #[cfg(feature = "storage-hdfs")]
@@ -138,41 +143,51 @@ impl Storage {
             }
             #[cfg(feature = "storage-s3")]
             "s3" | "s3a" => {
+                let user_agent = super::user_agent::storage_user_agent(&props);
                 let config = super::s3_config_parse(props)?;
                 Ok(Self::S3 {
                     config: Box::new(config),
+                    user_agent,
                     operators: Mutex::new(HashMap::new()),
                 })
             }
             #[cfg(feature = "storage-cos")]
             "cos" | "cosn" => {
+                let user_agent = super::user_agent::storage_user_agent(&props);
                 let config = super::cos_config_parse(props)?;
                 Ok(Self::Cos {
                     config: Box::new(config),
+                    user_agent,
                     operators: Mutex::new(HashMap::new()),
                 })
             }
             #[cfg(feature = "storage-azdls")]
             "abfs" | "abfss" | "az" | "azdfs" | "azdls" | "azure" => {
+                let user_agent = super::user_agent::storage_user_agent(&props);
                 let config = super::azdls_config_parse(props)?;
                 Ok(Self::Azdls {
                     config: Box::new(config),
+                    user_agent,
                     operators: Mutex::new(HashMap::new()),
                 })
             }
             #[cfg(feature = "storage-obs")]
             "obs" => {
+                let user_agent = super::user_agent::storage_user_agent(&props);
                 let config = super::obs_config_parse(props)?;
                 Ok(Self::Obs {
                     config: Box::new(config),
+                    user_agent,
                     operators: Mutex::new(HashMap::new()),
                 })
             }
             #[cfg(feature = "storage-gcs")]
             "gcs" | "gs" => {
+                let user_agent = super::user_agent::storage_user_agent(&props);
                 let config = super::gcs_config_parse(props)?;
                 Ok(Self::Gcs {
                     config: Box::new(config),
+                    user_agent,
                     operators: Mutex::new(HashMap::new()),
                 })
             }
@@ -222,45 +237,65 @@ impl Storage {
                 Ok((op, Cow::Borrowed(relative_path)))
             }
             #[cfg(feature = "storage-s3")]
-            Storage::S3 { config, operators } => {
+            Storage::S3 {
+                config,
+                user_agent,
+                operators,
+            } => {
                 let (bucket, relative_path) =
                     Self::bucket_and_relative_path(path, "S3", &["s3", "s3a"])?;
-                let op = Self::cached_s3_operator(config, operators, path, &bucket)?;
+                let op = Self::cached_s3_operator(config, user_agent, operators, path, &bucket)?;
                 Ok((op, Cow::Borrowed(relative_path)))
             }
             #[cfg(feature = "storage-cos")]
-            Storage::Cos { config, operators } => {
+            Storage::Cos {
+                config,
+                user_agent,
+                operators,
+            } => {
                 let (bucket, relative_path) =
                     Self::bucket_and_relative_path(path, "COS", &["cos", "cosn"])?;
                 let op = Self::cached_operator(operators, "COS", &bucket, || {
-                    super::cos_config_build(config, path)
+                    super::cos_config_build(config, path, user_agent)
                 })?;
                 Ok((op, Cow::Borrowed(relative_path)))
             }
             #[cfg(feature = "storage-azdls")]
-            Storage::Azdls { config, operators } => {
+            Storage::Azdls {
+                config,
+                user_agent,
+                operators,
+            } => {
                 let relative_path = super::azdls_relative_path(path)?;
                 let cache_key = super::azdls_operator_cache_key(config, path)?;
                 let op = Self::cached_operator(operators, "Azure", &cache_key, || {
-                    super::azdls_config_build(config, path)
+                    super::azdls_config_build(config, path, user_agent)
                 })?;
                 Ok((op, Cow::Borrowed(relative_path)))
             }
             #[cfg(feature = "storage-obs")]
-            Storage::Obs { config, operators } => {
+            Storage::Obs {
+                config,
+                user_agent,
+                operators,
+            } => {
                 let (bucket, relative_path) =
                     Self::bucket_and_relative_path(path, "OBS", &["obs"])?;
                 let op = Self::cached_operator(operators, "OBS", &bucket, || {
-                    super::obs_config_build(config, path)
+                    super::obs_config_build(config, path, user_agent)
                 })?;
                 Ok((op, Cow::Borrowed(relative_path)))
             }
             #[cfg(feature = "storage-gcs")]
-            Storage::Gcs { config, operators } => {
+            Storage::Gcs {
+                config,
+                user_agent,
+                operators,
+            } => {
                 let (bucket, relative_path) =
                     Self::bucket_and_relative_path(path, "GCS", &["gcs", "gs"])?;
                 let op = Self::cached_operator(operators, "GCS", &bucket, || {
-                    super::gcs_config_build(config, path)
+                    super::gcs_config_build(config, path, user_agent)
                 })?;
                 Ok((op, Cow::Borrowed(relative_path)))
             }
@@ -425,12 +460,13 @@ impl Storage {
     #[cfg(feature = "storage-s3")]
     fn cached_s3_operator(
         config: &S3Config,
+        user_agent: &str,
         operators: &Mutex<HashMap<String, Operator>>,
         path: &str,
         bucket: &str,
     ) -> crate::Result<Operator> {
         Self::cached_operator(operators, "S3", bucket, || {
-            super::s3_config_build(config, path)
+            super::s3_config_build(config, path, user_agent)
         })
     }
 }
@@ -482,6 +518,28 @@ mod scheme_tests {
         }
     }
 
+    #[cfg(feature = "storage-s3")]
+    #[test]
+    fn s3_storage_uses_generic_user_agent_keys() {
+        let storage = Storage::build(FileIOBuilder::new("s3").with_props([
+            ("user-agent.features", "Flink"),
+            ("user-agent.extended", "vvr"),
+            ("dlf.access-tracking.extended-info", "uid/123"),
+        ]))
+        .unwrap();
+        let Storage::S3 { user_agent, .. } = storage else {
+            panic!("expected S3 storage");
+        };
+        assert_eq!(
+            user_agent,
+            format!(
+                "paimon-rust/{}(opendal/{};Flink) vvr uid/123",
+                env!("CARGO_PKG_VERSION"),
+                opendal::raw::VERSION
+            )
+        );
+    }
+
     #[cfg(feature = "storage-cos")]
     #[test]
     fn cos_scheme_aliases_are_compatible() {

crates/paimon/src/io/storage_azdls.rs

@@ -64,11 +64,15 @@ pub(crate) fn azdls_config_parse(props: HashMap<String, String>) -> Result<Azdls
     Ok(AzdlsStorageConfig { config, normalized })
 }
 
-pub(crate) fn azdls_config_build(cfg: &AzdlsStorageConfig, path: &str) -> Result<Operator> {
+pub(crate) fn azdls_config_build(
+    cfg: &AzdlsStorageConfig,
+    path: &str,
+    user_agent: &str,
+) -> Result<Operator> {
     let (cfg, relative_path) = azdls_config_for_path(cfg, path)?;
 
     let builder = cfg.into_builder();
-    let op = super::with_http_transport(Operator::new(builder)?);
+    let op = super::with_http_transport(Operator::new(builder)?, user_agent);
 
     debug_assert_eq!(
         relative_path,
@@ -394,14 +398,19 @@ mod tests {
     fn test_azdls_config_build_hadoop_form() {
         let cfg = azdls_config_parse(HashMap::new()).unwrap();
 
-        let op = azdls_config_build(&cfg, "abfs://fs@account.dfs.core.windows.net/a/b").unwrap();
+        let op = azdls_config_build(
+            &cfg,
+            "abfs://fs@account.dfs.core.windows.net/a/b",
+            "paimon-rust/test",
+        )
+        .unwrap();
         assert_eq!(op.info().name(), "fs");
     }
 
     #[test]
     fn test_azdls_config_build_fsspec_form_requires_endpoint() {
         let cfg = azdls_config_parse(HashMap::new()).unwrap();
-        let result = azdls_config_build(&cfg, "abfs://fs/a/b");
+        let result = azdls_config_build(&cfg, "abfs://fs/a/b", "paimon-rust/test");
         assert!(result.is_err());
     }
 
@@ -413,7 +422,7 @@ mod tests {
         )]))
         .unwrap();
 
-        let op = azdls_config_build(&cfg, "abfs://fs/a/b").unwrap();
+        let op = azdls_config_build(&cfg, "abfs://fs/a/b", "paimon-rust/test").unwrap();
         assert_eq!(op.info().name(), "fs");
     }
 
@@ -439,7 +448,8 @@ mod tests {
     #[test]
     fn test_azdls_config_build_missing_filesystem() {
         let cfg = azdls_config_parse(HashMap::new()).unwrap();
-        let result = azdls_config_build(&cfg, "abfs:///path/without/filesystem");
+        let result =
+            azdls_config_build(&cfg, "abfs:///path/without/filesystem", "paimon-rust/test");
         assert!(result.is_err());
     }
 }

crates/paimon/src/io/storage_cos.rs

@@ -59,7 +59,7 @@ pub(crate) fn cos_config_parse(props: HashMap<String, String>) -> Result<CosConf
     Ok(cfg)
 }
 
-pub(crate) fn cos_config_build(cfg: &CosConfig, path: &str) -> Result<Operator> {
+pub(crate) fn cos_config_build(cfg: &CosConfig, path: &str, user_agent: &str) -> Result<Operator> {
     let url = Url::parse(path).map_err(|_| Error::ConfigInvalid {
         message: format!("Invalid COS url: {path}"),
     })?;
@@ -69,7 +69,10 @@ pub(crate) fn cos_config_build(cfg: &CosConfig, path: &str) -> Result<Operator>
     })?;
 
     let builder = cfg.clone().into_builder().bucket(bucket);
-    Ok(super::with_http_transport(Operator::new(builder)?))
+    Ok(super::with_http_transport(
+        Operator::new(builder)?,
+        user_agent,
+    ))
 }
 
 #[cfg(test)]
@@ -124,14 +127,14 @@ mod tests {
             ..Default::default()
         };
 
-        let op = cos_config_build(&cfg, "cosn://my-bucket/some/path").unwrap();
+        let op = cos_config_build(&cfg, "cosn://my-bucket/some/path", "paimon-rust/test").unwrap();
         assert_eq!(op.info().name(), "my-bucket");
     }
 
     #[test]
     fn test_cos_config_build_missing_bucket() {
         let cfg = CosConfig::default();
-        let result = cos_config_build(&cfg, "cosn:///path/without/bucket");
+        let result = cos_config_build(&cfg, "cosn:///path/without/bucket", "paimon-rust/test");
         assert!(result.is_err());
     }
 }

crates/paimon/src/io/storage_gcs.rs

@@ -81,7 +81,7 @@ pub(crate) fn gcs_config_parse(props: HashMap<String, String>) -> Result<GcsConf
     Ok(cfg)
 }
 
-pub(crate) fn gcs_config_build(cfg: &GcsConfig, path: &str) -> Result<Operator> {
+pub(crate) fn gcs_config_build(cfg: &GcsConfig, path: &str, user_agent: &str) -> Result<Operator> {
     let url = Url::parse(path).map_err(|_| Error::ConfigInvalid {
         message: format!("Invalid GCS url: {path}"),
     })?;
@@ -91,7 +91,10 @@ pub(crate) fn gcs_config_build(cfg: &GcsConfig, path: &str) -> Result<Operator>
     })?;
 
     let builder = cfg.clone().into_builder().bucket(bucket);
-    Ok(super::with_http_transport(Operator::new(builder)?))
+    Ok(super::with_http_transport(
+        Operator::new(builder)?,
+        user_agent,
+    ))
 }
 
 #[cfg(test)]
@@ -185,14 +188,14 @@ mod tests {
     fn test_gcs_config_build_extracts_bucket() {
         let cfg = GcsConfig::default();
 
-        let op = gcs_config_build(&cfg, "gs://my-bucket/some/path").unwrap();
+        let op = gcs_config_build(&cfg, "gs://my-bucket/some/path", "paimon-rust/test").unwrap();
         assert_eq!(op.info().name(), "my-bucket");
     }
 
     #[test]
     fn test_gcs_config_build_missing_bucket() {
         let cfg = GcsConfig::default();
-        let result = gcs_config_build(&cfg, "gs:///path/without/bucket");
+        let result = gcs_config_build(&cfg, "gs:///path/without/bucket", "paimon-rust/test");
         assert!(result.is_err());
     }
 }