[FSTORE-1536] Support delta table in python client (#347)

* clean up

add DeltaCachedFeatureGroups

read delta s3 python

read write delta s3

add delta to dev

add delta

read delta on s3

* clean up

* clean up

* fix style

* fix test

* fix region

* fix s3 path

* fix s3 path

Author: kennethmhc

python/hsfs/constructor/fs_query.py

@@ -35,6 +35,7 @@ def __init__(
         expand: Optional[List[str]] = None,
         items: Optional[List[Dict[str, Any]]] = None,
         type: Optional[str] = None,
+        delta_cached_feature_groups: Optional[List[Dict[str, Any]]] = None,
         **kwargs,
     ) -> None:
         self._query = query
@@ -60,6 +61,14 @@ def __init__(
         else:
             self._hudi_cached_feature_groups = []
 
+        if delta_cached_feature_groups is not None:
+            self._delta_cached_feature_groups = [
+                hudi_feature_group_alias.HudiFeatureGroupAlias.from_response_json(fg)
+                for fg in delta_cached_feature_groups
+            ]
+        else:
+            self._delta_cached_feature_groups = []
+
     @classmethod
     def from_response_json(cls, json_dict: Dict[str, Any]) -> "FsQuery":
         json_decamelized = humps.decamelize(json_dict)
@@ -127,7 +136,7 @@ def register_delta_tables(
         feature_store_name: str,
         read_options: Optional[Dict[str, Any]],
     ) -> None:
-        for hudi_fg in self._hudi_cached_feature_groups:
+        for hudi_fg in self._delta_cached_feature_groups:
             engine.get_instance().register_delta_temporary_table(
                 hudi_fg, feature_store_id, feature_store_name, read_options
             )

python/hsfs/core/arrow_flight_client.py

@@ -124,7 +124,13 @@ def _is_query_supported_rec(query: query.Query):
         and query._left_feature_group.storage_connector.type
         in ArrowFlightClient.SUPPORTED_EXTERNAL_CONNECTORS
     )
-    supported = hudi_no_time_travel or supported_connector
+    delta_s3 = (
+        isinstance(query._left_feature_group, feature_group.FeatureGroup)
+        and query._left_feature_group.time_travel_format == "DELTA"
+        and query._left_feature_group.storage_connector
+        and query._left_feature_group.storage_connector.type == StorageConnector.S3
+    )
+    supported = hudi_no_time_travel or supported_connector or delta_s3
     for j in query._joins:
         supported &= _is_query_supported_rec(j._query)
     return supported
@@ -549,6 +555,7 @@ def enabled_on_cluster(self) -> bool:
 def _serialize_featuregroup_connector(fg, query, on_demand_fg_aliases):
     connector = {}
     if isinstance(fg, feature_group.ExternalFeatureGroup):
+        connector["time_travel_type"] = None
         connector["type"] = fg.storage_connector.type
         connector["options"] = fg.storage_connector.connector_options()
         connector["query"] = fg.query[:-1] if fg.query.endswith(";") else fg.query
@@ -566,8 +573,25 @@ def _serialize_featuregroup_connector(fg, query, on_demand_fg_aliases):
                     connector["filters"] = _serialize_filter_expression(
                         join_obj._query._filter, join_obj._query, True
                     )
+    elif fg.time_travel_format == "DELTA":
+        connector["time_travel_type"] = "delta"
+        connector["type"] = fg.storage_connector.type
+        connector["options"] = fg.storage_connector.connector_options()
+        if fg.storage_connector.type == StorageConnector.S3:
+            connector["options"]["path"] = fg.location
+        connector["query"] = ""
+        if query._left_feature_group == fg:
+            connector["filters"] = _serialize_filter_expression(
+                query._filter, query, True
+            )
+        else:
+            for join_obj in query._joins:
+                if join_obj._query._left_feature_group == fg:
+                    connector["filters"] = _serialize_filter_expression(
+                        join_obj._query._filter, join_obj._query, True
+                    )
     else:
-        connector["type"] = "hudi"
+        connector["time_travel_type"] = "hudi"
     return connector
 
 

python/hsfs/feature_group.py

@@ -166,7 +166,6 @@ def __init__(
             )
         else:
             self._storage_connector: "sc.StorageConnector" = storage_connector
-
         self._online_config = (
             OnlineConfig.from_response_json(online_config)
             if isinstance(online_config, dict)

python/hsfs/storage_connector.py

@@ -369,6 +369,15 @@ def prepare_spark(self, path: Optional[str] = None) -> Optional[str]:
         """
         return engine.get_instance().setup_storage_connector(self, path)
 
+    def connector_options(self) -> Dict[str, Any]:
+        """Return options to be passed to an external S3 connector library"""
+        return {
+            "access_key": self.access_key,
+            "secret_key": self.secret_key,
+            "session_token": self.session_token,
+            "region": self.region,
+        }
+
     def read(
         self,
         query: Optional[str] = None,

python/pyproject.toml

@@ -74,6 +74,7 @@ dev-no-opt = [
     "pyspark==3.1.1",
     "moto[s3]==5.0.0",
     "typeguard==4.2.1",
+    "delta-spark==1.0.1"
 ]
 dev-pandas1 = [
     "hopsworks[python]",
@@ -84,6 +85,7 @@ dev-pandas1 = [
     "moto[s3]==5.0.0",
     "pandas<=1.5.3",
     "sqlalchemy<=1.4.48",
+    "delta-spark==1.0.1"
 ]
 dev = ["hopsworks[dev-no-opt,great-expectations,polars]"]
 polars=["polars>=0.20.18,<=0.21.0"]

python/tests/core/test_arrow_flight_client.py

@@ -252,7 +252,7 @@ def test_construct_query_object(self, mocker, backend_fixtures):
                     "right_filter": None,
                 },
             },
-            "connectors": {"test.fg_test_1": {"type": "hudi"}},
+            "connectors": {"test.fg_test_1": {"time_travel_type": "hudi"}},
         }
 
         query_object["features"] = {
@@ -293,7 +293,7 @@ def test_construct_query_object_datetime_filter(self, mocker, backend_fixtures):
                 },
                 "right_filter": None,
             },
-            "connectors": {"test.fg_test_1": {"type": "hudi"}},
+            "connectors": {"test.fg_test_1": {"time_travel_type": "hudi"}},
         }
 
         query_object["features"] = {
@@ -331,7 +331,7 @@ def test_construct_query_object_without_fs(self, mocker, backend_fixtures):
                 },
                 "right_filter": None,
             },
-            "connectors": {"test.fg_test_1": {"type": "hudi"}},
+            "connectors": {"test.fg_test_1": {"time_travel_type": "hudi"}},
         }
 
         query_object["features"] = {
@@ -369,7 +369,7 @@ def test_construct_query_object_without_fs_excluded(self, mocker, backend_fixtur
                 },
                 "right_filter": None,
             },
-            "connectors": {"test.fg_test_1": {"type": "hudi"}},
+            "connectors": {"test.fg_test_1": {"time_travel_type": "hudi"}},
         }
 
         query_object["features"] = {
@@ -430,7 +430,8 @@ def test_construct_query_object_snowflake(self, mocker, backend_fixtures):
             },
             "connectors": {
                 "test.tpch1snowflake_1": {
-                    "type": "SNOWFLAKE",
+                    "time_travel_type": None,
+                    "type": 'SNOWFLAKE',
                     "options": {
                         "user": "test_user",
                         "account": "test_url",