[FSTORE-1548] BigQuery connector is using the wrong project when querying arrowflight

Author: SirOibaf

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"