chore: hms/glue catalog create table should respect default location (#1302)

## What changes are included in this PR?

The glue/hms catalog will throw error `Can't create table without
location` if table did not have location option.

We need to inject the default location before
`TableMetadataBuilder::from_table_creation`

## Are these changes tested?

unit test

Author: sundy-li

crates/catalog/glue/src/catalog.rs

@@ -172,7 +172,7 @@ impl Catalog for GlueCatalog {
     ///
     /// - Errors from `validate_namespace` if the namespace identifier does not
     /// meet validation criteria.
-    /// - Errors from `convert_to_database` if the properties cannot be  
+    /// - Errors from `convert_to_database` if the properties cannot be
     /// successfully converted into a database configuration.
     /// - Errors from the underlying database creation process, converted using
     /// `from_sdk_error`.
@@ -259,7 +259,7 @@ impl Catalog for GlueCatalog {
     /// Asynchronously updates properties of an existing namespace.
     ///
     /// Converts the given namespace identifier and properties into a database
-    /// representation and then attempts to update the corresponding namespace  
+    /// representation and then attempts to update the corresponding namespace
     /// in the Glue Catalog.
     ///
     /// # Returns
@@ -295,7 +295,7 @@ impl Catalog for GlueCatalog {
     /// # Returns
     /// A `Result<()>` indicating the outcome:
     /// - `Ok(())` signifies successful namespace deletion.
-    /// - `Err(...)` signifies failure to drop the namespace due to validation  
+    /// - `Err(...)` signifies failure to drop the namespace due to validation
     /// errors, connectivity issues, or Glue Catalog constraints.
     async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
         let db_name = validate_namespace(namespace)?;
@@ -322,7 +322,7 @@ impl Catalog for GlueCatalog {
     /// A `Result<Vec<TableIdent>>`, which is:
     /// - `Ok(vec![...])` containing a vector of `TableIdent` instances, each
     /// representing a table within the specified namespace.
-    /// - `Err(...)` if an error occurs during namespace validation or while  
+    /// - `Err(...)` if an error occurs during namespace validation or while
     /// querying the database.
     async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
         let db_name = validate_namespace(namespace)?;
@@ -375,7 +375,7 @@ impl Catalog for GlueCatalog {
     async fn create_table(
         &self,
         namespace: &NamespaceIdent,
-        creation: TableCreation,
+        mut creation: TableCreation,
     ) -> Result<Table> {
         let db_name = validate_namespace(namespace)?;
         let table_name = creation.name.clone();
@@ -384,10 +384,12 @@ impl Catalog for GlueCatalog {
             Some(location) => location.clone(),
             None => {
                 let ns = self.get_namespace(namespace).await?;
-                get_default_table_location(&ns, &db_name, &table_name, &self.config.warehouse)
+                let location =
+                    get_default_table_location(&ns, &db_name, &table_name, &self.config.warehouse);
+                creation.location = Some(location.clone());
+                location
             }
         };
-
         let metadata = TableMetadataBuilder::from_table_creation(creation)?
             .build()?
             .metadata;

crates/catalog/glue/tests/glue_catalog_test.rs

@@ -105,7 +105,7 @@ async fn set_test_namespace(catalog: &GlueCatalog, namespace: &NamespaceIdent) -
     Ok(())
 }
 
-fn set_table_creation(location: impl ToString, name: impl ToString) -> Result<TableCreation> {
+fn set_table_creation(location: Option<String>, name: impl ToString) -> Result<TableCreation> {
     let schema = Schema::builder()
         .with_schema_id(0)
         .with_fields(vec![
@@ -114,20 +114,19 @@ fn set_table_creation(location: impl ToString, name: impl ToString) -> Result<Ta
         ])
         .build()?;
 
-    let creation = TableCreation::builder()
-        .location(location.to_string())
+    let builder = TableCreation::builder()
         .name(name.to_string())
         .properties(HashMap::new())
-        .schema(schema)
-        .build();
+        .location_opt(location)
+        .schema(schema);
 
-    Ok(creation)
+    Ok(builder.build())
 }
 
 #[tokio::test]
 async fn test_rename_table() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation = set_table_creation(None, "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_rename_table".into()));
 
     catalog
@@ -154,7 +153,7 @@ async fn test_rename_table() -> Result<()> {
 #[tokio::test]
 async fn test_table_exists() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation = set_table_creation(None, "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_table_exists".into()));
 
     catalog
@@ -178,7 +177,7 @@ async fn test_table_exists() -> Result<()> {
 #[tokio::test]
 async fn test_drop_table() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation = set_table_creation(None, "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_drop_table".into()));
 
     catalog
@@ -199,7 +198,7 @@ async fn test_drop_table() -> Result<()> {
 #[tokio::test]
 async fn test_load_table() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation = set_table_creation(None, "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_load_table".into()));
 
     catalog
@@ -227,8 +226,8 @@ async fn test_create_table() -> Result<()> {
     let catalog = get_catalog().await;
     let namespace = NamespaceIdent::new("test_create_table".to_string());
     set_test_namespace(&catalog, &namespace).await?;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
-
+    // inject custom location, ignore the namespace prefix
+    let creation = set_table_creation(Some("s3a://warehouse/hive".into()), "my_table")?;
     let result = catalog.create_table(&namespace, creation).await?;
 
     assert_eq!(result.identifier().name(), "my_table");

crates/catalog/hms/src/catalog.rs

@@ -154,11 +154,11 @@ impl Catalog for HmsCatalog {
     ///
     /// This function can return an error in the following situations:
     ///
-    /// - If `hive.metastore.database.owner-type` is specified without  
+    /// - If `hive.metastore.database.owner-type` is specified without
     /// `hive.metastore.database.owner`,
     /// - Errors from `validate_namespace` if the namespace identifier does not
     /// meet validation criteria.
-    /// - Errors from `convert_to_database` if the properties cannot be  
+    /// - Errors from `convert_to_database` if the properties cannot be
     /// successfully converted into a database configuration.
     /// - Errors from the underlying database creation process, converted using
     /// `from_thrift_error`.
@@ -238,7 +238,7 @@ impl Catalog for HmsCatalog {
     /// Asynchronously updates properties of an existing namespace.
     ///
     /// Converts the given namespace identifier and properties into a database
-    /// representation and then attempts to update the corresponding namespace  
+    /// representation and then attempts to update the corresponding namespace
     /// in the Hive Metastore.
     ///
     /// # Returns
@@ -276,7 +276,7 @@ impl Catalog for HmsCatalog {
     /// # Returns
     /// A `Result<()>` indicating the outcome:
     /// - `Ok(())` signifies successful namespace deletion.
-    /// - `Err(...)` signifies failure to drop the namespace due to validation  
+    /// - `Err(...)` signifies failure to drop the namespace due to validation
     /// errors, connectivity issues, or Hive Metastore constraints.
     async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
         let name = validate_namespace(namespace)?;
@@ -297,7 +297,7 @@ impl Catalog for HmsCatalog {
     /// A `Result<Vec<TableIdent>>`, which is:
     /// - `Ok(vec![...])` containing a vector of `TableIdent` instances, each
     /// representing a table within the specified namespace.
-    /// - `Err(...)` if an error occurs during namespace validation or while  
+    /// - `Err(...)` if an error occurs during namespace validation or while
     /// querying the database.
     async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
         let name = validate_namespace(namespace)?;
@@ -333,7 +333,7 @@ impl Catalog for HmsCatalog {
     async fn create_table(
         &self,
         namespace: &NamespaceIdent,
-        creation: TableCreation,
+        mut creation: TableCreation,
     ) -> Result<Table> {
         let db_name = validate_namespace(namespace)?;
         let table_name = creation.name.clone();
@@ -342,13 +342,15 @@ impl Catalog for HmsCatalog {
             Some(location) => location.clone(),
             None => {
                 let ns = self.get_namespace(namespace).await?;
-                get_default_table_location(&ns, &table_name, &self.config.warehouse)
+                let location = get_default_table_location(&ns, &table_name, &self.config.warehouse);
+                creation.location = Some(location.clone());
+                location
             }
         };
-
         let metadata = TableMetadataBuilder::from_table_creation(creation)?
             .build()?
             .metadata;
+
         let metadata_location = create_metadata_location(&location, 0)?;
 
         self.file_io

crates/catalog/hms/tests/hms_catalog_test.rs

@@ -101,7 +101,7 @@ async fn set_test_namespace(catalog: &HmsCatalog, namespace: &NamespaceIdent) ->
     Ok(())
 }
 
-fn set_table_creation(location: impl ToString, name: impl ToString) -> Result<TableCreation> {
+fn set_table_creation(location: Option<String>, name: impl ToString) -> Result<TableCreation> {
     let schema = Schema::builder()
         .with_schema_id(0)
         .with_fields(vec![
@@ -110,20 +110,19 @@ fn set_table_creation(location: impl ToString, name: impl ToString) -> Result<Ta
         ])
         .build()?;
 
-    let creation = TableCreation::builder()
-        .location(location.to_string())
+    let builder = TableCreation::builder()
         .name(name.to_string())
         .properties(HashMap::new())
-        .schema(schema)
-        .build();
+        .location_opt(location)
+        .schema(schema);
 
-    Ok(creation)
+    Ok(builder.build())
 }
 
 #[tokio::test]
 async fn test_rename_table() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation: TableCreation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation: TableCreation = set_table_creation(None, "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_rename_table".into()));
     set_test_namespace(&catalog, namespace.name()).await?;
 
@@ -143,7 +142,7 @@ async fn test_rename_table() -> Result<()> {
 #[tokio::test]
 async fn test_table_exists() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation = set_table_creation(None, "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_table_exists".into()));
     set_test_namespace(&catalog, namespace.name()).await?;
 
@@ -159,7 +158,7 @@ async fn test_table_exists() -> Result<()> {
 #[tokio::test]
 async fn test_drop_table() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation = set_table_creation(None, "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_drop_table".into()));
     set_test_namespace(&catalog, namespace.name()).await?;
 
@@ -177,7 +176,7 @@ async fn test_drop_table() -> Result<()> {
 #[tokio::test]
 async fn test_load_table() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation = set_table_creation(None, "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_load_table".into()));
     set_test_namespace(&catalog, namespace.name()).await?;
 
@@ -200,7 +199,8 @@ async fn test_load_table() -> Result<()> {
 #[tokio::test]
 async fn test_create_table() -> Result<()> {
     let catalog = get_catalog().await;
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    // inject custom location, ignore the namespace prefix
+    let creation = set_table_creation(Some("s3a://warehouse/hive".into()), "my_table")?;
     let namespace = Namespace::new(NamespaceIdent::new("test_create_table".into()));
     set_test_namespace(&catalog, namespace.name()).await?;
 
@@ -229,7 +229,7 @@ async fn test_list_tables() -> Result<()> {
 
     assert_eq!(result, vec![]);
 
-    let creation = set_table_creation("s3a://warehouse/hive", "my_table")?;
+    let creation = set_table_creation(None, "my_table")?;
     catalog.create_table(ns.name(), creation).await?;
     let result = catalog.list_tables(ns.name()).await?;