fix(planner): observe source freshness per source and fail plans on missing source tables

fensu.toml

@@ -93,6 +93,12 @@ rule = "FFR601"
 path = "src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py"
 reason = "The SQL Server adapter intentionally keeps overrideable backend behavior on one class."
 
+[[targets.python.rule_exceptions]]
+rule = "XSB051"
+path = "src/sqlbuild/compiler/planner/_helpers/warehouse/source_columns.py"
+symbols = ["gather_source_inspection"]
+reason = "Each type-enforced expression source is its own query whose output columns can only be read by running it; table sources are listed in one batch per database."
+
 [[targets.python.rule_exceptions]]
 rule = "XSB051"
 path = "src/sqlbuild/executor/migrations/_helpers/columns.py"

src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/freshness.md

@@ -74,6 +74,12 @@ Summary: observed=3 changed=0 unchanged=0 tolerated=0 unknown=0 errors=0
 
 Sources without explicit `freshness:` config are auto-observed using the `adapter` strategy if the adapter supports table metadata. Sources that can't be observed (expression sources, managed sources without freshness config on unsupported adapters) show as `unknown`.
 
+Each source is reported on its own. Adapter metadata is fetched in one batched lookup, but a
+source the metadata can't describe affects only that source: it shows as `unknown` with the
+reason (for example a view, an external table, a table with no last-modified time, or a table the
+metadata lookup did not find). A column or SQL freshness query that fails shows as `error`. The
+other sources are still observed.
+
 ## Comparing against state
 
 Use `--state` to compare current observations against the freshness state stored from the last successful build:
@@ -105,7 +111,7 @@ Summary: observed=0 changed=1 unchanged=1 tolerated=1 unknown=0 errors=0
 | `changed` | Data version differs from the stored state |
 | `unchanged` | Data version matches the stored state exactly |
 | `tolerated` | Data version differs but is within the `lag_tolerance` threshold |
-| `unknown` | No freshness config and adapter metadata unavailable, or no previous state to compare against |
+| `unknown` | No freshness config and adapter metadata unavailable, metadata unavailable for an existing table, or no previous state to compare against |
 | `error` | Observation failed (e.g. source table does not exist, query error) |
 
 ## CI integration

src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/plan.md

@@ -12,6 +12,7 @@ Online: https://sqlbuild.com/docs/cli/plan/
 - Flags
 - Output
 - Missing upstream dependencies
+- Missing source tables
 
 Shows the execution plan without making any changes. Useful for inspecting change detection, backfill policies, and selector scope before building.
 
@@ -122,3 +123,10 @@ error[S301]: cannot build selected scope: audit 'relationships' on 'stg_orders'
 
 Load or build the resource first (for a seed, `sqb seed`), or add it to the selection. When both
 kinds are missing, the error lists the missing upstream dependencies and then the audit reads.
+
+## Missing source tables
+
+When a selected model reads an external source whose table does not exist in the warehouse,
+planning fails with `S405` before anything runs. The error names each missing source, its table,
+the selected models that read it, and the source declaration file and line. See
+[Source freshness: Missing source tables](../concepts/planning/source-freshness.md#missing-source-tables).

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/planning/source-freshness.md

@@ -24,6 +24,52 @@ During planning, SQLBuild observes the current data version of each source that
 
 Sources without explicit `freshness:` config are auto-observed using the `adapter` strategy if the adapter supports table metadata and the source has a physical table (not an expression source, not a managed source).
 
+Adapter metadata for all sources is fetched in one batched lookup, but each source gets its own
+result. When one source can't be observed, only that source is unknown; the others keep their
+freshness. The `Freshness metadata` progress line reports the outcome, for example
+`Freshness metadata  OK (257 sources, 3 unknown)`. The plan adds one warning per reason, listing up
+to 10 sources, such as `source freshness unknown (unavailable) for raw_events, raw_sessions: ...`.
+Sources without freshness configuration that the adapter can't observe are not warned about.
+Every unknown source is listed with its reason and message under `unknown_source_details` in the
+`direct_source_freshness` metadata of `sqb plan --json`; run `sqb freshness` for the same detail
+in text form.
+
+## Missing source tables
+
+A source table that a selected model reads must exist before anything runs. If it doesn't,
+planning fails with `S405`, naming each missing source, its table, and the selected models that
+read it, and pointing at the source declaration:
+
+```
+error[S405]: cannot build selected scope: 1 source table read by selected resources does not exist in the warehouse:
+  - source 'raw_payments' (raw.payments), read by payments
+  = help: create or load the table, correct its database, schema, or table in the source declaration, or stop reading it from the selected models; declared 'raw_payments' at sources/raw.yml:16
+```
+
+Create the table, fix the declaration, or leave the reading models out of the selection. Managed
+sources and sources with a loader are exempt, because SQLBuild creates their tables when it loads
+them. Expression sources, sources without a schema, and wildcard tables are also exempt. A missing
+table that only unselected models read does not fail the plan; its freshness is reported as unknown.
+
+The check first looks for each source in the warehouse listing that planning already reads for
+source columns, so it normally issues no extra queries. A source the listing does not find (for
+example because of identifier case, a stream, a materialized view, or a table in another database)
+is then probed directly with `SELECT 1 FROM <relation> WHERE 1=0`, using the same relation name the
+model SQL reads. Probes for several such sources run concurrently, within the same limit as other
+warehouse inspection reads. Only a not-found error from that probe counts as missing, including a
+missing schema, catalog, or attached database; any other error (for example a connection error)
+fails planning with that error. On Snowflake the probe may need a running warehouse, but it only
+runs for sources the listing did not find.
+
+Snowflake reports a missing table and a table the current role can't read with the same error, and
+its listing hides tables the role can't access. For those sources `S405` says the table "does not
+exist or is not readable by" the role, names the role in use when the connection reports it, and the help suggests both fixes:
+create or load the table, or grant `SELECT` on it to that role.
+
+A table that a model's own `pre_hook` creates does not exist yet when the plan is made, so planning
+fails with `S405`. Declare such a table as a managed source or a source with a loader, or create it
+before planning.
+
 ## Lag tolerance
 
 For timestamp-based freshness, `lag_tolerance` controls how much the observed value can drift before being considered a real change. If the current timestamp is within the tolerance of the previous observation, the source is treated as unchanged. This is useful for sources where the freshness timestamp moves by seconds or minutes on every query but the underlying data hasn't meaningfully changed.

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

@@ -30,6 +30,7 @@
     AdapterUserError,
     UnsupportedTypedSqlRenderingError,
 )
+from sqlbuild.adapter.contract.main.run_relation_read_probe import run_relation_read_probe
 from sqlbuild.adapter.contract.main.same_view_definition import same_view_definition
 from sqlbuild.adapter.contract.models import (
     ColumnInfo,
@@ -41,6 +42,7 @@
     QueryResult,
     RelationGrant,
     RelationInfo,
+    RelationReadProbe,
     RowDiffCoverage,
     RowDiffPreparedRelations,
     RowDiffResult,
@@ -271,6 +273,16 @@ def describe_relation(self, *, connection: Any, relation: str) -> tuple[ColumnIn
         cursor: Any = self.execute(connection=connection, sql=f"DESCRIBE {relation}")
         return tuple(ColumnInfo(name=row[0], type=row[1]) for row in cursor.fetchall())
 
+    def probe_relation_read(self, *, connection: Any, relation: str) -> RelationReadProbe:
+        """Probe a rendered relation; only the adapter's not-found errors become missing."""
+
+        return run_relation_read_probe(
+            execute=self.execute,
+            connection=connection,
+            relation=relation,
+            classify_not_found=lambda error: None,
+        )
+
     def get_table_freshness_metadata(
         self,
         *,
@@ -292,12 +304,15 @@ def get_tables_freshness_metadata(
         results: dict[TableFreshnessRequest, TableFreshnessMetadata] = {}
         request: TableFreshnessRequest
         for request in requests:
-            results[request] = self.get_table_freshness_metadata(
-                connection=connection,
-                database=request.database,
-                schema=request.schema,
-                name=request.name,
-            )
+            try:
+                results[request] = self.get_table_freshness_metadata(
+                    connection=connection,
+                    database=request.database,
+                    schema=request.schema,
+                    name=request.name,
+                )
+            except AdapterUserError as error:
+                results[request] = TableFreshnessMetadata.unavailable(message=error.message)
         return results
 
     def query_column_names(self, *, connection: Any, sql: str) -> tuple[str, ...]:

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

@@ -32,9 +32,12 @@
 from sqlbuild.adapter.contract.constants import (
     DIFF_LEFT_SIDE,
     DIFF_RIGHT_SIDE,
+    DUCKDB_MISSING_OBJECT_MARKER,
+    DUCKDB_NOT_FOUND_ERROR_CLASS_NAMES,
     QUALIFIED_NAME_SEPARATOR,
 )
 from sqlbuild.adapter.contract.exceptions import AdapterUserError
+from sqlbuild.adapter.contract.main.run_relation_read_probe import run_relation_read_probe
 from sqlbuild.adapter.contract.main.same_view_definition import same_view_definition
 from sqlbuild.adapter.contract.models import (
     ColumnInfo,
@@ -46,6 +49,7 @@
     QueryResult,
     RelationGrant,
     RelationInfo,
+    RelationReadProbe,
     RowDiffColumnResult,
     RowDiffCoverage,
     RowDiffPreparedRelations,
@@ -66,6 +70,7 @@
     LoaderLogicalType,
     MigrationTransfer,
     PromotionStrategy,
+    RelationReadStatus,
     StatementSizeLimit,
     TablePromotionMode,
 )
@@ -120,6 +125,23 @@ def supports_relation_age_metadata(self) -> bool:
     def supports_table_freshness_metadata(self) -> bool:
         return False
 
+    def probe_relation_read(self, *, connection: Any, relation: str) -> RelationReadProbe:
+        return run_relation_read_probe(
+            execute=self.execute,
+            connection=connection,
+            relation=relation,
+            classify_not_found=self._classify_relation_not_found,
+        )
+
+    @staticmethod
+    def _classify_relation_not_found(error: BaseException) -> RelationReadStatus | None:
+        return (
+            RelationReadStatus.MISSING
+            if type(error).__name__ in DUCKDB_NOT_FOUND_ERROR_CLASS_NAMES
+            and DUCKDB_MISSING_OBJECT_MARKER in str(error)
+            else None
+        )
+
     def get_table_freshness_metadata(
         self,
         *,

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

@@ -17,6 +17,7 @@
     ExpressionInferenceProfile,
     MigrationStagePlan,
     RelationGrant,
+    RelationReadProbe,
     RowDiffTolerance,
     RowDiffTolerances,
     SnapshotChangeTarget,
@@ -134,6 +135,11 @@ def describe_relation(self, *, connection: Any, relation: str) -> tuple[ColumnIn
         """Return relation column metadata."""
         ...
 
+    @abstractmethod
+    def probe_relation_read(self, *, connection: Any, relation: str) -> RelationReadProbe:
+        """Probe a rendered relation; only the adapter's not-found errors become missing."""
+        ...
+
     @abstractmethod
     def get_table_freshness_metadata(
         self,
@@ -153,7 +159,7 @@ def get_tables_freshness_metadata(
         connection: Any,
         requests: tuple[TableFreshnessRequest, ...],
     ) -> dict[TableFreshnessRequest, TableFreshnessMetadata]:
-        """Return comparable freshness metadata for physical tables in batch."""
+        """Return one freshness outcome per requested table without failing the whole batch."""
         ...
 
     @abstractmethod

src/sqlbuild/adapter/contract/constants.py

@@ -29,3 +29,7 @@
 PYTHON_IDENTIFIER_REPLACEMENT: str = "_"
 
 SNAPSHOT_VERSION_START_COLUMN: str = "__version_start"
+DUCKDB_NOT_FOUND_ERROR_CLASS_NAMES: frozenset[str] = frozenset(
+    {"CatalogException", "BinderException"}
+)
+DUCKDB_MISSING_OBJECT_MARKER: str = "does not exist"

src/sqlbuild/adapter/contract/main/complete_table_freshness_results.py

@@ -0,0 +1,24 @@
+"""Complete batch table freshness results with an outcome for every request."""
+
+from __future__ import annotations
+
+from sqlbuild.adapter.contract.models import TableFreshnessMetadata, TableFreshnessRequest
+
+
+def complete_table_freshness_results(
+    *,
+    requests: tuple[TableFreshnessRequest, ...],
+    results: dict[TableFreshnessRequest, TableFreshnessMetadata],
+    adapter_label: str,
+) -> dict[TableFreshnessRequest, TableFreshnessMetadata]:
+    """Return one outcome per request; a request without a metadata row is unavailable."""
+
+    completed: dict[TableFreshnessRequest, TableFreshnessMetadata] = dict(results)
+    request: TableFreshnessRequest
+    for request in requests:
+        if request in completed:
+            continue
+        completed[request] = TableFreshnessMetadata.unavailable(
+            message=f"{adapter_label} table freshness metadata not found for {request.name}"
+        )
+    return completed

src/sqlbuild/adapter/contract/main/run_relation_read_probe.py

@@ -0,0 +1,48 @@
+"""Direct relation read probe shared by adapter contract implementations."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from sqlbuild.adapter.contract.models import RelationReadProbe
+from sqlbuild.adapter.contract.types import AdapterExecute, RelationReadStatus
+
+
+def run_relation_read_probe(
+    *,
+    execute: AdapterExecute[Any, Any],
+    connection: Any,
+    relation: str,
+    classify_not_found: Callable[[BaseException], RelationReadStatus | None],
+) -> RelationReadProbe:
+    """Read nothing from a relation; only adapter-classified not-found errors become outcomes."""
+
+    try:
+        cursor: object = execute(connection=connection, sql=f"SELECT 1 FROM {relation} WHERE 1=0")
+    except Exception as error:
+        link: BaseException
+        for link in _error_chain(error):
+            status: RelationReadStatus | None = classify_not_found(link)
+            if status is not None:
+                return RelationReadProbe(status=status)
+        raise
+    del cursor
+    return RelationReadProbe(status=RelationReadStatus.READABLE)
+
+
+def _error_chain(error: BaseException) -> tuple[BaseException, ...]:
+    chain: list[BaseException] = []
+    pending: list[BaseException] = [error]
+    while pending:
+        current: BaseException = pending.pop()
+        if any(current is seen for seen in chain):
+            continue
+        chain.append(current)
+        original: object = getattr(current, "__dict__", {}).get("original_error")
+        pending.extend(
+            link
+            for link in (current.__cause__, current.__context__, original)
+            if isinstance(link, BaseException)
+        )
+    return tuple(chain)

src/sqlbuild/adapter/contract/models.py

@@ -16,10 +16,12 @@
     HistoricalSnapshotInsertStyle,
     LifeCycleEventKind,
     MigrationTransfer,
+    RelationReadStatus,
     RetentionChangePhase,
     RetentionScope,
     SnapshotLatestVersionStyle,
     SnapshotUpdateStyle,
+    TableFreshnessStatus,
     TypeFamily,
 )
 from sqlbuild.compiler.compile.types import FunctionLanguage
@@ -135,13 +137,34 @@ def is_transient(self, *, database: str | None = None, schema: str | None, name:
         return bool(relation.is_transient) if relation is not None else False
 
 
+@dataclass(frozen=True)
+class RelationReadProbe:
+    """Whether a rendered relation is readable, and the role used when the adapter knows it."""
+
+    status: RelationReadStatus
+    role: str | None = None
+
+
 @dataclass(frozen=True)
 class TableFreshnessMetadata:
-    """Adapter-observed freshness metadata for one physical table source."""
+    """Adapter freshness outcome for one physical table: observed or unavailable."""
 
     data_version: object
     value_kind: str
     observed_at: datetime | None = None
+    status: TableFreshnessStatus = TableFreshnessStatus.OBSERVED
+    message: str | None = None
+
+    @classmethod
+    def unavailable(cls, *, message: str) -> TableFreshnessMetadata:
+        """Return the outcome for a table whose freshness metadata cannot be compared."""
+
+        return cls(
+            data_version=None,
+            value_kind="timestamp",
+            status=TableFreshnessStatus.UNAVAILABLE,
+            message=message,
+        )
 
 
 @dataclass(frozen=True)

src/sqlbuild/adapter/contract/types.py

@@ -26,6 +26,21 @@ class BuiltinAdapter(StrEnum):
     SQLSERVER = "sqlserver"
 
 
+class RelationReadStatus(StrEnum):
+    """Outcome of probing whether a rendered relation can be read."""
+
+    READABLE = "readable"
+    MISSING = "missing"
+    MISSING_OR_UNREADABLE = "missing_or_unreadable"
+
+
+class TableFreshnessStatus(StrEnum):
+    """Per-table outcome of an adapter freshness metadata lookup."""
+
+    OBSERVED = "observed"
+    UNAVAILABLE = "unavailable"
+
+
 class CursorKind(StrEnum):
     TIMESTAMP = "timestamp"
     INTEGER = "integer"