Added query_config parameter to read_gbq

doc/source/whatsnew/v0.20.0.txt

@@ -83,3 +83,10 @@ Performance Improvements
 
 Bug Fixes
 ~~~~~~~~~
+
+.. _whatsnew_0200.gbq:
+
+Google BigQuery Enhancements
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+- The :func:`read_gbq` method now allows query configuration preferences

pandas/io/gbq.py

@@ -375,7 +375,7 @@ def process_insert_errors(self, insert_errors):
 
         raise StreamingInsertError
 
-    def run_query(self, query):
+    def run_query(self, query, **kwargs):
         try:
             from googleapiclient.errors import HttpError
         except:
@@ -395,6 +395,9 @@ def run_query(self, query):
                 }
             }
         }
+        query_config = kwargs.get('query_config')
+        if query_config is not None:
+            job_data['configuration']['query'].update(query_config)
 
         self._start_timer()
         try:
@@ -622,7 +625,8 @@ def _parse_entry(field_value, field_type):
 
 
 def read_gbq(query, project_id=None, index_col=None, col_order=None,
-             reauth=False, verbose=True, private_key=None, dialect='legacy'):
+             reauth=False, verbose=True, private_key=None, dialect='legacy',
+             **kwargs):
     """Load data from Google BigQuery.
 
     THIS IS AN EXPERIMENTAL LIBRARY
@@ -682,6 +686,13 @@ def read_gbq(query, project_id=None, index_col=None, col_order=None,
 
         .. versionadded:: 0.19.0
 
+    **kwargs: Arbitrary keyword arguments
+        query_config (dict): query configuration parameters for job processing.
+            For more information see `BigQuery SQL Reference
+            <https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.query>`
+
+        .. versionadded:: 0.20.0
+
     Returns
     -------
     df: DataFrame
@@ -698,7 +709,7 @@ def read_gbq(query, project_id=None, index_col=None, col_order=None,
     connector = GbqConnector(project_id, reauth=reauth, verbose=verbose,
                              private_key=private_key,
                              dialect=dialect)
-    schema, pages = connector.run_query(query)
+    schema, pages = connector.run_query(query, **kwargs)
     dataframe_list = []
     while len(pages) > 0:
         page = pages.pop()

pandas/io/tests/test_gbq.py

@@ -707,10 +707,57 @@ def test_invalid_option_for_sql_dialect(self):
                          private_key=_get_private_key_path())
 
         # Test that a correct option for dialect succeeds
-        # to make sure ValueError was due to invalid dialect
+        # to make sure ValueError was due to invalid  dialect
         gbq.read_gbq(sql_statement, project_id=_get_project_id(),
                      dialect='standard', private_key=_get_private_key_path())
 
+    def test_query_with_parameters(self):
+        sql_statement = "SELECT @param1 + @param2 as VALID_RESULT"
+        query_config = {
+            "useLegacySql": False,
+            "parameterMode": "named",
+            "queryParameters": [
+                {
+                    "name": "param1",
+                    "parameterType": {
+                        "type": "INTEGER"
+                    },
+                    "parameterValue": {
+                        "value": 1
+                    }
+                },
+                {
+                    "name": "param2",
+                    "parameterType": {
+                        "type": "INTEGER"
+                    },
+                    "parameterValue": {
+                        "value": 2
+                    }
+                }
+            ]
+        }
+        # Test that an invalid query without query_config
+        with tm.assertRaises(ValueError):
+            gbq.read_gbq(sql_statement, project_id=_get_project_id(),
+                         private_key=_get_private_key_path())
+
+        # Test that a correct query with query config
+        df = gbq.read_gbq(sql_statement, project_id=_get_project_id(),
+                          private_key=_get_private_key_path(),
+                          query_config=query_config)
+        tm.assert_frame_equal(df, DataFrame({'VALID_RESULT': [3]}))
+
+    def test_query_no_cache(self):
+        query = 'SELECT "PI" as VALID_STRING'
+        query_config = {
+            "useQueryCache": False,
+        }
+        df = gbq.read_gbq(query, project_id=_get_project_id(),
+                          private_key=_get_private_key_path(),
+                          query_config=query_config)
+        tm.assert_frame_equal(df, DataFrame({'VALID_STRING': ['PI']}))
+
 
 class TestToGBQIntegration(tm.TestCase):
     # Changes to BigQuery table schema may take up to 2 minutes as of May 2015