`fs-filter`¶

Segmented ingestion with FilterManager — standard, hybrid, or clone modes.

name: "Feature Store: Filter" description: "Generate code for segmented feature ingestion with a FilterManager — one segment at a time (standard), segments under a TimeManager, hybrid (each segment a (value, date) pair), or a cloned manager. Invoke when a process must run one category / region / partition at a time, when the source is too large for one pass, or for a daily backfill per category." argument-hint: "[filter-column][mode]" allowed-tools: "Read, Edit, Write, Bash, Glob, Grep"

Generate Python code for segmented feature ingestion with a FilterManager.

User context: $ARGUMENTS

Step	Skill	Why
Trigger	`fs-analyze`	Decide which column to segment on (ideally a partitioning column of the source).
Sibling	`fs-upload`	`upload_features(..., filtermanager=fm)` attaches the manager to the process.
Sibling	`fs-rollout`	Segments under a TimeManager: `roll_out` loops over dates, each date over segments.
Inspect	`fs-inspect`	`follow_up_report()` has one row per segment (`APPLIED_FILTER`).

Demos: demos/notebooks/07 (FilterManager alone), 08 (with a TimeManager), 09 (hybrid).

The model¶

A FilterManager is a table of segments (one row each, numbered filter_id 1..n) and a one-row view showing the current one. The process view reads the current segment through a scalar subquery on that view. run(process_id) sees the manager attached to the process and loops over the segments: move the cursor, ingest, next segment. Each segment gets its own follow-up row, so a failed segment is visible on its own and the others still complete.

Mode	Segment	When
Standard	one value of a column (a category, a region)	One slice at a time: smaller spool and unpivot per ingestion, per-segment trace.
With a TimeManager	the same segments, at each business date	A history computed segment by segment (`roll_out` over dates).
Hybrid	a (value, date) pair; the view exposes the column and `BUSINESS_DATE`	Daily backfill per category; only the listed pairs are processed.
Clone	a copy of another manager's segments	Another process needs the same segments with its own cursor.

The segments must split the entity. Each segment writes its own rows, so no two segments may produce the same entity key. Segmenting by Category requires the category in the entity (['CustomerID', 'Category']), or entities that each fall in one segment. With CustomerID alone, the segment processed last overwrites the values of a customer active in several categories.

Ask for any missing values before generating code: - Database of the feature store, data domain, source table. - Filter column(s) — e.g. 'Category', or Category + a date column for a hybrid manager. - Entity — must contain the filter column (or be split by it, see above). - Mode — standard, with a TimeManager, hybrid, or clone.

Standard FilterManager¶

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.filter_management import FilterManager

SOURCE = '<SOURCE_DATABASE>.<SOURCE_TABLE>'

# 1. The segments: one row per value. The view shows the current one.
fm = FilterManager(table_name='<FILTER_TABLE_NAME>', schema_name='<DATABASE>')
fm.load_filter(tdml.DataFrame.from_query(f"SELECT DISTINCT <FILTER_COLUMN> FROM {SOURCE}"))
print(fm.nb_filters, 'segments')
fm.update(filter_id=2); print(fm.display())      # move the cursor by hand to look at one segment
fm.update(filter_id=1)

# 2. The process view reads the current segment through a SCALAR subquery
#    (never a join, never a hard-coded value). The filter column is in the entity.
sql = f"""
SELECT
    A.<ENTITY_COLUMN>
,   A.<FILTER_COLUMN>
,   SUM(A.<VALUE_COLUMN>) AS <FEATURE_1>
,   COUNT(*)              AS <FEATURE_2>
FROM {SOURCE} A
WHERE A.<FILTER_COLUMN> = (SELECT <FILTER_COLUMN> FROM {fm.sql_view})
GROUP BY A.<ENTITY_COLUMN>, A.<FILTER_COLUMN>"""

pid = tdfs4ds.upload_features(
    df=None, sql_query=sql, target_view_name='<DATABASE>.<VIEW_NAME>',
    entity_id=['<ENTITY_COLUMN>', '<FILTER_COLUMN>'],
    feature_names=['<FEATURE_1>', '<FEATURE_2>'],
    filtermanager=fm,                            # attaches the manager: run() loops over its segments
    run_now=False, return_process_id=True)

# 3. Run every segment; a failed segment raises once all have been attempted
tdfs4ds.run(pid, on_failure='raise')

# 4. One follow-up row per segment
fu = tdfs4ds.follow_up_report(process_id=pid).to_pandas().reset_index()
print(fu[['APPLIED_FILTER', 'STATUS', 'START_DATETIME']])

Check once that segment by segment equals all at once: build a dataset on the entity and compare it with one unsegmented GROUP BY of the source (same row count, no differing value). Demo 07 does it: 8,858 rows on both sides, 0 differing values.

Segments under a TimeManager¶

from tdfs4ds.utils.time_management import TimeManager
tm = TimeManager(table_name='<TIME_TABLE_NAME>', schema_name='<DATABASE>')   # see fs-rollout

sql = f"""
SELECT A.<FILTER_COLUMN>, SUM(A.<VALUE_COLUMN>) AS <FEATURE_1>
FROM {SOURCE} A
WHERE A.<DATE_COLUMN>   <= (SELECT BUSINESS_DATE   FROM {tm.sql_view})
  AND A.<FILTER_COLUMN>  = (SELECT <FILTER_COLUMN> FROM {fm.sql_view})
GROUP BY A.<FILTER_COLUMN>"""
# upload_features(..., filtermanager=fm, run_now=False, return_process_id=True) as above, then:
tdfs4ds.roll_out([pid], time_manager=tm)          # each date, every segment: one follow-up row per (segment, date)
# with a domain clock (0.3.6.0): tdfs4ds.roll_out(data_domain='<DOMAIN>') - each date, every process, every segment

Hybrid FilterManager (each segment a (value, date) pair)¶

hf = FilterManager(table_name='<HYBRID_FILTER_TABLE>', schema_name='<DATABASE>')
hf.load_filter(
    tdml.DataFrame.from_query(f"""
        SELECT DISTINCT <FILTER_COLUMN>, <DATE_COLUMN> FROM {SOURCE}
        WHERE <DATE_COLUMN> BETWEEN DATE '<FROM>' AND DATE '<TO>'"""),
    time_column='<DATE_COLUMN>')                 # the view exposes <FILTER_COLUMN> and BUSINESS_DATE

sql = f"""
SELECT A.<FILTER_COLUMN>, A.<DATE_COLUMN>, SUM(A.<VALUE_COLUMN>) AS <FEATURE_1>
FROM {SOURCE} A
WHERE A.<FILTER_COLUMN> = (SELECT <FILTER_COLUMN> FROM {hf.sql_view})
  AND A.<DATE_COLUMN>   = (SELECT BUSINESS_DATE   FROM {hf.sql_view})
GROUP BY A.<FILTER_COLUMN>, A.<DATE_COLUMN>"""
# entity_id=['<FILTER_COLUMN>', '<DATE_COLUMN>']: the pair is the entity, so each segment writes its own rows

Clone a FilterManager¶

fm_copy = FilterManager(table_name='<NEW_TABLE>', schema_name='<DATABASE>')
fm_copy.clone_filter(fm, if_exists='replace')    # independent cursor: moving the copy does not move fm

Rules¶

The segments must split the entity (see above). Put the filter column in entity_id unless every entity lies in exactly one segment.
Scalar subqueries on the manager view, never a join, never a literal. Teradata reads the one-row view once per request and passes the value to the plan as a parameter.
What the plan reads — measured, not assumed (EXPLAIN, source partitioned by day × category, 50,000 rows): | Filter in the process view | Source read | |---|---| | category and day, each a scalar on a hybrid manager | a single partition | | the same filter written as a join to the manager view | all-rows scan | | date <= literal date | 1,267 partitions | | date <= (SELECT BUSINESS_DATE ...) (TimeManager, a range) | all-rows scan | | category = (SELECT ...) (one level only) | all-rows scan | A scalar eliminates partitions when it pins every partitioning level; on a small table one level or a range gets a full scan. Read the EXPLAIN step that accesses the source; do not count steps.
Segmenting pays most when the filter column is a partitioning column of the source.
Always pass filtermanager=fm to upload_features; without it run() computes only the segment the view shows.
fm.update(filter_id=N) moves the cursor by hand (1-based), to inspect or test one segment.
Use on_failure='raise' under a scheduler: every segment is attempted, then FeatureIngestionError lists the failed ones. 'stop' halts at the first failure; the default 'continue' records it and goes on.
Hybrid managers expose the date as BUSINESS_DATE; filter both columns, each with its own scalar subquery.
Vantage Cloud Lake + tdstone2 Apply only — OFS manager table + JOIN the filter in. A view scored by Apply runs on the Analytical Compute Cluster (ACC): create the manager with FilterManager(..., storage='OFS') and reference it through a JOIN (the ACC rejects scalar subqueries with Error 6881). This is the one exception to the scalar-subquery rule; roll_out/run() cannot drive an Apply model themselves.

fs-filter¶

Related skills¶