tdfs4ds.scheduling.script¶
Generate a standalone Python script that executes a parallel-execution plan (v0.3.1.147).
The plan produced by :func:tdfs4ds.analyze_parallel_plan is turned into a
self-contained .py that runs the closure's processes wave by wave:
sequential across waves (dependency order), parallel within a wave. Parallelism
uses concurrent.futures.ProcessPoolExecutor — one OS process per unit — NOT
threads, because tdfs4ds.run and ingestion mutate module-level globals
(DATA_DOMAIN / FEATURE_STORE_TIME / RUN_ID / PROCESS_TYPE) and
use teradataml's shared SQLAlchemy pool, which corrupts under thread fan-out.
Each worker process bootstraps its own Teradata context from the
TDFS4DS_TD_* environment, so its global mutations never collide.
generate_execution_script returns the script as a string (inspect
without executing — the repo's return_query idiom); write_execution_script
saves it under tdfs4ds.SCHEDULING_SCRIPT_DIR.
module-attribute
¶
_TEMPLATE = '#!/usr/bin/env python\n"""\nAuto-generated by tdfs4ds {version} on {generated_at}.\n\nParallel-execution script: {n_processes} process(es) across {n_waves} wave(s).\nWaves run sequentially (dependency order); processes within a wave run in\nparallel, one OS process each (safe re: tdfs4ds module globals + teradataml\npool). Each worker opens its own Teradata context from TDFS4DS_TD_* env.\n\nMax parallel workers per wave: {max_workers} (None = one worker per parallel\nunit, i.e. unbounded).\n\nProcesses that share a TimeManager run in LOCKSTEP: this script moves the\nmanager\'s cursor one business date at a time and, for each date, runs those\nprocesses in parallel with run(pid, time_manager=...), which reads the cursor and\nnever moves it; the next date starts when all of them are done. So processes on\none manager always compute at the same business date. (A per-process roll_out\nwould move one shared cursor from several workers at once.) Each TimeManager group\nof a wave runs after that wave\'s other processes, one group at a time.\n\nFailure policy: {wave_on_failure}\n continue - run every wave; report failures at the end (exit code 1).\n skip_descendants - skip only what depends on a failed process; everything\n else still runs. This is the default: a failure poisons\n its own subtree, not the whole plan.\n stop_after_wave - finish the current wave, then stop before the next one.\n raise - like stop_after_wave, but ends with a RuntimeError.\n\nEach process is executed with on_failure=\'raise\', so a partial (per-filter)\ningestion failure is visible here instead of being silently absorbed. Edit that\nbelow if you deliberately want a process to tolerate failed filters.\n\nRun: python {basename}\n"""\nimport json\nimport sys\nfrom concurrent.futures import ProcessPoolExecutor, as_completed\n\nPLAN = json.loads({plan_literal})\nFORCE_COMPUTE = {force_compute}\nMAX_WORKERS = {max_workers} # None = one worker per parallel unit\nMODE = {mode!r}\nWAVE_ON_FAILURE = {wave_on_failure!r}\n\n_FAILURES = [] # [(pid, error message, with its causes)]\n_SKIPPED = {{}} # pid -> the upstream pid whose failure caused the skip\n\n\ndef _describe(exc):\n """The error and every error it was raised from, without driver stack frames.\n\n teradataml often re-raises a database error as a generic one ("Encountered\n problem with the query"): the first line alone hides the real cause.\n """\n parts, seen = [], set()\n while exc is not None and id(exc) not in seen:\n seen.add(id(exc))\n msg = " ".join(l.strip() for l in str(exc).splitlines()\n if l.strip() and not l.strip().startswith("at "))\n if msg and msg not in parts:\n parts.append(msg)\n exc = exc.__cause__ or exc.__context__\n return " <- caused by: ".join(parts) or "unknown error"\n\n\ndef _bootstrap():\n """Open this worker\'s own Teradata context + tdfs4ds config from env."""\n import tdfs4ds\n from tdfs4ds.utils.reconnect import reconnect_teradata\n ok = reconnect_teradata()\n if not ok:\n raise RuntimeError(\n "reconnect_teradata() failed - set TDFS4DS_TD_HOST/USERNAME/PASSWORD "\n "(and optionally _LOGMECH/_DATABASE) in the environment."\n )\n if PLAN.get("schema"):\n tdfs4ds.SCHEMA = PLAN["schema"]\n if PLAN.get("schema_view"):\n tdfs4ds.SCHEMA_VIEW = PLAN["schema_view"]\n return tdfs4ds\n\n\ndef _time_key(pid):\n """(database, view) of the process\'s TimeManager, or None."""\n tm = PLAN["processes"].get(pid, {{}}).get("time") or {{}}\n return (tm["database"], tm["view"]) if tm.get("view") else None\n\n\ndef _execute_pid(pid, step=None):\n """Run one process.\n\n step=None: no TimeManager, a plain run. Otherwise the parent has moved the\n TimeManager cursor to this step; run() reads it (and never moves it). The\n lineage check runs at the first step only.\n """\n tdfs4ds = _bootstrap()\n meta = PLAN["processes"].get(pid, {{}})\n dom = meta.get("data_domain")\n if dom:\n tdfs4ds.DATA_DOMAIN = dom\n key = _time_key(pid)\n if key and step is not None:\n from tdfs4ds.utils.time_management import TimeManager\n tmgr = TimeManager(table_name=key[1], schema_name=key[0])\n tdfs4ds.run(pid, force_compute=FORCE_COMPUTE, on_failure="raise",\n time_manager=tmgr, _time_manager_checked=(step > 1))\n else:\n tdfs4ds.run(pid, force_compute=FORCE_COMPUTE, on_failure="raise")\n return pid\n\n\ndef _run_group(pids, step=None):\n """Run a serialize-together group sequentially inside a single worker.\n\n Returns [(pid, error_or_None)]. Failures are RETURNED, not raised, so the\n parent can attribute each one to the process that actually failed - which\n is what the skip_descendants policy needs.\n """\n results = []\n for pid in pids:\n print(" [proc] running", pid, "" if step is None else "(step {{}})".format(step), flush=True)\n try:\n _execute_pid(pid) if step is None else _execute_pid(pid, step)\n results.append((pid, None))\n except Exception as e:\n results.append((pid, _describe(e)))\n # The group exists because its members write the same table; once one\n # has failed there is no value in racing the rest through it.\n break\n return results\n\n\ndef _descendants(pid):\n """Processes transitively downstream of `pid`, from the plan\'s edges."""\n down = {{}}\n for edge in PLAN.get("edges", []):\n up, dep = edge[0], edge[1]\n down.setdefault(up, []).append(dep)\n seen, stack = set(), [pid]\n while stack:\n cur = stack.pop()\n for nxt in down.get(cur, []):\n if nxt not in seen:\n seen.add(nxt)\n stack.append(nxt)\n return seen\n\n\ndef _record_failure(pid, err):\n _FAILURES.append((pid, err))\n print(" [FAIL] {{}} : {{}}".format(pid, err), flush=True)\n if WAVE_ON_FAILURE == "skip_descendants":\n for d in _descendants(pid):\n _SKIPPED.setdefault(d, pid)\n\n\ndef _should_stop():\n return bool(_FAILURES) and WAVE_ON_FAILURE in ("stop_after_wave", "raise")\n\n\ndef _wave_units(wave_idx, wave):\n """Group write-conflicting processes into one unit; others are singletons."""\n units, seen = [], set()\n for grp in PLAN["serialize_together"]:\n if grp.get("wave") == wave_idx:\n members = [p for p in grp["processes"] if p in wave and p not in seen]\n if len(members) > 1:\n units.append(members)\n seen.update(members)\n for pid in wave:\n if pid not in seen:\n units.append([pid])\n seen.add(pid)\n return units\n\n\ndef _filter_skipped(units):\n """Drop processes whose upstream failed; report each one."""\n kept = []\n for u in units:\n alive = []\n for p in u:\n if p in _SKIPPED:\n print(" [skip] {{}} (upstream {{}} failed)".format(p, _SKIPPED[p]),\n flush=True)\n else:\n alive.append(p)\n if alive:\n kept.append(alive)\n return kept\n\n\ndef _calendar_key(pid):\n """The table of dates the process\'s TimeManager reads: (database, table), or the\n manager (database, view) when the plan does not say. Soft clones of one manager\n share a table, so they form ONE lockstep group (their cursors move together)."""\n tm = PLAN["processes"].get(pid, {{}}).get("time") or {{}}\n if not tm.get("view"):\n return None\n if tm.get("table"):\n return (tm.get("table_database") or tm["database"], tm["table"])\n return (tm["database"], tm["view"])\n\n\ndef _split_by_time(units):\n """(units without a TimeManager, {{(db, table): units}}) - a unit belongs to the\n calendar of its first member that has a TimeManager."""\n plain, groups = [], {{}}\n for u in units:\n key = next((_calendar_key(p) for p in u if _calendar_key(p)), None)\n if key is None:\n plain.append(u)\n else:\n groups.setdefault(key, []).append(u)\n return plain, groups\n\n\ndef _run_units(units, step=None, parallel=True):\n """Run units (in parallel or not), record failures, return the failed pids."""\n failed = set()\n if not units:\n return failed\n if not parallel:\n for unit in units:\n for pid, err in _run_group(unit, step):\n if err:\n _record_failure(pid, err)\n failed.add(pid)\n return failed\n workers = MAX_WORKERS or max(1, len(units))\n with ProcessPoolExecutor(max_workers=workers) as ex:\n futs = {{ex.submit(_run_group, u, step): u for u in units}}\n for fut in as_completed(futs):\n unit = futs[fut]\n try:\n for pid, err in fut.result():\n if err:\n _record_failure(pid, err)\n failed.add(pid)\n except Exception as e:\n # The worker died before returning results (bootstrap failure,\n # killed process). Attribute it to the whole unit.\n for pid in unit:\n _record_failure(pid, _describe(e))\n failed.add(pid)\n return failed\n\n\ndef _run_time_group(key, units, parallel=True):\n """Lockstep: the parent moves the cursors, the units compute that date, barrier.\n\n The group is one table of dates; its members may read different views of it (the\n manager and its soft clones): every view is moved to the same step."""\n tdfs4ds = _bootstrap()\n from tdfs4ds.utils.time_management import TimeManager, point_time_manager_view\n views = sorted({{_time_key(p) for u in units for p in u if _time_key(p)}})\n tables = {{}}\n for u in units:\n for p in u:\n tm = PLAN["processes"].get(p, {{}}).get("time") or {{}}\n if tm.get("view"):\n tables[(tm["database"], tm["view"])] = (tm.get("table_database") or tm["database"],\n tm.get("table") or tm["view"] + "_HIDDEN")\n tmgr = TimeManager(table_name=views[0][1], schema_name=views[0][0])\n n = int(tmgr.nb_time_steps or 0)\n print(" TimeManager {{}}.{{}}: {{}} unit(s) in lockstep over {{}} date(s)".format(\n views[0][0], views[0][1], len(units), n), flush=True)\n alive = [list(u) for u in units]\n for step in range(1, n + 1):\n alive = [[p for p in u if p not in _SKIPPED and p not in dict(_FAILURES)] for u in alive]\n alive = [u for u in alive if u]\n if not alive:\n break\n tmgr.update(time_id=step)\n for v in views[1:]:\n tdb, tname = tables[v]\n point_time_manager_view(v[0], v[1], tdb, tname, step)\n print(" step {{}}/{{}}: {{}}".format(step, n, tmgr.get_date_in_the_past()), flush=True)\n _run_units(alive, step, parallel)\n\n\ndef _run_wave(k, wave, parallel):\n units = _filter_skipped(_wave_units(k, wave) if parallel else [[p] for p in wave])\n print("Wave {{}}: {{}} unit(s) / {{}} process(es){{}}".format(\n k, len(units), len(wave), "" if parallel else " (sequential)"), flush=True)\n plain, groups = _split_by_time(units)\n _run_units(plain, None, parallel)\n for key in sorted(groups):\n _run_time_group(key, _filter_skipped(groups[key]), parallel)\n\n\ndef _run_sequential():\n for k, wave in enumerate(PLAN["waves"]):\n # The wave is always finished, even once a failure is known: the policy\n # is \'stop AFTER wave\', and the parallel path cannot abort in-flight\n # units either - the two must not disagree on what a wave means.\n _run_wave(k, wave, parallel=False)\n if _should_stop():\n print("Stopping before the next wave (wave_on_failure={{}})."\n .format(WAVE_ON_FAILURE), flush=True)\n break\n _finish()\n\n\ndef _run_parallel():\n for k, wave in enumerate(PLAN["waves"]):\n _run_wave(k, wave, parallel=True)\n if _should_stop():\n print("Stopping before the next wave (wave_on_failure={{}})."\n .format(WAVE_ON_FAILURE), flush=True)\n break\n _finish()\n\n\ndef _finish():\n if _SKIPPED:\n print("", flush=True)\n print("Skipped because an upstream process failed: {{}}".format(len(_SKIPPED)),\n flush=True)\n for pid in sorted(_SKIPPED):\n print(" - {{}} (upstream {{}})".format(pid, _SKIPPED[pid]), flush=True)\n if _FAILURES:\n print("", flush=True)\n print("FAILED: {{}} process(es)".format(len(_FAILURES)), flush=True)\n for pid, err in _FAILURES:\n print(" - {{}} : {{}}".format(pid, err), flush=True)\n print("", flush=True)\n print("Each failure is also recorded in the feature store\'s follow-up "\n "table (STATUS starting with FAILED).", flush=True)\n if WAVE_ON_FAILURE == "raise":\n raise RuntimeError(\n "{{}} process(es) failed; see the list above.".format(len(_FAILURES)))\n sys.exit(1)\n print("All waves complete.", flush=True)\n\n\ndef main():\n cap = "unbounded" if MAX_WORKERS is None else str(MAX_WORKERS)\n print("tdfs4ds parallel execution | {n_waves} wave(s) | {n_processes} process(es) "\n "| max workers/wave: " + cap + " | wave_on_failure: " + WAVE_ON_FAILURE,\n flush=True)\n if MODE == "sequential":\n _run_sequential()\n else:\n _run_parallel()\n\n\nif __name__ == "__main__":\n main()\n'
module-attribute
¶
Keep only what the generated script needs (waves, per-pid time mgr, serialize).
generate_execution_script(plan: Dict[str, Any], *, out_path: Optional[str] = None, mode: str = 'process_pool', force_compute: bool = False, max_workers: Optional[int] = None, wave_on_failure: str = 'skip_descendants') -> str
Turn an :func:analyze_parallel_plan result into a runnable .py script.
Parameters¶
plan : dict
The plan returned by :func:tdfs4ds.analyze_parallel_plan.
out_path : str | None
If given, also write the script there (see also
:func:write_execution_script).
mode : {'process_pool', 'sequential'}
'process_pool' (default) parallelises each wave across processes;
'sequential' runs everything in order (debugging).
force_compute : bool
Passed to run in the generated worker.
Processes of a wave that share a TimeManager form one time group, run in
lockstep: the script moves the manager's cursor one date at a time, the group's
processes compute that date in parallel with run(pid, time_manager=tm) (which
reads the cursor and never moves it), and the next date starts when they are all
done. A per-process roll_out would move one shared cursor from several
workers at once. Give processes their own cursor with
:func:tdfs4ds.split_time_manager when they must run on different dates.
max_workers : int | None
Cap on parallel workers per wave. None (default) falls back to the
tdfs4ds.SCHEDULING_MAX_WORKERS global; when that is also None the
script runs unbounded (one worker per parallel unit). An explicit int here
overrides the global for this call.
wave_on_failure : {'skip_descendants', 'continue', 'stop_after_wave', 'raise'}
What the generated script does when a process fails. Named apart from the
per-process on_failure on run/roll_out deliberately: the two
take different value sets, and 'raise' is legal in both while meaning
something different, so one name for both would make an invalid value read
as a documentation error.
'skip_descendants' (default) skips only the failed process's
downstream closure and lets independent branches finish - a failure
poisons its own subtree, not the plan. 'raise' reproduces the
historical abort-everything behaviour.
Note the generated worker always calls ``run``/``roll_out`` with
``on_failure='raise'``: a wave policy can only react to failures it is
told about, and the per-filter loop does not report them otherwise.
Returns¶
str The script source.
Generate the script and write it to disk, returning the path.
out_path defaults to {tdfs4ds.SCHEDULING_SCRIPT_DIR}/tdfs4ds_execution_<ts>.py.
Extra keyword args are forwarded to :func:generate_execution_script.