Implement Async BigqueryOperator

[kaxil]

Build async version of https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/operators/bigquery.py

Acceptance Criteria:

• Unit Tests coverage in the PR (80% Code Coverage -- We will need to add CodeCov separately to measure code cov) with all of them passing
• Example DAG using the async Operator that can be used to run Integration tests that are parametrized via Environment variables. Example - https://github.com/apache/airflow/blob/8a03a505e1df0f9de276038c5509135ac569a667/airflow/providers/google/cloud/example_dags/example_bigquery_to_gcs.py#L33-L35
• Add proper docstrings for each of the methods and functions including Example DAG on how it should be used (populate
• Exception Handling in case of errors
• Improve the OSS Docs to make sure it covers the following:
  • Has an example DAG for the sync version
  • How to add a connection via Environment Variable & explain each of the fields. Example - https://airflow.apache.org/docs/apache-airflow-providers-postgres/stable/connections/postgres.html
  • How to use Guide for the Operator - example: https://airflow.apache.org/docs/apache-airflow-providers-postgres/stable/operators/postgres_operator_howto_guide.html

[phanikumv]

Below classes need to be converted to their Async versions ( based on feasibility )

Operator	API used	Assignee	Async Implementation Needed ?	Comments
BigQueryCheckOperator		@phanikumv	Yes	Performs checks against BigQuery. The BigQueryCheckOperator expects a sql query that will return a single row.
BigQueryValueCheckOperator		@pankajastro	Yes	Performs a simple value check using sql code.
BigQueryIntervalCheckOperator		@sunank200	Yes	Checks that the values of metrics given as SQL expressions are within a certain tolerance of the ones from days_back before.
BigQueryGetDataOperator	https://cloud.google.com/bigquery/docs/reference/rest/v2/tabledata/list	@rajaths010494	Yes	Would require async but not gcloud aio not directly usable.Cannot get pid or query id to get status of long running tasks
BigQueryExecuteQueryOperator			No	NA(Deprecated)
BigQueryCreateEmptyTableOperator			No
BigQueryCreateExternalTableOperator			No
BigQueryDeleteDatasetOperator			No
BigQueryCreateEmptyDatasetOperator			No
BigQueryGetDatasetOperator			No
BigQueryGetDatasetTablesOperator			No
BigQueryPatchDatasetOperator			No	It only replaces fields that are provided in the submitted dataset resource.
BigQueryUpdateTableOperator		@sunank200	No	Cannot get pid or query id to get status of long running tasks. Changes some field in table as specified
BigQueryUpdateDatasetOperator			No	Changes some field in dataset as specified
BigQueryDeleteTableOperator			No
BigQueryUpsertTableOperator	https://cloud.google.com/bigquery/docs/reference/v2/tables#resource		No	If the table already exists, update the existing table if not create new. Since BigQuery does not natively allow table upserts, this is not an atomic operation.
BigQueryUpdateTableSchemaOperator	https://cloud.google.com/bigquery/docs/reference/rest/v2/tables#TableSchema		No	Update fields within a schema for a given dataset and table.
BigQueryInsertJobOperator	https://cloud.google.com/bigquery/docs/reference/v2/jobs	@phanikumv	Yes	implementation completed

[phanikumv]

Getting the below api response from gcloud-aio library when I try to read a table. While the sync version of the BigQueryGetDataOperator is returning a list[Row] --> [Row(('100', '200'), {'col1': 0, 'col2': 1}), Row(('300', '400'), {'col1': 0, 'col2': 1})], this is returning a list --> [{'f': [{'v': '300'}, {'v': '400'}]}, {'f': [{'v': '100'}, {'v': '200'}]}]

Do we need to ensure that both Async and Sync versions of an Operator return same datatypes at the end of execution? Thoughts @kaxil

gcloud-aio response

{'kind': 'bigquery#getQueryResultsResponse', 'etag': 'Ycgra5bwYNpTA3Tgy9ICdw==', 'schema': {'fields': [{'name': 'col1', 'type': 'STRING', 'mode': 'NULLABLE'}, {'name': 'col2', 'type': 'STRING', 'mode': 'NULLABLE'}]}, 'jobReference': {'projectId': 'astronomer-airflow-providers', 'jobId': 'job_-2e9ok-6d9gWKmBnX3K92wLpF0km', 'location': 'US'}, 'totalRows': '2', 'rows': [{'f': [{'v': '300'}, {'v': '400'}]}, {'f': [{'v': '100'}, {'v': '200'}]}], 'totalBytesProcessed': '20', 'jobComplete': True, 'cacheHit': False}

[phanikumv]

Let's focus on the below 4 operators on priority:
BigQueryGetDataOperator
BigQueryExecuteQueryOperator
BigQueryCheckOperator
BigQueryInsertJobOperator

[phanikumv]

Only the jobs api of the BigQuery submits a job asynchronously. Refer https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/insert
It returns a job ID through which we can track the status of the request using the Triggerer.

This might become an issue for asynchronously calling the other BigQuery api's like https://cloud.google.com/bigquery/docs/reference/rest/v2/tabledata/list , because this api doesnt return any kind of processId or a jobId , once the request is submitted. Hence, need to find alternate approach for implementing asynchronous version of the BigQueryGetDataOperator

[phanikumv]

When multiple queries are passed within a list, it doesnt work in OSS version of the BigQueryInsertJobOperator. It does work when the queries are passed as a single string.

BigQueryInsertJobOperator uses the Google Jobs API. if the user passes the queries within a list, it wont work because the Google Jobs API doesnt allow using an array to pass multiple queries (please refer to screenshot below).

There used to be an operator called BigQueryExecuteQueryOperator , which supported passing multiple queries through a list, but this is deprecated now. @kaxil @dstandish - please opine on this observation

---

This is discussed and we agreed that we will add a docstring to the Operator to describe how to pass multiple queries - through a string separated by a semi-colon rather than a list.

[phanikumv]

BigQueryExecuteQueryOperator is deprecated. Hence no need to create an async version of it.

[phanikumv]

Draft #47 for the BigQueryInsertJobOperator

[phanikumv]

PR is ready for BigQueryInsertJobOperatorAsync

[sunank200]

BigQueryUpdateTableOperator uses Table object and change specified fields of a table. This operator doesn't update the data in the table but instead changes the fields of Table.

I have tested the Method: tables.update and have attached the screenshot. In my opinion, Async implementation for this BigQueryUpdateTableOperator or any other operator which uses Method: tables of google cloud would not improve the performance as it just deals with Table fields rather than Tabledata

Also, it doesn't return any job-id or process-id as such which can be tracked on the google-cloud side. This is only available for Operators which use Method:Jobs

Moreover, gcloud-aio class Table uses Table object instead of TableData

All the operators which use google-cloud jobs API would require async implementation.

@phanikumv , @rajaths010494 , @kaxil

[rajaths010494]

In OSS BigQueryGetDataOperator uses Table object using list_rows methods to fetch data.
In gcloud aio class Table uses Table object instead of TableData which doesn't have TableData List api implemented and also all the calls for table in gcloud uses aiohttp calls to the api where it awaits until the response is given rather than getting a pid or job-id (Method:Jobs responses with the id).
So if a long running get operation is made it awaits until the API get call is finished rather than giving a query id so that we can check for the status of the query if it's finished or not.

All the operators which use google-cloud jobs API would require async implementation.

[sunank200]

I have updated the table above with a field Async Implementation Needed and Comments with necessary details. Please let me know your views

[phanikumv]

My thought process for BigQueryGetDataOperator is that we’ll re-use the BigQueryInsertjobOperatorAsync to form a select * from table during run time, get a job id, and then poll it on the Trigger, because only the jobs API gives us the job ID. We can adopt same strategy for BigQueryCheckOperator, BigQueryValueCheckOperator and BigQueryIntervalCheckOperator

[phanikumv]

PR is ready for BigQueryCheckOperatorAsync

[sunank200]

PR for BigQueryIntervalCheckOperatorAsync

[kaxil]

Can this be closed now, or are there any other pending implementations @phanikumv ?

[phanikumv]

Implementation of the below operators is done. Hence closing the story.

1. BigQueryCheckOperator
2. BigQueryValueCheckOperator
3. BigQueryIntervalCheckOperator
4. BigQueryGetDataOperator
5. BigQueryInsertJobOperator