fs-rollout¶
Drive processes with a TimeManager: run() at the current business date, roll_out() over a range of dates,
and roll_out(force_compute=True) from the first date to rebuild a history.
From 0.3.6.0, also one clock per data domain (soft clones per process, roll_out / run over a
whole domain at one time), an alternative to a TimeManager attached per process.
name: "Feature Store: Rollout" description: "Generate code to drive feature processes with a TimeManager: attach it at registration, run() at the current business date, roll_out() over a range of dates to build a history, roll_out(force_compute=True) from the first date to rebuild one, and keep processes synchronised (one clock per data domain with automatic soft clones, roll_out / run over a whole domain or a list at one time, shared manager, lockstep parallel runs, split_time_manager for an own cursor). Invoke when the user wants to backfill features over past periods, schedule processes on a business date, recompute a history after a logic change, or check which manager drives a process." argument-hint: "[process-view-name][date-column] [start-id][end-id]" allowed-tools: "Read, Edit, Write, Bash, Glob, Grep"
Generate Python code that drives registered feature processes with a TimeManager: one business date with run(),
a range of dates with roll_out().
User context: $ARGUMENTS
Related skills¶
| Step | Skill | Why |
|---|---|---|
| Before | fs-upload |
Register the processes first, with time_manager=tm and run_now=False so they are only computed under the manager. |
| Sibling | fs-filter |
Combine a TimeManager with a FilterManager: each date is computed segment by segment. |
| After | fs-dataset |
Read the history: as_of=, time_manager=, build_historical_dataset. |
| Monitor | fs-inspect |
follow_up_report() has one row per process and date (VALIDTIME_DATE). |
| Promote | fs-migrate |
Production rebuilds its own history with the same roll_out. |
Demos: demos/notebooks/05 - Time management.ipynb, 05b - Synchronising processes.ipynb, 08 (a TimeManager
with a FilterManager) and A2 (lockstep parallel runs).
A dataset that follows the manager¶
upload_features(..., time_manager=tm, dataset_view_name="db.ds", dataset_time_manager=True) builds the dataset with
build_dataset(time_manager=tm) (recorded as a process setting; run/roll_out keep it, dataset_time_manager=False
overrides). After roll_out, tm.update(i) shows date i with no re-run. Without a TimeManager it warns and stays current;
a dataset already built current is not rebuilt: use tdfs4ds.rebuild_dataset("db.ds", time_manager=tm, replace=True).
The model: two calls, two jobs¶
| Call | What it does |
|---|---|
upload_features(..., time_manager=tm) |
Attaches the manager at registration, the way a FilterManager is attached: the pairing is recorded in FS_TIME_MANAGER, so run(process_id) needs no argument afterwards. |
run(process_id) / run(process_id, time_manager=tm) |
Computes one date: the attached (or passed) manager's current one. It does not loop over the manager's dates. The values are stored valid from that business date. This is the scheduler's daily call. |
roll_out(process_list, time_manager=tm, time_id_start=, time_id_end=) |
The loop. For each date in the range: move the cursor, run every process of the list at that date. This is how a history is built (backfill). |
Every process attached to the same manager is synchronised: - same ingestion time: values are stored valid from the business date, not from the clock time of each run; - same visibility: a value is visible from its business date, so a dataset read at a date sees every process at that same date.
Without a manager (or FEATURE_STORE_TIME), a run is stamped with the database clock: two processes run a minute
apart get two valid times. Real time is a fallback, never the recommended way.
What the view reads is checked. Before run(..., time_manager=), roll_out or upload_features(...,
time_manager=) use a manager, tdfs4ds looks in the lineage graph (through any intermediate views) for the manager
the process view really reads. A manager it does not read would stamp values with a date they were not computed at.
tdfs4ds.TIME_MANAGER_CHECK decides: 'warn' (default) runs as before, logs a warning and records no pairing;
'error' refuses before anything runs (generate code with 'error'); 'off' skips the check. A process missing
from the lineage graph is only warned about, with the rebuild_lineage_graph(objects=[...]) command to run.
tdfs4ds.time_manager_dependency(pid) shows declared vs read; tdfs4ds.link_time_managers(dry_run=True) records the
pairing for processes registered without it (older code, or a manager read through another view).
Ask for any missing values before generating code:
- Database of the feature store, and the data domain.
- Process view name(s) — to look the process ids up (or the ids returned by upload_features(..., return_process_id=True)).
- TimeManager table name (e.g. 'BUSINESS_DATE') and the dates: where they come from (e.g. month-ends of a date column).
- What to do: run the current date, build a history over a range, or rebuild a history after a logic change.
Emit the following pattern, substituting real values for every <PLACEHOLDER>:
import tdfs4ds # reads the TDFS4DS_* settings of .env
tdfs4ds.reconnect_teradata(raise_error=True) # teradataml context from TDFS4DS_TD_HOST / _USERNAME / _PASSWORD
tdfs4ds.connect(database='<DATABASE>')
tdfs4ds.select_data_domain('<DATA_DOMAIN>')
import teradataml as tdml
from tdfs4ds.utils.time_management import TimeManager
# ---------------------------------------------------------------------------
# 1. The TimeManager: a table of business dates and a one-row view showing the
# current one (the cursor). Creating it does not run anything.
# ---------------------------------------------------------------------------
tdfs4ds.TIME_MANAGER_CHECK = 'error' # refuse a manager the process view does not read
tm = TimeManager(table_name='<TIME_TABLE_NAME>', schema_name='<DATABASE>')
tm.load_time_steps( # first time only; later sessions reconnect to the table
tdml.DataFrame.from_query(
"SELECT DISTINCT LAST_DAY(<DATE_COLUMN>) AS business_date FROM <SOURCE_DATABASE>.<SOURCE_TABLE>"),
'business_date')
print(tm.nb_time_steps, 'dates')
# The process views read the cursor with a SCALAR subquery (see fs-upload / fs-filter):
# WHERE A.<DATE_COLUMN> <= (SELECT BUSINESS_DATE FROM <DATABASE>.<TIME_TABLE_NAME>)
# and measure recency from that date, not from CURRENT_DATE.
# ---------------------------------------------------------------------------
# 2. The processes (registered with upload_features(..., time_manager=tm, run_now=False,
# return_process_id=True)); for older processes: tdfs4ds.link_time_managers(dry_run=False)
# ---------------------------------------------------------------------------
cat = tdfs4ds.process_catalog().to_pandas()
def process_id_for(view): # exact match on the view name, never a substring
hits = cat[cat['VIEW_NAME'].str.split('.').str[-1].str.strip('"').str.upper() == view.upper()]
return hits['PROCESS_ID'].iloc[0]
process_list = [process_id_for('<PROCESS_VIEW_1>'), process_id_for('<PROCESS_VIEW_2>')]
# ---------------------------------------------------------------------------
# 3a. ONE date: run() at the manager's current date (the daily scheduler call)
# ---------------------------------------------------------------------------
tm.update(time_id=tm.nb_time_steps) # e.g. the latest business date
for pid in process_list:
tdfs4ds.run(pid, on_failure='raise') # the attached manager's current date
# ---------------------------------------------------------------------------
# 3b. A RANGE of dates: roll_out() is the loop that builds the history
# (replay in date order, from the first date)
# ---------------------------------------------------------------------------
tdfs4ds.roll_out(process_list, time_manager=tm, time_id_start=1, time_id_end=None)
# ---------------------------------------------------------------------------
# 3c. REBUILD a history after the process logic changed: replay every date
# from the first one, forcing the dates already COMPLETED
# ---------------------------------------------------------------------------
# tdfs4ds.roll_out(process_list, time_manager=tm, time_id_start=1, force_compute=True)
# ---------------------------------------------------------------------------
# 4. Check: one COMPLETED follow-up row per process and date
# ---------------------------------------------------------------------------
fu = tdfs4ds.follow_up_report().to_pandas().reset_index()
fu = fu[fu['PROCESS_ID'].isin(process_list) & (fu['STATUS'] == 'COMPLETED')]
print(fu.groupby('PROCESS_ID')['VALIDTIME_DATE'].nunique()) # = tm.nb_time_steps for each process
One clock per data domain (0.3.6.0 and later)¶
Check tdfs4ds.__version__ first. From 0.3.6.0, an alternative to attaching a manager per process: declare the
domain's calendar once. It is the simplest way to keep every process of a domain in step; a manager attached per
process (time_manager=) stays valid and gives more control (own calendar, hand-made clones):
tm = TimeManager('<DOMAIN>_CLOCK', '<DATABASE>')
tm.load_time_steps(dates_df, '<DATE_COLUMN>') # once
tdfs4ds.set_domain_time_manager(tm) # clock of tdfs4ds.DATA_DOMAIN
# the process SQL reads the clock itself:
# WHERE A.<DATE_COLUMN> <= (SELECT BUSINESS_DATE FROM <DATABASE>.<DOMAIN>_CLOCK)
pid = tdfs4ds.upload_features(None, sql_query=sql, target_view_name='<DATABASE>.<PROCESS_VIEW>',
entity_id=['<ENTITY_COLUMN>'], feature_names=['<FEATURE_1>'],
run_now=False, return_process_id=True)
# -> no time_manager= needed: the clock is found in the view's lineage, the process gets
# its own soft clone <PROCESS_VIEW>_TM and its view is rewritten to read it
tdfs4ds.roll_out(data_domain='<DOMAIN>') # history: every process, date by date
tdfs4ds.run(data_domain='<DOMAIN>') # every process at ONE "now" (database clock)
tdfs4ds.run(['<PID_1>', '<PID_2>'], at='2024-03-31', force_compute=True) # a list at a past date
runover a list or a domain computes at one time (at=, default the database clock read once): every clone is swapped for that literal date during the computation and put back on its cursor afterwards. No cursor moves, so a refresh "now" after a roll-out leaves the roll-out where it was. It returns{pid: 'COMPLETED' | 'FAILED: ...' | 'SKIPPED: ...'}.- Dependency order: both
run(list) androll_outrun upstream processes first (lineage graph, stable sort);order='given'keeps the list's order. A failed process's downstream processes are skipped byrun. - A process reading no TimeManager is computed on current data and stored at the same time (warning). A view reading
the clock only through an intermediate view keeps the clock (no clone;
NOT_DIRECT). - Processes registered before the clock:
tdfs4ds.align_domain_time_managers('<DOMAIN>', dry_run=True)lists them,dry_run=Falsegives each its clone. The workflow assessment reportsprocess_not_on_domain_clock. - An interrupted run can leave a clone on the literal date (only its selected value changes, it still reads its
table at its cursor); the next
run/roll_outrestores it, orTimeManager('<PROCESS_VIEW>_TM', '<DATABASE>').restore_cursor(). - An explicit
time_manager=tmat registration keeps the 0.3.5 pairing even whentmis the clock (no clone);soft_clone=Trueforces one.run(pid, time_manager=clock)on a clone reader sets the clone to the clock's date. - Document the clock once: clones copy its documentation. Migration carries the clock.
- In the admin app: the process card (Explore › Feature catalog) has a Domain clock row, the dataset builder lists the clocks first, the Report's Time drivers lists each clock with the processes on and off it.
- Without a domain clock nothing changes, except that
upload_featuresnow records the one manager the view reads (tdfs4ds.TIME_MANAGER_DETECT = Falseturns that off).
Parallel runs and own cursors¶
generate_execution_script(analyze_parallel_plan(seeds=...))runs the processes of a wave that share a manager in lockstep: the script moves the cursor one date at a time and runs them in parallel at that date, with a barrier before the next date. Never generate a per-workerroll_outover a shared manager: several workers would move one cursor at once.tdfs4ds.split_time_manager(process_ids, '<NEW_NAME>', clone_mode='soft', dry_run=True)gives processes their own cursor over the same dates (a soft clone, the views rewritten onto it, lineage refreshed);unify_time_managersis the inverse. A process that reads the manager only through an intermediate view is reportednot_direct.- The admin app's Report → Time drivers lists, per process, the declared and read manager (
OK,UNDECLARED,MISMATCH,DECLARED_UNUSED,NONE,UNKNOWN) and the groups: synchronised (one view), soft clones (one table of dates, own cursors), hard clones (copies), independent.
When to clone, and what to watch¶
Share one manager by default: one cursor, one business date for every process, consistent datasets. Clone only when a process must stand at another date: catching up a late process, back-filling a new one, another cadence on the same calendar, independent timelines in parallel.
| share | soft clone | hard clone | |
|---|---|---|---|
| same dates | yes | yes, always (reads the original's table) | only at copy time |
| own cursor | no | yes | yes |
- Soft clone:
load_time_stepson the original changes the clone's dates (that is the point); dropping the original's table breaks it. - Hard clone: drifts silently, a date added to the original is not in the copy.
- One
roll_outover processes on a manager AND on its soft clones (0.3.5.15): the clones' cursors move with the manager at each date, so all of them compute at the same date.roll_out(process_list)without a manager uses the attached one when they share one table of dates (the original drives). A process on a hard clone, or any other manager, is refused before anything runs: roll it out with its own manager. - Outside a shared
roll_out, moving one cursor does not move the other. Once cursors differ, a dataset joining their features at one instant mixes business dates: read itas_ofa date both computed, orunify_time_managersand re-run the late ones from the first divergent date withforce_compute=True(a history is built forward only). - Freshness is per cursor; the Report's Time drivers groups show which processes are still aligned.
History already in the source: valid_time_column¶
If the source holds one row per entity and valid time, you do not need one run per date. Register the process with
valid_time_column (see fs-upload); the first run writes the whole history:
pid = tdfs4ds.upload_features(
'<DATABASE>.<HISTORY_VIEW>',
entity_id=['<ENTITY_COLUMN>'],
feature_names=['<FEATURE_1>'],
valid_time_column='<EVENT_TS>',
return_process_id=True,
)
A re-run is a no-op; an incremental load rewrites only from its own earliest valid time; a backdated correction is a
full load. The TimeManager date does not set the values' valid time for such a process. Use roll_out instead when
each date must be recomputed with the view's logic as of that date.
Rules¶
run(time_manager=tm)computes one date;roll_outis the loop. Never describe or userunas iterating over the manager's dates. (With a FilterManager attached,runloops over the segments of that one date.)- Synchronise processes on one manager. Processes whose features are combined in a dataset must share the same TimeManager, so their values have the same valid times.
- Build a history forward, from its first date. Each run writes its value from its business date onward
(a sequenced MERGE over
[business date, end of time)). A run at a date earlier than the first value already stored adds nothing in front of it. Since 0.3.5.16roll_outchecks this before step 1: when a process's stored values start after the first date it would roll out, it raisesValueErrorand runs nothing.roll_out(..., rebuild_history=True)deletes those processes' stored values from the first date rolled out onward (all of them in this case; logged inFS_REMOVAL_LOG, kindVALUES) and replays every date withforce_compute=True. A process whose valid time comes from a source column is not checked; an unreadable start is a warning. - Rebuild =
roll_out(..., time_id_start=1, force_compute=True). Replayed in order from the first date, the runs rewrite every stored period with the new logic (checked live: same periods, every value replaced, no old value left). Withoutforce_compute=True,roll_outskips the dates alreadyCOMPLETEDin the follow-up table and nothing changes. - Entities that leave and come back: a process registered with
close_absent_entities=True(seefs-upload) ends, at each date, the validity of the entities its view no longer returns; one that comes back later gets a new period, the absence is a gap. Same forward rule as the values: replay from the first date.roll_out(..., close_absent_entities=True|False)overrides the recorded option for one roll-out. time_id_start/time_id_endare 1-based positions in the ordered dates, not timestamps.- A monthly calendar with lags: use month STARTS, not month ends. A lag of
-1 monthisADD_MONTHS, which clamps month ends (30 Apr - 1 month = 30 Mar, not 31 Mar), so a lag read at a month end misses the previous month's value. Month starts shift exactly; let the process read the month that just ended (< BUSINESS_DATE). - A process first run in real time blocks the backfill.
upload_featuresruns at once by default, storing values valid from today; a laterroll_outover past dates then adds nothing before them (forward rule above). Register withtime_manager=tm, run_now=False, thenroll_out. If it already happened: remove the process and its features, then register again and roll out from the first date. - Before 0.3.5.14 (check
tdfs4ds.__version__):load_time_stepsfailed with Error 3515 when the source frame already had aBUSINESS_DATE/TIME_IDcolumn (rename it), andupload_features(run_now=False, dataset_view_name=...)dropped the dataset (build it withbuild_datasetafter the roll-out). roll_outrestores the session'sFEATURE_STORE_TIMEat the end;run()restores it too.- Register the processes with
time_manager=tmandrun_now=Falsewhen they must only run under the manager: an immediate real-time run would store values valid from today, after the business dates. - Look processes up by an exact view name (as above), never with
'X' in view_name(it also matchesX_DAILY,X_2). - Pass
on_failure='raise'under a scheduler so a failed run raises (FeatureIngestionError, carryingprocess_id,n_failed,n_total,failures); the default'continue'records the failure in the follow-up table and continues,'stop'stops at the first failed segment. An ingestion of 0 rows is recordedCOMPLETED,0 rowsand is recomputed by the next run. - Don't compute on a failed upstream:
run(..., check_upstream=True)/roll_out(..., check_upstream=True)refuses a process when an upstream process (from the persisted lineage graph) isFAILEDat this valid time;strict_upstream=Truealso refuses onRUNNING. Check without running withtdfs4ds.check_upstream_health(pid): its verdict separatesokfromverified, and when lineage is missing or stale it says so and gives therebuild_lineage_graph(...)command (fail-open, never a silent pass). - Generated scripts:
generate_execution_script(plan, wave_on_failure='skip_descendants')(default) skips the processes downstream of a failure and lets independent branches finish;'continue','stop_after_wave'and'raise'are the alternatives. The script's workers callrun(..., on_failure='raise'). - Re-running a fixed upstream does not repair downstream processes that already ran
COMPLETEDon its empty or stale output: re-run them withforce_compute=True. - Vantage Cloud Lake + tdstone2 Apply only — OFS manager table + JOIN the date in. A view scored by Apply runs on
the Analytical Compute Cluster (ACC). Create the manager with
TimeManager(..., storage='OFS')and reference the date through a JOIN to the manager view: the ACC rejects the scalar subquery with Error 6881. This is the one exception to the scalar-subquery rule.roll_out/run()cannot drive a view that inlines an Apply invocation (its ingestion CTAS also hits 6881): call the model'sscore()first, thenupload_featuresover the scores. A view that merely reads a tdstone2 repository table is an ordinary read and is driven like any other.