Add priority parameter to BigQueryHook

[ying-w]

Currently priority= is available in run_query() function within BigQueryHook class but can't be set when class is created.

This causes problems when you're using SQLExecuteQueryOperator since priority is missing from class constructor, you won't be able to pass in with hook_params={"priority":"BATCH"}. Trying to workaround using hook_params={"api_resource_configs": {"query": {"priority": "BATCH"}}" will error on _api_resource_configs_duplication_check() since priority is defaulted to 'INTERACTIVE'.

This PR adds priority to BigQueryHook constructor and defaults to using that value for run_query()

[hussein-awala]

This need a unit test, especially for the missing location which is not detected in the current tests

[hussein-awala]

I think you should not update the class instance attribute:

# if you create a hook with default priority BATCH
hook = BigQueryHook(..., priority="BATCH")
# the you run a query with priority INTERACTIVE
hook.run_query(..., priority="INTERACTIVE")
# finally, if you run another query without priority argument, it should use the one
# provided on the instance creation (BATCH), but it is not the case
hook.run_query(....)  # self.priority is set to INTERACTIVE

[ying-w]

I'm OK working off BigQueryHook() but was wondering if the preferred way to submit arbitrary sql is to leverage SQLExecuteQueryOperator() under commons.

I'm pretty new to airflow and wasn't sure when I should start with operator vs starting with hook. It didn't seem like there was an obvious way of passing in a modified hook into SQLExecuteQueryOperator()

If I use BigQueryInsertJobOperator() (the bigquery specific way of doing SQLExecuteQueryOperator()) then I would need to create a custom operator based on BigQueryInsertJobOperator() that overwrites _submit_job() since that calls BigQueryHook() and operator class also doesn't have priority as a class attribute.

[vchiapaikeo]

Why not just add a priority parameter to BigQueryInsertJobOperator as well? That way, you can instantiate the hook with it? Please default it in such a way where it is backwards compatible.

[ying-w]

In BigQueryInsertJobOperator priority is set by configuration. This field is similar to the api_resource_configs field within BigQueryHook

[ying-w]

I'm going to revert the location change, in insert_job it's not used in configuration but rather jobReference

[ying-w]

@vchiapaikeo i think it's ready for another review

I think the ideal fix is that run_query() is deprecated and cursor should instead call insert_job(), however, need the former to parse api_resource_configs from hook and build this into configuration for insert_job()

[vchiapaikeo]

I think the ideal fix is that run_query() is deprecated and cursor should instead call insert_job(), however, need the former to parse api_resource_configs from hook and build this into configuration for insert_job()

This makes sense to me. You are suggesting we replace this call to run_query to use insert_job instead right? Would it be possible to refactor the api_resource_configs parsing done in run_query and call that within execute?

Also, can we add a unit test or modify an existing one that shows no priority passed to the hook, the value of priority passed to insert_job within configuration, if it was passed it, and if it is set on the method itself?

Nice work so far!

[vchiapaikeo]

I'm not sure if this is used in some other way to propagate location to the instance attribute. It isn't the right thing to do but at the same time, I'm worried that an operator was relying on this functionality - especially given the fact that there was a test around this. Can you check if callers of run_query were previously relying on it?

[ying-w]

i was just checking what other calls go to run_query() and in hooks/bigquery.py theres only 1 other call from deprecated BigQueryCursor.run_query which passes in *args, **kwargs so might rely on it

in operators/bigquery.py are 2 calls from within the deprecated BigQueryExecuteQueryOperator class just passing through values for execute() but doesn't pass in location. Many of these operators set self.location = location and then pass that to BigQueryHook

So getting rid of the call from cursor.execute() should remove calls to run_query() from non-deprecated functions

(i checked all references of run_query within airflow/providers/google/cloud/)

[ying-w]

This makes sense to me. You are suggesting we replace this call to run_query to use insert_job instead right?

yup

Would it be possible to refactor the api_resource_configs parsing done in run_query and call that within execute?

yeah, just wasn't sure how much to put in one PR

btw, is there a way to test changes kinda like pip install -e .? I couldn't figure out how to 'install' the provider and instead modifying the code in python site-packages

[potiuk]

Generally pip install -e should also include provider packages if you use INSTALL_PROVIDERS_FROM_SOURCES (see https://github.com/potiuk/airflow/blob/main/INSTALL) for example. And Any IDE will work fine wiht Providers automatically as they are part of the sources.

There is a bug about some problems with latest pip versions #30764 so it might not work for all cases or you might need to use older pip version.

Also you can use breeze environment if you want to use containerized environment - this is the same as used in CI.

All the options of setting up the env are also well described in CONTRIBUTING docs.

[ying-w]

This makes sense to me. You are suggesting we replace this call to run_query to use insert_job instead right? Would it be possible to refactor the api_resource_configs parsing done in run_query and call that within execute?

On second thought, a lot of what run_query() is doing is parsing and validating api_resource_configs so I think there would be a lot of duplication if i refactored it. I think what happened was they moved from an ExecuteSQL(sql, api_resource_configs) abstraction to InsertJob(configuration={query: sql, <other config>: api_resource_configs}) abstraction and hook.run_query() links the two

Also, can we add a unit test or modify an existing one that shows no priority passed to the hook, the value of priority passed to insert_job within configuration, if it was passed it, and if it is set on the method itself?

Done @vchiapaikeo

[potiuk]

LGTM: pending tests pass.

[potiuk]

Tests failed though :)

[ying-w]

I'll make some changes later tonight, it has to do with moving location out of configuration and being passed in as parameter