Skip to content

Parallel execution plan

Once you have registered many feature-engineering processes, you rarely want to run them one after another. tdfs4ds can read the persisted lineage graph, work out which processes are independent, and schedule them into waves — every process in a wave can run in parallel, and waves run in dependency order.

You seed the planner with the things you actually care about — some datasets, features, or processes — and it expands to the upstream closure: every process needed to (re)build those seeds, and nothing else. That closure is also a natural migration unit (it is self-contained).

Build a plan

import tdfs4ds

# Seed with datasets / features / processes (mix freely). None = whole store.
plan = tdfs4ds.analyze_parallel_plan(seeds=["customer_360", "churn_score"])

print("waves:", plan["waves"])          # [[pid, pid], [pid], ...]  each list = one parallel wave
print("processes:", plan["n_processes"])
print("unresolved seeds:", plan["unresolved_seeds"])

Each wave is a list of process ids that can run concurrently because none of them depends on another in the same wave. Wave 0 holds the leaf-most upstream processes; your seeds land in the last wave(s).

Existing processes from older versions — rebuild lineage first

The planner derives wave order from the persisted lineage graph (FS_LINEAGE_GRAPH). A process that isn't in the graph yet is flagged lineage_unknown and placed in wave 0 — assumed independent. That is only safe if it genuinely has no upstream process; if it does, dependent processes would be scheduled in parallel instead of in order. Processes registered by older tdfs4ds versions may not have their edges persisted, so before scheduling a legacy store run this once:

tdfs4ds.rebuild_lineage_graph(all_data_domains=True)

It (re)derives and persists the dependencies of every registered process and dataset. Then analyze_parallel_plan(...) orders them into correct waves.

The plan also reports:

  • wave_groups — per wave, processes grouped by their FilterManager / TimeManager.
  • unify_candidates — distinct filter/time-manager objects that hold equivalent content (candidates for unification).
  • serialize_together — processes in the same wave that write to the same storage table and therefore must not run concurrently (a MERGE locks its target FS_T_* table). Feature-store tables are keyed by (data_domain, entity, data type), so processes on different entities or different data types write different tables and can run in parallel. When a process's features are not yet registered (a first-time parallel build, so the exact target table can't be resolved from the feature catalog), the planner falls back to a conservative (data_domain, entity) conflict key — reported as entity:DOMAIN::ENTITY — so same-entity processes still serialize instead of racing the same table.
  • lineage_unknown — processes not yet in FS_LINEAGE_GRAPH (run tdfs4ds.rebuild_lineage_graph() to fill the gap; they are placed in wave 0).

Tip

A cycle in the process graph is reported (plan["cycle"]), never silently ignored — resolve the circular dependency before scheduling.

Generate an execution script

Turn a plan into a standalone, runnable Python script:

path = tdfs4ds.write_execution_script(plan, force_compute=False)
print("wrote", path)          # under tdfs4ds.SCHEDULING_SCRIPT_DIR

The generated script runs waves sequentially and, within each wave, runs the independent processes in parallel using separate OS processes (ProcessPoolExecutor). Each worker opens its own Teradata context from the TDFS4DS_TD_* environment variables, so the module-level configuration globals that run() mutates never collide between workers. Processes flagged serialize_together run one after another inside a single worker.

Processes on one TimeManager run in lockstep (0.3.1.240). Within a wave, the processes that share a TimeManager form a time group: the script moves the manager's cursor one date at a time, runs the group's processes in parallel at that date with run(pid, time_manager=tm) (which reads the cursor and never moves it), and starts the next date when they are all done. Earlier versions ran a roll_out per worker, which moved one shared cursor from several workers at once, so a process could compute at another's date. A wave's other processes run first; time groups then run one after another. To let processes run on different dates, give them their own cursor with split_time_manager.

First runs are serialized. A process whose features are not all registered yet registers them on its first run, and a new FEATURE_ID is allocated from the catalog's current maximum: two first runs in parallel would get the same ids. Such processes share the conflict key feature registration (first run) and run one after another inside one unit; once their features exist, the next plan runs them in parallel.

Prefer a string you can inspect first? Use tdfs4ds.generate_execution_script(plan) (returns the source without writing). Pass mode="sequential" for a plain, single-process debugging run.

Cap the parallelism

Each parallel unit is a full OS process running a heavy Teradata ingestion, so a wide wave can overwhelm the system. Cap how many run at once per wave with max_workers:

path = tdfs4ds.write_execution_script(plan, max_workers=4)   # at most 4 processes per wave

When a wave has more units than the cap, the extra units queue in a rolling ProcessPoolExecutor — at most max_workers run at once and each queued unit starts the instant a worker frees (continuous, not fixed batches; the only barrier is between waves, for dependency order). A write-conflict (serialize_together) group counts as one unit — its processes run back-to-back inside a single worker — so the cap bounds concurrent units, never letting two processes MERGE the same table at once. max_workers=None (the default) falls back to the global tdfs4ds.SCHEDULING_MAX_WORKERS (or TDFS4DS_SCHEDULING_MAX_WORKERS / JSON scheduling_max_workers); when that is also None the script is unbounded — one worker per parallel unit (the historical behaviour). An explicit max_workers= argument always overrides the global for that call.

What happens when a process fails

Added in v0.3.1.210.

A failure in one process does not invalidate the whole plan -- only what depends on it. wave_on_failure decides how much of the plan a failure takes with it. It is named apart from the per-process on_failure on run()/roll_out() deliberately: the two take different value sets, and 'raise' is valid in both while meaning something different, so a single name would make an invalid value look like a documentation error.

Policy Behaviour
'skip_descendants' (default) Skips only the failed process's downstream closure; independent branches still run.
'continue' Runs every wave, then reports the failures.
'stop_after_wave' Finishes the current wave, then stops before the next one.
'raise' Same, but ends with a RuntimeError. 
path = tdfs4ds.write_execution_script(plan, wave_on_failure='skip_descendants')

The downstream closure is computed inside the generated script from the plan's own dependency edges, so the script stays self-contained -- it does not import networkx.

Both runners finish the current wave before honouring 'stop_after_wave' or 'raise': the parallel path cannot abort units already in flight, and the two must not disagree about what a wave is.

The generated worker calls run(..., on_failure='raise'), because a wave policy can only react to failures it is told about -- a per-filter failure is otherwise absorbed by the process. Edit that line in the generated script if you deliberately want a process to tolerate failed filters.

The script exits non-zero when anything failed, listing the failed processes and those skipped because of them. Each failure is also recorded in the follow-up table -- see Failure handling.

Unify managers

When unify_candidates shows that several parallel processes use equivalent but distinct FilterManagers or TimeManagers, you can repoint them onto a single canonical manager so one driver drives the whole wave. The mutators are opt-in and default to a dry run:

group = plan["unify_candidates"]["filter"][0]        # {managers:[...], processes:[...]}
preview = tdfs4ds.unify_filter_managers(group["processes"])            # dry_run=True: returns SQL
tdfs4ds.unify_filter_managers(group["processes"], dry_run=False)       # actually repoint

Repointing rewrites each process view's manager reference (REPLACE VIEW), updates the FS_FILTER_MANAGER / FS_TIME_MANAGER catalog row and refreshes the view's lineage. Use unify_time_managers(...) for time drivers, and split_time_manager(...) for the inverse (an own cursor per process).

In the admin UI

The API server admin shell has an Execution plan panel: pick datasets/processes as seeds, click Build workflow to see an interactive wave-layered graph (columns = execution order; nodes in one column run in parallel; red border = write conflict, orange = unifiable manager), set an optional Max parallel workers cap, and use ⬇ Generate .py script to download the runnable script.

The FS_TIME_MANAGER catalog

To make "which TimeManager does this process use" discoverable (so it can be grouped/unified), each process's TimeManager is recorded in a new FS_TIME_MANAGER catalog (mirrored on FS_V_PROCESS_CATALOG as TIME_DATABASE_NAME / TIME_VIEW_NAME / TIME_TABLE_NAME).

The association is recorded at registration with upload_features(..., time_manager=tm) (0.3.1.240), by roll_out, by run(..., time_manager=tm), and by link_time_managers for processes registered without it. run(pid) then uses the recorded manager's current date. A manager the process view does not read is checked against the lineage graph (TIME_MANAGER_CHECK, see TimeManager).

tdfs4ds.roll_out(process_list, time_manager)   # records each process's TimeManager

Upgrading from an older version

The catalog is fully backward compatible:

  • connect() creates the (empty) table automatically on a pre-existing store — non-destructively, before it refreshes the catalog view. Nothing is required for your store to keep working; old processes simply read NULL in the TIME_* columns, which the planner treats as "no time manager" and still schedules into waves.
  • There is no backfill from history — the per-process TimeManager was never stored before this version. Instead the catalog is self-populating: the next time you roll_out your existing processes, their associations are recorded. (For a one-off, unify_time_managers(...) also writes the rows.)