Merge branch 'master' into refactor-type-convert

Author: vatj

java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java

@@ -1097,5 +1097,10 @@ protected String makeQueryName(String queryName, FeatureGroupBase featureGroup)
     }
     return queryName;
   }
-
+  
+  public void closeSparkSession() {
+    if (getSparkSession() != null) {
+      getSparkSession().stop();
+    }
+  }
 }

python/hsfs/constructor/join.py

@@ -47,7 +47,7 @@ def __init__(
         self._on = util.parse_features(on)
         self._left_on = util.parse_features(left_on)
         self._right_on = util.parse_features(right_on)
-        self._join_type = join_type or self.INNER
+        self._join_type = join_type or self.LEFT
         self._prefix = prefix
 
     def to_dict(self) -> Dict[str, Any]:

python/hsfs/constructor/query.py

@@ -239,7 +239,7 @@ def join(
         on: Optional[List[str]] = None,
         left_on: Optional[List[str]] = None,
         right_on: Optional[List[str]] = None,
-        join_type: Optional[str] = "inner",
+        join_type: Optional[str] = "left",
         prefix: Optional[str] = None,
     ) -> "Query":
         """Join Query with another Query.
@@ -769,7 +769,7 @@ def featuregroups(
         """List of feature groups used in the query"""
         featuregroups = {self._left_feature_group}
         for join_obj in self.joins:
-            featuregroups.add(join_obj.query._left_feature_group)
+            self._fg_rec_add(join_obj, featuregroups)
         return list(featuregroups)
 
     @property
@@ -809,6 +809,18 @@ def get_feature(self, feature_name: str) -> "Feature":
         """
         return self._get_feature_by_name(feature_name)[0]
 
+    def _fg_rec_add(self, join_object, featuregroups):
+        """
+        Recursively get a feature groups from nested join and add to featuregroups list.
+
+        # Arguments
+            join_object: `Join object`.
+        """
+        if len(join_object.query.joins) > 0:
+            for nested_join in join_object.query.joins:
+                self._fg_rec_add(nested_join, featuregroups)
+        featuregroups.add(join_object.query._left_feature_group)
+
     def __getattr__(self, name: str) -> Any:
         try:
             return self.__getitem__(name)

python/hsfs/engine/spark.py

@@ -1190,7 +1190,7 @@ def add_cols_to_delta_table(self, feature_group, new_features):
             "spark.databricks.delta.schema.autoMerge.enabled", "true"
         ).save(feature_group.location)
 
-    def _apply_transformation_function(self, transformation_functions, dataset):
+    def _apply_transformation_function(self, transformation_functions, dataset, **kwargs):
         # generate transformation function expressions
         transformed_feature_names = []
         transformation_fn_expressions = []

python/tests/constructor/test_join.py

@@ -55,7 +55,7 @@ def test_from_response_json_basic_info(self, mocker, backend_fixtures):
         assert len(j._on) == 0
         assert len(j._left_on) == 0
         assert len(j._right_on) == 0
-        assert j._join_type == "INNER"
+        assert j._join_type == "LEFT"
         assert j._prefix is None
 
     def test_from_response_json_left_join(self, mocker, backend_fixtures):

utils/java/src/main/java/com/logicalclocks/utils/MainClass.java

@@ -118,5 +118,9 @@ public static void main(String[] args) throws Exception {
     if (op.equals("offline_fg_materialization") || op.equals("offline_fg_backfill")) {
       SparkEngine.getInstance().streamToHudiTable(streamFeatureGroup, writeOptions);
     }
+    
+    LOGGER.info("Closing spark session...");
+    SparkEngine.getInstance().closeSparkSession();
+    System.exit(0);
   }
 }

utils/python/hdfs-file-operations.py

@@ -292,3 +292,6 @@ def parse_isoformat_date(da: str) -> datetime:
         import_fg(job_conf)
     elif args.op == "run_feature_monitoring":
         run_feature_monitoring(job_conf)
+
+    if spark is not None:
+        spark.stop()