feat: Adding mode='python' for get_historical_features on ODFVs (#4653)

sdk/python/feast/transformation/python_transformation.py

@@ -26,11 +26,11 @@ def __init__(self, udf: FunctionType, udf_string: str = ""):
         self.udf_string = udf_string
 
     def transform_arrow(
-        self, pa_table: pyarrow.Table, features: list[Field]
+        self,
+        pa_table: pyarrow.Table,
+        features: list[Field],
     ) -> pyarrow.Table:
-        raise Exception(
-            'OnDemandFeatureView with mode "python" does not support offline processing.'
-        )
+        return pyarrow.Table.from_pydict(self.udf(pa_table.to_pydict()))
 
     def transform(self, input_dict: dict) -> dict:
         # Ensuring that the inputs are included as well

sdk/python/tests/unit/test_on_demand_python_transformation.py

@@ -561,6 +561,24 @@ def pandas_view(features_df: pd.DataFrame) -> pd.DataFrame:
             assert resp is not None
             assert resp["conv_rate_plus_val1"].isnull().sum() == 0
 
+            batch_sample["avg_daily_trip_rank_thresholds"] = [
+                [100, 250, 500, 1000]
+            ] * batch_sample.shape[0]
+            batch_sample["avg_daily_trip_rank_names"] = [
+                ["Bronze", "Silver", "Gold", "Platinum"]
+            ] * batch_sample.shape[0]
+            resp_python = self.store.get_historical_features(
+                entity_df=batch_sample,
+                features=[
+                    "driver_hourly_stats:conv_rate",
+                    "driver_hourly_stats:acc_rate",
+                    "driver_hourly_stats:avg_daily_trips",
+                    "python_view:conv_rate_plus_acc",
+                ],
+            ).to_df()
+            assert resp_python is not None
+            assert resp_python["conv_rate_plus_acc"].isnull().sum() == 0
+
             # Now testing feature retrieval for driver ids not in the dataset
             missing_batch_sample = pd.DataFrame([1234567890], columns=["driver_id"])
             missing_batch_sample["val_to_add"] = 0