nailo2c opened a new pull request, #69234: URL: https://github.com/apache/airflow/pull/69234
closes: #47579 ## Summary `BigQueryInsertJobOperator` running a stored procedure / multi-statement query produced a single task event whose inputs and outputs were the **aggregation of all child query jobs**, so dataset-level lineage looked like every input fed every output. This adds a separate OpenLineage QUERY event per BigQuery child job, giving accurate per-statement lineage. + Test Dag run successfully (with connect to BigQuery). <img width="1426" height="803" alt="openlineage_af_ui" src="https://github.com/user-attachments/assets/c8676220-bcd9-4056-a9ae-19d72f385a3b"; /> + Open Lineage UI shows `input_table1` -> `query.1` -> `output_table1` <img width="1437" height="665" alt="Screenshot 2026-07-02 at 3 42 04 PM" src="https://github.com/user-attachments/assets/45714fce-35b9-4904-b248-4f91e7d3fb4f"; /> ## Before One COMPLETE event, child datasets flattened together, looks many-to-many: ```jsonc {"eventType": "COMPLETE", "job": {"name": "...call_stored_procedure"}, "inputs": ["...input_table1", "...input_table2"], "outputs": ["...output_table1", "...output_table2"]} // i1,i2 -> o1,o2 (misleading) ``` ## After The task event is **unchanged** (still the coarse, task-level summary, backward compatible), and each child query is additionally emitted as its own event with precise `input -> output` and a `parent` facet linking back to the task: ```jsonc // child event 1 {"eventType": "COMPLETE", "job": {"name": "...call_stored_procedure.query.1"}, "inputs": ["...input_table1"], "outputs": ["...output_table1"], "run": {"facets": {"parent": {"job": {"name": "...call_stored_procedure"}}, "externalQuery": {"externalQueryId": "script_job_..._0"}}}} // child event 2 -> query.2 : input_table2 -> output_table2 ``` ### Test Dag I Used ```python import os from datetime import datetime from airflow.models.dag import DAG from airflow.providers.google.cloud.operators.bigquery import ( BigQueryCreateEmptyDatasetOperator, BigQueryInsertJobOperator, ) PROJECT_ID = "{{ dag_run.conf.get('project_id', params.project_id) }}" DATASET_ID = "{{ dag_run.conf.get('dataset_id', params.dataset_id) }}" GCP_CONN_ID = "{{ dag_run.conf.get('gcp_conn_id', params.gcp_conn_id) }}" LOCATION = os.environ.get("BQ_OPENLINEAGE_LOCATION", "US") DAG_ID = "reproduce_issue_47579_bigquery_openlineage" ROUTINE_ID = "issue_47579_proc" INPUT_TABLE_1 = f"`{PROJECT_ID}.{DATASET_ID}.input_table1`" INPUT_TABLE_2 = f"`{PROJECT_ID}.{DATASET_ID}.input_table2`" OUTPUT_TABLE_1 = f"`{PROJECT_ID}.{DATASET_ID}.output_table1`" OUTPUT_TABLE_2 = f"`{PROJECT_ID}.{DATASET_ID}.output_table2`" ROUTINE = f"`{PROJECT_ID}.{DATASET_ID}.{ROUTINE_ID}`" with DAG( dag_id=DAG_ID, schedule=None, start_date=datetime(2026, 7, 1), catchup=False, max_active_runs=1, tags=["repro", "bigquery", "openlineage"], params={ "project_id": os.environ.get("BQ_OPENLINEAGE_PROJECT_ID", "SET_ME"), "dataset_id": os.environ.get("BQ_OPENLINEAGE_DATASET_ID", "issue_47579_openlineage_repro"), "gcp_conn_id": os.environ.get("BQ_OPENLINEAGE_CONN_ID", "google_cloud_default"), }, ) as dag: create_dataset = BigQueryCreateEmptyDatasetOperator( task_id="create_dataset", project_id=PROJECT_ID, dataset_id=DATASET_ID, location=LOCATION, gcp_conn_id=GCP_CONN_ID, if_exists="ignore", ) create_input_table1 = BigQueryInsertJobOperator( task_id="create_input_table1", project_id=PROJECT_ID, location=LOCATION, gcp_conn_id=GCP_CONN_ID, configuration={ "query": { "query": f""" CREATE OR REPLACE TABLE {INPUT_TABLE_1} AS SELECT 1 AS id, 'only_from_input_table1' AS payload """, "useLegacySql": False, } }, ) create_input_table2 = BigQueryInsertJobOperator( task_id="create_input_table2", project_id=PROJECT_ID, location=LOCATION, gcp_conn_id=GCP_CONN_ID, configuration={ "query": { "query": f""" CREATE OR REPLACE TABLE {INPUT_TABLE_2} AS SELECT 2 AS id, 'only_from_input_table2' AS payload """, "useLegacySql": False, } }, ) create_stored_procedure = BigQueryInsertJobOperator( task_id="create_stored_procedure", project_id=PROJECT_ID, location=LOCATION, gcp_conn_id=GCP_CONN_ID, configuration={ "query": { "query": f""" CREATE OR REPLACE PROCEDURE {ROUTINE}() BEGIN CREATE OR REPLACE TABLE {OUTPUT_TABLE_1} AS SELECT id, payload, 'first_ctas' AS marker FROM {INPUT_TABLE_1}; CREATE OR REPLACE TABLE {OUTPUT_TABLE_2} AS SELECT id, payload, 'second_ctas' AS marker FROM {INPUT_TABLE_2}; END """, "useLegacySql": False, } }, ) call_stored_procedure = BigQueryInsertJobOperator( task_id="call_stored_procedure", project_id=PROJECT_ID, location=LOCATION, gcp_conn_id=GCP_CONN_ID, configuration={ "query": { "query": f"CALL {ROUTINE}();", "useLegacySql": False, "priority": "INTERACTIVE", } }, ) create_dataset >> [create_input_table1, create_input_table2] >> create_stored_procedure create_stored_procedure >> call_stored_procedure ``` --- ##### Was generative AI tooling used to co-author this PR? <!-- If generative AI tooling has been used in the process of authoring this PR, please change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by". --> - [X] Yes (please specify the tool below) Generated-by: Claude Opus 4.8 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
