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

@@ -17,6 +17,7 @@
     "check_exclude",
     "excluded",
     "id_type",
+    "matched_relation",
     "merge_dictionary",
     "normalize_identifier",
     "search",
@@ -84,6 +85,25 @@ class MergedDictionary:
     definition_bindings: dict[str, Any] | None
 
 
+def matched_relation(relation: Relation, requested: TableId) -> Relation:
+    """An exact selection's match, under the label it was asked for.
+
+    `requested` is the authored name qualified with the namespace the lookup
+    ran in. The warehouse's own id is kept as the identity, since it carries
+    that backend's casing and is what later metadata queries name. A relation
+    the warehouse reports without a namespace has none to be labelled with,
+    and none to be queried under either: a Databricks temporary view answers
+    only to its bare name.
+    """
+    qualified = relation.id.schema is not None or relation.id.catalog is not None
+    return Relation(
+        id=requested if qualified else relation.id,
+        kind=relation.kind,
+        description=relation.description,
+        identity=relation.id,
+    )
+
+
 def normalize_identifier(value: Any, identifier_case: str | None) -> Any:
     """Fold an identifier the way the backend does, or leave it alone."""
     if isinstance(value, list):
@@ -184,12 +204,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))
@@ -236,7 +262,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(

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

@@ -11,7 +11,7 @@
 from typing import Any
 
 from .._data_source import TableId
-from ._core import Relation, Selector, id_type
+from ._core import Relation, Selector, id_type, matched_relation
 
 __all__ = [
     "columns_from_describe",
@@ -139,16 +139,36 @@ def _list_hive_relations(backend: Any, selector: Selector) -> list[Relation]:
     rows = backend.query(
         f"SHOW TABLES IN {_quote_path([selector.catalog or _HIVE, selector.schema])}"
     )
-    return [
-        Relation(
-            id=TableId(
-                catalog=selector.catalog,
-                schema=selector.schema,
-                table=_lower_keys(row)["tablename"],
+    relations = []
+    for row in rows:
+        values = _lower_keys(row)
+        temporary = _is_temporary(values)
+        relations.append(
+            Relation(
+                id=TableId(
+                    catalog=None if temporary else selector.catalog,
+                    schema=None if temporary else selector.schema,
+                    table=values["tablename"],
+                )
             )
         )
-        for row in rows
-    ]
+    return relations
+
+
+def _is_temporary(values: dict[str, Any]) -> bool:
+    """Whether a `SHOW TABLES` row is a session-scoped temporary view.
+
+    `SHOW TABLES` lists the session's temporary views alongside the schema's
+    own tables, reporting them with an empty database and `isTemporary` set. A
+    temporary view belongs to no schema and answers only to its bare name, so
+    qualifying it with the listed namespace would build an id naming nothing.
+    """
+    if "istemporary" in values:
+        return bool(values["istemporary"])
+    for column in ("database", "namespace"):
+        if column in values:
+            return not values[column]
+    return False
 
 
 def _is_hive(selector: Selector) -> bool:
@@ -179,12 +199,7 @@ def exact_relation(backend: Any, selector: Selector) -> Relation | None:
     )
     for relation in relations:
         if relation.id.table == requested.table:
-            return Relation(
-                id=requested,
-                kind=relation.kind,
-                description=relation.description,
-                identity=relation.id,
-            )
+            return matched_relation(relation, requested)
     return None