tdfs4ds.scheduling.upstream_check¶
Upstream health guard for feature-store process execution.
Why this exists¶
A process failure does not stay local. When level 2 fails, the level-3 view that
reads its features still compiles and still returns rows - just empty or stale
ones - so level 3 ingests, succeeds, and records COMPLETED. The skip predicate in
_upload_features then matches STATUS == 'COMPLETED' at that valid time and
refuses to recompute it, so the hollow values survive every later run until the
next real delta moves FEATURE_STORE_TIME. Nothing in the log says anything is
wrong: the level-3 run genuinely succeeded at what it was asked to do.
The orchestration-level answers to this (stop the wave, or skip the failed
process's descendants) only work for callers that own the whole DAG. This module
is the answer that works for any caller, including a hand-typed
run(process_id): before computing, ask whether this process's upstream
processes are healthy at this valid time, and say so clearly if they are not.
This guard is only possible because a failed ingestion now records FAILED. Before that fix there was no row to look for - the failure left no trace at all.
Predicate¶
The check is "no upstream FAILED at this valid time", deliberately NOT "every upstream COMPLETED at this valid time". Upstreams legitimately run on different cadences (a weekly reference process feeding a daily one), so the stricter reading would refuse to run half of any mixed-cadence store - a guard that cries wolf gets turned off, which is worse than no guard.
Fail-open, but never silent¶
This guard reads the persisted lineage graph, so it is only as good as that graph. Three situations must not be confused, and the verdict separates them:
FS_LINEAGE_GRAPHempty or unreadable -> nothing could be checked;- the process is absent from the graph (typically registered after the last refresh) -> its upstream is unknown;
- the process is known and simply has no upstream -> genuinely verified.
The first two are fail-open (ok=True, so the run proceeds) but report
verified=False together with the exact command that fixes them
(rebuild_lineage_graph). Reporting them as "nothing to check" would read as
an all-clear from a guard that never ran, which is worse than having no guard at
all - the user would believe they are protected.
module-attribute
¶
LINEAGE_HINTS = {LINEAGE_UNAVAILABLE: 'The lineage graph (%s) is empty or could not be read, so upstream health could NOT be verified - this is not an all-clear. Build it once with tdfs4ds.rebuild_lineage_graph(all_data_domains=True), then re-run with check_upstream=True.', LINEAGE_PROCESS_ABSENT: "Process %s is not present in the lineage graph, so its upstream dependencies could NOT be verified - this is not an all-clear. This usually means the process was registered after the last lineage refresh. Refresh just this one with tdfs4ds.rebuild_lineage_graph(objects=['%s'], check=True), or rebuild everything with tdfs4ds.rebuild_lineage_graph(all_data_domains=True)."}
Split follow-up rows into blocking failures and suspect in-flight runs.
rows is an iterable of dicts carrying at least PROCESS_ID and
STATUS. Only rows whose process is in upstream_pids are considered.
Returns (blocking, suspect):
- blocking - upstream processes with a recorded FAILED status. These produced no features for this valid time, so anything computed from them now is incomplete by construction.
- suspect - upstream processes sitting at RUNNING. That is ambiguous: it is either a process still executing in a parallel wave (fine) or one that died without recording anything (not fine). It is reported but does not block by default, because refusing on it would deadlock legitimate concurrent execution.
Render a verdict as a prescriptive, plain-ASCII explanation.
Written to be actionable on its own: it names the offending processes, says what running anyway would mean, and gives both ways forward - because the right call genuinely depends on context the package cannot see (a partially stale input may be perfectly acceptable for a backfill, and unacceptable for a published KPI).
VIEW_NAME of process_id from the process catalog, or None.
Return (upstream_pids, lineage_status, view_name) for process_id.
lineage_status is one of LINEAGE_UNAVAILABLE / LINEAGE_PROCESS_ABSENT
/ LINEAGE_OK. The distinction matters: a process legitimately at the top of
its chain has no upstream and is fully verified, whereas a process missing
from the graph has unknown upstream. Both yield an empty set, so collapsing
them would report an unverifiable process as healthy.
Membership is tested on the process's own node in the OBJECT graph, not on whether it appears in a process-to-process edge - a root process that reads only source tables appears in no such edge yet is perfectly well known.
Follow-up rows for process_ids at feature_store_time.
Assess whether process_id's upstream processes are healthy.
Parameters¶
process_id : str
The process about to run.
feature_store_time : str or None
Valid time to assess. None means the current-validtime rows.
all_domains : bool
Read every data domain when building the process DAG (default).
strict : bool
Also treat an upstream stuck at RUNNING as blocking. Off by default:
a RUNNING upstream is legitimate during parallel execution.
Returns¶
dict
process_id, ok, verified, blocking, suspect,
n_upstream, lineage_status, view_name, feature_store_time,
explanation, error.
ok and verified are different questions and both matter: ok=False
means "a problem was found", while verified=False means "no problem was
found because nothing could be checked". Only ok=False blocks a run, but a
caller that treats verified=False as an all-clear is trusting a guard that
never ran.
Never raises: any failure yields ok=True with error set, because a
guard that cannot read its own inputs must not be what stops a run.