FrankYang0529 opened a new pull request, #73247: URL: https://github.com/apache/airflow/pull/73247
## Why - Asset-triggered Dag runs emit a `jobDependencies` facet. For each asset event that triggered the Dag run, the facet lists the task and the OpenLineage run ID of the try that emitted the asset event. The run ID is built from the try number. - `_extract_ol_info_from_asset_event` takes the try number from `AssetEvent.source_task_instance`. That relationship looks up the `task_instance` row by Dag, run, task and map index. The row only holds the latest try, and the asset event does not record which try emitted it. - The facet is rebuilt for the START, COMPLETE and FAIL events of the Dag run. For example, try 1 of a task emits an asset event, and that asset event triggers a Dag run. Before that Dag run finishes, the task is cleared and scheduled again as try 2. From then on, every event of that Dag run points at try 2, even though try 1 emitted the asset event. If the Dag run is already running when the task is cleared, its START event points at try 1 and its COMPLETE event points at try 2. ## How - `DagRun.schedule_tis` sets `try_number` and `scheduled_dttm` in the same update. - If the source task instance was scheduled at or before the event, its current try emitted the event. In that case the run ID is unchanged and no extra query runs. - If the source task instance was scheduled after the event, a new helper looks up the highest try in `TaskInstanceHistory` that was scheduled at or before the event. ## Verification - Unit test: `TZ=UTC uv run --frozen --project providers/openlineage pytest providers/openlineage/tests/unit` - Integration test: 1. Setup ```sh uv sync --frozen --no-dev --package apache-airflow-providers-openlineage --package apache-airflow-providers-standard npx -y [email protected] -C airflow-core/src/airflow/ui install --frozen-lockfile npx -y [email protected] -C airflow-core/src/airflow/ui build mkdir -p /tmp/ol-job-dependencies-demo/dags cat > /tmp/ol-job-dependencies-demo/dags/orders.py <<'EOF' import time from pathlib import Path import pendulum from airflow.sdk import DAG, Asset, task orders = Asset("demo://warehouse/orders") with DAG( dag_id="orders_producer", schedule=None, start_date=pendulum.datetime(2025, 1, 1, tz="UTC"), is_paused_upon_creation=False, ): @task(outlets=[orders]) def build_orders(): return "orders table rebuilt" build_orders() with DAG( dag_id="orders_report", schedule=[orders], start_date=pendulum.datetime(2025, 1, 1, tz="UTC"), is_paused_upon_creation=False, ): @task def make_report(): while not Path("/tmp/ol-job-dependencies-demo/finish").exists(): time.sleep(1) make_report() EOF ``` ```sh cat > /tmp/ol-job-dependencies-demo/airflow.env <<'EOF' AIRFLOW_HOME=/tmp/ol-job-dependencies-demo/airflow-home AIRFLOW__CORE__DAGS_FOLDER=/tmp/ol-job-dependencies-demo/dags AIRFLOW__CORE__LOAD_EXAMPLES=False AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS=True AIRFLOW__OPENLINEAGE__TRANSPORT='{"type": "file", "log_file_path": "/tmp/ol-job-dependencies-demo/airflow-home/ol_events.jsonl", "append": true}' EOF ``` 2. Start Airflow ```sh set -a; . /tmp/ol-job-dependencies-demo/airflow.env; PATH="$PWD/.venv/bin:$PATH"; exec airflow standalone ``` 3. Run the Dag Trigger the Dag: ```sh curl -s -X POST "http://localhost:8080/api/v2/dags/orders_producer/dagRuns"; -H 'Content-Type: application/json' -d '{"logical_date": null}' | jq -c '{dag_run_id, state}' ``` Check the state of `build_orders`: ```sh curl -s "http://localhost:8080/api/v2/dags/orders_producer/dagRuns/~/taskInstances"; | jq -c '.task_instances[] | {task_id, try_number, state}' ``` Clear `build_orders` once it is `success`: ```sh curl -s -X POST "http://localhost:8080/api/v2/dags/orders_producer/clearTaskInstances"; -H 'Content-Type: application/json' -d '{"dry_run": false, "only_failed": false, "task_ids": ["build_orders"]}' | jq -c '.task_instances[] | {task_id, state}' ``` Let `orders_report` finish: ```sh touch /tmp/ol-job-dependencies-demo/finish ``` Check both `orders_report` runs are `success`: ```sh curl -s "http://localhost:8080/api/v2/dags/orders_report/dagRuns"; | jq -c '.dag_runs[] | {dag_run_id, state}' ``` 4. Check result ```sh jq -r 'select(.job.name == "orders_producer.build_orders" and .eventType == "COMPLETE") | "build_orders try \(.run.facets.airflow.taskInstance.try_number): run \(.run.runId)"' /tmp/ol-job-dependencies-demo/airflow-home/ol_events.jsonl jq -r 'select(.job.name == "orders_report" and .eventType == "COMPLETE") | .run.facets.jobDependencies.upstream[] | "asset event \(.airflow.asset_events[0].asset_event_id): upstream run \(.run.runId)"' /tmp/ol-job-dependencies-demo/airflow-home/ol_events.jsonl | sort ``` On `main`, asset event 1 shows the run ID of try 2: ```text build_orders try 1: run 01a0aa0b-f137-7d33-b737-24e810559276 build_orders try 2: run 01a0aa0b-f137-7f1d-975b-87e1216ea322 ``` ```text asset event 1: upstream run 01a0aa0b-f137-7f1d-975b-87e1216ea322 asset event 2: upstream run 01a0aa0b-f137-7f1d-975b-87e1216ea322 ``` On this branch, asset event 1 shows the run ID of try 1: ```text build_orders try 1: run 01a0a9d4-61ca-7dc3-b577-be0c3728e983 build_orders try 2: run 01a0a9d4-61ca-7716-92a2-1a0f5372c4d4 ``` ```text asset event 1: upstream run 01a0a9d4-61ca-7dc3-b577-be0c3728e983 asset event 2: upstream run 01a0a9d4-61ca-7716-92a2-1a0f5372c4d4 ``` <!-- SPDX-License-Identifier: Apache-2.0 https://www.apache.org/licenses/LICENSE-2.0 --> <!-- Thank you for contributing! Please provide above a brief description of the changes made in this pull request. Write a good git commit message following this guide: https://chris.beams.io/posts/git-commit/ Please make sure that your code changes are covered with tests. And in case of new features or big changes remember to adjust the documentation. For user-facing UI changes, please attach before/after screenshots (or a short screen recording) so reviewers can assess the visual impact. Feel free to ping (in general) for the review if you do not see reaction for a few days (72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications. In case of an existing issue, reference it using one of the following: * closes: #ISSUE * related: #ISSUE --> --- ##### 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 - Claude Code <!-- Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) --> --- * Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged. * For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed. * When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x). * For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number. -- 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]
