From 621c1b1bd5d4c8f0dc56d7c78ec7f8a94d1c4eb9 Mon Sep 17 00:00:00 2001 From: Sanskar Singh Date: Wed, 22 Jul 2026 11:06:43 +0530 Subject: [PATCH] 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. Fixes #6621 Signed-off-by: Sanskar Singh --- sdk/python/feast/utils.py | 29 ++++++- sdk/python/tests/unit/test_utils.py | 125 ++++++++++++++++++++++++++++ 2 files changed, 151 insertions(+), 3 deletions(-) diff --git a/sdk/python/feast/utils.py b/sdk/python/feast/utils.py index 4c10c1903e5..ae41a78f013 100644 --- a/sdk/python/feast/utils.py +++ b/sdk/python/feast/utils.py @@ -55,6 +55,7 @@ from feast.base_feature_view import BaseFeatureView from feast.feature_service import FeatureService from feast.feature_view import FeatureView + from feast.feature_view_projection import FeatureViewProjection from feast.infra.registry.base_registry import BaseRegistry from feast.on_demand_feature_view import OnDemandFeatureView @@ -1193,6 +1194,19 @@ def _list_feature_views( return feature_views +def _merge_projection_features( + target: "FeatureViewProjection", + other: "FeatureViewProjection", +) -> None: + existing_names = {feature.name for feature in target.features} + merged = list(target.features) + for feature in other.features: + if feature.name not in existing_names: + existing_names.add(feature.name) + merged.append(feature) + target.features = merged + + def _get_feature_views_to_use( registry: "BaseRegistry", project, @@ -1223,6 +1237,7 @@ def _get_feature_views_to_use( feature_views = parsed # type: ignore[assignment] fvs_to_use, od_fvs_to_use = [], [] + fvs_by_projection_key: Dict[str, "FeatureView"] = {} for name, version_num, projection in feature_views: if version_num is not None: if not getattr(registry, "enable_online_versioning", False): @@ -1286,9 +1301,17 @@ def _get_feature_views_to_use( source_fv.entities = [] # type: ignore[attr-defined] source_fv.entity_columns = [] # type: ignore[attr-defined] - if source_fv not in fvs_to_use: - fvs_to_use.append( - source_fv.with_projection(copy.copy(source_projection)) + projection_key = source_projection.name_to_use() + existing_source_fv = fvs_by_projection_key.get(projection_key) + if existing_source_fv is None: + new_source_fv = source_fv.with_projection( + copy.copy(source_projection) + ) + fvs_by_projection_key[projection_key] = new_source_fv + fvs_to_use.append(new_source_fv) + else: + _merge_projection_features( + existing_source_fv.projection, source_projection ) else: if ( diff --git a/sdk/python/tests/unit/test_utils.py b/sdk/python/tests/unit/test_utils.py index 7eebf46461a..789a7fe9c1d 100644 --- a/sdk/python/tests/unit/test_utils.py +++ b/sdk/python/tests/unit/test_utils.py @@ -266,3 +266,128 @@ def test_large_scale_correctness(self): assert ts.seconds == expected_ts for status in fv.statuses: assert status == FieldStatus.PRESENT + + +class TestGetFeatureViewsToUseSharedSource: + def _build_store(self, data_dir): + import os + from datetime import timedelta + + import pandas as pd + + from feast import Entity, FeatureStore, FeatureView, Field, FileSource + from feast.infra.online_stores.sqlite import SqliteOnlineStoreConfig + from feast.on_demand_feature_view import on_demand_feature_view + from feast.repo_config import RepoConfig + from feast.types import Float64 + + driver = Entity(name="driver", join_keys=["driver_id"]) + + df = pd.DataFrame( + { + "driver_id": [1001, 1002], + "event_timestamp": [ + datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc), + datetime(2024, 1, 1, 11, 0, 0, tzinfo=timezone.utc), + ], + "created": [ + datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc), + datetime(2024, 1, 1, 11, 0, 0, tzinfo=timezone.utc), + ], + "a": [1.0, 2.0], + "b": [10.0, 20.0], + } + ) + src_path = os.path.join(data_dir, "src.parquet") + df.to_parquet(path=src_path, allow_truncated_timestamps=True) + + src = FileSource( + name="src_source", + path=src_path, + timestamp_field="event_timestamp", + created_timestamp_column="created", + ) + + src_fv = FeatureView( + name="src_fv", + entities=[driver], + ttl=timedelta(days=0), + schema=[ + Field(name="a", dtype=Float64), + Field(name="b", dtype=Float64), + ], + online=True, + source=src, + ) + + @on_demand_feature_view( + sources=[src_fv[["a"]]], + schema=[Field(name="a_out", dtype=Float64)], + mode="pandas", + ) + def odfv_a(inputs): + return pd.DataFrame({"a_out": inputs["a"] + 1}) + + @on_demand_feature_view( + sources=[src_fv[["b"]]], + schema=[Field(name="b_out", dtype=Float64)], + mode="pandas", + ) + def odfv_b(inputs): + return pd.DataFrame({"b_out": inputs["b"] + 100}) + + store = FeatureStore( + config=RepoConfig( + project="test_shared_source", + registry=os.path.join(data_dir, "registry.db"), + provider="local", + entity_key_serialization_version=3, + online_store=SqliteOnlineStoreConfig( + path=os.path.join(data_dir, "online.db") + ), + ) + ) + store.apply([driver, src, src_fv, odfv_a, odfv_b]) + return store + + def test_shared_source_projections_are_merged(self): + import tempfile + + from feast.utils import _get_feature_views_to_use + + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as data_dir: + store = self._build_store(data_dir) + + fvs, odfvs = _get_feature_views_to_use( + store.registry, + store.project, + ["odfv_a:a_out", "odfv_b:b_out"], + ) + + src_entries = [fv for fv in fvs if fv.projection.name_to_use() == "src_fv"] + assert len(src_entries) == 1 + projected = sorted( + feature.name for feature in src_entries[0].projection.features + ) + assert projected == ["a", "b"] + assert {odfv.name for odfv in odfvs} == {"odfv_a", "odfv_b"} + + def test_shared_source_ref_order_independent(self): + import tempfile + + from feast.utils import _get_feature_views_to_use + + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as data_dir: + store = self._build_store(data_dir) + + fvs, _ = _get_feature_views_to_use( + store.registry, + store.project, + ["odfv_b:b_out", "odfv_a:a_out"], + ) + src_entries = [fv for fv in fvs if fv.projection.name_to_use() == "src_fv"] + assert len(src_entries) == 1 + projected = sorted( + feature.name for feature in src_entries[0].projection.features + ) + assert projected == ["a", "b"]