tdfs4ds.scheduling.parallel_plan¶
Parallel-execution planner for tdfs4ds feature-engineering processes (v0.3.1.147).
Given a seed list of processes / datasets / features, this module resolves each seed to its producing process(es), expands to the upstream process closure (every process the seeds depend on — the self-contained "build set", which doubles as a migration unit), and groups the closure into dependency waves: a topological layering of the process DAG where every process in a wave is mutually independent and can run in parallel, and waves run in order.
The heavy lifting reuses the persisted lineage graph
(:func:tdfs4ds.lineage.lineage_store.load_lineage_digraph) and the process /
feature / dataset catalogs. The graph algorithms (_build_process_dag /
_compute_waves / _upstream_closure) are pure functions over a
networkx.DiGraph so they can be unit-tested without a database.
Nothing here mutates the store — see :mod:tdfs4ds.scheduling.unify for the
opt-in manager-unification mutators and :mod:tdfs4ds.scheduling.script for
turning a plan into an executable script.
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.
The whole process catalog as a pandas frame (fail-open to empty).
None/NaN/blank -> None, else the stripped string (LEFT JOIN nulls).
Canonical 'DB.OBJECT' lineage node id for a catalog VIEW_NAME.
Map each process's lineage node id ('DB.OBJECT') -> PROCESS_ID.
Contract the object lineage graph G onto process nodes.
G edges point parent -> child = downstream -> upstream (a process
depends on its children). For each process p we take its transitive
upstream objects (networkx.descendants(G, node_p)) and, for every one
that is also a process q, emit an edge (q, p) meaning
"q is upstream, run q before p". Transitive edges are harmless for the
topological layering computed later.
Build the process DiGraph: nodes = all_pids, edges = upstream -> dependent.
Seed processes + every process they depend on (ancestors in PG).
Topological generations of PG -> (waves, cycle).
waves is a list of lists; each inner list is one parallel wave (a
topological generation, in_degree-0 first). cycle is None on a DAG,
else a list of edges forming a cycle (waves is then []).
Resolve a mixed list of process / dataset / feature names to producing
PROCESS_IDs.
Returns (resolved, unresolved, seed_pids) where resolved maps each
seed string to the list of process ids it resolved to. First matching
category wins per seed: PROCESS_ID -> process view -> dataset -> feature.
Producing process ids of a dataset, from its view DDL FEATURE_VERSION refs.
Process ids whose registered FEATURE_NAMES contains feature_name.
Stable content hash of a FilterManager (col_names + time flag + rows).
Stable content hash of a TimeManager (timeline rows + data type).
_entity_conflict_key(data_domain: Optional[str], entity_id: Optional[str]) -> Optional[Tuple[str, str]]
Synthetic conflict key for a process whose target FS_T_* table can't yet be resolved from the feature catalog (features not registered).
Physical tables are named FS_T_{domain}_{entity}{part}, so two processes on the same (DATA_DOMAIN, ENTITY_ID) may MERGE into the same table. This coarse key serialises them conservatively (it cannot distinguish data types, which the catalog would — see _pid_to_tables). Returns None when neither domain nor entity is known (nothing to key on).
Render a conflict key for display: real table as 'db.table', synthetic as 'entity:...'.
Map each PROCESS_ID -> set of (FEATURE_DATABASE, FEATURE_TABLE) it writes.
Conservative: joins the process's FEATURE_NAMES to the feature catalog on (FEATURE_NAME, DATA_DOMAIN). Over-inclusion only makes the serialize guard more cautious (safe). Fail-open to empty.
Fresh-build fallback (issue 1): when a process resolves to zero exact
tables (its features are not yet registered — the common case on a first
parallel build), it is keyed instead on a synthetic (DATA_DOMAIN, ENTITY_ID)
conflict key (:func:_entity_conflict_key) so it still serialises against
other same-domain-same-entity processes. The fallback fires only when the
exact set is empty, so once the catalog is populated the exact table drives
the decision and same-entity/different-data-type processes still parallelise.
Accepted rare gap: a registered process (exact table) and an unregistered one
(synthetic key) that write the same physical table are not matched.
refresh_tables (0.3.3.30): {PROCESS_ID: (DATABASE, TABLE)} of the admin FS_R_*
table a process with track_entity_refresh writes after its values. Its MERGE takes a table
lock too, so two tracked processes of one (domain, entity, primary index, placement tag) are
serialised even when their storage tables differ. Added after the fallback above, never in
place of it. None / {} (no tracked process) leaves the result unchanged.
Union-find pids in a wave that share a target table -> conflict groups.
Compute a parallel-execution plan for feature-engineering processes.
Parameters¶
seeds : list[str] | None
Mixed list of process (id or view name) / dataset / feature names.
The plan is the upstream closure of the seeds' producing processes
(everything needed to (re)build them). None = the whole store.
all_domains : bool
Read every data domain (default). False restricts to the active
tdfs4ds.DATA_DOMAIN.
time_managers : ignored placeholder
Reserved for supplying explicit TimeManagers to the content-equivalence
check when the catalog has no TIME_* association yet.
Returns¶
dict
seeds, resolved_seeds, unresolved_seeds, closure,
processes (per-pid metadata incl. filter/time manager + wave index
+ lineage_unknown), waves (ordered list of parallel batches),
wave_groups (per-wave grouping by (filter, time) manager),
serialize_together (write-conflict groups), unify_candidates
(equivalent-but-distinct managers), edges (process dependency
edges within the closure), cycle (None or a cycle), plus counts.
Group distinct filter/time manager objects (in the closure) by content.