tdfs4ds.workflow_assessment¶
Workflow assessment: is the pipeline well built? (0.3.3.25)
store_consistency answers "do the catalogs and the storage line up?". This
module answers a different question: is the workflow well implemented?
- Entity keys (area
entity). Users often put every id column they know intoentity_id. A column the others determine splits nothing, yet it widens the primary index of every storage table, so a lookup by the real key cannot use it, and an attribute that changes over time starts a new entity at each change. The scan counts distinct key tuples in the storage tables, over the whole history and now, with each column left out in turn. - Order and synchronisation (area
time). Did every process run after the processes it reads, at each business date? Do the two ends of a dependency follow the same TimeManager? Is a TimeManager-driven process reading a dataset that does not move with its valid time (point-in-time leakage)? - Runs (area
runs). FilterManager segments missing or failed at the latest date, hollow runs (COMPLETED,0 rows), rows stuck at RUNNING. - Process SQL (area
process). Live SQL that differs from the stored version, and histories computed under several versions. - Storage (area
storage). Primary-index skew, entity columns typed differently across tables, entity keys contained in other ones.
Three layers, like the other report sections: :func:collect_workflow_inputs
is the only I/O, :func:build_workflow_model is pure, :func:render_workflow_html
returns HTML. :func:assess_workflow runs the first two.
Rules:
- No LLM. Every check is a count, a dictionary read, a graph walk or a hash comparison; every text is a fixed template. A test guards the imports.
- Every finding names its fix and nothing here runs it.
- A check that cannot run is listed in
unverifiablewith its reason; it is never reported as passing. - A dependency between entity columns is observed in the data, never a rule: the finding says so and asks for the business to confirm.
- Primary store only.
module-attribute
¶
AREA_TITLES = {'lineage': 'Lineage integrity', 'entity': 'Entity keys', 'time': 'Order and synchronisation', 'runs': 'Runs and segments', 'process': 'Process SQL', 'storage': 'Storage layout'}
module-attribute
¶
WORKFLOW_KIND_LABELS = {'redundant_key_column': 'Entity column determined by the other key columns', 'key_column_changes_over_time': 'Entity column that changes over time', 'entity_key_superset': 'Entity key containing another entity key', 'entity_column_type_mismatch': 'Entity column typed differently across tables', 'storage_skew': 'Storage table skewed across AMPs', 'entity_pi_inconsistent': 'Storage tables of one entity with different primary indexes', 'pi_not_aligned_with_source': 'Primary index not aligned with the largest source', 'partition_not_aligned_with_source': 'Partitioning not aligned with the largest source', 'segments_rescan_source': 'FilterManager segments each reading the whole source', 'segment_filter_defeats_pruning': 'Segment filter on the partitioning column that the SQL keeps from pruning', 'time_filter_by_join': 'TimeManager read through a join instead of a scalar filter', 'time_filter_not_on_partitioning': 'Business date compared with a column the source is not partitioned on', 'time_filter_defeats_pruning': 'Business-date filter on the partitioning column that the SQL keeps from pruning', 'fs_partition_not_on_filter': 'Storage tables not partitioned on the segmentation columns', 'ran_before_upstream': 'Process started before its upstream had finished', 'upstream_recomputed_after': 'Upstream recomputed after the process ran', 'upstream_missing_at_date': "Upstream not computed at the process's business date", 'completed_on_failed_upstream': 'Process completed on a failed or empty upstream', 'ahead_of_upstream': 'Process ahead of its upstream in business time', 'mixed_time_modes': 'Dependency between real time and a TimeManager', 'different_time_drivers': 'Dependency between two independent TimeManagers', 'process_not_on_domain_clock': 'Process of a data domain not on the domain clock', 'processes_share_time_filtered_view': 'Processes sharing the domain clock through a view', 'point_in_time_leak': 'TimeManager process reading a current dataset', 'fixed_as_of_input': 'TimeManager process reading a fixed-date dataset', 'input_other_time_manager': 'Process reading a dataset on another TimeManager', 'segments_missing': 'FilterManager segments not computed at the latest date', 'segment_failed': 'FilterManager segment failed at the latest date', 'hollow_run': 'Run that wrote no rows', 'stuck_running': 'Run still marked RUNNING', 'sql_drift': 'Live SQL different from the stored version', 'history_mixes_versions': 'History computed under several SQL versions', 'dataset_reads_removed_process': 'Dataset reading the values of a removed process', 'dataset_reads_removed_feature': 'Dataset reading a feature removed from the catalog', 'dataset_reads_stale_table': 'Dataset reading a storage table the catalog no longer uses for the feature', 'dataset_reads_missing_table': 'Dataset reading a storage table that no longer exists', 'dataset_feature_no_rows': 'Dataset feature with no stored rows', 'relocation_reverted': 'Relocated feature registered or written in its old table again (older client)', 'feature_without_process': 'Feature no process computes any more', 'process_view_missing': 'Process whose view no longer exists', 'dataset_view_missing': 'Dataset whose view no longer exists', 'process_reads_missing_object': 'Process view reading an object that no longer exists', 'process_view_broken': 'Process view that no longer compiles', 'process_reads_unregistered_dataset': 'Process view reading a dataset no longer registered', 'lineage_edge_to_unknown': 'Lineage edge to a view that is no registered process', 'dataset_reads_other_store': 'Dataset reading another feature store (not checked from this one)', 'view_does_not_compile': 'View that does not compile (EXPLAIN fails)', 'view_not_verifiable': 'View this user may not EXPLAIN (missing right): not verified', 'process_reads_inaccessible_object': 'Process view reading an object this user has no right on', 'lineage_stops_no_access': 'Lineage stopping at a view whose definition this user cannot read', 'lineage_not_recorded': 'View whose inputs the lineage graph does not record', 'lineage_stops_unreadable': 'Lineage stopping at a view SHOW VIEW refuses', 'views_not_compiled': 'Views not compiled (over the limit)', 'lineage_split': 'Lineage made of separate networks', 'lineage_isolated': 'Process or dataset with no lineage link', 'process_not_read': 'Process nobody reads (dead end)', 'detached_branch': 'Lineage branch no process or dataset reads (dead branch)'}
module-attribute
¶
LINEAGE_KINDS = ('dataset_reads_removed_process', 'dataset_reads_removed_feature', 'feature_without_process', 'process_view_missing', 'dataset_view_missing', 'process_reads_missing_object', 'process_view_broken', 'process_reads_unregistered_dataset', 'lineage_edge_to_unknown', 'dataset_reads_stale_table', 'dataset_reads_missing_table', 'process_reads_inaccessible_object')
module-attribute
¶
module-attribute
¶
CHECKS = ('entity_scan', 'entity_layout', 'skew', 'distribution', 'order', 'leakage', 'segments', 'sql', 'integrity', 'network', 'dataset_rows')
module-attribute
¶
module-attribute
¶
module-attribute
¶
DIMENSIONS = [('lineage', 'Lineage', ('integrity', 'network')), ('entity', 'Entity keys', ('entity_scan',)), ('order', 'Order and time', ('order',)), ('inputs', 'Inputs in time', ('leakage',)), ('runs', 'Runs', ('segments',)), ('sql', 'Process SQL', ('sql',)), ('storage', 'Storage and pruning', ('distribution', 'skew'))]
module-attribute
¶
module-attribute
¶
_CHECK_DIMENSION = {'integrity': 'lineage', 'network': 'lineage', 'dataset_rows': 'lineage', 'entity_scan': 'entity', 'order': 'order', 'leakage': 'inputs', 'segments': 'runs', 'runs': 'runs', 'sql': 'sql', 'distribution': 'storage', 'storage': 'storage'}
module-attribute
¶
_EXEC_CSS = '<style>\n.twa-exec{margin:4px 0 18px}\n.twa-exec h3{margin:0 0 8px;font-size:1.05rem}\n.twa-cards{display:flex;flex-wrap:wrap;gap:10px;margin:6px 0 10px}\n.twa-card{border:1px solid rgba(127,127,127,.35);border-left:4px solid #FF5F00;border-radius:8px;padding:6px 12px;min-width:110px}\n.twa-card b{display:block;font-size:1.35rem;line-height:1.2}\n.twa-card span{font-size:.78rem;opacity:.8}\n.twa-card.e{border-left-color:#C62828}.twa-card.w{border-left-color:#EF6C00}\n.twa-card.i{border-left-color:#2F6E9E}.twa-card.o{border-left-color:#2E7D32}\n.twa-verdict{margin:4px 0 8px}\n.twa-top{margin:0 0 12px 18px;padding:0}\n.twa-gridwrap{overflow-x:auto;max-width:100%}\ntable.twa-grid{border-collapse:separate;border-spacing:3px;font-size:.82rem}\ntable.twa-grid th{font-weight:600;text-align:left;padding:2px 6px;white-space:nowrap}\ntable.twa-grid th.d{text-align:center;max-width:90px;white-space:normal}\ntable.twa-grid td.p{padding:2px 8px;white-space:nowrap;max-width:320px;overflow:hidden;text-overflow:ellipsis}\ntable.twa-grid tr.wave td{padding:8px 4px 2px;font-weight:600;color:#FF5F00;border-bottom:1px solid rgba(127,127,127,.3)}\ntd.twa-c{width:74px;text-align:center;color:#fff;border-radius:4px;font-weight:700}\ntd.twa-error{background:#C62828}td.twa-warning{background:#EF6C00}td.twa-info{background:#2F6E9E}\ntd.twa-ok{background:#2E7D32}td.twa-na{background:rgba(127,127,127,.35);color:inherit;font-weight:400}\n.twa-legend{font-size:.78rem;opacity:.85;margin-top:6px}\n.twa-legend i{display:inline-block;width:12px;height:12px;border-radius:3px;vertical-align:middle;margin:0 4px 0 10px}\n</style>'
module-attribute
¶
_CELL_MARK = {'error': '✖', 'warning': '▲', 'info': '●', 'ok': '✓', 'n/a': '–'}
module-attribute
¶
_CELL_WORD = {'error': 'error', 'warning': 'warning', 'info': 'info', 'ok': 'nothing to report', 'n/a': 'not checked'}
module-attribute
¶
Wrapper around the global logger that only emits logs when
tdfs4ds.DISPLAY_LOGS is True. level is a string like "info", "error", etc.
Build db.obj with each part quoted only as needed.
qualified('TDB', 'FS_X') == 'TDB.FS_X' (unchanged for ordinary names);
qualified('SVC@CORP', 'FS_X') == '"SVC@CORP".FS_X'. An empty/None
database yields the bare object name, so unqualified callers keep working.
force=True quotes both parts unconditionally — only for the sites that
have always emitted quotes and whose exact text is already persisted (dataset
view DDL, returned volatile-table names).
Escape value for use inside a single-quoted SQL string literal.
For the DBC probes and catalog predicates that compare a name as data
(WHERE DatabaseName = '…'). The caller supplies the surrounding quotes.
A finding. pids: the processes it concerns (the executive summary places
it on their rows); ref: entity=(DOMAIN, KEY) or table='DB.T' for the
findings that are mapped to processes later, in the model.
Place findings (a slice just produced) on the processes pids, unless a finding already names its own.
FS_FOLLOW_UP rows -> pid, run_id, start, end, status, vdate, dkey, seg.
vdate is the business date (VALIDTIME_DATE, NaT for a real-time run),
dkey its day as text ('' in real time), seg the applied filter ('' when
unsegmented). Pure.
{(pid, dkey): state} for the TimeManager rows (dkey != '').
A state describes the last computation at that business date, per segment:
ok_start / ok_end (first start and last end of the latest COMPLETED
row of each segment), failed / hollow (a segment whose latest row
failed / wrote nothing, with its start), completed, vdate. Pure.
{"driver": {pid: manager}, "group": {manager: group id}} from the
time-driver model. A manager outside any synchronised / clone group is its
own group. Pure.
True / False when both processes have a known driver, else None.
The process dependencies without the transitive ones (q -> p). A
cycle keeps every edge. Pure.
A process of a data domain that has a clock (0.3.6.0) but reads no manager on its
dates (or another one): run(data_domain=...) / roll_out(data_domain=...) cannot
keep it in step. Pure.
Findings for each dependency q -> p (q upstream). Pure.
TimeManager rows are compared business date by business date; real-time rows by overlap of the runs only (two real-time processes on different cadences are normal).
Dataset nodes a process view reads, walking through plain views and stopping at datasets and at other processes' views. Pure.
(findings, unverifiable). A process counts as TimeManager-driven when
its view reads a manager or its runs carry a business date. Pure.
segment_findings(fu: DataFrame, expected: Dict[str, int], names=None, now=None, stuck_hours: float = 12) -> List[dict]
Missing / failed segments and hollow runs at each process's latest date,
rows stuck at RUNNING. expected: {pid: number of FilterManager rows}. Pure.
Findings from :func:tdfs4ds.dependencies.lineage_gaps. Each names its
fix; a gap that leaves a dead branch also carries removal, the objects to
pass to :func:tdfs4ds.plan_removal. With a scope (processes or data
domains), only the gaps touching it are kept. Pure.
dataset_feature_no_rows findings from the storage-level check. rows:
{"reads": dependencies.dataset_feature_reads(H), "scan": scan_storage_orphans(
by_version=True), "completed": {process id: COMPLETED runs} or None}.
A feature read has no rows when the table holds none for its (FEATURE_ID, FEATURE_VERSION) at the valid time the dataset reads (now for a current dataset, any time otherwise). An INNER dataset then returns no row at all (error); a FULL OUTER one reads NULL for the column (warning). The follow-up tells a process that never completed from one whose rows are gone. A table that could not be scanned is skipped (the caller reports it). Pure.
relocation_reverted findings from the storage-level check. strays:
:func:tdfs4ds.feature_store.relocation.read_relocation_strays.
A relocated feature has a placement tag. A client older than 0.3.3.30 does not know it: when it registers the feature again (a type change, or removed and registered anew) it names the untagged table (the catalog row follows) and writes its values there, so the history is split over two tables and the datasets read either. Pure.
({node: {"pid", "label"}}, {node: {"label"}}) of the registered
process and dataset views, keyed like the persisted lineage graph and
labelled like :func:tdfs4ds.dependencies.build_dependency_graph. Pure.
:func:tdfs4ds.lineage.network_health.network_findings, without the
objects the gap check already reported. With a process scope, the findings
about other processes are dropped; those about the whole network stay. Pure.
sql_findings(versions: Dict[str, list], drift: Dict[str, dict], fu: DataFrame, names=None) -> List[dict]
versions: {pid: [{"PROCESS_VERSION", "VALID_FROM", "VALID_TO"}]};
drift: {pid: {"since", "changed_steps"}}. Pure.
Distinct entity tuples of a storage table, for the key and the key minus
each column, over the whole history (H:) and now (C:). One request.
The temporal qualifier sits inside each derived table (a derived table refuses CURRENT, and a UNION branch refuses a leading qualifier: Error 9345); the label is CAST because the first branch types the column.
(determined over the whole history, determined now only). Pure.
Greedy reduction: drop a column the others determine over the whole history, re-count, until none is left (at most one round per column). run_sql returns the rows of a statement. Pure apart from run_sql.
Findings from :func:reduce_entity_key results. pi: {(DB, TABLE):
[columns]}. Pure.
Each entity-scan finding with the scan result it came from (same object).
Two entity keys of one domain where one contains the other. Pure.
The family that decides whether two columns join without conversion.
columns: [(db, table, column, family)]; entity_cols: {(DB, TABLE):
{COL}}. Pure.
sizes: {(db, table): [CurrentPerm per AMP]}. Skew = 1 - avg / max. Pure.
Tables a process view reads: walk through plain views, stop at datasets, at other processes' views and at manager / FS objects. Pure.
The entity columns that carry the source's primary-index columns, in the
source's PI order; None unless every PI column maps. origins: {entity
col: {"node", "column"}} from :func:column_lineage. Pure.
How the rows reach the probe table of a staging CTAS EXPLAIN: "local"
(the spool feeding it is built locally), "redistributed" (by hash code),
"duplicated", or None when the text does not say. Read from the steps,
never counted. Pure.
Correlation names the view SQL gives source_node: EXPLAIN names a table
by them (DB_SOURCE.A in view ..., seen live), not by its name. Pure.
How an EXPLAIN reads source_node ('DB.TABLE'): "partitions" (partition
elimination, incl. a single partition), "all_rows" (no elimination), or
None when no step names it. aliases: correlation names of the table (see
:func:table_aliases). The worst step wins. Pure.
Columns of a table named in its partitioning expression (FEATURE_ID and the valid-time columns left out). Pure.
How the view SQL reads its TimeManager: {"scalar_columns": [...],
"join": bool}. Scalar = <column> <op> (SELECT ... FROM <manager>), the
form the optimizer can substitute and prune on (src.d <= (SELECT
BUSINESS_DATE FROM tm), seen live); join = the manager in a FROM/JOIN list.
Pure, on the flattened text.
The entity column the view passes source_column of the largest source as.
A FilterManager column names a partitioning column of the largest source, directly or as the entity column the view passes it as. Pure.
distribution_findings(rows: List[dict], *, align_min_mb: float = 1000, skew_threshold: float = 0.25, names=None) -> List[dict]
rows: one per process (see :func:collect_distribution). Grouped per
(domain, entity key): the recommendation is for the entity, never one
process. Pure.
The process id a finding's object ends with, as <view> (<process id>).
{pid: wave}: the execution waves of the process DAG (wave 0 = runs
first), the same layering as :func:tdfs4ds.analyze_parallel_plan. Empty on
a cycle; absent processes (unknown to the lineage graph) get no wave. Pure.
Give each finding the processes it concerns (pids), for the summary:
its own process (object <view> (<pid>)), the processes of its entity
(entity=(DOMAIN, KEY)) or those writing its table (table='DB.T'). A
store-level finding (an entity column typed differently) keeps none. Pure.
One row per process, ordered by execution wave, one cell per dimension:
the worst severity of its findings, "ok" when the dimension was checked
and found nothing, "n/a" when it was not (check not run, or not
verifiable for that process). Also the verdict counts and the most frequent
issues. Pure.
The dead branch each lineage gap leaves: [{"from", "kind", "branch"}],
only the gaps whose branch is not empty. Pure.
The lineage network section: compilation, connected networks, dead branches. '' when the network check did not run.
Whether the storage level of the dataset check ran, and when ('' when the check was not part of the assessment).
What the lineage integrity check read and looked for ('' when it did not run).
What the storage-level check covered (None when it did not run). Pure.
build_workflow_model(inputs: dict, *, skew_threshold: float = 0.25, skew_min_mb: float = 100, stuck_hours: float = 12, align_min_mb: float = 1000) -> dict
Findings, counts and what could not be checked. Pure.
The verdict, the most frequent issues and the health grid: one row per process, grouped by execution wave, one cell per dimension (worst severity; the issues in the tooltip). Scriptless HTML.
'Workflow assessment' body: the executive summary, then one group per area (the consistency style).
The assessment as Markdown: verdict, top issues, the health grid by wave, then the findings per area and what could not be verified.
The lineage network section of the Markdown report ([] when not checked).
A standalone HTML page (the panel's content, for export): offline, no script.
Write the assessment as html (standalone page), md or csv
(findings) and return the path. Default path: DASHBOARD_REPORT_DIR.
Follow-up rows of the processes started in the last days days (primary store).
{dataset node: {"mode", "as_of", "tm"}} from the build view, then the
view's DDL for a dataset without a build row (at most max_ddl reads).
({pid: FilterManager rows}, unverifiable) - one COUNT per manager table.
({pid: {"row", "chain"}}, unverifiable): the live SQL chain of each
process (SHOW VIEW of its view and of the views it reads), read once and
shared by the SQL drift and distribution checks; at most max_n.
(versions, drift, unverifiable) from FS_PROCESS_SQL (two reads) and the
live chains read by :func:read_chains.
The staging CTAS of an ingestion, for EXPLAIN only (never executed).
(plan text, dynamic): the plan of the process view for the active
FilterManager segment, with incremental planning forced (DYNAMIC EXPLAIN,
the segment's value known) when the session allows it, else the static plan.
A static plan leaves the segment value as a parameter and understates
partition elimination (seen live).
How the plan of the SELECT sql reads the table source ('DB.TABLE'):
"partitions" (partition elimination, or a PTI time range), "all_rows",
or None when no step reads it. dynamic=True plans with the subquery values
known (DYNAMIC EXPLAIN with incremental planning forced, the session's
query band restored); False is the static plan, which leaves a scalar
subquery's value as a parameter. EXPLAIN only: nothing runs.
collect_distribution(proc, feat, graph, chains, layout, expected_segments=None, max_explain: int = 30, emit=None, drivers=None) -> tuple
(rows, unverifiable): per denormalized-view process, its storage tables
(PI, partitioning), its largest source (size, skew, PI, partitioning), the
origin of each entity column, the aligned PI and the EXPLAIN evidence. Reads:
the dictionary for the sources (batched) and at most max_explain x 3 EXPLAINs.
Primary index, row counts, entity column types and AMP sizes of the
storage tables (batched DBC reads). {"ok": False} when DBC is unreadable.
One :func:reduce_entity_key per (domain, entity key) of two columns or
more, on its largest storage table (StatsV row count, else any).
The storage level of the dataset-feature check: rows per (FEATURE_ID,
FEATURE_VERSION) of every storage table the datasets read (one grouped pass
per table, now and over the whole history), and the COMPLETED runs of the
producing processes. (rows, unverifiable); each failed read is an
unverifiable entry, never an empty result read as "all fine".
collect_workflow_inputs(processes=None, data_domains=None, checks=None, days: int = 90, max_sql_processes: int = 100, emit=None) -> dict
Everything :func:build_workflow_model needs. Each read is fail-open: a
failure becomes an unverifiable entry, never an empty "all clear".
assess_workflow(processes=None, data_domains=None, checks=None, entity_scan: bool = True, days: int = 90, display: bool = True, skew_threshold: float = 0.25, skew_min_mb: float = 100, stuck_hours: float = 12, align_min_mb: float = 1000, progress=None, dataset_rows: bool = True) -> dict
Assess how the workflow is built: entity keys, run order and time synchronisation, point-in-time inputs, segments, process SQL, storage.
Parameters¶
processes : list of str, optional
Process ids or view names to assess (default: every process).
data_domains : str or list, optional
Restrict to these data domains.
checks : iterable, optional
A subset of :data:CHECKS (default: all of them).
entity_scan : bool
Read the storage tables to test the entity keys (one grouped pass per
entity over its largest table). False skips the only expensive read.
dataset_rows : bool
Count the stored rows behind every dataset feature (one grouped pass per
storage table the datasets read). False keeps the dictionary-level
checks only.
days : int
Follow-up window, in days, for the run checks.
display : bool
Print a summary and the findings.
progress : callable, optional
Called with one line per step.
Returns¶
dict
findings (area, kind, object, detail, severity, fix), counts,
unverifiable (checks that could not run, with the reason),
summary, entities (entity scan results), checks.
No language model is used, and nothing is changed: every finding names the command that fixes it.