Repository: Adyen/feast-spark-offline-store (language: python)

Issue:
point-in-time join breaks when joining on multiple feature views
When joining the `entity_df` on multiple feature views, the templated point-in-time join query renders in a way that creates all temporary `*__base` tables in one big go.

This makes the SQL parser raise an exception
```
pyspark.sql.utils.ParseException: 
extraneous input 'CREATE' expecting {<EOF>, ';'}(line 89, pos 0)
```


Fix is to move the `---EOS---` inside the Jinja loop that creates `*__base` tables. Will submit a PR.



Discussion / hints:

Fix patch:
diff --git a/example_feature_repo/example.py b/example_feature_repo/example.py
index d0bb2ac..302da37 100644
--- a/example_feature_repo/example.py
+++ b/example_feature_repo/example.py
@@ -6,7 +6,10 @@
 from feast_spark_offline_store import SparkSource
 from pyspark.sql import SparkSession
 from datetime import datetime, timedelta
-from feast.driver_test_data import create_driver_hourly_stats_df
+from feast.driver_test_data import create_driver_hourly_stats_df, create_customer_daily_profile_df
+
+# # we are loading a sparksession here, but should be configurable in the yaml
+spark = SparkSession.builder.getOrCreate()
 
 end_date = datetime.now().replace(microsecond=0, second=0, minute=0)
 start_date = end_date - timedelta(days=15)
@@ -14,8 +17,6 @@
 driver_entities = [1001, 1002, 1003, 1004, 1005]
 driver_df = create_driver_hourly_stats_df(driver_entities, start_date, end_date)
 
-# # we are loading a sparksession here, but should be configurable in the yaml
-spark = SparkSession.builder.getOrCreate()
 
 # this would just be a registered table
 spark.createDataFrame(driver_df).createOrReplaceTempView("driver_stats")
@@ -48,3 +49,36 @@
     batch_source=driver_hourly_stats,
     tags={},
 )
+
+
+customer_entities = [201, 202, 203, 204, 205]
+customer_df = create_customer_daily_profile_df(customer_entities, start_date, end_date)
+spark.createDataFrame(customer_df).createOrReplaceTempView("customer_daily_stats")
+
+customer_daily_stats = SparkSource(
+    table="customer_daily_stats",  # must be serializable so no support of DataFrame objects
+    event_timestamp_column="event_timestamp",
+    created_timestamp_column="created",
+)
+
+customer = Entity(
+    name="customer_id",
+    value_type=ValueType.INT64,
+    description="customer id",
+)
+
+# Define FeatureView
+customer_daily_profile_view = FeatureView(
+    name="customer_daily_profile",
+    entities=["customer_id"],
+    ttl=Duration(seconds=86400 * 1),
+    features=[
+        Feature(name="current_balance", dtype=ValueType.FLOAT),
+        Feature(name="avg_passenger_count", dtype=ValueType.FLOAT),
+        Feature(name="lifetime_trip_count", dtype=ValueType.INT64),
+    ],
+    online=True,
+    batch_source=customer_daily_stats,
+    tags={},
+)
+
diff --git a/feast_spark_offline_store/spark.py b/feast_spark_offline_store/spark.py
index 1260bae..d2697c2 100644
--- a/feast_spark_offline_store/spark.py
+++ b/feast_spark_offline_store/spark.py
@@ -344,10 +344,11 @@ def _format_datetime(t: datetime):
     {% endfor %}
 );
 
+---EOS---
+
 {% endfor %}
 -- End create temporary table *__base
 
----EOS---
 
 {% for featureview in featureviews %}
 


Test patch:
diff --git a/tests/test_spark_offline_store.py b/tests/test_spark_offline_store.py
index d96c7a5..3725132 100644
--- a/tests/test_spark_offline_store.py
+++ b/tests/test_spark_offline_store.py
@@ -4,10 +4,10 @@
 
 from feast import FeatureStore
 
-from example_feature_repo.example import driver, driver_hourly_stats_view
+from example_feature_repo.example import driver, driver_hourly_stats_view, customer, customer_daily_profile_view
 
 
-def test_end_to_end():
+def test_end_to_end_one_feature_view():
     fs = FeatureStore("example_feature_repo/")
 
     try:
@@ -36,6 +36,40 @@ def test_end_to_end():
         fs.teardown()
 
 
+def test_end_to_end_multiple_feature_views():
+    fs = FeatureStore("example_feature_repo/")
+
+    try:
+        # apply repository
+        fs.apply([driver, driver_hourly_stats_view, customer, customer_daily_profile_view])
+
+        # load data into online store (uses offline stores pull_latest_from_table_or_query)
+        fs.materialize_incremental(end_date=datetime.now())
+
+        entity_df = pd.DataFrame(
+            {"driver_id": [1001], "customer_id": [201], "event_timestamp": [datetime.now()]}
+        )
+
+        # Read features from offline store
+        feature_vector = (
+            fs.get_historical_features(
+                features=["driver_hourly_stats:conv_rate", "customer_daily_profile:lifetime_trip_count"],
+                entity_df=entity_df
+            )
+            .to_df()
+            .to_dict()
+        )
+        conv_rate = feature_vector["conv_rate"][0]
+        assert conv_rate > 0
+
+        lifetime_trip_count = feature_vector["lifetime_trip_count"][0]
+        assert lifetime_trip_count > 0
+
+    finally:
+        # tear down feature store
+        fs.teardown()
+
+
 def test_cli():
     repo_name = "example_feature_repo"
     os.system(f"PYTHONPATH=$PYTHONPATH:/$(pwd) feast -c {repo_name} apply")