feat: Add upstream FeatureView lineage support for PushSource

[mao-liu]

What this PR does / why we need it:

In production ML architectures, derived features are often pre-computed by external stream or batch computation engines (e.g., Spark, Flink) that read from existing upstream Feast FeatureViews and push the transformed features into Feast via a PushSource.

Previously, Feast's static registry lineage only tracked:

PushSource --> Target FeatureView

Because PushSource had no way to declare upstream FeatureView dependencies, the connection between the upstream feature views and the push pipeline was lost in both registry lineage metadata and the Feast Web UI graph.

This PR adds first-class support for declaring upstream FeatureView dependencies on PushSource.

Summary of Changes:

1. Protobuf (protos/feast/core/DataSource.proto):
   • Added repeated string upstream_feature_views = 2; to PushOptions.
   • Recompiled generated Python protobuf definitions.
   • See note below on protobuf regeneration
2. Python SDK (sdk/python/feast/data_source.py):
   • Updated PushSource constructor to accept source_views: Optional[Sequence[Union[BaseFeatureView, str]]].
   • Updated to_proto(), from_proto(), and __eq__() serialization and comparison logic.
3. Registry Lineage Engine (sdk/python/feast/lineage/registry_lineage.py):
   • Emits EntityRelation edges for upstream FeatureView / LabelView -> DataSource (PushSource).
   • Ensures streamSource -> FeatureView direct relationships are consistently registered for stream/push sources.
   • See note below on stream source rendering
4. UI Lineage Parser & Graph (ui/):
   • Updated parseEntityRelationships.ts to parse upstreamFeatureViews on dataSources and draw visual edges.
   • Updated RegistryVisualization.tsx to include stream sources and standalone data sources in the visual DAG.
5. OpenLineage event emission is deliberately not implemented
   • Granular relationships between data sources and feature views are not currently rendered in OpenLineage
   • See also: Granular OpenLineage event emission on feast apply for precise FeatureView lineage #6852

Protobuf regeneration notes

To avoid mass regeneration of proto files leading to a huge diff, the protos were
generated by pinning generator tools to a version matching the older tools
used when generating the existing files.

uv run \
  --with "grpcio-tools==1.62.2" \
  --with "protobuf==4.25.1" \
  --with "mypy-protobuf==3.3.0" \
  python infra/scripts/generate_protos.py

A prior commit on master (28bde01 which introduced ConnectionRef) updated DataSource.proto but did not commit the generated DataSource_pb2.py / .pyi bindings. Compiling DataSource.proto now brought in both upstream_feature_views and ConnectionRef.

Rendering of stream sources in Feast Lineage

Previously, there was an inconsistency between FeatureView and StreamFeatureView:

• StreamFeatureView: Always drew edges for both its streamSource and its batchSource:
  • streamSource -> StreamFeatureView
  • batchSource -> StreamFeatureView
• Standard FeatureView: Only inspected fv.spec.batchSource:
  • If a standard FeatureView used a PushSource, KafkaSource, or KinesisSource, Feast populated both stream_source and batch_source under the hood.
  • However, the lineage generator (both in Python registry_lineage.py and in UI parseEntityRelationships.ts) completely ignored fv.spec.streamSource.
  • Consequently, the stream/push source node was omitted, and only the batch source was connected.

Now, whenever a standard FeatureView has a streamSource defined (PushSource, KafkaSource, KinesisSource, etc.), the relationship streamSource -> FeatureView is always drawn. Specific unit tests in both Python (test_registry_lineage.py) and UI (RegistryVisualization.test.tsx) assert this behavior across PushSource, KafkaSource, and KinesisSource.

Which issue(s) this PR fixes:

Fixes #6839

Checks

• I've made sure the tests are passing.
• My commits are signed off (git commit -s)
• My PR title follows conventional commits format

Testing Strategy

• Unit tests
• Integration tests
• Manual tests
• Testing is not required for this change

Misc

Manual PoC Testing:

• Tested end-to-end with a sample feature definition containing upstream feature views, a push source with source_views, and a downstream feature view.

Before:

After:

PoC testing configuration

feature-store.yaml

project: push_source_poc
registry: data/registry.db
provider: local
offline_store:
  type: file
online_store:
  type: sqlite
  path: data/online_store.db
entity_key_serialization_version: 3

definitions.py

from datetime import timedelta
import pandas as pd

from feast import (
    BatchFeatureView,
    Entity,
    FeatureService,
    FeatureView,
    Field,
    FileSource,
    KafkaSource,
    KinesisSource,
    PushSource,
    StreamFeatureView,
)
from feast.data_format import JsonFormat
from feast.on_demand_feature_view import on_demand_feature_view
from feast.types import Float32, Int64
from feast.value_type import ValueType

# ---------------------------------------------------------------------------
# 1. Entity
# ---------------------------------------------------------------------------
user = Entity(
    name="user_id",
    join_keys=["user_id"],
    value_type=ValueType.INT64,
    description="Unique identifier for users",
)

# ---------------------------------------------------------------------------
# 2. Batch Data Sources (FileSource)
# ---------------------------------------------------------------------------
user_tx_batch_source = FileSource(
    name="user_tx_batch_source",
    path="data/transactions.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

user_credit_batch_source = FileSource(
    name="user_credit_batch_source",
    path="data/credit_profile.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

user_raw_push_batch_source = FileSource(
    name="user_raw_push_batch_source",
    path="data/raw_push.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

risk_calc_batch_source = FileSource(
    name="risk_calc_batch_source",
    path="data/risk_offline.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

# ---------------------------------------------------------------------------
# 3. Upstream Feature Views
# ---------------------------------------------------------------------------
user_transaction_stats = FeatureView(
    name="user_transaction_stats",
    entities=[user],
    ttl=timedelta(days=30),
    schema=[Field(name="tx_amount", dtype=Float32)],
    source=user_tx_batch_source,
    description="User transaction statistics",
)

user_credit_profile = FeatureView(
    name="user_credit_profile",
    entities=[user],
    ttl=timedelta(days=30),
    schema=[Field(name="credit_score", dtype=Int64)],
    source=user_credit_batch_source,
    description="User credit profile history",
)

# ---------------------------------------------------------------------------
# 4. Push & Stream Data Sources
# ---------------------------------------------------------------------------
# PushSource with upstream FeatureView dependencies
risk_calc_pipeline = PushSource(
    name="risk_calc_pipeline",
    batch_source=risk_calc_batch_source,
    source_views=[user_transaction_stats, user_credit_profile],
    description="Spark/Flink streaming pipeline reading tx stats and credit profile",
)

# PushSource without upstream FeatureView dependencies
raw_push_source = PushSource(
    name="raw_push_source",
    batch_source=user_raw_push_batch_source,
    description="Direct push source without upstream views",
)

# ---------------------------------------------------------------------------
# 5. Downstream & Target Feature Views
# ---------------------------------------------------------------------------
# Target FeatureView fed by PushSource with upstream lineage
user_risk_target_fv = FeatureView(
    name="user_risk_target_fv",
    entities=[user],
    ttl=timedelta(days=30),
    schema=[
        Field(name="composite_risk_score", dtype=Float32),
        Field(name="risk_tier", dtype=Int64),
    ],
    source=risk_calc_pipeline,
    description="User risk scores computed by external pipeline and pushed",
)

# FeatureView fed by PushSource without upstream dependencies
user_raw_push_fv = FeatureView(
    name="user_raw_push_fv",
    entities=[user],
    ttl=timedelta(days=30),
    schema=[Field(name="push_val", dtype=Float32)],
    source=raw_push_source,
    description="Raw push feature view",
)

[codecov-commenter]

⚠️ Please install the to ensure uploads and comments are reliably processed by Codecov.

Codecov Report

❌ Patch coverage is 77.77778% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 47.46%. Comparing base (fa8f06b) to head (e3566a3).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
sdk/python/feast/lineage/registry_lineage.py	70.00%	3 Missing and 3 partials ⚠️
❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #6853      +/-   ##
==========================================
+ Coverage   47.43%   47.46%   +0.02%     
==========================================
  Files         422      422              
  Lines       52263    52289      +26     
  Branches     7582     7594      +12     
==========================================
+ Hits        24791    24817      +26     
+ Misses      25708    25707       -1     
- Partials     1764     1765       +1     

Flag	Coverage Δ
go-feature-server	30.58% <ø> (ø)
python-unit	48.78% <77.77%> (+0.02%)	⬆️

Files with missing lines	Coverage Δ
sdk/python/feast/data_source.py	81.46% <100.00%> (+0.36%)	⬆️
sdk/python/feast/lineage/registry_lineage.py	58.73% <70.00%> (+3.31%)	⬆️

---

Continue to review full report in Codecov by Harness.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update fa8f06b...e3566a3. Read the comment docs.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[haoxu0]

Could you clarify why this needs a second source_views model on PushSource instead of reusing the existing derived-FeatureView relationship? FeatureView already accepts one or more FeatureViews as source, persists them in FeatureViewSpec.source_views, and uses sink_source for the output location.

I understand that the intended distinction may be that this PR describes an externally managed Spark/Flink pipeline and therefore treats the dependency as lineage-only metadata. However, the current API leaves two ways to express essentially the same upstream relationship:

• FeatureView.source_views: Feast-managed derived-feature semantics, serialized as FeatureView specs
• PushSource.source_views: external lineage metadata, serialized as names

Those representations can disagree, and the shared source_views name does not communicate their different semantics. Could we either reuse the existing relationship, or explicitly document why that is not suitable and give the new field a metadata-specific name such as upstream_feature_view_names or lineage_inputs? We should also define validation/precedence if both representations are present.