Skip to content

Commit 75b9463

Browse files
Vedant-AgarwalVedant-Agarwal
authored
fix: Do not pass undeclared feature view columns to ODFV UDFs (#6527)
* fix: do not pass undeclared feature view columns to ODFV UDFs An OnDemandFeatureView's UDF was receiving every feature column in the online response, including features from feature views that the ODFV did not list in its sources. That allowed a UDF to silently depend on an undeclared source. This filters the UDF input down to the ODFV's declared sources before the transform runs, so columns from undeclared feature views are hidden. Join keys, request data and the declared features are kept. substrait already restricts its inputs through its query plan, so it is left as is. Fixes #6158. Signed-off-by: Vedant Agarwal <[email protected]> * test: cover ODFV source isolation for an aliased source Add a regression test that sources a pandas ODFV from an aliased feature view (with_name) and asserts the declared feature still reaches the UDF while an unrelated feature view stays hidden. The isolation filter keys columns by projection.name, which is what the retrieval path emits regardless of the alias; the test fails if that is switched to name_to_use(). Signed-off-by: Vedant Agarwal <[email protected]> --------- Signed-off-by: Vedant Agarwal <[email protected]>
1 parent 3c2ae3c commit 75b9463

2 files changed

Lines changed: 299 additions & 2 deletions

File tree

‎sdk/python/feast/utils.py‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -812,17 +812,58 @@ def _is_metrics_active():
812812
# Apply transformation. Note: aggregations and transformation configs are mutually exclusive
813813
# TODO: Fix to make it work for having both aggregation and transformation
814814
# ticket: https://github.com/feast-dev/feast/issues/5689
815+
# A UDF should only get features from the sources it declared. Drop
816+
# features that belong to other feature views so it can't read them by
817+
# accident (#6158). Join keys, request data and the declared features
818+
# stay. substrait filters on its own, so leave it alone.
819+
declared_source_names = {
820+
projection.name
821+
for projection in odfv.source_feature_view_projections.values()
822+
}
823+
undeclared_columns: set[str] = set()
824+
# features the caller asked for that aren't one of our sources
825+
for feature_ref in feature_refs:
826+
ref_view_name, _, ref_feature_name = _parse_feature_ref(feature_ref)
827+
if ref_view_name in requested_odfv_feature_names:
828+
continue
829+
if ref_view_name not in declared_source_names:
830+
undeclared_columns.add(ref_feature_name)
831+
undeclared_columns.add(f"{ref_view_name}__{ref_feature_name}")
832+
# features that are only in the response because another requested
833+
# ODFV needs them as input
834+
for other_odfv in requested_on_demand_feature_views:
835+
for projection in other_odfv.source_feature_view_projections.values():
836+
if projection.name in declared_source_names:
837+
continue
838+
for feature in projection.features:
839+
undeclared_columns.add(feature.name)
840+
undeclared_columns.add(f"{projection.name}__{feature.name}")
841+
815842
if odfv.mode == "python":
816843
if initial_response_dict is None:
817844
initial_response_dict = initial_response.to_dict()
845+
odfv_input_dict = {
846+
k: v
847+
for k, v in initial_response_dict.items()
848+
if k not in undeclared_columns
849+
}
818850
transformed_features_dict: Dict[str, List[Any]] = odfv.transform_dict(
819-
initial_response_dict
851+
odfv_input_dict
820852
)
821853
elif odfv.mode in {"pandas", "substrait"}:
822854
if initial_response_arrow is None:
823855
initial_response_arrow = initial_response.to_arrow()
856+
odfv_input_arrow = initial_response_arrow
857+
if odfv.mode == "pandas":
858+
odfv_input_arrow = initial_response_arrow.select(
859+
[
860+
c
861+
for c in initial_response_arrow.column_names
862+
if c not in undeclared_columns
863+
]
864+
)
824865
transformed_features_arrow = odfv.transform_arrow(
825-
initial_response_arrow, full_feature_names
866+
odfv_input_arrow, full_feature_names
826867
)
827868
else:
828869
raise Exception(

‎sdk/python/tests/unit/test_on_demand_pandas_transformation.py‎

Lines changed: 256 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -341,3 +341,259 @@ def pandas_view(inputs: pd.DataFrame) -> pd.DataFrame:
341341
),
342342
):
343343
store.apply([request_source, pandas_view])
344+
345+
346+
def test_odfv_udf_does_not_receive_undeclared_source_columns():
347+
"""#6158: a pandas ODFV's UDF should only see its own source columns, not
348+
features from another feature view that was requested too."""
349+
with tempfile.TemporaryDirectory() as data_dir:
350+
store = FeatureStore(
351+
config=RepoConfig(
352+
project="test_odfv_source_isolation",
353+
registry=os.path.join(data_dir, "registry.db"),
354+
provider="local",
355+
entity_key_serialization_version=3,
356+
online_store=SqliteOnlineStoreConfig(
357+
path=os.path.join(data_dir, "online.db")
358+
),
359+
)
360+
)
361+
362+
end_date = datetime.now().replace(microsecond=0, second=0, minute=0)
363+
start_date = end_date - timedelta(days=15)
364+
driver_df = create_driver_hourly_stats_df([1001], start_date, end_date)
365+
driver_stats_path = os.path.join(data_dir, "driver_stats.parquet")
366+
driver_df.to_parquet(path=driver_stats_path, allow_truncated_timestamps=True)
367+
368+
driver = Entity(name="driver", join_keys=["driver_id"])
369+
source = FileSource(
370+
name="driver_hourly_stats_source",
371+
path=driver_stats_path,
372+
timestamp_field="event_timestamp",
373+
created_timestamp_column="created",
374+
)
375+
376+
# Two feature views from the same source. The ODFV declares only fv_declared.
377+
fv_declared = FeatureView(
378+
name="fv_declared",
379+
entities=[driver],
380+
ttl=timedelta(days=0),
381+
schema=[Field(name="conv_rate", dtype=Float32)],
382+
online=True,
383+
source=source,
384+
)
385+
fv_undeclared = FeatureView(
386+
name="fv_undeclared",
387+
entities=[driver],
388+
ttl=timedelta(days=0),
389+
schema=[Field(name="avg_daily_trips", dtype=Int64)],
390+
online=True,
391+
source=source,
392+
)
393+
394+
@on_demand_feature_view(
395+
sources=[fv_declared],
396+
schema=[
397+
Field(name="saw_undeclared", dtype=Bool),
398+
Field(name="saw_declared", dtype=Bool),
399+
Field(name="saw_join_key", dtype=Bool),
400+
],
401+
mode="pandas",
402+
)
403+
def guard_view(inputs: pd.DataFrame) -> pd.DataFrame:
404+
out = pd.DataFrame()
405+
n = range(len(inputs))
406+
out["saw_undeclared"] = ["avg_daily_trips" in inputs.columns for _ in n]
407+
out["saw_declared"] = ["conv_rate" in inputs.columns for _ in n]
408+
out["saw_join_key"] = ["driver_id" in inputs.columns for _ in n]
409+
return out
410+
411+
store.apply([driver, source, fv_declared, fv_undeclared, guard_view])
412+
store.write_to_online_store(feature_view_name="fv_declared", df=driver_df)
413+
store.write_to_online_store(feature_view_name="fv_undeclared", df=driver_df)
414+
415+
response = store.get_online_features(
416+
entity_rows=[{"driver_id": 1001}],
417+
features=[
418+
"fv_declared:conv_rate",
419+
"fv_undeclared:avg_daily_trips",
420+
"guard_view:saw_undeclared",
421+
"guard_view:saw_declared",
422+
"guard_view:saw_join_key",
423+
],
424+
).to_dict()
425+
426+
# the other FV's column should be hidden; ours and the join key should not
427+
assert response["saw_undeclared"] == [False]
428+
assert response["saw_declared"] == [True]
429+
assert response["saw_join_key"] == [True]
430+
431+
432+
def _two_fv_store(data_dir):
433+
store = FeatureStore(
434+
config=RepoConfig(
435+
project="test_odfv_source_isolation",
436+
registry=os.path.join(data_dir, "registry.db"),
437+
provider="local",
438+
entity_key_serialization_version=3,
439+
online_store=SqliteOnlineStoreConfig(
440+
path=os.path.join(data_dir, "online.db")
441+
),
442+
)
443+
)
444+
end_date = datetime.now().replace(microsecond=0, second=0, minute=0)
445+
start_date = end_date - timedelta(days=15)
446+
driver_df = create_driver_hourly_stats_df([1001], start_date, end_date)
447+
path = os.path.join(data_dir, "driver_stats.parquet")
448+
driver_df.to_parquet(path=path, allow_truncated_timestamps=True)
449+
driver = Entity(name="driver", join_keys=["driver_id"])
450+
src = FileSource(
451+
name="driver_hourly_stats_source",
452+
path=path,
453+
timestamp_field="event_timestamp",
454+
created_timestamp_column="created",
455+
)
456+
fv1 = FeatureView(
457+
name="fv1",
458+
entities=[driver],
459+
ttl=timedelta(days=0),
460+
schema=[Field(name="conv_rate", dtype=Float32)],
461+
online=True,
462+
source=src,
463+
)
464+
fv2 = FeatureView(
465+
name="fv2",
466+
entities=[driver],
467+
ttl=timedelta(days=0),
468+
schema=[Field(name="avg_daily_trips", dtype=Int64)],
469+
online=True,
470+
source=src,
471+
)
472+
return store, driver, src, fv1, fv2, driver_df
473+
474+
475+
def test_odfv_python_mode_does_not_receive_undeclared_source_columns():
476+
"""Same as above but for python mode, where the UDF gets a dict (#6158)."""
477+
with tempfile.TemporaryDirectory() as data_dir:
478+
store, driver, src, fv1, fv2, driver_df = _two_fv_store(data_dir)
479+
480+
@on_demand_feature_view(
481+
sources=[fv1],
482+
schema=[
483+
Field(name="saw_undeclared", dtype=Bool),
484+
Field(name="saw_declared", dtype=Bool),
485+
],
486+
mode="python",
487+
)
488+
def guard_py(inputs: dict) -> dict:
489+
return {
490+
"saw_undeclared": ["avg_daily_trips" in inputs],
491+
"saw_declared": ["conv_rate" in inputs],
492+
}
493+
494+
store.apply([driver, src, fv1, fv2, guard_py])
495+
store.write_to_online_store(feature_view_name="fv1", df=driver_df)
496+
store.write_to_online_store(feature_view_name="fv2", df=driver_df)
497+
498+
response = store.get_online_features(
499+
entity_rows=[{"driver_id": 1001}],
500+
features=[
501+
"fv1:conv_rate",
502+
"fv2:avg_daily_trips",
503+
"guard_py:saw_undeclared",
504+
"guard_py:saw_declared",
505+
],
506+
).to_dict()
507+
508+
assert response["saw_undeclared"] == [False]
509+
assert response["saw_declared"] == [True]
510+
511+
512+
def test_odfv_does_not_see_other_requested_odfv_source_columns():
513+
"""#6158: one ODFV's UDF should not see another ODFV's source column, even
514+
when that feature view is only in the response as the other ODFV's input."""
515+
with tempfile.TemporaryDirectory() as data_dir:
516+
store, driver, src, fv1, fv2, driver_df = _two_fv_store(data_dir)
517+
518+
@on_demand_feature_view(
519+
sources=[fv1],
520+
schema=[Field(name="a_saw_fv2", dtype=Bool)],
521+
mode="pandas",
522+
)
523+
def odfv_a(inputs: pd.DataFrame) -> pd.DataFrame:
524+
out = pd.DataFrame()
525+
out["a_saw_fv2"] = [
526+
"avg_daily_trips" in inputs.columns for _ in range(len(inputs))
527+
]
528+
return out
529+
530+
@on_demand_feature_view(
531+
sources=[fv2],
532+
schema=[Field(name="b_out", dtype=Int64)],
533+
mode="pandas",
534+
)
535+
def odfv_b(inputs: pd.DataFrame) -> pd.DataFrame:
536+
out = pd.DataFrame()
537+
out["b_out"] = list(inputs["avg_daily_trips"])
538+
return out
539+
540+
store.apply([driver, src, fv1, fv2, odfv_a, odfv_b])
541+
store.write_to_online_store(feature_view_name="fv1", df=driver_df)
542+
store.write_to_online_store(feature_view_name="fv2", df=driver_df)
543+
544+
# ask for just the two ODFVs - fv2 is only here as odfv_b's input
545+
response = store.get_online_features(
546+
entity_rows=[{"driver_id": 1001}],
547+
features=["odfv_a:a_saw_fv2", "odfv_b:b_out"],
548+
).to_dict()
549+
550+
# odfv_a shouldn't see fv2's column; odfv_b still uses it fine
551+
assert response["a_saw_fv2"] == [False]
552+
assert response["b_out"] is not None
553+
554+
555+
def test_odfv_udf_receives_aliased_declared_source_columns():
556+
"""A declared source referenced under an alias (``with_name``) must still reach
557+
the UDF. The isolation filter keys columns by ``projection.name``, which is what
558+
the retrieval path emits regardless of the alias, so the aliased source's feature
559+
is retained (not dropped) while an unrelated feature view stays hidden (#6158)."""
560+
with tempfile.TemporaryDirectory() as data_dir:
561+
store, driver, src, fv1, fv2, driver_df = _two_fv_store(data_dir)
562+
563+
# ODFV declares fv1 *under an alias*; fv2 is requested but undeclared.
564+
aliased_source = fv1.with_name("aliased_fv1")
565+
assert aliased_source.projection.name == "fv1"
566+
assert aliased_source.projection.name_to_use() == "aliased_fv1"
567+
568+
@on_demand_feature_view(
569+
sources=[aliased_source],
570+
schema=[
571+
Field(name="saw_declared", dtype=Bool),
572+
Field(name="saw_undeclared", dtype=Bool),
573+
],
574+
mode="pandas",
575+
)
576+
def aliased_guard(inputs: pd.DataFrame) -> pd.DataFrame:
577+
out = pd.DataFrame()
578+
n = range(len(inputs))
579+
out["saw_declared"] = ["conv_rate" in inputs.columns for _ in n]
580+
out["saw_undeclared"] = ["avg_daily_trips" in inputs.columns for _ in n]
581+
return out
582+
583+
store.apply([driver, src, fv1, fv2, aliased_guard])
584+
store.write_to_online_store(feature_view_name="fv1", df=driver_df)
585+
store.write_to_online_store(feature_view_name="fv2", df=driver_df)
586+
587+
response = store.get_online_features(
588+
entity_rows=[{"driver_id": 1001}],
589+
features=[
590+
"fv1:conv_rate",
591+
"fv2:avg_daily_trips",
592+
"aliased_guard:saw_declared",
593+
"aliased_guard:saw_undeclared",
594+
],
595+
).to_dict()
596+
597+
# the aliased-but-declared source must survive; the unrelated FV stays hidden
598+
assert response["saw_declared"] == [True]
599+
assert response["saw_undeclared"] == [False]

0 commit comments

Comments
 (0)