Emit per-statement OpenLineage events for BigQuery script jobs

[nailo2c]

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).

• Open Lineage UI shows input_table1 -> query.1 -> output_table1

Before

One COMPLETE event, child datasets flattened together, looks many-to-many:

{"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:

// 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

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?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.8 following the guidelines

[kacpermuda]

Left few comments, thanks for taking a look at this !

[nailo2c]

Thanks for the review! I believe this version is better than before :)

[kacpermuda]

Looks good, thanks. One more important change around import error catching, no other blocking issues. Please try running this code with older OpenLineage provider version to make sure it works (this import). You can use breeze release-management prepare-provider-distributions command to generate google provider wheel and install it somewhere where you can test it.

[nailo2c]

Thanks for the suggestion. I built the Google provider wheel from this branch and installed it with apache-airflow-providers-openlineage==2.15.0 and test it. Everything looks good :)