Skip to main content

staleness

Read-before-compute helper shared by the per-file calculation steps.

A calculation step writes one row per (task_hash, file_id), derived from its upstream prediction rows plus its own config. It can therefore skip any file whose row is already present and no older than the predictions it was computed from. This module holds that rule, and the chunked prediction read that goes with it, so the five calculation steps cannot drift apart.

The rule is sound only because task_hash is a per-step Merkle hash over the step's own semantic config and its parents' hashes (bitfount.flows.dag.hashing, and see bitfount.steps.cache_view): a config change or a model-version change lands in a fresh partition, so neither needs detecting here. What is left to detect is freshness relative to the upstream rows, which is what stale_file_ids does.

The rule also requires every stale file with nothing recorded for it to end up with a row, or it stays stale for ever; backfill_missing_data over unrecorded_file_ids is the other half of the contract, and each of the five steps applies it before persisting. The one thing that must never be written off is a batch the datasource failed on rather than the file — see metadata_backfill_reason and DATASOURCE_METADATA_UNAVAILABLE — and no datasource miss, wholesale or per-file, may replace a metric already on file: preserve_existing_metrics is that guard for the reasons a step's own loop records, as unrecorded_file_ids is for the backfill.

Accepted limitation — the read-then-persist window​

A row's processed_at is the wall clock when the row was written, which is later than the source read it was computed from. A prediction written inside that window carries a timestamp older than the row that does not include it, so the source_timestamp > processed_at branch of stale_file_ids never sees it: the file keeps metrics derived from the older prediction, silently and for ever. The window is as wide as the step's own runtime — seconds on an incremental run, much longer on a first pass over a large inventory.

Reaching the window needs a second writer on the same source partition while a calculation step's body runs. Two mechanisms rule that out for every DAG shape in the tree today: the executor awaits every FromRef dependency before a task body starts, so a step's own inference parents are complete before it reads; and flow-run dedup stops two runs of the same DAG overlapping. What remains is two concurrent runs of different DAGs whose inference steps hash to the same partition, one re-inferencing a file edited in place, or a write to the partition from outside the DAG.

Module​

Functions​

backfill_missing_data​

def backfill_missing_data(    output: dict[str, _MetricT | str | None], candidates: Iterable[str], reason: str,) ‑> dict[str, typing.Union[~_MetricT, str, NoneType]]:

Record reason against every candidate output holds no entry for.

A calculation step is asked for candidates (its stale set) but only produces an entry for the files its inputs could carry all the way through: a file the datasource cannot load never reaches get_data_for_files's output, and a file with no prediction row is scoped out by the thickness runner. Without a row of its own such a file trips the row is None branch of stale_file_ids on every run for ever — the empty-stale-set short-circuit can then never fire, and the file's bytes are re-read on every tick where anything else is stale. That is exactly the cost the missing_data: non-retry rule exists to avoid, so the file is written off explicitly here.

Writing the row does NOT abandon a file that is merely awaiting inference. The row makes the file settled under the not error.startswith(...) test only; source_timestamp > processed_at still revives it the moment any source holds a processed_at newer than the row's, so a prediction landing after this row recomputes the file — with the one exception the module docstring records, a prediction written while this very run was between its source read and this write.

Arguments

  • output: The step's per-file result map, modified in place.
  • candidates: The stale ids that hold no result worth keeping — pass unrecorded_file_ids(stale, own), never the raw stale set, or a run that fails to recompute an already-computed file replaces its metric with a permanent null.
  • reason: The missing_data: reason to record — MISSING_DATA_NO_PREDICTION or MISSING_DATA_METADATA.

Returns The same output mapping, for use as an expression.

Raises

  • ValueError: If reason is not a missing_data: reason. Any other prefix is retried by the not error.startswith(...) branch of stale_file_ids, which would defeat the purpose of the backfill and re-read every unloadable file on every run.

metadata_backfill_reason​

def metadata_backfill_reason(metadata_df: pd.DataFrame) ‑> str | None:

Return the reason to write off files the datasource metadata frame lacks.

MISSING_DATA_METADATA settles a file permanently, and that is only correct when the datasource answered for the batch and dropped this file — an unloadable scan, a non-scan artifact. When it returned nothing at all the same write would settle every stale file in the batch on one transient failure (see DATASOURCE_METADATA_UNAVAILABLE), so no backfill is done and the files stay stale for the next run.

Arguments

  • metadata_df: The frame get_data_for_files returned for the stale set.

Returns MISSING_DATA_METADATA when the frame holds at least one row, else None — pass straight to a step's backfill, which skips on None.

preserve_existing_metrics​

def preserve_existing_metrics(    output: dict[str, _MetricT | str | None], unrecorded: Iterable[str],) ‑> dict[str, typing.Union[~_MetricT, str, NoneType]]:

Drop metadata-miss reasons output holds for files that have a metric.

unrecorded_file_ids keeps a failed run from stamping a missing_data: reason over a metric already on file, but it guards only the backfill: a reason the step's own loop placed in output directly bypasses it, and bulk_merge writes the whole row, so persisting it takes the metric to NULL. The thickness runner does exactly that — it records MISSING_DATA_METADATA / DATASOURCE_METADATA_UNAVAILABLE for any file the datasource returned no row for, including one that is stale only because an upstream prediction became newer (source_timestamp > processed_at) and still holds a perfectly good metric from a previous run.

Neither reason says anything about the file: the datasource simply did not answer for it this time. Removing the entry leaves the previous row intact and the file stale for the next run, which is the recoverable direction — and because the file is recorded, the caller's tail MISSING_DATA_NO_PREDICTION backfill skips it too. A computed metric and a calculation_error: from the per-file maths are both left alone: those are this run's answer for the file, not the absence of one.

Arguments

  • output: The step's per-file result map, modified in place.
  • unrecorded: The stale ids that hold no metric — pass the same unrecorded_file_ids(stale, own) list given to the tail backfill. Any id carrying a metadata-miss reason and absent from it already has a metric, and its entry is dropped.

Returns The same output mapping, for use as an expression.

read_for_file_ids​

def read_for_file_ids(    accessor: CacheAccessor, file_ids: Sequence[str], *, chunk_size: int = 500,) ‑> pandas.core.frame.DataFrame:

Read accessor's rows for file_ids only, in bounded chunks.

The filter is built from a free-standing column("file_id") rather than an ORM attribute: an accessor always queries exactly one table, so the unqualified name resolves, and a calculation step therefore never has to import another step's versioned ORM (model_inferences is at v2 while the calculation records are at v1).

An empty file_ids still issues one read, so the returned frame carries the partition's columns — callers guard on .empty and index file_id, and a column-less frame would break them.

Arguments

  • accessor: The partition to read.
  • file_ids: The ids to restrict the read to.
  • chunk_size: Ids per IN (...) clause.

Returns A pandas.DataFrame of the matching rows, with a fresh index.

stale_file_ids​

def stale_file_ids(    candidates: Iterable[str], own: CacheAccessor, sources: Sequence[CacheAccessor],) ‑> list[str]:

Return the subset of candidates whose result is missing or stale.

A file_id is stale when any of the branches in the loop below adds it — each is named here by its own condition, and the rest of this module refers to them that way:

  • row is None — the file has no row in own at all.
  • not error.startswith(...) — its row carries an error that is not a missing_data: reason. The inverse, a row whose error is a missing_data: reason, is the settled case and falls through.
  • not has_metric — its row records neither a metric nor a reason, so it says nothing about the file and cannot be a settled result. Every table here holds exactly one of metrics_json and error, and the current writers keep that invariant, but the first cut of the shared thickness runner returned a bare None for an uncomputable file and its writer set no error at all (SDK 12.0.0-beta.74 to .77). A pod cache that predates the reason strings can therefore hold such rows. Skipping them would strand the file at a null metric for ever, because nothing else in this rule inspects the metric.
  • processed_at is None — its row's timestamp will not parse, so no comparison is possible and the file recomputes.
  • source_timestamp > processed_at — a row for it in sources is strictly newer than its own row's. See the read-then-persist window in the module docstring for the one case this comparison cannot see.

source_timestamp > processed_at is what a presence-only check lacks, and it covers three cases a calculation step actually meets: a prediction that had not landed when the row was first written (these steps left-merge, so such a file still gets a placeholder row); a file edited in place, which re-inference rewrites; and any upstream re-run.

Arguments

  • candidates: The file ids the step was asked to process.
  • own: An accessor over the step's own output partition.
  • sources: Accessors over the upstream prediction partitions this step computes from.

Returns The stale ids, sorted and deduplicated, so the caller's downstream ordering is deterministic.

unrecorded_file_ids​

def unrecorded_file_ids(candidates: Iterable[str], own: CacheAccessor) ‑> list[str]:

Return the candidates whose own row holds no result worth keeping.

These are the ids a missing_data: backfill may be written for: each has either no row at all (row is None in stale_file_ids) or a row carrying no computed metric — whether that is a row whose error is set (settled or not), or one of the pre-reason-string rows that hold neither (see the not has_metric branch of stale_file_ids). The test is metric presence rather than error presence precisely so that second shape is covered: it is stale on every run and, without a backfill, has no way to settle.

A candidate whose row carries a computed metric is deliberately excluded. It is in the stale set only because an upstream row became newer (source_timestamp > processed_at), and a run that then fails to recompute it — a datasource that cannot load the file this time — must not stamp missing_data: over the metric it already has: bulk_merge writes the whole row, so the metric would go to NULL, and a missing_data: row is never retried, so it would stay NULL for ever. Leaving such a file without a row this run keeps it stale for the next one, which is the recoverable direction.

Arguments

  • candidates: The stale ids the step was asked to process.
  • own: An accessor over the step's own output partition.

Returns The subset of candidates safe to write a missing_data: reason for, in the order given.