DAG with BigQueryToGCSOperator or GCSToBigQueryOperator not generating job_id when triggered by asset event in Airflow 3

[EliveltonRepolho]

Apache Airflow Provider(s)

google

Versions of Apache Airflow Providers

apache-airflow-providers-google==18.0.0

Apache Airflow version

3.1.0+astro.1

Operating System

Linux

Deployment

Astronomer

Deployment details

No response

What happened

When a DAG is triggered by Asset event, due to some changes in Airflow 3 the logical_date is not passed through the context anymore with asset triggered dagruns.

Operators BigQueryToGCSOperator and are broken GCSToBigQueryOperator due this issue:

KeyError: 'logical_date'
File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 920 in run

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1302 in _execute_task

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 416 in wrapper

File "/usr/local/airflow/plugins/operators/gcs_to_bigquery_operator.py", line 27 in execute

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 416 in wrapper

File "/usr/local/lib/python3.12/site-packages/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py", line 340 in execute

What you think should happen instead

BigQueryToGCSOperator and GCSToBigQueryOperator should be able to generate a job_id regardless of whether it is triggered in a scheduled / manually triggered DAG or through an asset event.

I don't know much about Airflow models, but possibly changing logical_date by dag_run_id would do the work ?

How to reproduce

Create DAG triggered by Asset that has BigQueryToGCSOperator and GCSToBigQueryOperator

Anything else

this issue was fixed for BigQueryInsertJobOperator in #55092, but we have the other 2 with same issue.

Are you willing to submit 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! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[kevinhongzl]

Update: This issue will be addressed once we merge #55110

---

The bug can reproduced with the following code:

from datetime import datetime

from airflow.datasets import Dataset
from airflow.decorators import dag, task
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator

# Define LOCAL_FILE_DATASET, GCS_BUCKET, GCS_OBJECT, BQ_DATASET, BQ_TABLE

@dag(dag_id="producer_dag", schedule=None, catchup=False)
def producer_dag_local():
    @task(outlets=[LOCAL_FILE_DATASET])
    def update_local_file():
        with open(LOCAL_FILE_DATASET.uri.replace("file://", ""), "w") as f:
            f.write(f"Data updated at {datetime.now()}")
        print(f"Local asset updated: {LOCAL_FILE_DATASET.uri}")

    update_local_file()

producer_dag_local()

@dag(
    dag_id="asset_consumer_gcs_dag",
    schedule=[LOCAL_FILE_DATASET],
    catchup=False,
)
def asset_consumer_gcs_dag():
    gcs_to_bq_failing = GCSToBigQueryOperator(
        task_id="gcs_to_bigquery_failing",
        bucket=GCS_BUCKET,
        source_objects=[GCS_OBJECT],
        destination_project_dataset_table=f"{BQ_DATASET}.{BQ_TABLE}",
        gcp_conn_id="google_cloud_default",
        source_format="CSV",
    )

asset_consumer_gcs_dag()