fix(incremental): reject cursor models with no inputs and allow empty cursor inputs

Makefile

@@ -102,6 +102,7 @@ E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_concurrent_microbatch_build.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_cursor_runtime_failures.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_dropped_relation_rebuild.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_empty_cursor_inputs.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_loader_watermark_build.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_aware_cursor_date_input.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_direct_lifecycle.py \

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/incremental.md

@@ -162,7 +162,33 @@ MODEL (
 
 SQLBuild uses these to determine the replay window. With multiple listed inputs, the end is the conservative common watermark: the minimum of their maxima. This prevents a faster input from advancing the model beyond data available from a slower input.
 
-On a first build, SQLBuild derives the interval from the declared cursor inputs and cursor policy. If it cannot establish a valid interval, the build fails before mutating the destination.
+On a first build, SQLBuild derives the interval from the declared cursor inputs and cursor policy. SQLBuild decides the window while planning, before any staging or delta relation is created. When an input is built earlier in the same run, the window is decided just before the model runs, still before any staging work.
+
+#### Models with no inputs
+
+A cursor-based model must read at least one input that can bound its window, either listed in `cursor_inputs` or read through `__ref()`, `__source()` or `__seed()`. A model that reads none of them could build once but could never work out a window afterwards, so compile rejects it with `P011`:
+
+```
+error[P011]: incremental model 'daily_revenue' reads no inputs, so builds after the first cannot work out their cursor window
+  --> models/finance/daily_revenue.sql:5:3
+    |
+  5 |   cursor revenue_date,
+    |   ^^^^^^^^^^^^^^^^^^^
+  = help: read the data through __ref(), __source() or __seed() so its cursor can bound the window, or use materialized table if the model has no upstream data
+```
+
+A `rolling_window` microbatch model takes its window from the invocation time, so it may read no inputs.
+
+#### Inputs with no rows
+
+Empty inputs are normal: a new development or CI environment, a source in the middle of a reload, or an upstream model that is legitimately empty. When a listed input has no rows, there is no new data to process:
+
+- On an incremental run, the model succeeds without changing its destination, and the plan and build output show the reason, for example `no input rows (stg_payments.paid_at)`. The model's audits still run against the unchanged destination, and downstream models continue. The `--json` build output lists the inputs as `empty_cursor_inputs`, and `sqb plan --json` reports `"resolution_status": "no_input_rows"` with the same inputs in `empty_inputs`.
+- When only some listed inputs are empty, the window still waits for them, so the model makes no progress even though other inputs have new rows. The plan and build output show this as a warning, for example `waiting on empty input stg_refunds.refunded_at; other inputs have new rows`. The build counts it in the summary's `WARN=` total and lists it in the asset's `warnings` in `--json` output, and the JSON output adds `"waiting_on_empty_inputs": true`. The exit code and model status do not change. A watermark microbatch model with `cursor_watermark_mode any` instead progresses from the inputs that have rows.
+- On a first build, the destination is created empty so that models that reference it can build. The next build derives its window from the inputs as usual.
+- A rebuild that would replace an existing destination, such as a query change with `replay_on_change full` or a microbatch `--full-refresh`, fails instead of leaving the table empty. Load the inputs first, or set the window explicitly with `--start-cursor-ts` and `--end-cursor-ts` (or the `-int` forms).
+
+Use [source freshness](sources.md) checks and source audits to catch an input that should not be empty. If SQLBuild cannot read the cursor bounds of an input at all, the plan fails with `S302`, naming the model and its inputs.
 
 ### Cursor bounds in model SQL
 

src/sqlbuild/cli/commands/classes/build_progress_callbacks.py

@@ -22,6 +22,10 @@
 from sqlbuild.cli.progress.models import AuditDisplayEntry, ExecutionCounts
 from sqlbuild.compiler.auditing.types import AuditEvaluationMode, AuditOutcome, AuditRunScope
 from sqlbuild.compiler.compile.types import CompiledResourceType
+from sqlbuild.compiler.planner.main.cursor_window.no_input_rows_display import format_no_input_rows
+from sqlbuild.compiler.planner.main.cursor_window.waiting_on_empty_inputs_display import (
+    format_waiting_on_empty_inputs,
+)
 from sqlbuild.compiler.planner.main.execution.cursor_bound_display import cursor_bound_display
 from sqlbuild.compiler.planner.main.execution.inclusive_cursor_end import inclusive_cursor_end
 from sqlbuild.compiler.planner.main.execution.materialization_type_display import (
@@ -597,10 +601,18 @@ def _write_model_result(self, *, ctr: str, model_result: ModelExecutionResult) -
                 elif event.kind == LifeCycleEventKind.LOG:
                     self._write_log_block(event.content)
 
-        batch_line: str | None = _format_streaming_batch_summary(model_result=model_result)
+        batch_line: str | None = _format_streaming_batch_summary(
+            model_result=model_result
+        ) or _format_empty_cursor_inputs(model_result=model_result)
         if batch_line is not None:
             self._stream.write(f"{self._style.muted(batch_line)}\n")
             self._stream.flush()
+        if model_result.waiting_on_empty_inputs and model_result.empty_cursor_inputs:
+            waiting_line: str = (
+                f"         {format_waiting_on_empty_inputs(model_result.empty_cursor_inputs)}"
+            )
+            self._stream.write(f"{self._style.warning(waiting_line)}\n")
+            self._stream.flush()
 
         sub_pad: str = " " * (self._prefix_width + _SUB_INDENT)
         sub_nw: int = self._name_width - _SUB_INDENT
@@ -1524,7 +1536,11 @@ def _format_streaming_batch_summary(*, model_result: ModelExecutionResult) -> st
     if model_result.batch_size is not None:
         count_text = f"{count_text} ({model_result.batch_size})"
     parts: list[str] = [count_text]
-    if model_result.cursor_range_start is not None and model_result.cursor_range_end is not None:
+    if (
+        model_result.cursor_range_start is not None
+        and model_result.cursor_range_end is not None
+        and not model_result.empty_cursor_inputs
+    ):
         start: str = cursor_bound_display(
             value=model_result.cursor_range_start,
             cursor_type=model_result.cursor_type,
@@ -1538,6 +1554,8 @@ def _format_streaming_batch_summary(*, model_result: ModelExecutionResult) -> st
         parts.append(f"range {start} \u2192 {end}")
     if model_result.rows_affected is not None:
         parts.append(_format_abbreviated_rows(count=model_result.rows_affected))
+    if model_result.empty_cursor_inputs and not model_result.waiting_on_empty_inputs:
+        parts.append(format_no_input_rows(model_result.empty_cursor_inputs))
     if model_result.microbatch_recovery_batch_count:
         parts.append(f"{model_result.microbatch_recovery_batch_count} recovery")
     if model_result.microbatch_synthetic_completion_count:
@@ -1547,6 +1565,14 @@ def _format_streaming_batch_summary(*, model_result: ModelExecutionResult) -> st
     return f"         {'    '.join(parts)}"
 
 
+def _format_empty_cursor_inputs(*, model_result: ModelExecutionResult) -> str | None:
+    """Return why a cursor window selected no rows, if its inputs were empty."""
+
+    if not model_result.empty_cursor_inputs or model_result.waiting_on_empty_inputs:
+        return None
+    return f"         {format_no_input_rows(model_result.empty_cursor_inputs)}"
+
+
 def _format_abbreviated_rows(*, count: int) -> str:
     """Format a row count with K/M/B abbreviation above 10,000."""
 

src/sqlbuild/cli/output/_helpers/cursor_plan.py

@@ -60,8 +60,10 @@ def build_cursor_plan_details(*, entry: ModelPlanEntry) -> CursorPlanDetails | N
             batch_size=effective_batch_size,
             cursor_type=entry.cursor_type,
         )
-    if resolved_bounds is not None:
-        resolution_status: CursorResolutionStatus = CursorResolutionStatus.RESOLVED
+    if entry.empty_cursor_inputs:
+        resolution_status: CursorResolutionStatus = CursorResolutionStatus.NO_INPUT_ROWS
+    elif resolved_bounds is not None:
+        resolution_status = CursorResolutionStatus.RESOLVED
     elif runtime_owned:
         resolution_status = CursorResolutionStatus.DEFERRED
     else:
@@ -77,6 +79,8 @@ def build_cursor_plan_details(*, entry: ModelPlanEntry) -> CursorPlanDetails | N
         declared_batch_size=entry.batch_size,
         effective_batch_size=effective_batch_size,
         planned_batch_count=planned_batch_count,
+        empty_inputs=entry.empty_cursor_inputs,
+        waiting_on_empty_inputs=entry.waiting_on_empty_inputs,
     )
 
 

src/sqlbuild/cli/output/_helpers/execution_result_document.py

@@ -623,6 +623,8 @@ def _format_model_assets(
                     else None
                 ),
                 "warnings": result.warning_messages,
+                "empty_cursor_inputs": list(result.empty_cursor_inputs) or None,
+                "waiting_on_empty_inputs": result.waiting_on_empty_inputs or None,
                 "future_cursor_safety": serialize_future_cursor_safety(result.future_cursor_safety),
                 "maximum_start_safety": serialize_maximum_start_safety(result.maximum_start_safety),
                 "microbatch": _format_microbatch_result(result),

src/sqlbuild/cli/output/_helpers/plan_json.py

@@ -231,6 +231,8 @@ def _serialize_cursor_details(*, details: CursorPlanDetails) -> dict[str, object
         "declared_batch_size": details.declared_batch_size,
         "effective_batch_size": details.effective_batch_size,
         "planned_batch_count": details.planned_batch_count,
+        "empty_inputs": list(details.empty_inputs),
+        "waiting_on_empty_inputs": details.waiting_on_empty_inputs,
     }
 
 

src/sqlbuild/cli/output/_helpers/plan_text.py

@@ -19,6 +19,10 @@
 from sqlbuild.compiler.pipeline.models import PythonPlanEntry
 from sqlbuild.compiler.planner.constants import SOURCE_FRESHNESS_UNKNOWN_WARNING_TITLE
 from sqlbuild.compiler.planner.main.changes.query_diff import format_query_diff
+from sqlbuild.compiler.planner.main.cursor_window.no_input_rows_display import format_no_input_rows
+from sqlbuild.compiler.planner.main.cursor_window.waiting_on_empty_inputs_display import (
+    format_waiting_on_empty_inputs,
+)
 from sqlbuild.compiler.planner.main.execution.cursor_bound_display import cursor_bound_display
 from sqlbuild.compiler.planner.main.execution.inclusive_cursor_end import inclusive_cursor_end
 from sqlbuild.compiler.planner.main.execution.model_materialization_label import (
@@ -942,6 +946,14 @@ def _append_cursor_detail(
         requested_start: str = details.requested_start or "earliest available"
         requested_end: str = details.requested_end or "latest available"
         lines.append(f"    requested: {requested_start} -> {requested_end}")
+    if details.empty_inputs and details.waiting_on_empty_inputs:
+        lines.append(
+            CliStyle(use_color=True).warning(
+                f"    window: {format_waiting_on_empty_inputs(details.empty_inputs)}"
+            )
+        )
+    elif details.empty_inputs:
+        lines.append(f"    window: {format_no_input_rows(details.empty_inputs)}")
     if show_range and details.resolved_bounds is not None:
         lines.append(
             f"    range: {_format_cursor_range(bounds=details.resolved_bounds, entry=entry)}"

src/sqlbuild/cli/output/models.py

@@ -91,6 +91,8 @@ class CursorPlanDetails:
     declared_batch_size: str | None
     effective_batch_size: str | None
     planned_batch_count: int | None
+    empty_inputs: tuple[str, ...] = ()
+    waiting_on_empty_inputs: bool = False
 
 
 @dataclass(frozen=True)

src/sqlbuild/cli/output/types.py

@@ -25,6 +25,7 @@ class CursorResolutionStatus(StrEnum):
     DEFERRED = "deferred"
     RESOLVED = "resolved"
     UNAVAILABLE = "unavailable"
+    NO_INPUT_ROWS = "no_input_rows"
 
 
 class PlanRowKind(StrEnum):

src/sqlbuild/compiler/compile/_helpers/config/cursor_inputs.py

@@ -0,0 +1,103 @@
+"""Reject cursor-based incremental models that have no input to bound their cursor window."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from sqlbuild.compiler.compile.constants import (
+    CURSOR_INPUTS_CONFIG_KEY,
+    CURSOR_MODEL_WITHOUT_INPUTS_CODE,
+)
+from sqlbuild.compiler.compile.models import CompileModelInput, CompilerDiagnostic
+from sqlbuild.compiler.compile.types import (
+    CompiledResourceType,
+    DiagnosticPhase,
+    DiagnosticSeverity,
+)
+from sqlbuild.compiler.discovery.main._model_header_entry_span import get_model_header_entry_span
+from sqlbuild.compiler.planner.types import MaterializationType, MicrobatchStrategy
+from sqlbuild.compiler.references.types import SqlReferenceKind
+from sqlbuild.spec.contracts.main.get_config_str import get_config_str
+from sqlbuild.spec.contracts.models import SourceLocation
+
+_CURSOR_CONFIG_KEY: str = "cursor"
+_RELATION_REFERENCE_KINDS: frozenset[SqlReferenceKind] = frozenset(
+    {
+        SqlReferenceKind.REF,
+        SqlReferenceKind.SOURCE,
+        SqlReferenceKind.SEED,
+        SqlReferenceKind.DBT_REF,
+    }
+)
+_HELP: str = (
+    "read the data through __ref(), __source() or __seed() so its cursor can bound the window, "
+    "or use materialized table if the model has no upstream data"
+)
+
+
+def cursor_model_without_inputs_diagnostics(
+    *, model_inputs: tuple[CompileModelInput, ...]
+) -> tuple[CompilerDiagnostic, ...]:
+    """Return one error per cursor-based incremental model with no cursor inputs."""
+
+    return tuple(
+        _diagnostic(model_input)
+        for model_input in model_inputs
+        if _lacks_cursor_inputs(model_input)
+    )
+
+
+def _lacks_cursor_inputs(model_input: CompileModelInput) -> bool:
+    values: dict[str, object] = model_input.config.values
+    if get_config_str(values=values, key="materialized") != MaterializationType.INCREMENTAL:
+        return False
+    if get_config_str(values=values, key=_CURSOR_CONFIG_KEY) is None:
+        return False
+    if (
+        get_config_str(values=values, key="microbatch_strategy")
+        == MicrobatchStrategy.ROLLING_WINDOW
+    ):
+        return False
+    if values.get(CURSOR_INPUTS_CONFIG_KEY) is not None:
+        return False
+    return not any(
+        reference.ref_kind in _RELATION_REFERENCE_KINDS for reference in model_input.references
+    )
+
+
+def _diagnostic(model_input: CompileModelInput) -> CompilerDiagnostic:
+    name: str = model_input.model_file.file_path.stem
+    return CompilerDiagnostic(
+        phase=DiagnosticPhase.COMPILE,
+        severity=DiagnosticSeverity.ERROR,
+        code=CURSOR_MODEL_WITHOUT_INPUTS_CODE,
+        message=(
+            f"incremental model '{name}' reads no inputs, so builds after the first "
+            "cannot work out their cursor window"
+        ),
+        resource_type=CompiledResourceType.MODEL,
+        resource_name=name,
+        location=_cursor_location(
+            path=model_input.model_file.relative_path,
+            contents=model_input.model_file.contents,
+        ),
+        help=_HELP,
+    )
+
+
+def _cursor_location(*, path: Path, contents: str) -> SourceLocation:
+    span: tuple[int, int] | None = get_model_header_entry_span(
+        contents=contents, key=_CURSOR_CONFIG_KEY
+    )
+    start: int = 0 if span is None else span[0]
+    line: int = contents.count("\n", 0, start) + 1
+    column: int = start - (contents.rfind("\n", 0, start) + 1) + 1
+    if span is None:
+        return SourceLocation(path=path, line=line, column=column)
+    return SourceLocation(
+        path=path,
+        line=line,
+        column=column,
+        end_line=line,
+        end_column=column + span[1] - span[0],
+    )

src/sqlbuild/compiler/compile/constants.py

@@ -20,6 +20,7 @@
 HARD_CODED_PROJECT_RELATION_CODE: str = "P008"
 UNNEEDED_SQL_ANALYSIS_OPT_OUT_CODE: str = "P009"
 MISSING_DESCRIPTION_CODE: str = "P010"
+CURSOR_MODEL_WITHOUT_INPUTS_CODE: str = "P011"
 DESCRIPTION_REQUIRED_INPUT_KINDS: dict[str, str] = {
     "model_files": "model",
     "scenario_files": "scenario",

src/sqlbuild/compiler/compile/main/_build_compile_inputs.py

@@ -22,6 +22,9 @@
 from sqlbuild.compiler.compile._helpers.audit_factories.core import (
     build_audit_factory_orphan_diagnostics,
 )
+from sqlbuild.compiler.compile._helpers.config.cursor_inputs import (
+    cursor_model_without_inputs_diagnostics,
+)
 from sqlbuild.compiler.compile._helpers.diagnostics.collector import (
     with_collected_compile_diagnostics,
 )
@@ -197,6 +200,7 @@ def build_compile_inputs(
             source_inputs=source_inputs,
             function_inputs=sql_function_inputs,
         ),
+        *cursor_model_without_inputs_diagnostics(model_inputs=model_build.inputs),
     )
     return CompileProjectInputs(
         project_config=discovered_inputs.project_config,

src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py

@@ -287,6 +287,33 @@ def model_header_columns_span(
     )
 
 
+def model_header_entry_span(*, contents: str, key: str) -> tuple[int, int] | None:
+    """Return authored byte offsets for one top-level `key value` MODEL header entry."""
+
+    header_match: re.Match[str] | None = _MODEL_HEADER_PATTERN.match(contents)
+    if header_match is None:
+        return None
+    header_start: int = header_match.start("header")
+    tokens: list[_ModelHeaderToken] = _tokenize_model_header_for_spans(header_match.group("header"))
+    depth: int = 0
+    for index, token in enumerate(tokens):
+        if token.kind == _MODEL_HEADER_SYMBOL_TOKEN and token.value == _MODEL_HEADER_OPEN_PAREN:
+            depth += 1
+        elif token.kind == _MODEL_HEADER_SYMBOL_TOKEN and token.value == _MODEL_HEADER_CLOSE_PAREN:
+            depth -= 1
+        elif (
+            depth == 0
+            and token.kind == _MODEL_HEADER_WORD_TOKEN
+            and token.value == key
+            and index + 2 < len(tokens)
+        ):
+            entry_end: int = header_start + tokens[index + 2].position
+            while entry_end > header_start + token.position and contents[entry_end - 1].isspace():
+                entry_end -= 1
+            return header_start + token.position, entry_end
+    return None
+
+
 def model_header_body_span(*, contents: str) -> tuple[int, int] | None:
     """Return authored byte offsets for the MODEL header body."""
 

src/sqlbuild/compiler/discovery/main/_model_header_entry_span.py

@@ -0,0 +1,13 @@
+"""Public authored MODEL header entry span query."""
+
+from __future__ import annotations
+
+from sqlbuild.compiler.discovery._helpers.sql.model_files import (
+    model_header_entry_span as _model_header_entry_span,
+)
+
+
+def get_model_header_entry_span(*, contents: str, key: str) -> tuple[int, int] | None:
+    """Return authored byte offsets for one top-level `key value` MODEL header entry."""
+
+    return _model_header_entry_span(contents=contents, key=key)