'Executes a BigQuery SQL query. Optionally persists results in a BigQuery
table. See here:
https://cloud.google.com/bigquery/docs/reference/v2/jobs
For more details about these parameters.
:param bql: The BigQuery SQL to execute.
:type bql: string
:param destination_dataset_table: The dotted <dataset>.<table>
BigQuery table to save the query results.
:param write_disposition: What to do if the table already exists in
BigQuery.
:type write_disposition: string
:param create_disposition: Specifies whether the job is allowed to create new tables.
:type create_disposition: string
:param allow_large_results: Whether to allow large results.
:type allow_large_results: boolean
:param udf_config: The User Defined Function configuration for the query.
See https://cloud.google.com/bigquery/user-defined-functions for details.
:type udf_config: list
:param use_legacy_sql: Whether to use legacy SQL (true) or standard SQL (false).
:type use_legacy_sql: boolean
:param maximum_billing_tier: Positive integer that serves as a multiplier of the basic price.
:type maximum_billing_tier: integer'

def run_query(self, bql, destination_dataset_table=False, write_disposition='WRITE_EMPTY', allow_large_results=False, udf_config=False, use_legacy_sql=True, maximum_billing_tier=None, create_disposition='CREATE_IF_NEEDED'):

    configuration = {'query': {'query': bql, 'useLegacySql': use_legacy_sql, 'maximumBillingTier': maximum_billing_tier}}
    if destination_dataset_table:
        assert ('.' in destination_dataset_table), 'Expected    destination_dataset_table    in    the    format    of    <dataset>.<table>.    Got:    {}'.format(destination_dataset_table)
        (destination_project, destination_dataset, destination_table) = _split_tablename(table_input=destination_dataset_table, default_project_id=self.project_id)
        configuration['query'].update({'allowLargeResults': allow_large_results, 'writeDisposition': write_disposition, 'createDisposition': create_disposition, 'destinationTable': {'projectId': destination_project, 'datasetId': destination_dataset, 'tableId': destination_table}})
    if udf_config:
        assert isinstance(udf_config, list)
        configuration['query'].update({'userDefinedFunctionResources': udf_config})
    return self.run_with_configuration(configuration)
