fix: Dask pulling of latest data (#5229)

patilsuraj767 · web-flow · commit 571d81f30578 · 2025-05-05T08:39:27.000-04:00
Signed-off-by: Suraj Patil &lt;patilsuraj767@gmail.com&gt;
diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py
@@ -365,13 +365,13 @@ def evaluate_offline_job():
                 if start_date and end_date:
                     source_df = source_df[
                         source_df[timestamp_field].between(
-                            start_date, end_date, inclusive="left"
+                            start_date, end_date, inclusive="both"
                         )
                     ]
                 elif start_date:
                     source_df = source_df[source_df[timestamp_field] >= start_date]
                 elif end_date:
-                    source_df = source_df[source_df[timestamp_field] < end_date]
+                    source_df = source_df[source_df[timestamp_field] <= end_date]
 
             source_df = source_df.persist()
 
diff --git a/sdk/python/tests/utils/feature_records.py b/sdk/python/tests/utils/feature_records.py
@@ -519,7 +519,7 @@ def get_last_feature_row(df: pd.DataFrame, driver_id, max_date: datetime):
     """Manually extract last feature value from a dataframe for a given driver_id with up to `max_date` date"""
     filtered = df[
         (df["driver_id"] == driver_id)
-        & (df["event_timestamp"] < max_date.replace(tzinfo=timezone.utc))
+        & (df["event_timestamp"] <= max_date.replace(tzinfo=timezone.utc))
     ]
     max_ts = filtered.loc[filtered["event_timestamp"].idxmax()]["event_timestamp"]
     filtered_by_ts = filtered[filtered["event_timestamp"] == max_ts]

Original file line number	Diff line number	Diff line change
`@@ -365,13 +365,13 @@ def evaluate_offline_job():`
`365`	`365`	`if start_date and end_date:`
`366`	`366`	`source_df = source_df[`
`367`	`367`	`source_df[timestamp_field].between(`
`368`		`- start_date, end_date, inclusive="left"`
	`368`	`+ start_date, end_date, inclusive="both"`
`369`	`369`	`)`
`370`	`370`	`]`
`371`	`371`	`elif start_date:`
`372`	`372`	`source_df = source_df[source_df[timestamp_field] >= start_date]`
`373`	`373`	`elif end_date:`
`374`		`- source_df = source_df[source_df[timestamp_field] < end_date]`
	`374`	`+ source_df = source_df[source_df[timestamp_field] <= end_date]`
`375`	`375`
`376`	`376`	`source_df = source_df.persist()`
`377`	`377`
Original file line number	Diff line number	Diff line change
`@@ -519,7 +519,7 @@ def get_last_feature_row(df: pd.DataFrame, driver_id, max_date: datetime):`
`519`	`519`	"""Manually extract last feature value from a dataframe for a given driver_id with up to `max_date` date"""
`520`	`520`	`filtered = df[`
`521`	`521`	`(df["driver_id"] == driver_id)`
`522`		`- & (df["event_timestamp"] < max_date.replace(tzinfo=timezone.utc))`
	`522`	`+ & (df["event_timestamp"] <= max_date.replace(tzinfo=timezone.utc))`
`523`	`523`	`]`
`524`	`524`	`max_ts = filtered.loc[filtered["event_timestamp"].idxmax()]["event_timestamp"]`
`525`	`525`	`filtered_by_ts = filtered[filtered["event_timestamp"] == max_ts]`