Check warehouse session identity and query access

pkg-py/src/commons/_catalog/__init__.py

@@ -4,9 +4,11 @@
 types are authoritative, identifier case normalizes per backend, and an
 ambiguous relative name is an error rather than a guess.
 
-Everything here is a pure function over the rows a warehouse listing returns.
-Running the queries that produce those rows belongs to the per-backend
-readers, which keeps this testable without a warehouse.
+Interpreting a warehouse listing is kept to pure functions over the rows it
+returns, and running the queries that produce them belongs to the per-backend
+readers, which keeps the interpretation testable without a warehouse. The
+session and access checks in `_security` are the exception, since asking the
+warehouse is the whole point of them.
 """
 
 from . import _databricks, _snowflake
@@ -23,19 +25,41 @@
     search,
     table_registry,
 )
+from ._security import (
+    CatalogAccessError,
+    CatalogAuthorizationError,
+    CatalogSessionChangedError,
+    CatalogTransientError,
+    SessionSnapshot,
+    check_session,
+    ensure_queryable,
+    require_queryable,
+    require_queryable_relations,
+    session_snapshot,
+)
 
 __all__ = [
+    "CatalogAccessError",
+    "CatalogAuthorizationError",
+    "CatalogSessionChangedError",
+    "CatalogTransientError",
     "Manifest",
     "MergedDictionary",
     "Relation",
     "Selector",
+    "SessionSnapshot",
     "_databricks",
     "_snowflake",
     "check_exclude",
+    "check_session",
+    "ensure_queryable",
     "excluded",
     "id_type",
     "merge_dictionary",
     "normalize_identifier",
+    "require_queryable",
+    "require_queryable_relations",
     "search",
+    "session_snapshot",
     "table_registry",
 ]

pkg-py/src/commons/_catalog/_core.py

@@ -174,12 +174,18 @@ def table_registry(
             # An entry naming a table is kept whether or not the warehouse
             # has it, and is always validated. Dropping a missing one turns
             # "that table is not there" into a quietly smaller selection.
-            table_id = _selector_id(selector)
             found = exact_relation(selector)
-            relations.append(
-                found if found is not None else Relation(id=table_id, discovered=False)
+            # Keyed by the relation's own id rather than the selector's: an
+            # entry naming a bare table is qualified with the connection's
+            # namespace once the warehouse answers, and the two lists have to
+            # agree on the label or the access check cannot pair them up.
+            relation = (
+                found
+                if found is not None
+                else Relation(id=_selector_id(selector), discovered=False)
             )
-            validate.append(table_id)
+            relations.append(relation)
+            validate.append(relation.id)
             continue
         namespace_selected = True
         relations.extend(list_relations(selector))
@@ -220,7 +226,9 @@ class Manifest:
     objects: dict[str, Relation]
     searchable: bool = False
     access: dict[str, str] = field(default_factory=dict)
-    access_errors: dict[str, str] = field(default_factory=dict)
+    # The driver's own failure, kept for the relations whose refusal is
+    # cached, so a later refusal can still be raised from what caused it.
+    access_errors: dict[str, BaseException] = field(default_factory=dict)
 
     @classmethod
     def build(