Repository navigation
get_historical_features on DuckDB ignores join keys that with_join_key_map does not rename #7158
Description
Expected Behavior
with_join_key_map can rename some of a feature view's join keys and leave the others as they are. get_historical_features should then join on every key: the renamed ones under their new name, the others under their own. The Dask/file offline store and get_online_features do this, and the SQL offline stores build their join keys the same way (offline_utils.get_feature_view_query_context).
Current Behavior
With the DuckDB offline store, the join keys that the map does not rename are dropped from the point-in-time join. Each entity row gets the latest value for the renamed key alone, taken from any value of the other keys, and there is no error.
In the example below, user 1 spent 10 at merchant 100 and 99 at merchant 200. get_historical_features returns 99 for both merchants, while get_online_features on the same store returns 10 and 99, so the training data and the served features disagree.
Steps to reproduce
import tempfile
from pathlib import Path
import pandas as pd
from feast import Entity, FeatureService, FeatureStore, FeatureView, Field, FileSource, RepoConfig
from feast.types import Int64
from feast.value_type import ValueType
repo = Path(tempfile.mkdtemp())
pd.DataFrame(
{
"user_id": [1, 1],
"merchant_id": [100, 200],
"event_timestamp": pd.to_datetime(["2024-01-01", "2024-01-02"], utc=True),
"spend": [10, 99],
}
).to_parquet(repo / "spend.parquet")
store = FeatureStore(
config=RepoConfig(
project="repro",
provider="local",
registry=str(repo / "registry.db"),
offline_store={"type": "duckdb"},
online_store={"type": "sqlite", "path": str(repo / "online.db")},
entity_key_serialization_version=3,
repo_path=str(repo),
)
)
user = Entity(name="user", join_keys=["user_id"], value_type=ValueType.INT64)
merchant = Entity(name="merchant", join_keys=["merchant_id"], value_type=ValueType.INT64)
spend = FeatureView(
name="user_merchant_spend",
entities=[user, merchant],
schema=[
Field(name="user_id", dtype=Int64),
Field(name="merchant_id", dtype=Int64),
Field(name="spend", dtype=Int64),
],
source=FileSource(path=str(repo / "spend.parquet"), timestamp_field="event_timestamp"),
)
# The entity data calls the user "sender_id"; merchant_id keeps its name.
service = FeatureService(
name="spend_by_sender",
features=[spend.with_join_key_map({"user_id": "sender_id"})],
)
store.apply([user, merchant, spend, service])
entity_df = pd.DataFrame(
{
"sender_id": [1, 1],
"merchant_id": [100, 200],
"event_timestamp": pd.to_datetime(["2024-01-03", "2024-01-03"], utc=True),
}
)
offline = store.get_historical_features(entity_df=entity_df, features=service).to_df()
print("# get_historical_features")
print(offline[["sender_id", "merchant_id", "spend"]].sort_values("merchant_id").to_string(index=False))
store.materialize(
start_date=pd.Timestamp("2024-01-01", tz="UTC"),
end_date=pd.Timestamp("2024-01-03", tz="UTC"),
)
online = store.get_online_features(
features=service,
entity_rows=[{"sender_id": 1, "merchant_id": 100}, {"sender_id": 1, "merchant_id": 200}],
).to_df()
print("\n# get_online_features")
print(online[["sender_id", "merchant_id", "spend"]].to_string(index=False))Output on master (cc17aad):
# get_historical_features
sender_id merchant_id spend
1 100 99
1 200 99
# get_online_features
sender_id merchant_id spend
1 100 10
1 200 99
0.68.0 from PyPI gives the same get_historical_features result. (On 0.68.0 the materialize step then fails with ConversionException: Unimplemented type for cast (TIMESTAMPTZ_NS -> TIMESTAMP), which does not happen on master.)
Specifications
- Version: master (cc17aad) and 0.68.0
- Platform: macOS, Python 3.12, DuckDB 2.0.0, ibis 12.0.0
- Subsystem: offline store, ibis point-in-time join (the DuckDB store; the contrib Oracle and MSSQL stores use the same code)
Possible Solution
get_historical_features_ibis passes feature_view.projection.join_key_map or {e.name: e.name for e in feature_view.entity_columns} to point_in_time_join, and _generate_row_id uses the same or. A non-empty partial map therefore replaces the whole list of join keys, both in the join predicates and in entity_row_id. Building the map from the entity columns with join_key_map.get(name, name), as the Dask store and offline_utils.get_feature_view_query_context do, fixes it. I have a PR ready with a regression test.
Activity