Skip to content

Commit d269946

Browse files
sanskar-singh-2403ntkathole
sanskar-singh-2403
authored and
ntkathole
committed
fix: Merge shared ODFV source projections in feature resolution
When several on demand feature views share a source feature view but select different input columns, _get_feature_views_to_use appended the shared source view once per ODFV, each copy carrying only that ODFV's projection. The membership check 'if source_fv not in fvs_to_use' never matched because BaseFeatureView.__eq__ compares projections and the freshly fetched full-projection view never equals a stored narrowed copy. Downstream, _group_feature_refs keys its view index by projection.name_to_use(), so the last duplicate won and the other ODFVs' source projections were silently dropped, producing missing inputs (NaN in pandas mode) during retrieval. Index accumulated source views by projection.name_to_use() and merge projections (union of features, deduped by name) instead of appending a duplicate, so a shared source view resolves to a single entry carrying the union of the required input columns. The merge helper reassigns projection.features to a fresh list rather than mutating in place, because the projections in play are shallow copies (copy.copy) that share their features list with the registry object. Adds TestGetFeatureViewsToUseSharedSource in tests/unit/test_utils.py covering the merged-projection result and order independence. Fixes #6621 Signed-off-by: Sanskar Singh <[email protected]>
1 parent d81d4e3 commit d269946

2 files changed

Lines changed: 210 additions & 5 deletions

File tree

‎sdk/python/feast/utils.py‎

Lines changed: 36 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@
5555
from feast.base_feature_view import BaseFeatureView
5656
from feast.feature_service import FeatureService
5757
from feast.feature_view import FeatureView
58+
from feast.feature_view_projection import FeatureViewProjection
5859
from feast.infra.registry.base_registry import BaseRegistry
5960
from feast.on_demand_feature_view import OnDemandFeatureView
6061

@@ -1193,6 +1194,19 @@ def _list_feature_views(
11931194
return feature_views
11941195

11951196

1197+
def _merge_projection_features(
1198+
target: "FeatureViewProjection",
1199+
other: "FeatureViewProjection",
1200+
) -> None:
1201+
existing_names = {feature.name for feature in target.features}
1202+
merged = list(target.features)
1203+
for feature in other.features:
1204+
if feature.name not in existing_names:
1205+
existing_names.add(feature.name)
1206+
merged.append(feature)
1207+
target.features = merged
1208+
1209+
11961210
def _get_feature_views_to_use(
11971211
registry: "BaseRegistry",
11981212
project,
@@ -1223,6 +1237,25 @@ def _get_feature_views_to_use(
12231237
feature_views = parsed # type: ignore[assignment]
12241238

12251239
fvs_to_use, od_fvs_to_use = [], []
1240+
fvs_by_projection_key: Dict[str, "FeatureView"] = {}
1241+
1242+
def _append_or_merge_source_fv(fv_with_projection: "FeatureView") -> None:
1243+
# Index every (source) feature view by its effective projection key so
1244+
# that a regular FeatureView and an ODFV source projection resolving to
1245+
# the same key are merged into a single entry instead of duplicated.
1246+
# Without this, requesting ["src_fv:a", "odfv_b:b_out"] where odfv_b
1247+
# sources src_fv appends two src_fv entries and later raises
1248+
# "KeyError: Feature a not found in projection src_fv".
1249+
projection_key = fv_with_projection.projection.name_to_use()
1250+
existing = fvs_by_projection_key.get(projection_key)
1251+
if existing is None:
1252+
fvs_by_projection_key[projection_key] = fv_with_projection
1253+
fvs_to_use.append(fv_with_projection)
1254+
else:
1255+
_merge_projection_features(
1256+
existing.projection, fv_with_projection.projection
1257+
)
1258+
12261259
for name, version_num, projection in feature_views:
12271260
if version_num is not None:
12281261
if not getattr(registry, "enable_online_versioning", False):
@@ -1286,10 +1319,8 @@ def _get_feature_views_to_use(
12861319
source_fv.entities = [] # type: ignore[attr-defined]
12871320
source_fv.entity_columns = [] # type: ignore[attr-defined]
12881321

1289-
if source_fv not in fvs_to_use:
1290-
fvs_to_use.append(
1291-
source_fv.with_projection(copy.copy(source_projection))
1292-
)
1322+
new_source_fv = source_fv.with_projection(copy.copy(source_projection))
1323+
_append_or_merge_source_fv(new_source_fv)
12931324
else:
12941325
if (
12951326
hide_dummy_entity
@@ -1302,7 +1333,7 @@ def _get_feature_views_to_use(
13021333
fv = fv.with_projection(copy.copy(projection))
13031334
if version_num is not None:
13041335
fv.projection.version_tag = version_num
1305-
fvs_to_use.append(fv)
1336+
_append_or_merge_source_fv(fv) # type: ignore[arg-type]
13061337

13071338
return (fvs_to_use, od_fvs_to_use)
13081339

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

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -266,3 +266,177 @@ def test_large_scale_correctness(self):
266266
assert ts.seconds == expected_ts
267267
for status in fv.statuses:
268268
assert status == FieldStatus.PRESENT
269+
270+
271+
class TestGetFeatureViewsToUseSharedSource:
272+
def _build_store(self, data_dir):
273+
import os
274+
from datetime import timedelta
275+
276+
import pandas as pd
277+
278+
from feast import Entity, FeatureStore, FeatureView, Field, FileSource
279+
from feast.infra.online_stores.sqlite import SqliteOnlineStoreConfig
280+
from feast.on_demand_feature_view import on_demand_feature_view
281+
from feast.repo_config import RepoConfig
282+
from feast.types import Float64
283+
284+
driver = Entity(name="driver", join_keys=["driver_id"])
285+
286+
df = pd.DataFrame(
287+
{
288+
"driver_id": [1001, 1002],
289+
"event_timestamp": [
290+
datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc),
291+
datetime(2024, 1, 1, 11, 0, 0, tzinfo=timezone.utc),
292+
],
293+
"created": [
294+
datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc),
295+
datetime(2024, 1, 1, 11, 0, 0, tzinfo=timezone.utc),
296+
],
297+
"a": [1.0, 2.0],
298+
"b": [10.0, 20.0],
299+
}
300+
)
301+
src_path = os.path.join(data_dir, "src.parquet")
302+
df.to_parquet(path=src_path, allow_truncated_timestamps=True)
303+
304+
src = FileSource(
305+
name="src_source",
306+
path=src_path,
307+
timestamp_field="event_timestamp",
308+
created_timestamp_column="created",
309+
)
310+
311+
src_fv = FeatureView(
312+
name="src_fv",
313+
entities=[driver],
314+
ttl=timedelta(days=0),
315+
schema=[
316+
Field(name="a", dtype=Float64),
317+
Field(name="b", dtype=Float64),
318+
],
319+
online=True,
320+
source=src,
321+
)
322+
323+
@on_demand_feature_view(
324+
sources=[src_fv[["a"]]],
325+
schema=[Field(name="a_out", dtype=Float64)],
326+
mode="pandas",
327+
)
328+
def odfv_a(inputs):
329+
return pd.DataFrame({"a_out": inputs["a"] + 1})
330+
331+
@on_demand_feature_view(
332+
sources=[src_fv[["b"]]],
333+
schema=[Field(name="b_out", dtype=Float64)],
334+
mode="pandas",
335+
)
336+
def odfv_b(inputs):
337+
return pd.DataFrame({"b_out": inputs["b"] + 100})
338+
339+
store = FeatureStore(
340+
config=RepoConfig(
341+
project="test_shared_source",
342+
registry=os.path.join(data_dir, "registry.db"),
343+
provider="local",
344+
entity_key_serialization_version=3,
345+
online_store=SqliteOnlineStoreConfig(
346+
path=os.path.join(data_dir, "online.db")
347+
),
348+
)
349+
)
350+
store.apply([driver, src, src_fv, odfv_a, odfv_b])
351+
return store
352+
353+
def test_shared_source_projections_are_merged(self):
354+
import tempfile
355+
356+
from feast.utils import _get_feature_views_to_use
357+
358+
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as data_dir:
359+
store = self._build_store(data_dir)
360+
361+
fvs, odfvs = _get_feature_views_to_use(
362+
store.registry,
363+
store.project,
364+
["odfv_a:a_out", "odfv_b:b_out"],
365+
)
366+
367+
src_entries = [fv for fv in fvs if fv.projection.name_to_use() == "src_fv"]
368+
assert len(src_entries) == 1
369+
projected = sorted(
370+
feature.name for feature in src_entries[0].projection.features
371+
)
372+
assert projected == ["a", "b"]
373+
assert {odfv.name for odfv in odfvs} == {"odfv_a", "odfv_b"}
374+
375+
def test_regular_fv_and_odfv_source_are_merged(self):
376+
# Regression: a regular FeatureView requested alongside an ODFV that
377+
# sources it must resolve to a single, merged src_fv entry. Previously
378+
# the regular FV was appended but not indexed in fvs_by_projection_key,
379+
# so the ODFV source appended a second partial src_fv projection and a
380+
# later lookup raised "KeyError: Feature a not found in projection
381+
# src_fv".
382+
import tempfile
383+
384+
from feast.utils import _get_feature_views_to_use
385+
386+
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as data_dir:
387+
store = self._build_store(data_dir)
388+
389+
fvs, odfvs = _get_feature_views_to_use(
390+
store.registry,
391+
store.project,
392+
["src_fv:a", "odfv_b:b_out"],
393+
)
394+
395+
src_entries = [fv for fv in fvs if fv.projection.name_to_use() == "src_fv"]
396+
assert len(src_entries) == 1
397+
projected = sorted(
398+
feature.name for feature in src_entries[0].projection.features
399+
)
400+
assert projected == ["a", "b"]
401+
assert {odfv.name for odfv in odfvs} == {"odfv_b"}
402+
403+
def test_regular_fv_and_odfv_source_merge_order_independent(self):
404+
import tempfile
405+
406+
from feast.utils import _get_feature_views_to_use
407+
408+
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as data_dir:
409+
store = self._build_store(data_dir)
410+
411+
fvs, _ = _get_feature_views_to_use(
412+
store.registry,
413+
store.project,
414+
["odfv_b:b_out", "src_fv:a"],
415+
)
416+
417+
src_entries = [fv for fv in fvs if fv.projection.name_to_use() == "src_fv"]
418+
assert len(src_entries) == 1
419+
projected = sorted(
420+
feature.name for feature in src_entries[0].projection.features
421+
)
422+
assert projected == ["a", "b"]
423+
424+
def test_shared_source_ref_order_independent(self):
425+
import tempfile
426+
427+
from feast.utils import _get_feature_views_to_use
428+
429+
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as data_dir:
430+
store = self._build_store(data_dir)
431+
432+
fvs, _ = _get_feature_views_to_use(
433+
store.registry,
434+
store.project,
435+
["odfv_b:b_out", "odfv_a:a_out"],
436+
)
437+
src_entries = [fv for fv in fvs if fv.projection.name_to_use() == "src_fv"]
438+
assert len(src_entries) == 1
439+
projected = sorted(
440+
feature.name for feature in src_entries[0].projection.features
441+
)
442+
assert projected == ["a", "b"]

0 commit comments

Comments
 (0)