chore: Provide related exceptions for empty data frame from entity sql execution in spark sql query execution. (feast-dev#3323)

amithadiraju1694 · web-flow · commit 891231af38cd · 2022-11-20T22:47:33.000-08:00
1. Modified spark.py, to raise error from current file regarding empty data from entity sql execution, than from
separate module.
2. Removed spark_utils.py
3. Removed un-necessary files from git repo and changed gitignore accordingly.

Formatted spark.py with make for better linting

Signed-off-by: amithadiraju1694 &lt;amith.adiraju@gmail.com&gt;

Signed-off-by: amithadiraju1694 &lt;amith.adiraju@gmail.com&gt;
diff --git a/.gitignore b/.gitignore
@@ -223,4 +223,4 @@ ui/.vercel
 **/yarn-error.log*
 
 # Go subprocess binaries (built during feast pip package building)
-sdk/python/feast/binaries/
+sdk/python/feast/binaries/
diff --git a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py
@@ -18,7 +18,7 @@
 
 from feast import FeatureView, OnDemandFeatureView
 from feast.data_source import DataSource
-from feast.errors import InvalidEntityType
+from feast.errors import EntitySQLEmptyResults, InvalidEntityType
 from feast.feature_view import DUMMY_ENTITY_ID, DUMMY_ENTITY_VAL
 from feast.infra.offline_stores import offline_utils
 from feast.infra.offline_stores.contrib.spark_offline_store.spark_source import (
@@ -449,7 +449,13 @@ def _get_entity_df_event_timestamp_range(
         # If the entity_df is a string (SQL query), determine range
         # from table
         df = spark_session.sql(entity_df).select(entity_df_event_timestamp_col)
+
+        # Checks if executing entity sql resulted in any data
+        if df.rdd.isEmpty():
+            raise EntitySQLEmptyResults(entity_df)
+
         # TODO(kzhang132): need utc conversion here.
+
         entity_df_event_timestamp_range = (
             df.agg({entity_df_event_timestamp_col: "min"}).collect()[0][0],
             df.agg({entity_df_event_timestamp_col: "max"}).collect()[0][0],
