feast-dev · feast-ci-bot · May 23, 2022 · May 14, 2022 · May 14, 2022 · May 18, 2022
diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml
@@ -55,6 +55,7 @@ jobs:
         run: make install-python-ci-dependencies
       - name: Test Python
         env:
+          IS_TEST: "True"
           SNOWFLAKE_CI_DEPLOYMENT: ${{ secrets.SNOWFLAKE_CI_DEPLOYMENT }}
           SNOWFLAKE_CI_USER: ${{ secrets.SNOWFLAKE_CI_USER }}
           SNOWFLAKE_CI_PASSWORD: ${{ secrets.SNOWFLAKE_CI_PASSWORD }}

@@ -24,7 +24,7 @@
 from feast.field import Field
 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
 from feast.repo_config import RepoConfig, get_data_source_class_from_type
-from feast.types import VALUE_TYPES_TO_FEAST_TYPES
+from feast.types import from_value_type
 from feast.value_type import ValueType
 
 
@@ -557,12 +557,10 @@ def __init__(
                 "Please use List[Field] instead for the schema",
                 DeprecationWarning,
             )
-            schemaList = []
-            for key, valueType in _schema.items():
-                schemaList.append(
-                    Field(name=key, dtype=VALUE_TYPES_TO_FEAST_TYPES[valueType])
-                )
-            self.schema = schemaList
+            schema_list = []
+            for key, value_type in _schema.items():
+                schema_list.append(Field(name=key, dtype=from_value_type(value_type)))
+            self.schema = schema_list
         elif isinstance(_schema, List):
             self.schema = _schema
         else:
@@ -641,9 +639,7 @@ def to_proto(self) -> DataSourceProto:
         if isinstance(self.schema, Dict):
             for key, value in self.schema.items():
                 schema_pb.append(
-                    Field(
-                        name=key, dtype=VALUE_TYPES_TO_FEAST_TYPES[value.value]
-                    ).to_proto()
+                    Field(name=key, dtype=from_value_type(value.value)).to_proto()
                 )
         else:
             for field in self.schema:

@@ -674,5 +674,5 @@ def feature_view_to_batch_feature_view(fv: FeatureView) -> BatchFeatureView:
         online=fv.online,
         owner=fv.owner,
         schema=fv.schema,
-        source=fv.source,
+        source=fv.batch_source,
     )
@@ -1,6 +1,10 @@
 from datetime import timedelta
 
+import pandas as pd
+
 from feast import Entity, Feature, FeatureView, FileSource, ValueType
+from feast.data_source import RequestDataSource
+from feast.on_demand_feature_view import on_demand_feature_view
 
 driver_hourly_stats = FileSource(
     path="%PARQUET_PATH%",  # placeholder to be replaced by the test
@@ -50,3 +54,27 @@
     batch_source=global_daily_stats,  # Changed to `source` in 0.20
     tags={},
 )
+
+
+request_source = RequestDataSource(
+    name="conv_rate_input", schema={"val_to_add": ValueType.INT64},
+)
+
+
+@on_demand_feature_view(
+    inputs={
+        "conv_rate_input": request_source,
+        "driver_hourly_stats": driver_hourly_stats_view,
+    },
+    features=[
+        Feature(name="conv_rate_plus_100", dtype=ValueType.DOUBLE),
+        Feature(name="conv_rate_plus_val_to_add", dtype=ValueType.DOUBLE),
+    ],
+)
+def conv_rate_plus_100(features_df: pd.DataFrame) -> pd.DataFrame:
+    df = pd.DataFrame()
+    df["conv_rate_plus_100"] = features_df["conv_rate"] + 100
+    df["conv_rate_plus_val_to_add"] = (
+        features_df["conv_rate"] + features_df["val_to_add"]
+    )
+    return df
@@ -41,7 +41,7 @@ def _assert_online_features(
         full_feature_names=True,
     )
 
-    # Float features should still be floats from the online store...
+    # Float features should still be floats.
     assert (
         response.proto.results[
             list(response.proto.metadata.feature_names.val).index(
@@ -67,6 +67,49 @@ def _assert_online_features(
     assert "global_daily_stats__num_rides" in result
     assert "global_daily_stats__avg_ride_length" in result
 
+    # Test the ODFV if it exists.
+    odfvs = store.list_on_demand_feature_views()
+    if odfvs and odfvs[0].name == "conv_rate_plus_100":
+        response = store.get_online_features(
+            features=[
+                "conv_rate_plus_100:conv_rate_plus_100",
+                "conv_rate_plus_100:conv_rate_plus_val_to_add",
+            ],
+            entity_rows=[{"driver_id": 1001, "val_to_add": 100}],
+            full_feature_names=True,
+        )
+
+        # Check that float64 feature is stored correctly in proto format.
+        assert (
+            response.proto.results[
+                list(response.proto.metadata.feature_names.val).index(
+                    "conv_rate_plus_100__conv_rate_plus_100"
+                )
+            ]
+            .values[0]
+            .double_val
+            > 0
+        )
+
+        result = response.to_dict()
+        assert len(result) == 3
+        assert "conv_rate_plus_100__conv_rate_plus_100" in result
+        assert "conv_rate_plus_100__conv_rate_plus_val_to_add" in result
+        assert (
+            abs(
+                result["conv_rate_plus_100__conv_rate_plus_100"][0]
+                - (_get_last_feature_row(driver_df, 1001, max_date)["conv_rate"] + 100)
+            )
+            < 0.01
+        )
+        assert (
+            abs(
+                result["conv_rate_plus_100__conv_rate_plus_val_to_add"][0]
+                - (_get_last_feature_row(driver_df, 1001, max_date)["conv_rate"] + 100)
+            )
+            < 0.01
+        )
+
 
 def _test_materialize_and_online_retrieval(
     runner: CliRunner,