get_online_features silently drops a join key value when it shares a name with an OnDemandFeatureView's RequestSource field #6790

for join_key_or_entity_name, values in entity_proto_values.items():
    # Found request data
    if join_key_or_entity_name in needed_request_data:
        request_data_features[join_key_or_entity_name] = values
    elif join_key_or_entity_name in join_keys_set:
        # It's a join key
        join_key = join_key_or_entity_name
        requested_result_row_names.add(join_key)
        join_key_values[join_key] = values
    elif join_key_or_entity_name in entity_name_to_join_key_map:
        ...
    else:
        continue

import os
import shutil
import tempfile
from datetime import datetime, timedelta, timezone

import pandas as pd
from feast import Entity, FeatureService, FeatureStore, FeatureView, Field, FileSource, RequestSource
from feast.on_demand_feature_view import on_demand_feature_view
from feast.types import Float32, String

tmp_dir = tempfile.mkdtemp()
repo_path = tmp_dir + "/repo"
os.makedirs(repo_path, exist_ok=True)

with open(repo_path + "/feature_store.yaml", "w") as f:
    f.write(
        "project: repro\n"
        "provider: local\n"
        "registry: registry.db\n"
        "online_store:\n"
        "  type: sqlite\n"
        "offline_store:\n"
        "  type: file\n"
        "entity_key_serialization_version: 3\n"
    )

data_path = repo_path + "/region_stats.parquet"
now = datetime.now(timezone.utc)
pd.DataFrame({
    "region": ["EU", "US"],
    "avg_price": [10.0, 20.0],
    "event_timestamp": [now - timedelta(days=1), now - timedelta(days=1)],
}).to_parquet(data_path)

region_source = FileSource(path=data_path, timestamp_field="event_timestamp")
region = Entity(name="region", join_keys=["region"])

region_stats_view = FeatureView(
    name="region_stats_view",
    entities=[region],
    schema=[Field(name="avg_price", dtype=Float32)],
    source=region_source,
    online=True,
)

# RequestSource field is deliberately named "region" too — same name as the join key above.
request_source = RequestSource(
    name="context",
    schema=[
        Field(name="region", dtype=String),
        Field(name="multiplier", dtype=Float32),
    ],
)


@on_demand_feature_view(
    sources=[request_source],
    schema=[Field(name="region_upper", dtype=String)],
)
def region_odfv(features_df: pd.DataFrame) -> pd.DataFrame:
    out = pd.DataFrame()
    out["region_upper"] = features_df["region"].str.upper() + "!" + features_df["multiplier"].astype(str)
    return out


service = FeatureService(
    name="demo_service",
    features=[region_stats_view[["avg_price"]], region_odfv],
)

store = FeatureStore(repo_path=repo_path)
store.apply([region, region_source, region_stats_view, region_odfv, service])

entity_rows = [{"region": "EU", "multiplier": 2.0}]

print("=== get_historical_features (works) ===")
entity_df = pd.DataFrame({"region": ["EU"], "multiplier": [2.0], "event_timestamp": [now]})
print(store.get_historical_features(entity_df=entity_df, features=service).to_df())

print("\n=== get_online_features (fails) ===")
print(store.get_online_features(features=service, entity_rows=entity_rows).to_df())

shutil.rmtree(tmp_dir, ignore_errors=True)

for join_key_or_entity_name, values in entity_proto_values.items():
    is_request_data = join_key_or_entity_name in needed_request_data
    is_join_key = join_key_or_entity_name in join_keys_set
    if is_request_data:
        request_data_features[join_key_or_entity_name] = values
    if is_join_key:
        requested_result_row_names.add(join_key_or_entity_name)
        join_key_values[join_key_or_entity_name] = values
    elif not is_request_data:
        if join_key_or_entity_name in entity_name_to_join_key_map:
            join_key = entity_name_to_join_key_map[join_key_or_entity_name]
            warnings.warn("Using entity name is deprecated. Use join_key instead.")
            requested_result_row_names.add(join_key)
            join_key_values[join_key] = values
        else:
            continue