feat(glue): make warehouse optional when loading the catalog

crates/catalog/glue/src/catalog.rs

@@ -52,7 +52,7 @@ use crate::{
 pub const GLUE_CATALOG_PROP_URI: &str = "uri";
 /// Glue catalog id
 pub const GLUE_CATALOG_PROP_CATALOG_ID: &str = "catalog_id";
-/// Glue catalog warehouse location
+/// Glue catalog warehouse location, used for new tables whose namespace has no location
 pub const GLUE_CATALOG_PROP_WAREHOUSE: &str = "warehouse";
 
 /// Builder for [`GlueCatalog`].
@@ -90,12 +90,6 @@ impl CatalogBuilder for GlueCatalogBuilder {
 
         async move {
             let catalog_properties = GlueCatalogProperties::from_properties(&props)?;
-            if catalog_properties.warehouse.is_empty() {
-                return Err(Error::new(
-                    ErrorKind::DataInvalid,
-                    "Catalog warehouse is required",
-                ));
-            }
 
             let runtime = match self.runtime {
                 Some(rt) => rt,
@@ -589,7 +583,7 @@ impl Catalog for GlueCatalog {
         if creation.location.is_none() {
             let ns = self.get_namespace(namespace).await?;
             let location =
-                get_default_table_location(&ns, &db_name, &table_name, &self.properties.warehouse);
+                get_default_table_location(&ns, &db_name, &table_name, &self.properties.warehouse)?;
             creation.location = Some(location);
         }
         let metadata = TableMetadataBuilder::from_table_creation(creation)?
@@ -963,18 +957,17 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_load_rejects_missing_or_empty_warehouse() {
+    async fn test_load_without_warehouse() {
         for props in [
             HashMap::new(),
             HashMap::from([(GLUE_CATALOG_PROP_WAREHOUSE.to_string(), String::new())]),
         ] {
-            let error = GlueCatalogBuilder::default()
+            let catalog = GlueCatalogBuilder::default()
                 .load("glue", props)
                 .await
-                .unwrap_err();
+                .unwrap();
 
-            assert_eq!(error.kind(), ErrorKind::DataInvalid);
-            assert_eq!(error.message(), "Catalog warehouse is required");
+            assert_eq!(catalog.properties.warehouse, "");
         }
     }
 }

crates/catalog/glue/src/utils.rs

@@ -208,20 +208,26 @@ pub(crate) fn get_default_table_location(
     db_name: impl AsRef<str>,
     table_name: impl AsRef<str>,
     warehouse: impl AsRef<str>,
-) -> String {
+) -> Result<String> {
     let properties = namespace.properties();
 
     match properties.get(LOCATION) {
-        Some(location) => format!("{}/{}", location, table_name.as_ref()),
+        Some(location) => Ok(format!("{}/{}", location, table_name.as_ref())),
         None => {
             let warehouse_location = warehouse.as_ref().trim_end_matches('/');
+            if warehouse_location.is_empty() {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    "Cannot derive default table location: catalog warehouse is not set and the namespace has no location",
+                ));
+            }
 
-            format!(
+            Ok(format!(
                 "{}/{}.db/{}",
                 warehouse_location,
                 db_name.as_ref(),
                 table_name.as_ref()
-            )
+            ))
         }
     }
 }
@@ -401,13 +407,30 @@ mod tests {
 
         let expected = "db_location/my_table";
         let result =
-            get_default_table_location(&namespace, db_name, table_name, "warehouse_location");
+            get_default_table_location(&namespace, db_name, table_name, "warehouse_location")?;
 
         assert_eq!(expected, result);
 
         Ok(())
     }
 
+    #[test]
+    fn test_get_default_table_location_without_warehouse() -> Result<()> {
+        let properties = HashMap::from([(LOCATION.to_string(), "db_location".to_string())]);
+        let namespace =
+            Namespace::with_properties(NamespaceIdent::new("default".into()), properties);
+        let db_name = validate_namespace(namespace.name())?;
+
+        let result = get_default_table_location(&namespace, &db_name, "my_table", "")?;
+        assert_eq!("db_location/my_table", result);
+
+        let namespace = Namespace::new(NamespaceIdent::new("default".into()));
+        let error = get_default_table_location(&namespace, &db_name, "my_table", "").unwrap_err();
+        assert_eq!(error.kind(), ErrorKind::DataInvalid);
+
+        Ok(())
+    }
+
     #[test]
     fn test_get_default_table_location_warehouse() -> Result<()> {
         let namespace = Namespace::new(NamespaceIdent::new("default".into()));
@@ -416,7 +439,7 @@ mod tests {
 
         let expected = "warehouse_location/default.db/my_table";
         let result =
-            get_default_table_location(&namespace, db_name, table_name, "warehouse_location");
+            get_default_table_location(&namespace, db_name, table_name, "warehouse_location")?;
 
         assert_eq!(expected, result);