[KED-1876] Allow GBQTableDataSet to optionally accept a sql query to load data

RELEASE.md

@@ -15,11 +15,12 @@
 ## Major features and improvements
 
 ## Bug fixes and other changes
+* Modified `GBQTableDataSet` to load customized results using customized queries from Google Big Query tables.
 
 ## Breaking changes to the API
 
 ## Thanks for supporting contributions
-
+[Ajay Bisht](https://github.com/ajb7)
 
 # Release 0.16.3
 

kedro/extras/datasets/pandas/gbq_dataset.py

@@ -137,12 +137,18 @@ def _describe(self) -> Dict[str, Any]:
 
     def _load(self) -> pd.DataFrame:
         sql = "select * from {}.{}".format(self._dataset, self._table_name)  # nosec
-        return pd.read_gbq(
-            sql,
-            project_id=self._project_id,
-            credentials=self._credentials,
-            **self._load_args
-        )
+        if "query" not in self._load_args:
+            self._load_args.update({"query": sql})
+
+        self._load_args["query"] = str(self._load_args["query"])
+        try:
+            return pd.read_gbq(
+                project_id=self._project_id,
+                credentials=self._credentials,
+                **self._load_args
+            )
+        except ValueError:
+            raise ValueError()
 
     def _save(self, data: pd.DataFrame) -> None:
         data.to_gbq(

tests/extras/datasets/pandas/test_gbq_dataset.py

@@ -135,7 +135,6 @@ def test_str_representation(self, gbq_dataset, save_args, load_args):
 
     def test_save_load_data(self, gbq_dataset, dummy_dataframe, mocker):
         """Test saving and reloading the data set."""
-        sql = "select * from {}.{}".format(DATASET, TABLE_NAME)
         table_id = "{}.{}".format(DATASET, TABLE_NAME)
         mocked_read_gbq = mocker.patch(
             "kedro.extras.datasets.pandas.gbq_dataset.pd.read_gbq"
@@ -150,7 +149,7 @@ def test_save_load_data(self, gbq_dataset, dummy_dataframe, mocker):
             table_id, project_id=PROJECT, credentials=None, progress_bar=False
         )
         mocked_read_gbq.assert_called_once_with(
-            sql, project_id=PROJECT, credentials=None
+            project_id=PROJECT, credentials=None, **gbq_dataset._load_args
         )
         assert_frame_equal(dummy_dataframe, loaded_data)