Skip to content

FilterManager — segmented ingestion

FilterManager lets upload_features iterate through a source one partition at a time: one date, one category, or one (date × category) combination per pass. It is the right tool whenever a single full-source ingestion would exhaust spool, exceed a maintenance window, or scan way more data than necessary.

When to use it

  • A source table is partitioned but the feature engineering view does not filter on the partition columns. The query optimizer / fs-analyze will flag this and propose a FilterManager rewrite.
  • The source is too large for a single MERGE pass.
  • The business logic should ingest one region / one segment / one date at a time for traceability.

Three modes

Mode Filter columns Typical use
Standard One categorical column Iterate by region / segment / category
Hybrid Date + categorical Iterate by (date × category) — BUSINESS_DATE always present
Clone Copy an existing filter Re-use the same filter sequence across multiple processes

How FilterManager works internally

FilterManager maintains two database objects:

  1. Hidden table (<view_name>_HIDDEN) — stores every filter row, each assigned a deterministic filter_id via ROW_NUMBER(). This table is written once by load_filter() and never mutated during ingestion.
  2. Public view (<view_name>) — always exposes exactly one row: WHERE filter_id = N. Switching to a different partition rewrites only this view definition — no data is touched.

Your feature engineering view references the public view with scalar subqueries. Teradata resolves those at plan time and pushes the constant into the source scan, enabling partition elimination.

load_filter()          → writes hidden table + creates view at filter_id=1
update(filter_id=N)    → REPLACE VIEW … WHERE filter_id = N
display()              → SELECT * FROM public view  (shows current partition)

When upload_features(filtermanager=fm) is called, the framework calls fm.update(filter_id) for each iteration internally.

Backing-table storage on Vantage Cloud Lake (storage='OFS')

By default the hidden table is created on block storage — unchanged. On Vantage Cloud Lake you can create it on the Object File System (OFS) instead:

fm = FilterManager(
    table_name  = 'FS_FILTER_REGIONS',
    schema_name = tdfs4ds.SCHEMA,
    storage     = 'OFS',        # None (default) = block storage, unchanged
)

This emits MAP = TD_MAP2, STORAGE = TD_OFSSTORAGE. The choice is per manager (block and OFS managers coexist in one session) and only affects table creation; ofs_map= overrides the map (default TD_MAP2).

tdstone2 Apply on the ACC: OFS manager table and a JOIN (not a scalar subquery)

A tdstone2 model scored via the Apply backend runs on the Analytical Compute Cluster (ACC), which reads from OFS. For a scoring view that references the FilterManager, two conditions must both hold: (1) storage='OFS' so the ACC can read the manager table (block storage → Error 6881); (2) reference the filter/date values via a JOIN to the FilterManager view, not correlated scalar subqueries — the ACC rejects scalar subqueries with 6881 but accepts a JOIN. Confirmed live: OFS + JOIN scores; OFS + scalar subquery still 6881s. Writing the filter as a JOIN is the user's responsibility. Note this is the opposite of the block-storage default guidance elsewhere in this page (scalar subqueries for partition elimination) — it applies only to the tdstone2-Apply/ACC path. See the TimeManager guide for a worked JOIN example.

Standard FilterManager

from tdfs4ds.utils.filter_management import FilterManager

fm = FilterManager(
    table_name  = 'FS_FILTER_REGIONS',
    schema_name = tdfs4ds.SCHEMA,
)

# Populate with the distinct REGION values from the source
fm.load_filter(tdml.DataFrame.from_query("""
    SELECT DISTINCT REGION
    FROM source_db.TRANSACTIONS
    ORDER BY REGION
"""))

print('Iterations:', fm.nb_filters)
fm.display()   # shows the current (first) partition

Wire the filter into the feature view

The source query must reference the FilterManager view with scalar subqueries — the optimizer resolves the value before scanning the source table, enabling partition elimination:

df_features_raw = tdml.DataFrame.from_query(f"""
    SELECT A.*
    FROM source_db.TRANSACTIONS A
    WHERE A.REGION = (SELECT REGION FROM {fm.schema_name}.{fm.view_name})
""")

# ... your feature engineering ...
df_features_proc = crystallize_view(
    df_features_raw.groupby('CUSTOMER_ID').agg(...),
    view_name   = 'FEAT_ENG_TX',
    schema_name = tdfs4ds.SCHEMA,
)

tdfs4ds.upload_features(
    df_features_proc,
    entity_id     = ['CUSTOMER_ID'],
    feature_names = ['nb_tx', 'total_amount'],
    filtermanager = fm,            # ← framework iterates over all partitions
)

Hybrid FilterManager (time × partition)

The hybrid mode adds a BUSINESS_DATE column so one iteration covers both a calendar date and a categorical dimension. The public view always exposes BUSINESS_DATE as the time column — your feature view reads it with a scalar subquery just like any other filter column.

fm = FilterManager(table_name='FS_FILTER_REGION_DATE', schema_name=tdfs4ds.SCHEMA)

fm.load_filter(
    tdml.DataFrame.from_query("""
        SELECT DISTINCT REGION, TX_DATE
        FROM source_db.TRANSACTIONS
        ORDER BY TX_DATE DESC, REGION
    """),
    time_column='TX_DATE',      # → stored as BUSINESS_DATE
)

print(fm.nb_filters)            # total iterations = |dates| × |regions|

df_features_raw = tdml.DataFrame.from_query(f"""
    SELECT A.*
    FROM source_db.TRANSACTIONS A
    WHERE A.TX_DATE = (SELECT BUSINESS_DATE FROM {fm.schema_name}.{fm.view_name})
      AND A.REGION  = (SELECT REGION        FROM {fm.schema_name}.{fm.view_name})
""")

The BUSINESS_DATE column is also consumed by roll_out to set tdfs4ds.FEATURE_STORE_TIME automatically on each step.

Cloning a hybrid filter works identically to a standard filter — clone_filter() preserves the BUSINESS_DATE column in the shared hidden table. See Cloning.

Inspection methods

Method Returns Description
display() teradataml.DataFrame Current partition row from the public view
get_all_filters() teradataml.DataFrame All rows in the hidden table
get_current_filterid() int Active filter_id parsed from view DDL
get_date_in_the_past() str BUSINESS_DATE of the current partition (hybrid only)
get_filter_condition() dict or str Active filter as a Python dict or JSON string
print_view_ddl() — Logs the view DDL for troubleshooting 
# Inspect active filter as a dictionary
print(fm.get_filter_condition(output_type="dict"))
# → {'REGION': 'WEST'}

# Jump manually to partition 5 for debugging
fm.update(filter_id=5)
fm.display()

# Roll back to the beginning
fm.update(filter_id=1)

Pruning

prune_filter() removes all partitions with filter_id < N and renumbers the remaining ones so the smallest id becomes 1. Use it to discard already-ingested partitions and keep the table lean.

# Remove all partitions before the current one (compact to where we are)
fm.prune_filter()                  # uses get_current_filterid() automatically

# Or remove everything before partition 10
fm.prune_filter(filter_id=10)

After pruning the public view is repointed to filter_id=1.

Cloning

Cloning lets one process re-use the filter definitions owned by another without duplicating the hidden table data. There are two clone modes:

Soft clone (default)

The new manager's view points directly to the source hidden table. No data is copied. Both managers remain in sync: if the source is repopulated with load_filter, both managers automatically reflect the new data (though their active filter_id is independent).

# Source manager already exists and has been populated
fm_source = FilterManager(table_name='FS_FILTER_REGIONS', schema_name=tdfs4ds.SCHEMA)

# New manager for a second feature process — shares the same hidden table
fm_new = FilterManager(table_name='FS_FILTER_REGIONS_PROCESS2', schema_name=tdfs4ds.SCHEMA)
fm_new.clone_filter(
    source_filtermanager = fm_source,
    filter_id_to_apply   = 1,         # start at the first partition
    clone_mode           = 'soft',    # default
)

The new FS_FILTER_REGIONS_PROCESS2 view is created (or replaced) in the same schema. The underlying hidden table remains FS_FILTER_REGIONS_HIDDEN in the source schema.

Ownership note: by default take_ownership=False so _drop() will not touch the hidden table when called on the new manager. Pass take_ownership=True if this manager should be responsible for cleanup (rare for soft clones).

Hard clone

The new manager gets its own physical copy of the hidden table. Changes to the source afterwards do not affect the clone.

fm_copy = FilterManager(table_name='FS_FILTER_REGIONS_COPY', schema_name=tdfs4ds.SCHEMA)
fm_copy.clone_filter(
    source_filtermanager = fm_source,
    filter_id_to_apply   = 1,
    clone_mode           = 'hard',
)
# fm_copy._owns_hidden is now True — it may drop its table if needed

A CREATE TABLE … AS (SELECT * FROM …) WITH DATA is issued into fm_copy.schema_name. The copy is fully independent.

Handling existing tables with if_exists

if_exists controls what happens when the destination hidden table already exists. It only has effect in hard clone mode — soft clones create only a view, never a table:

Value Behaviour
"error" (default) Raises RuntimeError if the destination already exists
"replace" Drops and recreates the destination table
"skip" Reuses the existing table without re-copying 
fm_copy.clone_filter(
    source_filtermanager = fm_source,
    clone_mode           = 'hard',
    if_exists            = 'replace',   # safe for repeated pipeline runs
)

Taking ownership after a soft clone

If you soft-cloned and later decide this manager should own cleanup:

fm_new.take_ownership()   # sets _owns_hidden = True

This does not copy any data; it only marks the instance. Use with care — two managers that both believe they own the same hidden table will both try to drop it.

Choosing between soft and hard

Soft clone Hard clone
Data footprint No extra storage Doubles the hidden table
Sync with source Automatic Independent snapshot
Can drop hidden table Only with take_ownership=True Always (owns the copy)
Typical use Multiple processes, same filter sequence, same shard window Archived copy for audit / replay

Operational notes

  • Seed small, expand later. Start with 3–5 partitions to validate the pipeline. Remove any QUALIFY ROW_NUMBER() cap before a full roll-out.
  • Order by date descending in hybrid mode so the first ingestion run reflects the most recent business state.
  • Move the manager manually with fm.update(filter_id=N) when debugging a single partition. filter_id is 1-based.
  • Errors do not abort the loop. Each partition is wrapped in try/except — upload_features records FAILED,… in the follow-up table and continues with the next filter.

See the fs-filter skill for a copy-paste agent template.