feat!(runtime): Support custom Runtime in Catalog

crates/catalog/glue/src/catalog.rs

@@ -33,7 +33,7 @@ use iceberg::spec::{TableMetadata, TableMetadataBuilder};
 use iceberg::table::Table;
 use iceberg::{
     Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
-    TableCommit, TableCreation, TableIdent,
+    Runtime, TableCommit, TableCreation, TableIdent,
 };
 use iceberg_storage_opendal::OpenDalStorageFactory;
 
@@ -58,6 +58,7 @@ pub const GLUE_CATALOG_PROP_WAREHOUSE: &str = "warehouse";
 pub struct GlueCatalogBuilder {
     config: GlueCatalogConfig,
     storage_factory: Option<Arc<dyn StorageFactory>>,
+    runtime: Runtime,
 }
 
 impl Default for GlueCatalogBuilder {
@@ -71,6 +72,7 @@ impl Default for GlueCatalogBuilder {
                 props: HashMap::new(),
             },
             storage_factory: None,
+            runtime: Runtime::default(),
         }
     }
 }
@@ -83,6 +85,11 @@ impl CatalogBuilder for GlueCatalogBuilder {
         self
     }
 
+    fn with_runtime(mut self, runtime: Runtime) -> Self {
+        self.runtime = runtime;
+        self
+    }
+
     fn load(
         mut self,
         name: impl Into<String>,
@@ -129,7 +136,7 @@ impl CatalogBuilder for GlueCatalogBuilder {
                 ));
             }
 
-            GlueCatalog::new(self.config, self.storage_factory).await
+            GlueCatalog::new(self.config, self.storage_factory, self.runtime).await
         }
     }
 }
@@ -151,6 +158,7 @@ pub struct GlueCatalog {
     config: GlueCatalogConfig,
     client: GlueClient,
     file_io: FileIO,
+    runtime: Runtime,
 }
 
 impl Debug for GlueCatalog {
@@ -166,6 +174,7 @@ impl GlueCatalog {
     async fn new(
         config: GlueCatalogConfig,
         storage_factory: Option<Arc<dyn StorageFactory>>,
+        runtime: Runtime,
     ) -> Result<Self> {
         let sdk_config = create_sdk_config(&config.props, config.uri.as_ref()).await;
         let mut file_io_props = config.props.clone();
@@ -215,6 +224,7 @@ impl GlueCatalog {
             config,
             client: GlueClient(client),
             file_io,
+            runtime,
         })
     }
     /// Get the catalogs `FileIO`
@@ -273,6 +283,7 @@ impl GlueCatalog {
                 NamespaceIdent::new(db_name),
                 table_name.to_owned(),
             ))
+            .runtime(self.runtime.clone())
             .build()?;
 
         Ok((table, version_id))
@@ -612,6 +623,7 @@ impl Catalog for GlueCatalog {
             .metadata_location(metadata_location_str)
             .metadata(metadata)
             .identifier(TableIdent::new(NamespaceIdent::new(db_name), table_name))
+            .runtime(self.runtime.clone())
             .build()
     }
 
@@ -846,6 +858,7 @@ impl Catalog for GlueCatalog {
             .metadata_location(metadata_location)
             .metadata(metadata)
             .file_io(self.file_io())
+            .runtime(self.runtime.clone())
             .build()?)
     }
 

crates/catalog/hms/src/catalog.rs

@@ -31,7 +31,7 @@ use iceberg::spec::{TableMetadata, TableMetadataBuilder};
 use iceberg::table::Table;
 use iceberg::{
     Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
-    TableCommit, TableCreation, TableIdent,
+    Runtime, TableCommit, TableCreation, TableIdent,
 };
 use volo_thrift::MaybeException;
 
@@ -56,6 +56,7 @@ pub const HMS_CATALOG_PROP_WAREHOUSE: &str = "warehouse";
 pub struct HmsCatalogBuilder {
     config: HmsCatalogConfig,
     storage_factory: Option<Arc<dyn StorageFactory>>,
+    runtime: Runtime,
 }
 
 impl Default for HmsCatalogBuilder {
@@ -69,6 +70,7 @@ impl Default for HmsCatalogBuilder {
                 props: HashMap::new(),
             },
             storage_factory: None,
+            runtime: Runtime::default(),
         }
     }
 }
@@ -81,6 +83,11 @@ impl CatalogBuilder for HmsCatalogBuilder {
         self
     }
 
+    fn with_runtime(mut self, runtime: Runtime) -> Self {
+        self.runtime = runtime;
+        self
+    }
+
     fn load(
         mut self,
         name: impl Into<String>,
@@ -133,7 +140,7 @@ impl CatalogBuilder for HmsCatalogBuilder {
                     "Catalog warehouse is required",
                 ))
             } else {
-                HmsCatalog::new(self.config, self.storage_factory)
+                HmsCatalog::new(self.config, self.storage_factory, self.runtime)
             }
         };
 
@@ -169,6 +176,7 @@ pub struct HmsCatalog {
     config: HmsCatalogConfig,
     client: HmsClient,
     file_io: FileIO,
+    runtime: Runtime,
 }
 
 impl Debug for HmsCatalog {
@@ -184,6 +192,7 @@ impl HmsCatalog {
     fn new(
         config: HmsCatalogConfig,
         storage_factory: Option<Arc<dyn StorageFactory>>,
+        runtime: Runtime,
     ) -> Result<Self> {
         let address = config
             .address
@@ -223,6 +232,7 @@ impl HmsCatalog {
             config,
             client: HmsClient(client),
             file_io,
+            runtime,
         })
     }
     /// Get the catalogs `FileIO`
@@ -529,6 +539,7 @@ impl Catalog for HmsCatalog {
             .metadata_location(metadata_location_str)
             .metadata(metadata)
             .identifier(TableIdent::new(NamespaceIdent::new(db_name), table_name))
+            .runtime(self.runtime.clone())
             .build()
     }
 
@@ -567,6 +578,7 @@ impl Catalog for HmsCatalog {
                 NamespaceIdent::new(db_name),
                 table.name.clone(),
             ))
+            .runtime(self.runtime.clone())
             .build()
     }