Skip to content

Commit c9b9d72

Browse files
mao-liumao-liu
authored
feat: Add upstream FeatureView lineage support for PushSource (#6853)
* feat: implementing registry lineage for PushSource Signed-off-by: Mao Liu <[email protected]> * rename source_views to upstream_feature_view_names, add docs and validation Signed-off-by: Mao Liu <[email protected]> --------- Signed-off-by: Mao Liu <[email protected]>
1 parent 386759c commit c9b9d72

12 files changed

Lines changed: 1077 additions & 53 deletions

File tree

‎docs/reference/data-sources/push.md‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,3 +84,22 @@ streamingDF.writeStream.foreachBatch(feast_writer).start()
8484
```
8585

8686
This can also be used under the hood by a contrib stream processor (see [Tutorial: Building streaming features](../../tutorials/building-streaming-features.md))
87+
88+
## External Pipeline Lineage (upstream_feature_view_names)
89+
90+
When feature transformations occurring outside Feast consume upstream Feast feature views before pushing results into Feast, this dependency metadata can be captured by setting `upstream_feature_view_names` on `PushSource` to visualize the relationship in the Feast Web UI lineage graph:
91+
92+
```python
93+
push_source2 = PushSource(
94+
name="push_source2",
95+
upstream_feature_view_names=[fv],
96+
)
97+
98+
fv2 = FeatureView(
99+
name="feature view 2",
100+
entities=[user],
101+
schema=[Field(name="derived_value", dtype=Int64)],
102+
source=push_source2,
103+
)
104+
```
105+

‎protos/feast/core/DataSource.proto‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -299,6 +299,9 @@ message DataSource {
299299
// the online store on-demand, such as by stream consumers.
300300
message PushOptions {
301301
reserved 1;
302+
// Names of upstream FeatureViews/LabelViews consumed by the external process feeding this PushSource.
303+
// Represents external pipeline lineage metadata; does not trigger Feast-managed DAG execution.
304+
repeated string upstream_feature_view_names = 2;
302305
}
303306

304307

‎sdk/python/feast/data_source.py‎

Lines changed: 46 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
import warnings
1616
from abc import ABC, abstractmethod
1717
from datetime import datetime, timedelta, timezone
18-
from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
18+
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union
1919

2020
from google.protobuf.duration_pb2 import Duration
2121
from google.protobuf.json_format import MessageToJson
@@ -876,12 +876,14 @@ class PushSource(DataSource):
876876
# TODO(adchia): consider adding schema here in case where Feast manages pushing events to the offline store
877877
# TODO(adchia): consider a "mode" to support pushing raw vs transformed events
878878
batch_source: Optional[DataSource] = None
879+
upstream_feature_view_names: List[str] = []
879880

880881
def __init__(
881882
self,
882883
*,
883884
name: str,
884885
batch_source: Optional[DataSource] = None,
886+
upstream_feature_view_names: Optional[Sequence[Union[Any, str]]] = None,
885887
description: Optional[str] = "",
886888
tags: Optional[Dict[str, str]] = None,
887889
owner: Optional[str] = "",
@@ -890,9 +892,14 @@ def __init__(
890892
Creates a PushSource object.
891893
892894
Args:
893-
name: Name of the push source
894-
batch_source: The batch source that backs this push source. It's used when materializing from the offline
895-
store to the online store, and when retrieving historical features.
895+
name: Name of the push source.
896+
batch_source (optional): The batch source that backs this push source. It's used when
897+
materializing from the offline store to the online store, and when retrieving
898+
historical features.
899+
upstream_feature_view_names (optional): Sequence of upstream FeatureView/LabelView
900+
objects or names consumed by the external process feeding this push source.
901+
Represents external pipeline lineage metadata and does not trigger Feast-managed
902+
DAG execution.
896903
description (optional): A human-readable description.
897904
tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
898905
owner (optional): The owner of the data source, typically the email of the primary
@@ -901,6 +908,27 @@ def __init__(
901908
super().__init__(name=name, description=description, tags=tags, owner=owner)
902909
self.batch_source = batch_source
903910

911+
normalized_upstream_names: List[str] = []
912+
if upstream_feature_view_names is not None:
913+
for item in upstream_feature_view_names:
914+
if hasattr(item, "name"):
915+
name_val = item.name
916+
if not isinstance(name_val, str) or not name_val.strip():
917+
raise ValueError(
918+
f"Upstream feature view object {item} has an invalid or empty name"
919+
)
920+
normalized_upstream_names.append(name_val.strip())
921+
elif isinstance(item, str):
922+
if not item.strip():
923+
raise ValueError("Upstream feature view name cannot be empty")
924+
normalized_upstream_names.append(item.strip())
925+
else:
926+
raise TypeError(
927+
f"Unsupported type for upstream feature view: {type(item)}. "
928+
"Expected str or object with a 'name' attribute (e.g. FeatureView, LabelView)."
929+
)
930+
self.upstream_feature_view_names = normalized_upstream_names
931+
904932
def __eq__(self, other):
905933
if not isinstance(other, PushSource):
906934
return False
@@ -911,6 +939,9 @@ def __eq__(self, other):
911939
if self.batch_source != other.batch_source:
912940
return False
913941

942+
if self.upstream_feature_view_names != other.upstream_feature_view_names:
943+
return False
944+
914945
return True
915946

916947
def __hash__(self):
@@ -932,9 +963,16 @@ def from_proto(data_source: DataSourceProto):
932963
else None
933964
)
934965

966+
upstream_feature_view_names = (
967+
list(data_source.push_options.upstream_feature_view_names)
968+
if data_source.HasField("push_options")
969+
else []
970+
)
971+
935972
return PushSource(
936973
name=data_source.name,
937974
batch_source=batch_source,
975+
upstream_feature_view_names=upstream_feature_view_names,
938976
description=data_source.description,
939977
tags=dict(data_source.tags),
940978
owner=data_source.owner,
@@ -949,6 +987,10 @@ def _to_proto_impl(self) -> DataSourceProto:
949987
owner=self.owner,
950988
)
951989

990+
data_source_proto.push_options.upstream_feature_view_names.extend(
991+
self.upstream_feature_view_names
992+
)
993+
952994
# Only set timestamp fields if we have a batch source and this PushSource doesn't have its own fields
953995
if self.batch_source and not (
954996
self.timestamp_field or self.created_timestamp_column or self.field_mapping

‎sdk/python/feast/feature_store.py‎

Lines changed: 49 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1474,7 +1474,16 @@ def plan(
14741474
desired_repo_contents.stream_feature_views,
14751475
desired_repo_contents.label_views,
14761476
)
1477-
_validate_data_sources(desired_repo_contents.data_sources)
1477+
all_fvs_for_validation = [
1478+
*desired_repo_contents.feature_views,
1479+
*desired_repo_contents.on_demand_feature_views,
1480+
*desired_repo_contents.stream_feature_views,
1481+
*desired_repo_contents.label_views,
1482+
]
1483+
_validate_data_sources(
1484+
desired_repo_contents.data_sources,
1485+
all_feature_views=all_fvs_for_validation,
1486+
)
14781487
self._make_inferences(
14791488
desired_repo_contents.data_sources,
14801489
desired_repo_contents.entities,
@@ -1749,14 +1758,24 @@ def apply(
17491758
# Handle all entityless feature views by using DUMMY_ENTITY as a placeholder entity.
17501759
entities_to_update.append(DUMMY_ENTITY)
17511760

1752-
# Validate all feature views and make inferences.
1761+
# Validate all feature views and data sources, then make inferences.
17531762
if not skip_feature_view_validation:
17541763
self._validate_all_feature_views(
17551764
views_to_update,
17561765
odfvs_to_update,
17571766
sfvs_to_update,
17581767
lvs_to_update,
17591768
)
1769+
all_fvs_for_validation = [
1770+
*views_to_update,
1771+
*odfvs_to_update,
1772+
*sfvs_to_update,
1773+
*(lvs_to_update or []),
1774+
]
1775+
_validate_data_sources(
1776+
data_sources_to_update,
1777+
all_feature_views=all_fvs_for_validation,
1778+
)
17601779
self._make_inferences(
17611780
data_sources_to_update,
17621781
entities_to_update,
@@ -5156,12 +5175,38 @@ def _validate_feature_views(feature_views: List[BaseFeatureView]):
51565175
fv_by_name[case_insensitive_fv_name] = fv
51575176

51585177

5159-
def _validate_data_sources(data_sources: List[DataSource]):
5160-
"""Verify data sources have case-insensitively unique names."""
5178+
def _validate_data_sources(
5179+
data_sources: List[DataSource],
5180+
all_feature_views: Optional[
5181+
Sequence[
5182+
Union[
5183+
BaseFeatureView,
5184+
FeatureView,
5185+
StreamFeatureView,
5186+
OnDemandFeatureView,
5187+
LabelView,
5188+
]
5189+
]
5190+
] = None,
5191+
):
5192+
"""Verify data sources have case-insensitively unique names and validate PushSource upstream references."""
51615193
ds_names = set()
51625194
for ds in data_sources:
51635195
case_insensitive_ds_name = ds.name.lower()
51645196
if case_insensitive_ds_name in ds_names:
51655197
raise DataSourceRepeatNamesException(case_insensitive_ds_name)
51665198
else:
51675199
ds_names.add(case_insensitive_ds_name)
5200+
5201+
if all_feature_views is not None:
5202+
known_fv_names = {fv.name for fv in all_feature_views if hasattr(fv, "name")}
5203+
for ds in data_sources:
5204+
if isinstance(ds, PushSource) and ds.upstream_feature_view_names:
5205+
for upstream_name in ds.upstream_feature_view_names:
5206+
if upstream_name not in known_fv_names:
5207+
warnings.warn(
5208+
f"PushSource '{ds.name}' references upstream feature view '{upstream_name}' "
5209+
"which was not found in the current feature repository.",
5210+
UserWarning,
5211+
stacklevel=2,
5212+
)

‎sdk/python/feast/lineage/registry_lineage.py‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -228,6 +228,27 @@ def _parse_direct_relationships(self, registry: Registry) -> List[EntityRelation
228228
)
229229
relationships.append(rel)
230230

231+
# Stream source -> FeatureView relationships
232+
if (
233+
hasattr(feature_view.spec, "stream_source")
234+
and feature_view.spec.stream_source
235+
):
236+
stream_source_name = getattr(
237+
feature_view.spec.stream_source, "name", None
238+
)
239+
if stream_source_name:
240+
relationships.append(
241+
EntityRelation(
242+
source=EntityReference(
243+
FeastObjectType.DATA_SOURCE, stream_source_name
244+
),
245+
target=EntityReference(
246+
FeastObjectType.FEATURE_VIEW,
247+
feature_view.spec.name,
248+
),
249+
)
250+
)
251+
231252
# Batch source relationship
232253
if (
233254
hasattr(feature_view.spec, "batch_source")
@@ -455,6 +476,72 @@ def _parse_direct_relationships(self, registry: Registry) -> List[EntityRelation
455476
)
456477
)
457478

479+
# Upstream FeatureView -> DataSource (PushSource) relationships
480+
push_sources = {
481+
ds.name: ds
482+
for ds in registry.data_sources
483+
if hasattr(ds, "name")
484+
and ds.name
485+
and hasattr(ds, "push_options")
486+
and ds.push_options
487+
}
488+
for fv in registry.feature_views:
489+
if (
490+
hasattr(fv, "spec")
491+
and fv.spec
492+
and hasattr(fv.spec, "stream_source")
493+
and fv.spec.stream_source
494+
and hasattr(fv.spec.stream_source, "name")
495+
and fv.spec.stream_source.name
496+
and hasattr(fv.spec.stream_source, "push_options")
497+
and fv.spec.stream_source.push_options
498+
):
499+
push_sources[fv.spec.stream_source.name] = fv.spec.stream_source
500+
for sfv in registry.stream_feature_views:
501+
if (
502+
hasattr(sfv, "spec")
503+
and sfv.spec
504+
and hasattr(sfv.spec, "stream_source")
505+
and sfv.spec.stream_source
506+
and hasattr(sfv.spec.stream_source, "name")
507+
and sfv.spec.stream_source.name
508+
and hasattr(sfv.spec.stream_source, "push_options")
509+
and sfv.spec.stream_source.push_options
510+
):
511+
push_sources[sfv.spec.stream_source.name] = sfv.spec.stream_source
512+
for lv in registry.label_views:
513+
if (
514+
hasattr(lv, "spec")
515+
and lv.spec
516+
and hasattr(lv.spec, "source")
517+
and lv.spec.source
518+
and hasattr(lv.spec.source, "name")
519+
and lv.spec.source.name
520+
and hasattr(lv.spec.source, "push_options")
521+
and lv.spec.source.push_options
522+
):
523+
push_sources[lv.spec.source.name] = lv.spec.source
524+
525+
for ds in push_sources.values():
526+
if (
527+
not hasattr(ds, "push_options")
528+
or not hasattr(ds.push_options, "upstream_feature_view_names")
529+
or not ds.push_options.upstream_feature_view_names
530+
):
531+
continue
532+
for upstream_fv in ds.push_options.upstream_feature_view_names:
533+
source_type = (
534+
FeastObjectType.LABEL_VIEW
535+
if upstream_fv in label_view_names
536+
else FeastObjectType.FEATURE_VIEW
537+
)
538+
relationships.append(
539+
EntityRelation(
540+
source=EntityReference(source_type, upstream_fv),
541+
target=EntityReference(FeastObjectType.DATA_SOURCE, ds.name),
542+
)
543+
)
544+
458545
# SavedDataset relationships
459546
ds_location_index = _build_datasource_location_index(registry)
460547

0 commit comments

Comments
 (0)