Use the storage location name as the key in the clade schema

clade/proto/schema.proto

@@ -11,16 +11,18 @@ message TableObject {
   string name = 1;
   // Path within the provided storage location, if any
   string path = 2;
-  // URL of the root storage location
-  optional string location = 3;
+  // Storage location identifier
+  optional string store = 3;
 }
 
 // A single root storage location, hosting many individual tables
 message StorageLocation {
-  // URL of the root storage location
-  string location = 1;
+  // Unique storage location identifier
+  string name = 1;
+  // URL of the storage location root
+  string location = 2;
   // Connection options for the object store client
-  map<string, string> options = 2;
+  map<string, string> options = 3;
 }
 
 message ListSchemaRequest {

src/catalog/metastore.rs

@@ -97,7 +97,7 @@ impl Metastore {
         let store_options = catalog_schemas
             .stores
             .into_iter()
-            .map(|store| (store.location, store.options))
+            .map(|store| (store.name, (store.location, store.options)))
             .collect();
 
         // Turn the list of all collections, tables and their columns into a nested map.
@@ -119,7 +119,7 @@ impl Metastore {
     async fn build_schema(
         &self,
         schema: SchemaObject,
-        store_options: &HashMap<String, HashMap<String, String>>,
+        store_options: &HashMap<String, (String, HashMap<String, String>)>,
     ) -> CatalogResult<(Arc<str>, Arc<SeafowlSchema>)> {
         let schema_name = schema.name;
 
@@ -140,21 +140,21 @@ impl Metastore {
     async fn build_table(
         &self,
         table: TableObject,
-        store_options: &HashMap<String, HashMap<String, String>>,
+        store_options: &HashMap<String, (String, HashMap<String, String>)>,
     ) -> CatalogResult<(Arc<str>, Arc<dyn TableProvider>)> {
         // Build a delta table but don't load it yet; we'll do that only for tables that are
         // actually referenced in a statement, via the async `table` method of the schema provider.
         // TODO: this means that any `information_schema.columns` query will serially load all
         // delta tables present in the database. The real fix for this is to make DF use `TableSource`
         // for the information schema, and then implement `TableSource` for `DeltaTable` in delta-rs.
 
-        let table_log_store = match table.location {
+        let table_log_store = match table.store {
             // Use the provided customized location
-            Some(location) => {
-                let this_store_options = store_options
-                    .get(&location)
+            Some(name) => {
+                let (location, this_store_options) = store_options
+                    .get(&name)
                     .ok_or(CatalogError::Generic {
-                        reason: format!("Object store for location {location} not found"),
+                        reason: format!("Object store with name {name} not found"),
                     })?
                     .clone();
 

src/catalog/repository.rs

@@ -130,7 +130,7 @@ impl SchemaStore for RepositoryStore {
                             Some(TableObject {
                                 name: name.clone(),
                                 path: uuid.to_string(),
-                                location: None,
+                                store: None,
                             })
                         } else {
                             None

tests/fixtures.rs

@@ -29,28 +29,29 @@ pub fn schemas() -> ListSchemaResponse {
                 tables: vec![TableObject {
                     name: "file".to_string(),
                     path: "delta-0.8.0-partitioned".to_string(),
-                    location: None,
+                    store: None,
                 }],
             },
             SchemaObject {
                 name: "s3".to_string(),
                 tables: vec![TableObject {
                     name: "minio".to_string(),
                     path: "test-data/delta-0.8.0-partitioned".to_string(),
-                    location: Some("s3://seafowl-test-bucket".to_string()),
+                    store: Some("minio".to_string()),
                 }],
             },
             SchemaObject {
                 name: "gcs".to_string(),
                 tables: vec![TableObject {
                     name: "fake".to_string(),
                     path: "delta-0.8.0-partitioned".to_string(),
-                    location: Some("gs://test-data".to_string()),
+                    store: Some("fake-gcs".to_string()),
                 }],
             },
         ],
         stores: vec![
             StorageLocation {
+                name: "minio".to_string(),
                 location: "s3://seafowl-test-bucket".to_string(),
                 options: HashMap::from([
                     (
@@ -74,6 +75,7 @@ pub fn schemas() -> ListSchemaResponse {
                 ]),
             },
             StorageLocation {
+                name: "fake-gcs".to_string(),
                 location: "gs://test-data".to_string(),
                 options: HashMap::from([(
                     GoogleConfigKey::ServiceAccount.as_ref().to_string(),