tdfs4ds.process_store.process_sql¶

tdfs4ds.process_store.process_sql ¶

Process versions from the SQL beneath a process view (0.3.3.12, optional).

A PROCESS_ID belongs to its view name: registering the view again keeps it, and it is the FEATURE_VERSION of every value the process writes. The SQL beneath the view - the view and every view it reads, down to the source tables - can change without the PROCESS_ID changing (a REPLACE VIEW by the optimizer, a migration or a person). The process version tracks that SQL:

FS_PROCESS_SQL (VALIDTIME, optional, no ALTER of any existing table) holds the current chain of every process, one row per step and 32,000-character chunk, with a hash per step and for the whole chain. Its valid time is real time: AS OF t gives the SQL in force at t, NONSEQUENCED every version.
:func:capture reads the chain from the live view definitions, hashes it, and compares the hash with the stored one (one single-AMP read). Same hash: nothing is written. A change closes the stored rows, writes the new chain under version v+1, and moves the process catalog's version counter (the base column FEATURE_VERSION, public PROCESS_VERSION) to v+1 at the current valid time. The first capture of a process records its current counter, unchanged.
It is called by upload_features (the view final), by run() before the ingestion, and by the optimizer's update_view. It never fails them.

The FEATURE_VERSION of the storage tables (FS_T_*) is the PROCESS_ID and is never touched here. tdfs4ds.PROCESS_SQL_VERSIONING = False turns all of this off.

logger `module-attribute` ¶

logger = getLogger(__name__)

TABLE_NAME `module-attribute` ¶

TABLE_NAME = 'FS_PROCESS_SQL'

VIEW_NAME `module-attribute` ¶

VIEW_NAME = 'FS_V_PROCESS_SQL'

CHUNK `module-attribute` ¶

CHUNK = 32000

_UNAVAILABLE `module-attribute` ¶

_UNAVAILABLE = {'flag': False}

_IN_ROLL_OUT `module-attribute` ¶

_IN_ROLL_OUT: dict = {}

COLUMNS `module-attribute` ¶

COLUMNS = ('PROCESS_ID', 'PROCESS_VERSION', 'STEP_NO', 'CHUNK_NO', 'DATABASE_NAME', 'OBJECT_NAME', 'OBJECT_KIND', 'STEP_HASH', 'CHAIN_HASH', 'SQL_TEXT', 'DB_USER')

_TOKEN_RE `module-attribute` ¶

_TOKEN_RE = compile('\'(?:[^\']|\'\')*\'|\\"(?:[^\\"]|\\"\\")*\\"|--[^\\n]*|/\\*.*?\\*/|\\s+|[^\'\\"\\s/-]+|.', DOTALL)

MANAGER_TOKEN `module-attribute` ¶

MANAGER_TOKEN = '<MANAGER>'

_FROM_RE `module-attribute` ¶

_FROM_RE = compile('(?i)\\b(?:FROM|JOIN)\\s+((?:"[^"]+"|[\\w$#@]+)(?:\\s*\\.\\s*(?:"[^"]+"|[\\w$#@]+))?)')

captured_for ¶

Context of a roll_out: each process is captured at its first date only (with that date's valid time); the later dates of the replay skip it.

process_ids `instance-attribute` ¶

process_ids = [p for p in (process_ids or [])]

init ¶

__init__(process_ids)

enter ¶

__enter__()

exit ¶

__exit__(*exc)

enabled ¶

enabled() -> bool

process_sql_ddl ¶

process_sql_ddl() -> tuple

(create_table, create_view). The metadata columns are LATIN or short UNICODE so that a 32,000-character UNICODE text fits the 64 KB row.

process_sql_creation ¶

process_sql_creation() -> bool

Create the table and its view if missing. Never raises; False when unavailable.

normalise_sql ¶

normalise_sql(text: str) -> str

The text a hash is taken on: comments removed, whitespace collapsed, case folded outside string literals (a quoted identifier keeps its case).

_sha ¶

_sha(text: str) -> str

canonical_name ¶

canonical_name(name: str) -> str

DB.OBJ upper-cased, quotes removed: one spelling per object.

manager_names ¶

manager_names(process_row: Optional[dict]) -> list

[(kind, 'DB.VIEW')] of the managers recorded for a process (catalog columns).

chain_managers ¶

chain_managers(chain: dict, process_row: Optional[dict] = None) -> list

[(kind, 'DB.VIEW')]: the managers a chain reads - recorded for the process, stopped at as manager inputs, or of the manager shape among its steps.

is_manager_step ¶

is_manager_step(step: dict, managers=()) -> bool

A recorded manager view, or a view of the manager shape: it reads one *_HIDDEN table only.

neutral_text ¶

neutral_text(text: str, managers=()) -> str

text with the references to the process's managers replaced by tokens.

versioned_steps ¶

versioned_steps(steps: list, managers=()) -> list

The steps a version is made of: the chain without the manager views.

step_hash ¶

step_hash(name: str, text: str, managers=()) -> str

chain_hash ¶

chain_hash(steps: list, managers=()) -> str

One hash for the chain, independent of the order the steps were listed in.

chain_rows ¶

chain_rows(process_id: str, version: int, steps: list, inputs=(), managers=()) -> list

The table rows of a chain: one per step and chunk; the inputs (source tables, managers) as rows of kind INPUT without text. Names are stored unquoted (canonical_name); manager views are never steps.

steps_from_rows ¶

steps_from_rows(rows: list) -> list

The steps of a stored version, chunks joined (inverse of :func:chain_rows).

changed_steps ¶

changed_steps(old: list, new: list, managers=()) -> list

Names of the steps added, removed or whose SQL changed between two chains.

A stored step carries its hash (taken with the managers of its time), which is used as is; a live step is hashed with managers.

_execute ¶

_execute(sql, params=None)

_process_row ¶

_process_row(process_id: str) -> Optional[dict]

live_chain ¶

live_chain(view_name: str, process_row: Optional[dict] = None) -> dict

The chain as it is now, from the view definitions (never the persisted lineage graph, which can lag a REPLACE VIEW): read_sql_chain over a graph built from the process view's DDL.

_validtime ¶

_validtime(at) -> str

The temporal qualifier of a DELETE / UPDATE at valid time at, as the ingestion MERGE does: CURRENT VALIDTIME in real time, else from at until changed. The period must reach UNTIL_CHANGED (9999-12-31): rows written in real time end there, and a period ending at 9999-01-01 left a sliver of them.

_as_of ¶

_as_of(at) -> str

current_version ¶

current_version(process_id: str, at=None) -> Optional[dict]

{"version", "hash"} in force at valid time at (now by default), or None.

known_versions ¶

known_versions(process_id: str) -> dict

{chain hash: version} of every version ever stored for the process.

_stored_rows ¶

_stored_rows(process_id: str, *, version: Optional[int] = None, at: Optional[str] = None) -> list

_write_version ¶

_write_version(process_id: str, version: int, chain: dict, *, at=None, managers=()) -> None

The chain under version from valid time at on (what was stored from at on is replaced; earlier valid times keep their version).

_set_catalog_version ¶

_set_catalog_version(process_id: str, version: int, at=None) -> bool

The catalog's version counter (base column FEATURE_VERSION, public PROCESS_VERSION) from valid time at on - the valid time registration and ingestion write at. The only statement touching it.

capture ¶

capture(process_id: str, origin: str = 'api', *, force: bool = False, at=None) -> dict

Record the SQL beneath a process; a change makes a new process version.

at is the valid time of the write: the valid time the process's values are written at (FEATURE_STORE_TIME under a TimeManager / roll_out, the run's real-time pin otherwise; None = now). The version is stamped from at on, in FS_PROCESS_SQL and in the catalog, so a value's version is the one in force at its ValidStart.

same SQL as the version in force at at: nothing is written;
nothing stored yet: the chain is stored under the catalog's current counter, which does not move;
SQL equal to an earlier version (a replay of past dates, a revert): that version is stamped again from at;
new SQL: version max+1 from at, catalog counter included.

Returns {"status": "first" | "unchanged" | "changed" | "reused" | "skipped" | "unavailable" | "error", "version", "changed_steps", "error"}. Never raises. Inside a roll_out only the first date captures; force ignores that.

capture_process_sql ¶

capture_process_sql(process_ids=None, *, all_processes: bool = False, dry_run: bool = True, progress=None) -> list

Capture the SQL of several processes (the backfill after an upgrade).

dry_run reports what would happen (first / unchanged / changed) without writing. Returns one dict per process. progress: a callable given "process SQL PROGRESS k/N <process id> <status>" after each process.

process_sql_versions ¶

process_sql_versions(process_id: str)

Every stored version of a process: PROCESS_VERSION, VALID_FROM, VALID_TO, STEPS, CHAIN_HASH, DB_USER, CHANGED_STEPS. A pandas DataFrame (empty when nothing is stored), or None when the table is unavailable.

process_sql ¶

process_sql(process_id: str, version: Optional[int] = None, *, as_of=None) -> Optional[dict]

The stored chain of a version (current when version and as_of are None): {"version", "steps": [{"name", "text"}], "inputs": [...]} or None.

process_sql_diff ¶

process_sql_diff(process_id: str, v1: int, v2: int) -> str

A unified diff of two stored versions, step by step.

chain_from_stored ¶

chain_from_stored(stored: dict) -> dict

A stored version in the shape of read_sql_chain (read_by recomputed from the references each step's text makes to the others).

runs_per_version ¶

runs_per_version(versions: list, runs: list) -> dict

{version: {"runs", "first", "last"}} (key None = before versioning).

versions: [{"PROCESS_VERSION", "VALID_FROM", "VALID_TO"}]; runs: follow-up rows {"VALIDTIME_DATE", "START_DATETIME"}. A run's valid time is its business date, else its start (the real-time pin is read just before it), and it belongs to the version whose period contains it. Pure.

chain_drift ¶

chain_drift(process_id: str, live: Optional[dict], process_row: Optional[dict] = None, current: Optional[dict] = None) -> Optional[dict]

{"since", "changed_steps"} when the live chain differs from the stored current version, else None (nothing stored, nothing read, or the same hash). current: :func:current_version when the caller already read it.

version_at ¶

version_at(process_id: str, when) -> Optional[int]

The process version in force at a real time (a run's start), or None when nothing was stored then (before versioning).

tdfs4ds.process_store.process_sql¶