Set warehouse location based on storage profile path

crates/catalog/src/service.rs

@@ -188,7 +188,7 @@ impl Catalog for CatalogImpl {
         let base_part = storage_profile
             .get_base_url()
             .context(error::ControlPlaneSnafu)?;
-        let table_part = format!("{}/{}", warehouse.location, commit.ident.table);
+        let table_part = format!("{}/{}", warehouse.path(), commit.ident.table);
         let metadata_part = format!("metadata/{}", Self::generate_metadata_filename());
 
         let mut properties = table.properties.clone();
@@ -339,7 +339,7 @@ impl Catalog for CatalogImpl {
         let base_part = storage_profile
             .get_base_url()
             .context(error::ControlPlaneSnafu)?;
-        let table_part = format!("{}/{}", warehouse.location, table_creation.name);
+        let table_part = format!("{}/{}", warehouse.path(), table_creation.name);
         let metadata_part = format!("metadata/{}", Self::generate_metadata_filename());
 
         let table_creation = {

crates/control_plane/src/models/mod.rs

@@ -378,29 +378,29 @@ impl Warehouse {
         storage_profile_id: Uuid,
     ) -> ControlPlaneModelResult<Self> {
         let id = Uuid::new_v4();
-        let location = format!("{prefix}/{id}");
         let now = Utc::now().naive_utc();
         Ok(Self {
             id,
             prefix,
             name,
-            location,
+            location: String::new(),
             storage_profile_id,
             created_at: now,
             updated_at: now,
         })
     }
-}
 
-impl TryFrom<WarehouseCreateRequest> for Warehouse {
-    type Error = ControlPlaneModelError;
+    pub fn with_location(&mut self, location: String) {
+        self.location = location;
+    }
 
-    fn try_from(value: WarehouseCreateRequest) -> ControlPlaneModelResult<Self> {
-        Self::new(
-            value.prefix.clone(),
-            value.name.clone(),
-            value.storage_profile_id,
-        )
+    #[must_use]
+    pub fn path(&self) -> String {
+        if self.prefix.is_empty() {
+            format!("{}", self.id)
+        } else {
+            format!("{}/{}", self.prefix, self.id)
+        }
     }
 }
 

crates/control_plane/src/service.rs

@@ -231,14 +231,19 @@ impl ControlService for ControlServiceImpl {
         &self,
         params: &WarehouseCreateRequest,
     ) -> ControlPlaneResult<Warehouse> {
-        // TODO: Check if storage profile exists
+        let profile = self.get_profile(params.storage_profile_id).await?;
+        let base_url = profile
+            .get_base_url()
+            .context(error::InvalidStorageProfileSnafu)?;
+
         // - Check if its valid
         // - Generate id, update created_at and updated_at
-        // - Try create Warehouse from WarehouseCreateRequest
-        let wh: Warehouse = params
+        // - Try to create Warehouse from WarehouseCreateRequest
+        let mut wh: Warehouse = params
             .try_into()
             .context(error::InvalidCreateWarehouseSnafu)?;
-        let _ = self.get_profile(wh.storage_profile_id).await?;
+        wh.with_location(base_url);
+
         self.warehouse_repo.create(&wh).await?;
         Ok(wh)
     }
@@ -394,7 +399,7 @@ impl ControlService for ControlServiceImpl {
 
         // this path also computes inside catalog service (create_table)
         // TODO need to refactor the code so this path calculation is in one place
-        let table_part = format!("{}/{}", warehouse.location, table_name);
+        let table_part = format!("{}/{}", warehouse.path(), table_name);
         let path_string = format!("{table_part}/tmp/{unique_file_id}/{file_name}");
 
         let path = Path::from(path_string.clone());

crates/runtime/src/datafusion/execution.rs

@@ -380,7 +380,6 @@ impl SqlExecutor {
                     .await
                     .context(ih_error::IcebergSnafu)?;
             };
-
             // Create new table
             rest_catalog
                 .create_table(