tdfs4ds.feature_store.feature_data_processing¶
module-attribute
¶
module-attribute
¶
Bases: ValueError
The process returns several rows for one entity key.
A ValueError, so existing handlers keep working. It carries what is
needed to fix the process view:
n_keys— number of entity key values found on several rows;entity— the entity key columns;sample— pandas DataFrame: every row of a few duplicated keys (N_ROWS= rows for that key), or None when it could not be read;differing— feature columns whose value differs between the rows of a sampled key (empty: the rows are identical copies);query— SQL listing every duplicated key of the process view.
Return name as a Teradata identifier, quoting it only if required.
Idempotent: an already-quoted name is returned unchanged rather than being
wrapped a second time. Embedded quotes are escaped as "".
Parameters¶
name : str
The raw object/database name.
force : bool, default False
Quote even when the name does not need it. Use only at the few sites
that have always emitted unconditional quotes and whose exact output is
already baked into persisted DDL (see
feature_store/feature_query_retrieval.py) — elsewhere forcing would
change the SQL of existing stores for no reason.
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).
Return the Teradata database user for volatile-table scoping.
The logon name is used here as a DATABASE QUALIFIER (<user>.<volatile>),
i.e. an identifier slot, so it must be double-quoted whenever it is not a
bare-legal Teradata identifier. LDAP/AD logins routinely are not: a dotted
name ('DEINF.CARLOSN') unquoted is an invalid three-part name (Error 3707),
stripping it to the last segment points at the wrong database (Error 3807),
and a name with an '@' ('SVC@CORP') is a syntax error outright. Quoting makes
it one valid identifier: "SVC@CORP".table.
An ordinary name is returned unchanged, so nothing about the generated SQL changes for a normal login. This is the identifier form only -- the raw name is what belongs in a credential or a DBC string predicate.
Display a TeradataML DataFrame as a formatted table in the console.
This function formats each row of the TeradataML DataFrame according to specified maximum column widths. It prints out the DataFrame in a tabular format with a header.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
df
|
DataFrame
|
The TeradataML DataFrame to be displayed. |
required |
max_widths
|
list of int
|
Maximum width of each column in characters. Defaults to [25, 55, 10]. |
[25, 55, 10]
|
header
|
list of str
|
Column headers to be displayed. Defaults to ["Feature Database", "Feature Table", "# rows"]. |
['Feature Database', 'Feature Table', '# rows']
|
Returns:
| Name | Type | Description |
|---|---|---|
None |
This function only prints the table and does not return anything. |
Example
To display a TeradataML DataFrame with custom column widths and headers:
display_table(my_teradataml_dataframe, max_widths=[30, 60, 12], header=["DB Name", "Table Name", "Row Count"])
Note
- The 'max_widths' list specifies the maximum width of each column in characters.
- The 'header' list defines the column headers to be displayed.
- The function formats each row of the TeradataML DataFrame to fit within the specified column widths and prints the table.
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
|
()
|
log_errors
|
bool
|
False when the caller re-raises ( |
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.
Converts a duration in seconds to a formatted string representing days, hours, minutes, and seconds with three decimal places.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
seconds
|
float
|
The duration in seconds. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
str |
A formatted string representing the duration in days, hours, minutes, and seconds with three decimal places. |
Wrapper around the global logger that only emits logs when
tdfs4ds.DISPLAY_LOGS is True. level is a string like "info", "error", etc.
Every row of up to n_keys duplicated entity keys of table (pandas), or None.
NULL keys are matched too (a NULL key repeated is a duplicate as well). Fail-open: this only serves the error message.
Columns whose value differs between the rows of one sampled key.
Build the DuplicateEntityKeysError for nb_duplicates duplicated keys.
The check runs on a volatile table created AS ... WITH DATA, a SET
table in Teradata mode: rows that are exact copies were already dropped
there (harmless, they carry the same values). The listing query counts
DISTINCT rows of the same columns, so it reports the same keys.
Raise if any entity / primary-index key column contains NULL values.
NULL values in an entity (primary-index) key column are invalid: the MERGE ON clause uses plain equality, so an un-substituted NULL key never matches an existing row and the rows fan out into a cartesian-like product (severe skew, runtime and storage blow-up). This mirrors the duplicate-key guard, which also hard-fails rather than silently dropping rows.
Parameters¶
null_counts : dict
Mapping of entity key column name -> number of NULL rows (as read from the
materialized volatile table, where any column covered by
entity_null_substitute is already COALESCE'd and therefore reports 0).
Raises¶
ValueError If one or more columns have a NULL count greater than 0. The message names the offending column(s) and their counts.
The ON condition joining two relations on the entity columns.
entity_id: {column: SQL type}; entity_null_substitute: {column: value} for
the columns whose NULLs are matched through COALESCE (quoted for a VARCHAR column);
left_name / right_name: the two relations' aliases. Returns the conditions joined
with AND.
Generate a COLLECT STATISTICS SQL query based on specified entity identifiers, primary index, and partitioning information. The query collects statistics for columns relevant to the data structure, including partitioning and entity identifiers, for optimized query performance.
entity_id (list or dict): A list of entity IDs or a dictionary where keys represent entity IDs. Used to identify the columns requiring statistical collection. primary_index (str or list, optional): The primary index column(s) for the table, provided as a single string or list of strings. Defaults to an empty string. partitioning (str, optional): A string defining partitioning information, used to identify additional columns for statistical collection. Defaults to an empty string.
tuple: A tuple containing two strings: - The base COLLECT STATISTICS SQL query. - The extended COLLECT STATISTICS SQL query with sampling and thresholds applied.
The entity columns a partitioning expression names, in entity_columns order.
One rule for every reader of a storage table's partitioning: statistics
collection (expected_statistics) and batch retrieval, which adds a
partition-eliminating predicate on exactly these columns. Whole-word,
case-insensitive, so SITE does not match SITE_OID_CD. FEATURE_ID
is not an entity column and is never returned.
The statistics a feature storage table should carry, as column lists.
The single authority: generate_collect_stats renders its statements from
it, and check_storage_statistics compares a table's defined statistics
against it — so the check can never disagree with what ingestion collects.
Returns (base, extended):
base— single-column statistics collected plainly:PARTITION, the partitioning columns (alwaysFEATURE_ID, plus any entity column the partitioning expression names),FEATURE_VERSION,ValidStart,ValidEnd.extended— multi-column statistics collectedUSING SAMPLE 25 PERCENT AND THRESHOLD 15 PERCENT: the primary index, the entity columns when they differ from it, and the entity columns +FEATURE_ID— what every dataset sub-query filters on (one entity, one feature). Without that last one DIAGNOSTIC HELPSTATS flagged it and the optimizer misestimated a single-entity lookup on a one-sub-query-per-feature dataset view by ~400x (7.79 s vs 0.02 s on a 10-feature dataset, measured live).
Fetch feature ID and conversion mapping from feature catalog.
Parameters: - list_entity_id: List of entity IDs. - feature_names: List of feature names.
Returns: - feature_id_names: List of tuples containing feature details (FEATURE_ID, FEATURE_NAME, FEATURE_TABLE, FEATURE_DATABASE). - conversion_name2id: Dictionary mapping FEATURE_NAME to FEATURE_ID.
prepare_feature_ingestion(df, entity_id, feature_names, feature_versions=None, primary_index=None, partitioning='', entity_null_substitute={}, valid_time_column=None, **kwargs)
Transforms and prepares a DataFrame for feature ingestion into a feature store by unpivoting it.
This function accepts a DataFrame along with the specifications for entity IDs, feature names, and optionally, feature versions. It generates a new DataFrame with the data unpivoted to align with the requirements of a feature store ingestion process. Additionally, it creates a volatile table in the database to facilitate the transformation.
Parameters:
- df (tdml.DataFrame): The input DataFrame containing the feature data.
- entity_id (list/dict/other): Specifies the entity IDs. If a dictionary, the keys are used as entity IDs.
If a list or another type, it is used directly as the entity ID.
- feature_names (list): Names of features to be unpivoted.
- feature_versions (dict, optional): Maps feature names to their versions. If not provided, a default version is applied.
- primary_index (list/str, optional): Primary index(es) for the volatile table. If not specified, entity IDs are used.
- valid_time_column (str, optional): a DATE / TIMESTAMP column of df carrying each row's valid
time (0.3.5.0). Staged as TDFS_VALID_TIME; the unique key and the NULL check then include it.
None (default): the staging SQL is unchanged.
- **kwargs: Additional keyword arguments for customization.
tuple: A tuple containing: 1. A tdml.DataFrame with the transformed data ready for ingestion. 2. The name of the volatile table created in the database.
Raises: Exception: If the function encounters an issue during the database operations or data transformations.
Note: - The function handles different data types for 'entity_id' to ensure correct SQL query formation. - It assumes a specific structure for 'df' to properly execute SQL commands for data transformation.
Example:
df = tdml.DataFrame(...) entity_id = {'customer_id': 'INTEGER'} feature_names = ['feature1', 'feature2'] transformed_df, volatile_table_name = prepare_feature_ingestion(df, entity_id, feature_names)
Error 3624, "There are no statistics defined for the table".
Matched as the bracketed code or by its phrase, never as a bare number.
apply_collect_stats(entity_id: Mapping[str, Any] | Iterable[str], primary_index: Optional[str], partitioning: Optional[str], feature_infos: DataFrame) -> Dict[str, Any]
Run COLLECT STATS on all target feature tables, with fallbacks and timing.
Steps:
1) Determine a stable ordering of entity IDs (for deterministic query gen).
2) Group feature_infos by FEATURE_DATABASE + FEATURE_TABLE to get unique targets.
3) Generate COLLECT STATS statements via generate_collect_stats(...) for fallback use.
4) For each target table:
- Try a simple COLLECT STATS ON <db>.<table>.
- On failure, retry with generated statements (and optional extension).
5) Log a compact summary (counts + total duration) and return it as a dict.
Parameters¶
entity_id : Mapping[str, Any] | Iterable[str]
Entity identifiers used to parameterize collect-stat statements.
If a mapping (e.g., dict), its keys are used and sorted.
If an iterable (e.g., list/tuple), it’s sorted directly.
primary_index : Optional[str]
Primary index used by generate_collect_stats (may be None).
partitioning : Optional[str]
Partitioning clause used by generate_collect_stats (may be None).
feature_infos : pd.DataFrame
Must contain columns: 'FEATURE_TABLE', 'FEATURE_DATABASE', 'FEATURE_ID'.
Returns¶
Dict[str, Any] Summary with keys: - total_tables: int - ok: int - retried: int - failed: int - duration_seconds: float - duration_hms: str - details: list[dict] # per-table status entries
The valid start of the rows a store call writes, or None for real time.
tdfs4ds.FEATURE_STORE_TIME when set (a TimeManager / roll_out run, or an
explicit choice), else the run's pinned valid_start — one database
timestamp read once by run() so that every feature of a real-time run
shares one valid start (a CURRENT VALIDTIME MERGE per feature gives each
its own clock reading, ~1 s apart). None keeps the historical
CURRENT VALIDTIME statements, byte-identical.
_store_feature_update_insert(entity_id, volatile_table_name, entity_null_substitute={}, primary_index=None, partitioning='', features_infos=None, **kwargs)
Stores prepared feature data in specific feature tables within a Teradata database. This function performs updates and inserts on the feature tables based on the provided entity IDs and prepared feature data.
Parameters: - entity_id (dict): A dictionary representing the entity ID, where keys are used to identify the entity. - prepared_features (tdml.DataFrame): A DataFrame containing the prepared feature data for ingestion. - **kwargs: Additional keyword arguments that can include: - schema (str): The schema name where the feature tables are located. - feature_catalog_name (str, optional): The name of the feature catalog table, defaulting to 'FS_FEATURE_CATALOG'.
Returns: None: This function does not return any value but performs database operations.
Note: - The function uses the Teradata temporal feature (VALIDTIME) for handling the valid time of the records. - It constructs SQL queries to update existing feature values and insert new ones based on the entity ID. - The function handles the entity ID, feature IDs, and versions for inserting and updating records. - It utilizes additional settings from the tdfs4ds module for configuring the feature store time and display logs. - This function assumes that the necessary database tables and schema are correctly set up and accessible.
Example Usage:
entity_id_dict = {'customer_id': 'INTEGER'} prepared_features = tdml.DataFrame(...) store_feature(entity_id_dict, prepared_features)
_store_feature_merge(entity_id, volatile_table_name, entity_null_substitute={}, primary_index=None, partitioning='', features_infos=None, **kwargs)
Stores prepared feature data in specific feature tables within a Teradata database. This function performs updates and inserts on the feature tables based on the provided entity IDs and prepared feature data.
Parameters: - entity_id (dict): A dictionary representing the entity ID, where keys are used to identify the entity. - prepared_features (tdml.DataFrame): A DataFrame containing the prepared feature data for ingestion. - **kwargs: Additional keyword arguments that can include: - schema (str): The schema name where the feature tables are located. - feature_catalog_name (str, optional): The name of the feature catalog table, defaulting to 'FS_FEATURE_CATALOG'.
Returns: None: This function does not return any value but performs database operations.
Note: - The function uses the Teradata temporal feature (VALIDTIME) for handling the valid time of the records. - It constructs SQL queries to update existing feature values and insert new ones based on the entity ID. - The function handles the entity ID, feature IDs, and versions for inserting and updating records. - It utilizes additional settings from the tdfs4ds module for configuring the feature store time and display logs. - This function assumes that the necessary database tables and schema are correctly set up and accessible.
Example Usage:
entity_id_dict = {'customer_id': 'INTEGER'} prepared_features = tdml.DataFrame(...) store_feature(entity_id_dict, prepared_features)
store_feature(entity_id, volatile_table_name, entity_null_substitute={}, primary_index=None, partitioning='', features_infos=None, **kwargs)
Conditionally stores prepared feature data in specific feature tables within a Teradata database. This function decides whether to perform updates and inserts on the feature tables based on a global configuration.
Parameters: - entity_id (dict): A dictionary representing the entity ID, where keys are used to identify the entity. - prepared_features (tdml.DataFrame): A DataFrame containing the prepared feature data for ingestion. - **kwargs: Additional keyword arguments for further customization and configurations.
Returns: None: This function does not return any value but performs database operations based on the specified condition.
Example Usage:
entity_id_dict = {'customer_id': 'INTEGER'} prepared_features = tdml.DataFrame(...) store_feature(entity_id_dict, prepared_features)
_close_absent_queries(entity_columns, volatile_ref, absent_ref, tables, store_time=None, filter_view=None, filter_columns=None)
SQL that ends the validity of the entities a run no longer produces. Pure.
tables: [(feature_database, feature_table, feature_ids, feature_versions)],
the storage tables this run wrote, with the same FEATURE_ID / FEATURE_VERSION
scope as the ingestion MERGE. An entity is absent when no row of volatile_ref
(the run's staged values, one row per entity) carries its keys; with a
FilterManager (filter_view / filter_columns) only the stored entities of
the active segment are considered.
The rule is the ingestion MERGE's, applied to the absent: a run states the
truth from its valid time t onward. With a valid time (a TimeManager /
roll_out date, an explicit FEATURE_STORE_TIME, or the run's pinned real time)
the delete is sequenced over VALIDTIME PERIOD '(t, <END_PERIOD>)', the same
period as the MERGE; without one, CURRENT VALIDTIME closes at now.
Two statements per table, because a sequenced DELETE refuses every subquery (Error 9347 for NOT EXISTS, NOT IN and IN alike, verified on Vantage 20) but accepts a join to a non-temporal table:
CREATE VOLATILE TABLE <absent_ref> AS (NONSEQUENCED VALIDTIME SELECT DISTINCT <keys> ...): the stored keys of this process, absent from the run, in the active segment (a superset is harmless: the delete only touches periods overlapping its own);- the temporal DELETE joined to it.
A DROP TABLE <absent_ref> follows each pair. A re-run at the same date finds
nothing left and is a no-op; an ordered replay converges. Re-running an EARLIER
date alone rewrites the future of absent and present entities alike, exactly
as the MERGE already does for the present ones (verified live): a history is
replayed from its first date. Ending only the period in force at t was tried
and rejected: whether a later period survived then depended on whether the
value had changed at a later date.
close_absent_entities(entity_id, volatile_table_name, features_infos, filtermanager=None, valid_start=None, at=None)
End the validity of the entities this run no longer produces.
Called by the ingestion right after store_feature, in the same session (the
staged volatile table must still exist), for a process registered with
close_absent_entities=True. The close sits at the run's valid time, the
same instant the MERGE wrote from (_ingestion_time), or at at when given
(a valid_time_column load: its earliest valid time). Raises on a database
error so the caller records the segment FAILED.
Returns {"tables", "rows", "valid_time"} (rows: the stored rows the
DELETE touched, as reported by the driver) or {"skipped": reason} when a
FilterManager segment cannot tell which stored entities are its own.
Prepares a DataFrame for feature ingestion into a tdstone2 feature store by transforming the data structure. This function unpivots the DataFrame and adds necessary columns for entity IDs, feature names, feature values, and a predefined feature version. It creates a volatile table in the database to facilitate this transformation.
- df (tdml.DataFrame): The input DataFrame containing the feature data. The DataFrame should be structured in a way that is compatible with tdstone2 feature store requirements.
- entity_id (list/dict/other): A representation of the entity ID. If a dictionary, the keys are used as entity IDs. If a list or another data type, it is used directly as the entity ID.
Returns: - tdml.DataFrame: A transformed DataFrame suitable for feature store ingestion. - str: The name of the volatile table created in the database for storing the transformed data.
Note: - The function requires that the input DataFrame 'df' has a valid table name and is compatible with tdml operations. - It automatically handles the creation and management of a volatile table in the database. - 'ID_PROCESS' is used as the default feature version identifier. - The transformed DataFrame includes columns for each entity ID, 'FEATURE_NAME', 'FEATURE_VALUE', and 'FEATURE_VERSION'.
Example Usage:
input_df = tdml.DataFrame(...) entity_id_dict = {'customer_id': 'INTEGER'} transformed_df, table_name = prepare_feature_ingestion_tdstone2(input_df, entity_id_dict)