Merge remote-tracking branch 'origin/main' into prefix-msg

Author: aversey

java/beam/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>hsfs-parent</artifactId>
     <groupId>com.logicalclocks</groupId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

java/flink/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>hsfs-parent</artifactId>
     <groupId>com.logicalclocks</groupId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

java/hsfs/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>hsfs-parent</artifactId>
     <groupId>com.logicalclocks</groupId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

java/pom.xml

@@ -7,7 +7,7 @@
     <groupId>com.logicalclocks</groupId>
     <artifactId>hsfs-parent</artifactId>
     <packaging>pom</packaging>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
     <modules>
         <module>hsfs</module>
         <module>spark</module>

java/spark/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <artifactId>hsfs-parent</artifactId>
         <groupId>com.logicalclocks</groupId>
-        <version>4.0.0-SNAPSHOT</version>
+        <version>4.1.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

python/hopsworks_common/version.py

@@ -14,4 +14,4 @@
 #   limitations under the License.
 #
 
-__version__ = "4.0.0.dev1"
+__version__ = "4.1.0.dev1"

python/hsfs/core/delta_engine.py

@@ -172,6 +172,8 @@ def _generate_merge_query(self, source_alias, updates_alias):
     @staticmethod
     def _get_last_commit_metadata(spark_context, base_path):
         fg_source_table = DeltaTable.forPath(spark_context, base_path)
+
+        # Get info about the latest commit
         last_commit = fg_source_table.history(1).first().asDict()
         version = last_commit["version"]
         commit_timestamp = util.convert_event_time_to_timestamp(
@@ -180,6 +182,12 @@ def _get_last_commit_metadata(spark_context, base_path):
         commit_date_string = util.get_hudi_datestr_from_timestamp(commit_timestamp)
         operation_metrics = last_commit["operationMetrics"]
 
+        # Get info about the oldest remaining commit
+        oldest_commit = fg_source_table.history().orderBy("version").first().asDict()
+        oldest_commit_timestamp = util.convert_event_time_to_timestamp(
+            oldest_commit["timestamp"]
+        )
+
         if version == 0:
             fg_commit = feature_group_commit.FeatureGroupCommit(
                 commitid=None,
@@ -188,7 +196,7 @@ def _get_last_commit_metadata(spark_context, base_path):
                 rows_inserted=operation_metrics["numOutputRows"],
                 rows_updated=0,
                 rows_deleted=0,
-                last_active_commit_time=commit_timestamp,
+                last_active_commit_time=oldest_commit_timestamp,
             )
         else:
             fg_commit = feature_group_commit.FeatureGroupCommit(
@@ -198,7 +206,7 @@ def _get_last_commit_metadata(spark_context, base_path):
                 rows_inserted=operation_metrics["numTargetRowsInserted"],
                 rows_updated=operation_metrics["numTargetRowsUpdated"],
                 rows_deleted=operation_metrics["numTargetRowsDeleted"],
-                last_active_commit_time=commit_timestamp,
+                last_active_commit_time=oldest_commit_timestamp,
             )
 
         return fg_commit

python/hsfs/storage_connector.py

@@ -1575,9 +1575,8 @@ def connector_options(self) -> Dict[str, Any]:
         """Return options to be passed to an external BigQuery connector library"""
         props = {
             "key_path": self._key_path,
-            "project_id": self._query_project,
+            "project_id": self._parent_project,
             "dataset_id": self._dataset,
-            "parent_project": self._parent_project,
         }
         return props
 

python/tests/fixtures/storage_connector_fixtures.json

@@ -495,7 +495,7 @@
         },
         "headers": null
     },
-    "get_big_query": {
+    "get_big_query_table": {
         "response": {
             "type": "featurestoreBigQueryConnectorDTO",
             "description": "BigQuery connector description",
@@ -508,6 +508,32 @@
             "dataset": "test_dataset",
             "query_table": "test_query_table",
             "query_project": "test_query_project",
+            "arguments": [{"name":  "test_name", "value":  "test_value"}]
+        },
+        "method": "GET",
+        "path_params": [
+            "project",
+            "119",
+            "featurestores",
+            67,
+            "storageconnectors",
+            "test_big_query"
+        ],
+        "query_params": {
+            "temporaryCredentials": true
+        },
+        "headers": null
+    },
+    "get_big_query_query": {
+        "response": {
+            "type": "featurestoreBigQueryConnectorDTO",
+            "description": "BigQuery connector description",
+            "featurestoreId": 67,
+            "id": 1,
+            "name": "test_big_query",
+            "storageConnectorType": "BIGQUERY",
+            "key_path": "test_key_path",
+            "parent_project": "test_parent_project",
             "materialization_dataset": "test_materialization_dataset",
             "arguments": [{"name":  "test_name", "value":  "test_value"}]
         },

python/tests/test_storage_connector.py

@@ -800,7 +800,7 @@ def test_default_path(self, mocker):
 class TestBigQueryConnector:
     def test_from_response_json(self, backend_fixtures):
         # Arrange
-        json = backend_fixtures["storage_connector"]["get_big_query"]["response"]
+        json = backend_fixtures["storage_connector"]["get_big_query_table"]["response"]
 
         # Act
         sc = storage_connector.StorageConnector.from_response_json(json)
@@ -815,7 +815,6 @@ def test_from_response_json(self, backend_fixtures):
         assert sc.dataset == "test_dataset"
         assert sc.query_table == "test_query_table"
         assert sc.query_project == "test_query_project"
-        assert sc.materialization_dataset == "test_materialization_dataset"
         assert sc.arguments == {"test_name": "test_value"}
 
     def test_from_response_json_basic_info(self, backend_fixtures):
@@ -850,7 +849,7 @@ def test_credentials_base64_encoded(self, mocker, backend_fixtures, tmp_path):
         credentialsFile = tmp_path / "bigquery.json"
         credentialsFile.write_text(credentials)
 
-        json = backend_fixtures["storage_connector"]["get_big_query"]["response"]
+        json = backend_fixtures["storage_connector"]["get_big_query_table"]["response"]
         if isinstance(tmp_path, WindowsPath):
             json["key_path"] = "file:///" + str(credentialsFile.resolve()).replace(
                 "\\", "/"
@@ -891,9 +890,7 @@ def test_query_validation(self, mocker, backend_fixtures, tmp_path):
         credentials = '{"type": "service_account", "project_id": "test"}'
         credentialsFile = tmp_path / "bigquery.json"
         credentialsFile.write_text(credentials)
-        json = backend_fixtures["storage_connector"]["get_big_query"]["response"]
-        # remove property for query
-        json.pop("materialization_dataset")
+        json = backend_fixtures["storage_connector"]["get_big_query_table"]["response"]
         if isinstance(tmp_path, WindowsPath):
             json["key_path"] = "file:///" + str(credentialsFile.resolve()).replace(
                 "\\", "/"
@@ -905,3 +902,15 @@ def test_query_validation(self, mocker, backend_fixtures, tmp_path):
         # Assert
         with pytest.raises(ValueError):
             sc.read(query="select * from")
+
+    def test_connector_options(self, backend_fixtures):
+        # Arrange
+        engine.set_instance("python", python.Engine())
+        json = backend_fixtures["storage_connector"]["get_big_query_query"]["response"]
+        sc = storage_connector.StorageConnector.from_response_json(json)
+
+        # Act
+        options = sc.connector_options()
+
+        # Assert
+        assert options["project_id"] == "test_parent_project"

utils/java/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.logicalclocks</groupId>
     <artifactId>hsfs-utils</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>
 
     <properties>
         <hops.version>3.2.0.0-SNAPSHOT</hops.version>