Data Quality — Contracts and Freshness

Triggers and datasets tell the platform when data moves. Contracts and freshness SLAs tell it whether that data is right and on time — turning run-failure alerts into actual data reliability. Both attach to the dataset, the platform's unit of data-awareness; the platform stores only metadata and validation outcomes, never the data itself.

Contracts: Validate at the Write

Declare the shape your data must have; the SDK enforces it inside your run, before anything reaches the warehouse:

import rebase as rb

prices = rb.Dataset.from_name(
    "nordpool/prices",
    contract=rb.Contract(
        columns=[
            rb.Column("delivery_start", "timestamp", not_null=True),
            rb.Column("price_eur_mwh", "float", between=(-500, 4000)),
            rb.Column("area", "string", isin=["SE1", "SE2", "SE3", "SE4"]),
            rb.Column("volume_mw", "float"),          # nullable by default
        ],
        primary_key=["delivery_start", "area"],       # uniqueness within each written batch
        min_rows=1,
        on_violation="fail",                          # or "warn"
        watermark_column="delivery_start",            # signals carry max(delivery_start)
    ),
    freshness=rb.Freshness(max_age="45m"),
)


@project.workflow(schedule=rb.Cron("*/10 * * * *"))
def ingest_prices():
    df = fetch_nordpool()
    src.write(df, "raw.prices", dataset=prices)
    # validate -> write -> signal (report attached), in that order

Column options

Contract options: primary_key (per-batch uniqueness), min_rows, extra ("ignore" default, or "forbid" undeclared columns), on_violation ("fail" default, or "warn"), watermark_column (derive the signal watermark from the data itself — recommended for time series, so ctx.since becomes a data boundary rather than an ingestion timestamp).

The write pipeline

src.write(df, table, dataset=...) runs validate → write → signal:

Escape hatches: src.write(..., on_violation="warn") overrides per call; validate=False skips validation for known-dirty backfills (the signal is then marked skipped and doesn't count as valid). During a replay the signal step is suppressed entirely (validation still runs) — check ctx.is_replay if the write itself should not happen either. Validate without writing anywhere:

report = prices.validate(df)      # ValidationReport: passed, failures with row samples
prices.validate(df, raise_on_failure=True)

# contracts double as CI assertions:
def test_transform_meets_contract():
    assert prices.validate(transform(FIXTURE)).passed

rebase dataset contract show displays the stored version, and rebase dataset validate <name> file.parquet pre-flights a local file against it.

Contract evolution and compile checks

Contracts have a deliberate lifecycle — code and platform are never allowed to drift silently:

1. Declaring a contract in code registers it in-process the moment the module imports; conflicting definitions of the same dataset warn immediately.
2. First publication is automatic: deploying (or the first write()) publishes a contract the platform doesn't have yet.
3. Changing a published contract is explicit: a deploy whose in-code contract differs from the stored one fails with a field-level diff — run rebase dataset sync <file> to review and apply the change, then deploy again. Runtime writes never overwrite a stored contract either; they warn and keep using the in-code version locally.

rebase dataset check datasets.py     # CI gate: field-level diff, exit 1 on drift
rebase dataset sync  datasets.py     # review the diff, confirm, publish

check is read-only and belongs in CI next to your tests — together with dataset.validate(fixture) assertions it gives Twirl-style compile checks: contract mistakes and code↔platform drift surface before anything runs.

Consumers: don't propagate bad data

@project.workflow(
    trigger=rb.OnUpdate([prices, weather], require="all", only_valid=True,
                        deadline=rb.Cron("0 9 * * *", timezone="Europe/Stockholm")),
)
def da_forecast(ctx: rb.TriggerContext = None):
    ctx.latest["nordpool/prices"]["validation"]   # {"passed": true, "checks": 6, ...}

With only_valid=True, an update whose validation failed (or was skipped) never satisfies the trigger — the next clean signal does. Deadline firings still run regardless (gate closure waits for no one), but ctx shows exactly which inputs were missing or invalid so the workflow can run degraded deliberately.

Freshness: Alert When Data Goes Stale

A freshness SLA alerts on the failure class run-webhooks can't see: nothing failed, but the data stopped arriving — hung schedules, upstream vendor gaps, silently disabled producers.

# Continuous: this feed should tick steadily
prices = rb.Dataset.from_name("nordpool/prices", freshness=rb.Freshness(max_age="45m"))

# Checkpoint: staleness only matters after gate closure
dayahead = rb.Dataset.from_name(
    "forecasts/dayahead",
    freshness=rb.Freshness(max_age="24h", check_at=rb.Cron("15 9 * * *", timezone="Europe/Stockholm")),
)

• Continuous (no check_at): the dataset is stale whenever its last update is older than max_age. Checked every minute by the platform sweeper.
• Checkpoint (check_at cron): evaluated once per occurrence — was the dataset updated within max_age of the check time? A forecast table that's "stale" at 3 a.m. never alerts; one that misses the 09:15 check does.

Alerts are edge-triggered: one dataset.stale webhook when a dataset transitions to stale, one dataset.fresh when it recovers — not one per minute. Enable them on the workspace webhook (same endpoint and HMAC signature as run-failure alerts):

rebase workspace notifications set --webhook-url https://hooks.slack.com/... --on-stale
rebase dataset freshness set forecasts/dayahead --max-age 24h --check-at "15 9 * * *" --timezone Europe/Stockholm
rebase dataset get forecasts/dayahead     # freshness: fresh (age 12m / max 24h) · last validation: passed

Put freshness on externally consumed leaf datasets — one SLA on the forecast output covers every silent failure upstream of it. Datasets never signaled stay unknown rather than alerting, so enabling freshness on a new dataset is safe.

Putting It Together

The reliability loop for a forecasting pipeline, end to end: producers enforce shape at the write (contract), signals carry watermarks and validation reports, consumers react only to clean data (only_valid) but never miss gate closure (deadline), and a freshness SLA on the final output alerts a human when the whole chain silently stops. Every piece is metadata — your data never leaves your warehouse.