tdfs4ds.feature_store.relocation¶
Relocate features to their own storage table, with their history (0.3.3.30).
plan_relocation is read only and returns a plan, as plan_removal does;
relocate_features(plan, dry_run=True) executes it, generate_relocation_script
writes it as a script that is a dry run unless started with --apply, and
verify_relocation checks the result. rewrite_dataset_tables is step 5 alone: it
rebuilds a dataset view so that some of its features read another storage table.
What a relocation is¶
A feature (a (data domain, entity, name) triplet) moves from the table it shares with
the other features of its entity and type to FS_T_<hash>__<TAG>. The tag is recorded in
FS_FEATURE_PLACEMENT and followed by register_features (see
feature_store/placement.py), so the next run writes to the new table instead of putting
the feature back. The value type is never converted (one tagged table per type), and the
new table has the source table's real primary index and partitioning, so rows land on the
same AMPs and partitions; primary_index= / partitioning= redistribute instead, in
which case the unit is the whole process (see :func:plan_relocation).
Order of the steps (a failure before the last one leaves the old table intact)::
create_table, copy_rows, verify_copy, place, rewrite_dataset*, check_dataset*,
lineage, statistics, clear_cache, delete_old
The delete is last: everything that can fail has run before the only irreversible step.
Older clients¶
A client older than 0.3.3.30 knows no placement. Its run of an already registered feature
leaves the catalog alone (ingestion reads FEATURE_TABLE from it), so the values keep going
to the tagged table - checked live. But whenever it registers the feature again (a type change,
or the feature removed and registered anew) it computes the untagged name, writes the catalog row
back to the old table and the values land there. This cannot be prevented from here, so every
client that runs these processes must be at least this version (the plan report says so). A new
client puts the tag back when it registers the feature and warns; the finding
relocation_reverted of the workflow assessment lists the rows an older client wrote to
the old table, and running :func:relocate_features on the same plan again moves them (a
sequenced merge: the periods of the rows already moved do not overlap).
module-attribute
¶
_TEMPLATE = '#!/usr/bin/env python\n"""Relocation script generated by tdfs4ds {version} on {generated_at}.\n\nMoves features to their own storage table (tag {tag}), in this order:\n{summary}\n\nRun it as is for a DRY RUN: each step says what it would do, nothing is changed.\nRun it with --apply to relocate. It stops at the first failed step; the old table\nis untouched until the last step (the delete). Afterwards it verifies the result\nand exits with code 2 if something is not as planned. Running it again after an\nolder client wrote to the old table moves those rows.\n\n{precondition}\n\nConnection: TDFS4DS_TD_HOST / USERNAME / PASSWORD (or LOGDATA) in the\nenvironment or a .env file, as tdfs4ds.reconnect_teradata() reads them.\n"""\nimport json\nimport sys\n\nPLAN = json.loads({plan_literal})\n\n\ndef main():\n apply = "--apply" in sys.argv[1:]\n force = "--force" in sys.argv[1:]\n import tdfs4ds\n from tdfs4ds.utils.reconnect import reconnect_teradata\n if not reconnect_teradata():\n sys.exit("reconnect_teradata() failed: set TDFS4DS_TD_HOST / USERNAME / PASSWORD.")\n if PLAN.get("schema"):\n tdfs4ds.SCHEMA = PLAN["schema"]\n if PLAN.get("schema_view"):\n tdfs4ds.SCHEMA_VIEW = PLAN["schema_view"]\n if PLAN.get("data_domain"):\n tdfs4ds.DATA_DOMAIN = PLAN["data_domain"]\n from tdfs4ds.feature_store.relocation import relocate_features, verify_relocation\n for b in PLAN.get("blocking") or []:\n print("BLOCKING (" + b["kind"] + "): " + b["detail"])\n print(("APPLY" if apply else "DRY RUN") + f": {{len(PLAN[\'steps\'])}} step(s)")\n res = relocate_features(PLAN, dry_run=not apply, force=force,\n progress=lambda m: print(" " + m))\n for s in res["steps"]:\n print(f" {{s[\'action\']}} {{s[\'object\']}}: {{s[\'result\']}}")\n if res.get("failed"):\n f = res["failed"]\n print(f"FAILED at {{f[\'action\']}} {{f[\'object\']}}: {{f[\'error\']}}")\n for x in f["not_run"]:\n print(" not run: " + x)\n sys.exit(1)\n if not apply:\n print("Nothing was changed. Run again with --apply to relocate.")\n return\n check = verify_relocation(PLAN)\n for c in check["checks"]:\n print((" ok " if c["ok"] else " FAIL ") + c["check"] + " " + c["object"] + (": " + c["detail"] if c["detail"] else ""))\n if not check["clean"]:\n sys.exit(2)\n print("Clean: relocation verified.")\n\n\nif __name__ == "__main__":\n try:\n main()\n finally:\n try: # end the session: an open one keeps a slot on the PE\n import teradataml as tdml\n tdml.remove_context()\n except Exception:\n pass\n'
Wrapper around the global logger that only emits logs when
tdfs4ds.DISPLAY_LOGS is True. level is a string like "info", "error", etc.
Build db.obj with each part quoted only as needed.
qualified('TDB', 'FS_X') == 'TDB.FS_X' (unchanged for ordinary names);
qualified('SVC@CORP', 'FS_X') == '"SVC@CORP".FS_X'. An empty/None
database yields the bare object name, so unqualified callers keep working.
force=True quotes both parts unconditionally — only for the sites that
have always emitted quotes and whose exact text is already persisted (dataset
view DDL, returned volatile-table names).
Split db.obj into (database, object), both de-quoted.
Quote-aware, so a dot inside a quoted part does not split the name:
split_qualified('"DEINF.CARLOSN"."FS_X"') == ('DEINF.CARLOSN', 'FS_X').
The naive rsplit('.', 1) this replaces returns ('"DEINF.CARLOSN"',
'"FS_X"') only by luck and ('"DEINF', 'CARLOSN"."FS_X"') in the
split('.') variant.
An unqualified name returns ('', name).
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.
select_features(snapshot: dict, features=None, process_id=None, include_shared: bool = False) -> dict
Which feature triplets a relocation moves, and which processes write them. Pure.
features: names (matched in tdfs4ds.DATA_DOMAIN when set, else in every domain;
a name that is several features is reported, never guessed), (name, entity,
data_domain) tuples or dicts with those keys. process_id: all the features of that
process. Both may be given.
A feature another process also computes is refused (shared blocking) unless
include_shared: it moves for both, and the other process is listed in processes.
collect_relocation_info(sel: dict, snapshot: dict, *, read_graph: bool = True, value_types=None) -> dict
The storage facts a plan needs, for the tables of the selected features: primary
index and partitioning (DBC), row counts per (FEATURE_ID, FEATURE_VERSION), perm, free
perm, RUNNING follow-up rows, the dataset views that read them and the views that read
the old per-table view. Every read is fail-open: what could not be read is listed in
unverified, never taken as "nothing there".
The placement tag of a storage table name (FS_T_<hash>__<TAG>), or None.
build_relocation_plan(sel: dict, info: dict, table_tag, *, primary_index=None, partitioning=None, include_shared: bool = False, value_types: Optional[dict] = None) -> dict
The ordered relocation steps, the blocking items and the cost. Pure.
value_types (0.3.5.1): {FEATURE_NAME: new SQL type}; the features move to the table of
that value type (their current placement tag kept unless table_tag is given), and the
catalog FEATURE_TYPE follows.
plan_relocation(features=None, process_id=None, table_tag=None, *, include_shared: bool = False, primary_index=None, partitioning=None, snapshot: dict = None, info: dict = None, value_types: Optional[dict] = None) -> dict
The plan to move features to their own storage table. Read only: nothing is executed.
Parameters¶
features : list, optional
Feature names, (name, entity, data_domain) tuples or dicts.
process_id : str, optional
All the features of that process.
table_tag : str
The tag of the new table(s): FS_T_<hash>__<TAG> (one per value type).
include_shared : bool
Move a feature another process also computes, for both. Default: refused.
primary_index, partitioning : optional
Redistribute: the new table gets this primary index (entity columns) and partitioning
instead of the source table's. The unit is then the whole process (its staging table
uses the process primary index), and the process's FOR_PRIMARY_INDEX /
FOR_DATA_PARTITIONING are updated with the catalog.
Returns a plan dict: moves, datasets (rewritten), blocking (nothing is
applied while it lists a fact; force=True overrides only :data:FORCEABLE kinds),
cost (rows, perm, free perm, whether the copy redistributes), steps (the delete
last) and the precondition about older clients.
clone_table_ddl(show_ddl: str, new_db: str, new_table: str, pi=None, partitioning=None, value_type=None) -> str
The CREATE TABLE of the new storage table from SHOW TABLE of the source:
same columns, same primary index and partitioning (pi / partitioning replace
them); value_type replaces the type of FEATURE_VALUE (0.3.5.1). Pure.
The statements copying the rows of ids from old to new. Pure.
The first copy is one nonsequenced INSERT ... SELECT (the history as stored). When the new table already holds rows of these features - a relocation applied again after an older client wrote to the old table - the same INSERT would overlap the periods already moved, so it is replaced by a sequenced merge, per (entity, feature, version), starting at the first period the old table holds for it: the new table's periods from there on are removed, the one in force is cut at that instant, and the old rows are inserted.
Run one plan step. Called by :func:relocate_features and by the generated script
with --apply only. ctx carries what a later step needs from an earlier one (the
rows of a dataset before its rewrite).
Execute a :func:plan_relocation plan, step by step.
dry_run=True (default) executes nothing and returns what each step would do. Otherwise
the steps run in order and stop at the first failure; the old table is untouched until the
last step. A plan with blocking items is refused; force=True overrides the kinds in
:data:FORCEABLE only. Running the same plan again after an older client wrote to the old
table moves those rows (a sequenced merge, not a plain insert).
Returns {"dry_run", "steps": [{action, object, result}], "failed": step or None, "clean"}.
After a relocation: the old table holds no row of the moved features, the new table
holds rows, the datasets compile, the catalog points at the new table and no dataset
still reads a stale table. Read only. Returns {"clean", "checks": [{"check",
"object", "ok", "detail"}]}.
A runnable .py for a :func:plan_relocation plan: a dry run unless started with
--apply. Pure ASCII. Refuses a plan with no step.
Write the relocation script of plan (plan_relocation / plan_type_change) to
out_path (default <SCHEDULING_SCRIPT_DIR>/tdfs4ds_relocation_<timestamp>.py) and
return its path. The script runs a dry run by default; --apply executes it.
generate_relocation_script returns the text without writing it.
{FEATURE_ID: 'DB.TABLE'} the catalog stores the features in (current valid time).
Rebuild a dataset view so that some of its features read another storage table.
The view is read back into its specification (storage table, FEATURE_ID, FEATURE_VERSION, alias, join type, strategy, offsets, history window, TimeManager), the table of the features concerned is swapped and the view is written again under the same name; the catalog row and DATASET_ID are kept. The features that do not move keep their table: a rewrite of every occurrence of a table name would break them.
dataset: 'database.view' (or the view name in the view database).
mapping: {feature_id: 'DB.TABLE'} (int key: that feature) or {'OLDDB.OLDTABLE':
'DB.TABLE'} (str key: every feature read from that table). None (default) points each
feature at the table the feature catalog stores it in now: the repair of a feature whose type
changed or that was relocated, i.e. the fix of dataset_reads_stale_table.
dry_run=True (default) changes nothing and returns the DDL.
Returns {"dataset", "moved": [(alias, old, new)], "ddl", "applied"}. Raises
ValueError when the view cannot be read back or the rewritten text does not parse back
to the same specification (nothing is changed then).
Rows of relocated features found in a table other than their tagged one (storage level, one grouped read per candidate table). A client older than 0.3.3.30 does this: it registers the feature in the untagged table again and writes there.
Returns [{"data_domain", "entity", "feature", "ids", "tag", "table" (the tagged one),
"stray_table", "rows", "last_start"}]. Candidates are the other storage tables of the same
entity (catalog). Fail-open: a table that cannot be read is skipped with a warning.
plan_type_change(process_id=None, *, types: Optional[dict] = None, entity=None, data_domain=None, snapshot: dict = None, info: dict = None) -> dict
The plan widening the stored type of features, with their history. Read only.
Either types ({feature: new SQL type}) with entity (list of entity columns) and
data_domain - the form the upload_features refusal prints - or process_id: then the
features of that process whose source type no longer fits the stored one, each to the
narrowest type holding the stored type and every producer's type
(type_changes.common_type). A feature computed by several processes moves for all of
them (one catalog entry); plan["type_changes"] lists the casts the other processes need
when their own type does not fit the new one.
Run it like a relocation: write_relocation_script(plan) (dry run, then --apply) or
relocate_features(plan, dry_run=False); verify_relocation(plan) checks the result.