fix: Fix ODFV bug

Signed-off-by: Danny Chiao <danny@tecton.ai>
feast-dev · adchia · Aug 11, 2022 · Aug 11, 2022 · Aug 11, 2022 · Aug 11, 2022
commit a02b99776c3d0a0679dee6adab1e7ab6dca59869
@@ -136,4 +136,6 @@ Unit &amp; Integration Tests can be used to verify functionality:
 mvn test -pl serving --also-make
 # run integration tests
 mvn verify -pl serving --also-make
+# run integration tests with debugger
+mvn -Dmaven.failsafe.debug verify -pl serving --also-make
 ```
@@ -82,6 +82,27 @@
         </configuration>
       </plugin>
 
+      <plugin>
+        <groupId>org.codehaus.mojo</groupId>
+        <artifactId>exec-maven-plugin</artifactId>
+        <version>1.6.0</version>
+        <executions>
+          <execution>
+            <configuration>
+              <executable>python</executable>
+              <workingDirectory>src/test/resources/docker-compose/feast10/</workingDirectory>
+              <arguments>
+                <argument>setup_it.py</argument>
+              </arguments>
+            </configuration>
+            <id>feast_test_apply</id>
+            <phase>process-test-resources</phase>
+            <goals>
+              <goal>exec</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
 

@@ -75,17 +75,17 @@ def transformed_conv_rate(features_df: pd.DataFrame) -> pd.DataFrame:
 
 entity = Entity(name="entity")
 
-benchmark_feature_views = [
-    FeatureView(
+benchmark_feature_views = []
+for i in range(25):
+    fv = FeatureView(
         name=f"feature_view_{i}",
         entities=[entity],
         ttl=timedelta(seconds=86400),
         schema=[Field(name=f"feature_{10 * i + j}", dtype=Int64) for j in range(10)],
         online=True,
         source=generated_data_source,
     )
-    for i in range(25)
-]
+    benchmark_feature_views.append(fv)
 
 benchmark_feature_service = FeatureService(
     name=f"benchmark_feature_service", features=benchmark_feature_views,

@@ -0,0 +1,84 @@
+from feast.repo_config import load_repo_config
+from datetime import datetime, timedelta
+
+import numpy as np
+import pandas as pd
+
+from definitions import (
+    benchmark_feature_service,
+    benchmark_feature_views,
+    driver,
+    driver_hourly_stats_view,
+    entity,
+    transformed_conv_rate,
+)
+
+from feast import FeatureStore
+
+
+def setup_data():
+    start = datetime.now() - timedelta(days=10)
+
+    df = pd.DataFrame()
+    df["driver_id"] = np.arange(1000, 1010)
+    df["created"] = datetime.now()
+    df["conv_rate"] = np.arange(0, 1, 0.1)
+    df["acc_rate"] = np.arange(0.5, 1, 0.05)
+    df["avg_daily_trips"] = np.arange(0, 1000, 100)
+
+    # some of rows are beyond 7 days to test OUTSIDE_MAX_AGE status
+    df["event_timestamp"] = start + pd.Series(np.arange(0, 10)).map(
+        lambda days: timedelta(days=days)
+    )
+
+    # Store data in parquet files. Parquet is convenient for local development mode. For
+    # production, you can use your favorite DWH, such as BigQuery. See Feast documentation
+    # for more info.
+    df.to_parquet("driver_stats.parquet")
+
+    # For Benchmarks
+    # Please read more in Feast RFC-031
+    # (link https://docs.google.com/document/d/12UuvTQnTTCJhdRgy6h10zSbInNGSyEJkIxpOcgOen1I/edit)
+    # about this benchmark setup
+    def generate_data(
+        num_rows: int, num_features: int, destination: str
+    ) -> pd.DataFrame:
+        features = [f"feature_{i}" for i in range(num_features)]
+        columns = ["entity", "event_timestamp"] + features
+        df = pd.DataFrame(0, index=np.arange(num_rows), columns=columns)
+        df["event_timestamp"] = datetime.utcnow()
+        for column in features:
+            df[column] = np.random.randint(1, num_rows, num_rows)
+
+        df["entity"] = "key-" + pd.Series(np.arange(1, num_rows + 1)).astype(
+            pd.StringDtype()
+        )
+
+        df.to_parquet(destination)
+
+    generate_data(10**3, 250, "benchmark_data.parquet")
+
+
+def main():
+    print("Running setup_it.py")
+
+    setup_data()
+    existing_repo_config = load_repo_config(Path("."))
+
+    fs = FeatureStore(config=existing_repo_config.copy(update={"online_store": {}}))
+    fs.apply(
+        [
+            driver_hourly_stats_view,
+            transformed_conv_rate,
+            driver,
+            entity,
+            benchmark_feature_service,
+            *benchmark_feature_views,
+        ]
+    )
+
+    print("setup_it finished")
+
+
+if __name__ == "__main__":
+    main()
@@ -144,8 +144,8 @@ def diff_registry_objects(
                 continue
             elif getattr(current_spec, _field.name) != getattr(new_spec, _field.name):
                 if _field.name == "user_defined_function":
-                    current_spec = cast(OnDemandFeatureViewSpec, current_proto)
-                    new_spec = cast(OnDemandFeatureViewSpec, new_proto)
+                    current_spec = cast(OnDemandFeatureViewSpec, current_proto.spec)
+                    new_spec = cast(OnDemandFeatureViewSpec, new_proto.spec)
                     current_udf = current_spec.user_defined_function
                     new_udf = new_spec.user_defined_function
                     for _udf_field in current_udf.DESCRIPTOR.fields:

diff --git a/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py b/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py
@@ -11,6 +11,7 @@
 from feast.feature_view import FeatureView
 from feast.field import Field
 from feast.infra.online_stores.sqlite import SqliteOnlineStoreConfig
+from feast.on_demand_feature_view import on_demand_feature_view
 from feast.repo_config import RepoConfig
 from feast.types import Array, Bytes, Int64, String
 from tests.utils.data_source_test_creator import prep_file_source
@@ -97,6 +98,63 @@ def test_apply_feature_view_success(test_feature_store):
     test_feature_store.teardown()
 
 
+@pytest.mark.parametrize(
+    "test_feature_store",
+    [lazy_fixture("feature_store_with_local_registry")],
+)
+def test_apply_on_demand_feature_view_success(test_feature_store):
+    # Create Feature Views
+    batch_source = FileSource(
+        file_format=ParquetFormat(),
+        path="file://feast/*",
+        timestamp_field="ts_col",
+        created_timestamp_column="timestamp",
+    )
+
+    entity = Entity(name="fs1_my_entity_1", join_keys=["entity_id"])
+
+    fv1 = FeatureView(
+        name="my_feature_view_1",
+        schema=[
+            Field(name="fs1_my_feature_1", dtype=Int64),
+            Field(name="fs1_my_feature_2", dtype=String),
+            Field(name="fs1_my_feature_3", dtype=Array(String)),
+            Field(name="fs1_my_feature_4", dtype=Array(Bytes)),
+            Field(name="entity_id", dtype=Int64),
+        ],
+        entities=[entity],
+        tags={"team": "matchmaking"},
+        source=batch_source,
+        ttl=timedelta(minutes=5),
+    )
+
+    @on_demand_feature_view(
+
+    )
+
+    # Register Feature View
+    test_feature_store.apply([entity, fv1])
+
+    feature_views = test_feature_store.list_feature_views()
+
+    # List Feature Views
+    assert (
+            len(feature_views) == 1
+            and feature_views[0].name == "my_feature_view_1"
+            and feature_views[0].features[0].name == "fs1_my_feature_1"
+            and feature_views[0].features[0].dtype == Int64
+            and feature_views[0].features[1].name == "fs1_my_feature_2"
+            and feature_views[0].features[1].dtype == String
+            and feature_views[0].features[2].name == "fs1_my_feature_3"
+            and feature_views[0].features[2].dtype == Array(String)
+            and feature_views[0].features[3].name == "fs1_my_feature_4"
+            and feature_views[0].features[3].dtype == Array(Bytes)
+            and feature_views[0].entities[0] == "fs1_my_entity_1"
+    )
+
+    test_feature_store.teardown()
+
+
 @pytest.mark.parametrize(
     "test_feature_store",
     [lazy_fixture("feature_store_with_local_registry")],