Allow partition-wise copy from bq to bq tables.

[bhankit1410]

Description

As of now, there is no support to copy a partition from a source table to destination table's partition in the existing code

bq cp cli gives an option to do this with a decorator ( $ ).

Use case/motivation

we have many use cases everyday where we keep different tables across projects in sync by materialising new partitions with a frequency. This feature could be great. other options for now would be using DBT, but to use it just for this will be an added task.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[o-nikolas]

@eladkal, @kaxil
Can you add the area:providers/provider:Google label to this one?

Also, it seems like a somewhat straightforward update to the operator (though GCP is not my area of expertise) so perhaps Good First Issue as well?

[kaxil]

@o-nikolas done

[eladkal]

@bhankit1410 We only expose capabilities available in job AP:
https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.copy
Can you reference to this ability in the API?

[bugraoz93]

Hello, I think I can give the reference. I think @bhankit1410 is talking about the following operation. It copies individual partitions to another table or another table's partition.
https://cloud.google.com/bigquery/docs/managing-partitioned-tables##copying_partitions
This is the same job triggered by the following part of the BigQuery Hook which is used in the BigQueryToBigQuery Operator.
Hook:

airflow/airflow/providers/google/cloud/hooks/bigquery.py

Line 1805 in 093345c

def run_copy(

Operator:

airflow/airflow/providers/google/cloud/transfers/bigquery_to_bigquery.py

Line 129 in 093345c

job_id = hook.run_copy(

The only difference is sourceTables.tableId and destinationTable.tableId can contain partition information with a $ sign followed by PARTITION_IDENTIFIER for the same copy job.
https://cloud.google.com/bigquery/docs/reference/rest/v2/TableReference

I am willing to submit a PR! If anyone can assign me the issue, that would be great! Thanks in advance!

[eladkal]

Assigned

[bugraoz93]

Thank you very much @eladkal!

[bugraoz93]

Hey @bhankit1410, I think this feature has already been supported.

I have tested on BigQueryToBigQueryOperator. I created a simple DAG. You can see the DAG as follows. I generated sample 1000 records with multiple types of column types such as integer, UUID and timestamp. I have inserted data into a table. I have tested with every three Write Dispositions. Every case I have tested has been working for me so far.

https://gist.github.com/bugraoz93/d3ee6d2d03d1881de4614d1e7c3b8234

import datetime

from airflow import DAG
from airflow.providers.google.cloud.transfers.bigquery_to_bigquery import BigQueryToBigQueryOperator

with DAG(
    dag_id="test_dag",
    max_active_runs=1,
    start_date=datetime.datetime(2022, 11, 24),
    schedule_interval="@once",
    catchup=False,
    concurrency=1,
) as dag:
    project = "test_project"
    dataset = "bugraoz93_test"
    source_table = "test_table$20221125"
    destination_table = "test_table_dest$20221124"

    copy_table_to_fact = BigQueryToBigQueryOperator(
        task_id="copy_test",
        gcp_conn_id="gcp_conn",
        source_project_dataset_tables="{project}.{dataset}.{table}".format(
            project=project, dataset=dataset, table=source_table),
        destination_project_dataset_table="{project}.{dataset}.{table}".format(
            project=project, dataset=dataset, table=destination_table
        ),
        write_disposition="WRITE_APPEND",
        create_disposition="CREATE_IF_NEEDED",
        dag=dag,
    )

I have also checked the code. I have tested the individual methods to ensure that they can process the $ sign within the table name without any exceptions. There is no part within the code that prevents this feature to work.

Could you please expand your case a little bit? Which Apache Airflow version are you using to achieve it? Which provider version are you using for google (apache-airflow-providers-google)?

[eladkal]

Thank you @bugraoz93
I'm closing this issue as completed then

[bhankit1410]

Yes. I realised it works with a '$' decorator and went ahead. It is in production for 2 months now. My sheer stupidity to miss out on updating this thread and closing it. Sorry about that. Cheers.