tdfs4ds.process_store.process_followup¶

tdfs4ds.process_store.process_followup ¶

logger `module-attribute` ¶

logger = getLogger(__name__)

STATUS_MAX_LEN `module-attribute` ¶

STATUS_MAX_LEN = 2048

EMPTY_INGESTION_STATUS `module-attribute` ¶

EMPTY_INGESTION_STATUS = 'COMPLETED,0 rows'

_FOLLOWUP_UNION_COLUMNS `module-attribute` ¶

_FOLLOWUP_UNION_COLUMNS = ['RUN_ID', 'PROCESS_TYPE', 'PROCESS_ID', 'START_DATETIME', 'END_DATETIME', 'VALIDTIME_DATE', 'CAST(APPLIED_FILTER AS VARCHAR(2000) CHARACTER SET LATIN) AS APPLIED_FILTER', 'STATUS']

RUN_GAP_MINUTES `module-attribute` ¶

RUN_GAP_MINUTES = 10

execute_query_wrapper ¶

execute_query_wrapper(f)

Decorator to execute a query. It wraps around the function and adds exception handling.

This decorator is used to wrap around a function that generates a query and execute that query. It handles exceptions during query execution and prints the error message along with the query. The decorator works with both TeradataML's tdml and legacy tdml.get_context() execution methods, depending on the TeradataML version.

Parameters:

Name	Type	Description	Default
`f`	`function`	Function to be decorated.	required

Returns:

Name	Type	Description
`function`		Decorated function.

Example

To decorate a function that generates and executes a query:

@execute_query_wrapper def my_query_function(): return "SELECT * FROM my_table"

This decorator can be applied to functions that generate and execute queries with ease.

execute_query ¶

execute_query(query, raise_error=False, fetch_results=False, quiet_errors=(), log_errors=True)

Execute a SQL query or a list of queries using the tdml module.

Parameters:

Name	Type	Description	Default
`query`	`str or list`	A single SQL query string or a list of SQL query strings.	required
`raise_error`	`bool`	If True, re-raise exceptions after printing them. Default is False.	`False`
`fetch_results`	`bool`	If True, fetch and return the results of the SQL execution. Default is False.	`False`
`quiet_errors`	`tuple`	Lowercase substrings identifying failures the caller EXPECTS and will handle itself. A matching failure is logged at DEBUG instead of ERROR; whether it is re-raised is still governed by `raise_error` alone. Default `()` — every existing call site keeps byte-identical behaviour.	`()`
`log_errors`	`bool`	False when the caller re-raises (`raise_error=True`) and logs the failure itself, so each failure is logged once.	`True`

Returns: The result of the SQL execution if a single query is passed. None if a list of queries is passed or an exception occurs.

create_object_if_missing ¶

create_object_if_missing(create_query, label, follow_up=())

Run an idempotent CREATE and report honestly what happened.

This exists because of a structural trap: execute_query defaults to raise_error=False, so it logs the failure itself and returns normally — an except wrapped around it never sees anything. Every catalog-creation block written that way therefore logged an ERROR for a store that was simply already provisioned, then went on to log "has been created" for a table it had not created, while its own already-exists handling sat unreachable.

Parameters¶

create_query : str The CREATE statement. label : str Human-readable object name for the log lines (e.g. "TABLE DB.FS_X"). follow_up : iterable of str Statements to run once the object is known to be present, whether this call created it or found it. They must be idempotent — COMMENT ON and REPLACE VIEW are the intended uses.

Returns¶

bool True if this call created the object, False if it was already there.

Raises¶

Exception The original driver exception for any failure that is NOT "already exists" — logged at ERROR with the full message, because the half that names the fault is often the later half.

sql_literal ¶

sql_literal(value)

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.

logger_safe ¶

logger_safe(level, message, *args, **kwargs)

Wrapper around the global logger that only emits logs when tdfs4ds.DISPLAY_LOGS is True. level is a string like "info", "error", etc.

follow_up_table_queries ¶

follow_up_table_queries()

Build the follow-up table's CREATE, its COMMENT and its public view.

Pure builder: returns the SQL without executing it. Exposed as follow_up_table_creation.__wrapped__ so the package-wide "give me the SQL without running it" convention still holds now that the function no longer carries @execute_query_wrapper.

Returns:

Name	Type	Description
`list`		the three SQL strings (table, comment, view).

follow_up_table_creation ¶

follow_up_table_creation()

Create the follow-up table, its comment and its public view, idempotently.

Not decorated with @execute_query_wrapper since 0.3.1.211: that decorator printed the first error line and the entire CREATE statement to stdout whenever the table was already there -- the nominal case on any provisioned store. The comment and the REPLACE VIEW are idempotent and replayed either way, exactly as the decorator did.

Returns:

Type	Description
	None

followup_open ¶

followup_open(run_id, process_type, process_id, timemanager=None, filtermanager=None)

Generates an SQL query to insert a new record into the follow-up table for a given run.

Parameters:

Name	Type	Description	Default
`run_id`	`str`	The unique identifier for the run.	required
`process_type`	`str`	The type of process being executed.	required
`process_id`	`str`	The unique identifier for the process.	required
`timemanager`	`object`	An object managing time-related data. Defaults to None.	`None`
`filtermanager`	`object`	An object managing filter-related data. Defaults to None.	`None`

Returns:

Name	Type	Description
`str`		The generated SQL query string.

_status_literal ¶

_status_literal(status)

Return status truncated to the STATUS column width and escaped for SQL.

Truncation happens before escaping because escaping doubles quotes: the value that has to fit in VARCHAR(2048) is the stored (un-escaped) one. Escaping matters because a FAILED status embeds a raw Teradata error message, and several common ones quote the offending object (Object 'X' does not exist) — unescaped, that produced a malformed UPDATE whose failure was in turn swallowed, leaving the row RUNNING.

_validtime_predicate ¶

_validtime_predicate(feature_store_time)

Return the VALIDTIME_DATE predicate matching what followup_open wrote.

followup_open writes a literal SQL NULL when FEATURE_STORE_TIME is unset and the quoted timestamp otherwise, so the close must mirror both forms. Without this predicate the UPDATE matched every RUNNING row for the same (RUN_ID, PROCESS_TYPE, PROCESS_ID, APPLIED_FILTER) whatever its valid time — and because RUN_ID is generated once per import rather than per run (see tdfs4ds.RUN_ID), a success at one time step could silently close, as COMPLETED, the stale RUNNING row left by a failed earlier time step.

_followup_close_query ¶

_followup_close_query(run_id, process_type, process_id, status='COMPLETED', filtermanager=None, feature_store_time=None)

Build the UPDATE that closes a follow-up record. Pure — executes nothing.

followup_close ¶

followup_close(run_id, process_type, process_id, status='COMPLETED', filtermanager=None)

Close a follow-up record: set its STATUS and END_DATETIME.

A status starting with 'FAIL' records the failure — it is an ordinary write like any other. This function does not raise on a FAILED status.

History (why this is not decorated with @execute_query_wrapper): from v0.3.1.x until 0.3.1.210 the body ended with ::

if status.startswith('FAIL'):
    raise
return query

A bare raise placed before return query, inside a function whose decorator is query = f(...) and only then execute_sql(query). The exception escaped f(...) before the wrapper reached execution, so the UPDATE was built and thrown away: FAILED could never be written, on any install. Callers then wrapped the call in except Exception: pass to absorb that raise, hiding it completely, and the row stayed RUNNING forever.

Executing through execute_query rather than the decorator is deliberate: the decorator swallows an execution error with a bare print, and this write happens precisely when something has already gone wrong — a silent failure here reproduces the original symptom. execute_query logs at ERROR and, on a dropped connection, reconnects and retries once.

Returns:

Name	Type	Description
`bool`		True when the UPDATE was executed without error.

_followup_source ¶

_followup_source() -> Optional[str]

The run history of every attached store that offers one, as a derived table (FS_V_FOLLOW_UP of each, tagged FEATURE_STORE), or None with a single store -- then the primary's table is read exactly as before.

Run rows are keyed by PROCESS_ID, a UUID, so a row always belongs to the process of the store it came from.

follow_up_report ¶

follow_up_report(filtermanager: Optional[object] = None, process_id: Optional[str] = None)

Return a follow-up report as a tdml.DataFrame, optionally filtered by process_id and/or a filtermanager's applied filter.

Behavior by arguments

process_id is None and filtermanager is None: Return all rows from SCHEMA.FOLLOW_UP_NAME, sorted by START_DATETIME desc.
process_id is not None and filtermanager is None: Return rows for the given PROCESS_ID.
process_id is not None and filtermanager is not None: Return rows for the given PROCESS_ID whose APPLIED_FILTER matches the JSON_AGG of filtermanager's columns coming from its schema/view.
process_id is None and filtermanager is not None: Return rows whose APPLIED_FILTER matches the JSON_AGG of filtermanager (no PROCESS_ID constraint).

Parameters:

Name	Type	Description	Default
`filtermanager`	`Optional[object]`	An object exposing `col_names`, `schema_name`, and `view_name`. Its columns are aggregated via `JSON_AGG(col1, col2, ...)` to compare against A.APPLIED_FILTER.	`None`
`process_id`	`Optional[str]`	Optional process identifier used to filter by PROCESS_ID.	`None`

Returns:

Type	Description
	tdml.DataFrame: The resulting dataframe sorted by START_DATETIME (descending).

Raises:

Type	Description
`ValueError`	If `filtermanager` is provided but is missing required attributes or has an empty `col_names` list.
`RuntimeError`	If the query fails.

follow_up_report_many ¶

follow_up_report_many(process_ids)

Return follow-up rows for several process_ids in a single query.

Batched counterpart to follow_up_report(process_id=…): issues one WHERE PROCESS_ID IN (…) query instead of one round-trip per process. Callers group the result by PROCESS_ID in pandas to reproduce the per-process aggregation. Returns a tdml.DataFrame sorted by START_DATETIME descending.

Parameters:

Name	Type	Description	Default
`process_ids`		Iterable of process identifier strings. Empty/None entries are dropped; duplicates are collapsed.	required

Returns:

Type	Description
	tdml.DataFrame with the same columns as the follow-up table. When the
	id list is empty, an empty result (`WHERE 1=0`) is returned so the
	caller's downstream pandas logic still works.

Raises:

Type	Description
`RuntimeError`	If the query fails.

execution_run_ids ¶

execution_run_ids(g, gap_minutes: int = RUN_GAP_MINUTES)

The run each follow-up row belongs to, as an integer Series aligned on g.

A run is one call: a FilterManager run writes one row per segment and a roll_out one row per business date, back to back, so counting rows overstates how often a process runs (seen live: a 10-date roll-out read as "5,634 runs a day"). A run = the rows of one process with the same RUN_ID, each starting within gap_minutes of the previous row's end. The gap matters because RUN_ID is drawn once per Python process: a long-lived session (the app, a scheduler) that runs a process every day keeps one RUN_ID. Pure.

execution_starts ¶

execution_starts(g, gap_minutes: int = RUN_GAP_MINUTES)

Start times of a process's runs (see :func:execution_run_ids), sorted.

_segment_key ¶

_segment_key(row) -> str

The segment of a follow-up row: its applied filter, '' when unsegmented.

followup_business_dates ¶

followup_business_dates(followup, as_of=None) -> dict

What the follow-up says about each process, in business dates (0.3.1.240).

followup: the FS_FOLLOW_UP rows (pandas). FS_FOLLOW_UP is an event log, not a temporal table; every run row carries two times:

START_DATETIME / END_DATETIME - when the run happened;
VALIDTIME_DATE - the business date it computed (the TimeManager's date or FEATURE_STORE_TIME). A real-time run leaves it NULL: its business date is then its own start.

as_of: only runs started at or before that instant count ("what was known then"). Returns {PROCESS_ID: {"last_business_date", "last_run_at", "last_failure_at", "n_runs", "segments": {segment: {"last_business_date", "last_run_at", "failed"}}}} - timestamps as pandas Timestamps (UTC). Pure.

tdfs4ds.process_store.process_followup¶