deduplicate_keep_latest fills null features of the latest row with values from older rows #7049

Expected Behavior

deduplicate_keep_latest (in feast/aggregation/tiling/tile_subtraction.py) should return the latest row for each entity as-is. If a feature is null in the latest row, it should still be null in the output.

Current Behavior

The function sorts by timestamp and then calls groupby(entity_keys).first(). GroupBy.first() returns the first non-null value per column, not the first row. When the latest row has a null feature, the value comes from an older row, so the output row mixes the latest timestamp with stale feature values.

This affects tiled aggregations materialized by the Spark and Ray compute engines, which both call this function before writing to the online store. For example, std/var are NaN when the latest window has fewer than two events (compute_holistic_value_from_irs does this on purpose to match the non-tiled result). Instead of NaN, the online store gets the std of an older window.

Steps to reproduce

import pandas as pd
from feast.aggregation.tiling import deduplicate_keep_latest

df = pd.DataFrame(
    {
        "driver_id": [1, 1],
        "event_timestamp": pd.to_datetime(["2024-01-01 00:00", "2024-01-01 00:05"]),
        # std over a window holding a single event is NaN
        "std_trips_1200s": [2.5, float("nan")],
    }
)
print(deduplicate_keep_latest(df, ["driver_id"], "event_timestamp"))
#    driver_id     event_timestamp  std_trips_1200s
# 0          1 2024-01-01 00:05:00              2.5

Expected std_trips_1200s to be NaN for the 00:05 row, but got 2.5, which comes from the 00:00 row.

Specifications

• Version: master (e304a84)
• Platform: macOS arm64, Python 3.11, pandas 2.3.3
• Subsystem: aggregation tiling (Spark / Ray compute engines)

Possible Solution

Keep the whole latest row per entity with drop_duplicates(subset=entity_keys, keep="first") after sorting, instead of groupby().first().