Allow configuring a default_table_root_location in MemoryCatalog

crates/catalog/memory/src/catalog.rs

@@ -38,14 +38,16 @@ use crate::namespace_state::NamespaceState;
 pub struct MemoryCatalog {
     root_namespace_state: Mutex<NamespaceState>,
     file_io: FileIO,
+    default_table_root_location: String,
 }
 
 impl MemoryCatalog {
     /// Creates an memory catalog.
-    pub fn new(file_io: FileIO) -> Self {
+    pub fn new(file_io: FileIO, default_table_root_location: String) -> Self {
         Self {
             root_namespace_state: Mutex::new(NamespaceState::default()),
             file_io,
+            default_table_root_location,
         }
     }
 }
@@ -166,7 +168,8 @@ impl Catalog for MemoryCatalog {
             Some(location) => (table_creation, location),
             None => {
                 let location = format!(
-                    "{}/{}",
+                    "{}/{}/{}",
+                    self.default_table_root_location,
                     table_ident.namespace().join("/"),
                     table_ident.name()
                 );
@@ -279,7 +282,11 @@ mod tests {
 
     fn new_memory_catalog() -> impl Catalog {
         let file_io = FileIOBuilder::new_fs_io().build().unwrap();
-        MemoryCatalog::new(file_io)
+
+        let tmp_dir = TempDir::new().unwrap();
+        let default_table_root_location = tmp_dir.path().to_str().unwrap().to_string();
+
+        MemoryCatalog::new(file_io, default_table_root_location)
     }
 
     async fn create_namespace<C: Catalog>(catalog: &C, namespace_ident: &NamespaceIdent) {
@@ -312,16 +319,12 @@ mod tests {
     }
 
     async fn create_table<C: Catalog>(catalog: &C, table_ident: &TableIdent) {
-        let tmp_dir = TempDir::new().unwrap();
-        let location = tmp_dir.path().to_str().unwrap().to_string();
-
         let _ = catalog
             .create_table(
                 &table_ident.namespace,
                 TableCreation::builder()
                     .name(table_ident.name().into())
                     .schema(simple_table_schema())
-                    .location(location)
                     .build(),
             )
             .await
@@ -956,6 +959,46 @@ mod tests {
             .unwrap());
     }
 
+    #[tokio::test]
+    async fn test_create_table_without_location_creates_table_at_catalog_default_table_root_location(
+    ) {
+        let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+        let tmp_dir = TempDir::new().unwrap();
+        let default_table_root_location = tmp_dir.path().to_str().unwrap().to_string();
+        let catalog = MemoryCatalog::new(file_io, default_table_root_location.clone());
+
+        let namespace_ident = NamespaceIdent::new("a".into());
+        create_namespace(&catalog, &namespace_ident).await;
+
+        let table_name = "tbl1";
+        let expected_table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
+
+        assert_table_eq(
+            &catalog
+                .create_table(
+                    &namespace_ident,
+                    TableCreation::builder()
+                        .name(table_name.into())
+                        .schema(simple_table_schema())
+                        .build(),
+                )
+                .await
+                .unwrap(),
+            &expected_table_ident,
+            &simple_table_schema(),
+        );
+
+        let table = &catalog.load_table(&expected_table_ident).await.unwrap();
+
+        assert_table_eq(table, &expected_table_ident, &simple_table_schema());
+
+        assert!(table
+            .metadata_location()
+            .unwrap()
+            .to_string()
+            .starts_with(&default_table_root_location));
+    }
+
     #[tokio::test]
     async fn test_create_table_with_location() {
         let tmp_dir = TempDir::new().unwrap();
@@ -990,13 +1033,7 @@ mod tests {
             .metadata_location()
             .unwrap()
             .to_string()
-            .starts_with(&location));
-
-        assert_table_eq(
-            &catalog.load_table(&expected_table_ident).await.unwrap(),
-            &expected_table_ident,
-            &simple_table_schema(),
-        )
+            .starts_with(&location))
     }
 
     #[tokio::test]