It shows how to use tags. For the second part, I'll sketch some code tonight.

here’s an example sketch — in this example we only pull data from the DB that’s relevant to the features we want:

# functions.py
import pandas as pd
from hamilton.function_modifiers import tag, tag_outputs, extract_columns

COL_NAMES = ["a", "b", "c"] # hardcoded list of the schema

@tag_outputs(**{c: dict(attributes=["source"]) for c in COL_NAMES})
@extract_columns(*COL_NAMES)
def raw_dataset(table_name: str, columns: list[str], db_client: object) -> pd.DataFrame:
    query = " ... " # use table_name and columns to create the query
    return pd.read_sql(query, con=db_client)

@tag(feature="prod")
def transform_a(a: pd.Series) -> pd.Series:
   ...

@tag(feature="staging")
def transform_b(b: pd.Series) -> pd.Series:
   ...

then in the driver we could do something like:

from hamilton import driver, base
import functions
# create a driver
dr = ( 
  driver.Builder()
    .with_modules(functions)
    .with_adapters(base.PandasDataFrameResult())
    .build()
)
# get prod features
prod_features = dr.list_available_variables(tag_filter={"feature":"prod"})
print(prod_features)
# find the sources that feed these features
all_nodes = dr.what_is_upstream_of(*[f.name for f in prod_features])
print(all_nodes)
# get the source column names
just_sources = [s.name for s in all_nodes if "source" in s.tags.get("attributes", [])]
print(just_sources)

# execute it
result = dr.execute(
  [f.name for f in prod_features], 
  inputs={'columns': just_sources, ...}
)

So the technical term of what we’ve down above is a lightweight “predicate pushdown”. From a feature store perspective it would in general be similar: 1. tag features 2. tag sources 3. given requested features, walk the dag to find sources starting from what you want to compute 4. using those sources, create the relevant input for the DAG, in the example above we passed in

columns

as a value… 5. execute things… Does this help? make sense?