perf(snowflake): plan from SHOW metadata and run inspection reads concurrently

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters/snowflake.md

@@ -41,6 +41,19 @@ On connect, SQLBuild applies the connection's role and warehouse, then applies t
 target's authoritative database and schema. These ensure the session context is set correctly
 regardless of the user's default settings.
 
+## Planning metadata
+
+Planning reads relation and column metadata with `SHOW TABLES`, `SHOW VIEWS`, and `SHOW COLUMNS`
+once per schema, several schemas at a time. `SHOW` commands are served by Snowflake's metadata
+service, so they don't wait for, or resume, a warehouse. Two reads still use `INFORMATION_SCHEMA`
+and need a running warehouse:
+
+- `LAST_ALTERED` for source freshness, read only for the sources the plan observes.
+- Columns for a schema whose `SHOW COLUMNS` output is exactly 10,000 rows, the documented
+  limit at which the output may be cut off.
+
+Run `sqb plan -v` to see each metadata query with its elapsed time and row count.
+
 ## Shared connections across targets
 
 Use multiple targets to reuse one Snowflake connection while selecting different authoritative

src/sqlbuild/adapter/contract/classes/base_adapter.py

@@ -107,6 +107,7 @@ class BaseAdapter(RetentionAdapterMixin, StrictAdapter):
     views_read_with_reader_access: ClassVar[bool] = False
     allows_implicit_managed_write_schema: ClassVar[bool] = False
     execution_duration_limit_seconds: ClassVar[int | None] = None
+    metadata_inspection_concurrency: ClassVar[int] = 1
     _snapshot_sql_dialect: ClassVar[SnapshotSqlDialect] = SnapshotSqlDialect(
         timestamp_type="TIMESTAMP",
         distinct_condition=render_is_distinct_from,

src/sqlbuild/adapter/relations/_helpers/inspection_context.py

@@ -0,0 +1,87 @@
+"""Invocation-local inspection catalog and verbose query sink state."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from contextvars import ContextVar, Token
+from typing import TYPE_CHECKING, Any
+
+from sqlbuild.adapter.relations.models import InspectionQueryRecord
+from sqlbuild.diagnostics.main.log_debug_event import log_debug_event
+from sqlbuild.diagnostics.main.log_sql import log_sql
+
+if TYPE_CHECKING:
+    from sqlbuild.adapter.relations.classes.inspection_catalog import InspectionCatalog
+
+_LOGGER: logging.Logger = logging.getLogger("sqlbuild.inspection")
+_ACTIVE_CATALOG: ContextVar[InspectionCatalog | None] = ContextVar(
+    "sqlbuild_inspection_catalog", default=None
+)
+_INSIDE_WORKER: ContextVar[bool] = ContextVar("sqlbuild_inside_inspection_worker", default=False)
+_QUERY_SINK: ContextVar[Callable[[InspectionQueryRecord], None] | None] = ContextVar(
+    "sqlbuild_inspection_query_sink", default=None
+)
+
+
+def current_catalog(*, adapter: Any, connection: Any) -> InspectionCatalog | None:
+    """Return the open catalog for exactly this adapter and connection, if any."""
+
+    catalog: InspectionCatalog | None = _ACTIVE_CATALOG.get()
+    if catalog is None or catalog.adapter is not adapter or catalog.connection is not connection:
+        return None
+    return catalog
+
+
+def emit_inspection_record(*, record: InspectionQueryRecord) -> None:
+    """Log one inspection read and forward it to the active verbose sink."""
+
+    log_sql(logger=_LOGGER, sql=record.sql, action="inspection")
+    log_debug_event(
+        logger=_LOGGER,
+        message="warehouse inspection query",
+        sqlbuild_elapsed_seconds=f"{record.elapsed_seconds:.3f}",
+        sqlbuild_row_count=record.row_count,
+        sqlbuild_error=record.error,
+    )
+    sink: Callable[[InspectionQueryRecord], None] | None = _QUERY_SINK.get()
+    if sink is not None:
+        sink(record)
+
+
+def activate_catalog(catalog: InspectionCatalog) -> Token[InspectionCatalog | None]:
+    """Make ``catalog`` the open catalog for this context."""
+
+    return _ACTIVE_CATALOG.set(catalog)
+
+
+def deactivate_catalog(token: Token[InspectionCatalog | None]) -> None:
+    """Restore the catalog that was open before ``activate_catalog``."""
+
+    _ACTIVE_CATALOG.reset(token)
+
+
+def activate_query_sink(
+    sink: Callable[[InspectionQueryRecord], None],
+) -> Token[Callable[[InspectionQueryRecord], None] | None]:
+    """Deliver recorded inspection reads in this context to ``sink``."""
+
+    return _QUERY_SINK.set(sink)
+
+
+def deactivate_query_sink(token: Token[Callable[[InspectionQueryRecord], None] | None]) -> None:
+    """Restore the sink that was active before ``activate_query_sink``."""
+
+    _QUERY_SINK.reset(token)
+
+
+def inside_inspection_worker() -> bool:
+    """Return whether this thread is running one bounded inspection read."""
+
+    return _INSIDE_WORKER.get()
+
+
+def mark_inspection_worker() -> None:
+    """Mark this worker context so nested bounded reads run serially."""
+
+    _ = _INSIDE_WORKER.set(True)