fix: resolve aliased catalogs

crates/catalog/src/session_catalog.rs

@@ -47,6 +47,8 @@ pub struct ResolveConfig {
 /// from the remote state provided by metastore.
 #[derive(Clone, Debug)]
 pub struct SessionCatalog {
+    /// Optional alias for referencing objects in this catalog
+    alias: Option<String>,
     /// The state retrieved from a remote Metastore.
     state: Arc<CatalogState>,
     /// Map database names to their ids.
@@ -69,6 +71,7 @@ impl SessionCatalog {
     /// Create a new session catalog with an initial state.
     pub fn new(state: Arc<CatalogState>, resolve_conf: ResolveConfig) -> SessionCatalog {
         let mut catalog = SessionCatalog {
+            alias: None,
             state,
             database_names: HashMap::new(),
             tunnel_names: HashMap::new(),
@@ -82,6 +85,20 @@ impl SessionCatalog {
         catalog
     }
 
+    pub fn new_with_alias(
+        state: Arc<CatalogState>,
+        resolve_conf: ResolveConfig,
+        alias: String,
+    ) -> SessionCatalog {
+        let mut catalog = Self::new(state, resolve_conf);
+        catalog.alias = Some(alias);
+        catalog
+    }
+
+    pub fn alias(&self) -> Option<&str> {
+        self.alias.as_deref()
+    }
+
     /// Get the version of this catalog state.
     pub fn version(&self) -> u64 {
         self.state.version

crates/sqlexec/src/remote/client.rs

@@ -281,12 +281,13 @@ impl RemoteClient {
 
         Ok((
             remote_sess_client,
-            SessionCatalog::new(
+            SessionCatalog::new_with_alias(
                 Arc::new(resp.catalog),
                 ResolveConfig {
                     default_schema_oid: SCHEMA_DEFAULT.oid,
                     session_schema_oid: SCHEMA_CURRENT_SESSION.oid,
                 },
+                self.get_deployment_name().to_string(),
             ),
         ))
     }

crates/sqlexec/src/resolve.rs

@@ -108,6 +108,19 @@ impl<'a> EntryResolver<'a> {
                 schema,
                 table,
             } => {
+                // if the catalog has an alias, we need to check if the
+                // reference is to the alias and if so, resolve it to the
+                // actual catalog name.
+                if let Some(catalog_alias) = self.catalog.alias() {
+                    if catalog == catalog_alias {
+                        if let Some(ent) =
+                            self.catalog.resolve_entry(DEFAULT_CATALOG, schema, table)
+                        {
+                            return Ok(ResolvedEntry::Entry(ent.clone()));
+                        }
+                    }
+                }
+
                 // If catalog is anything but "default", we know we need to do
                 // external resolution since we don't store info about
                 // individual tables.