You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ashb (via GitHub)" <gi...@apache.org> on 2026/10/06 15:06:38 UTC
[PR] Attribute log, asset-event and asset-state writes to the task attempt [airflow]
ashb opened a new pull request, #74347:
URL: https://github.com/apache/airflow/pull/74347
<!-- 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".
-->
- [ ] Yes (please specify the tool below)
<!--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Attribute log, asset-event and asset-state writes to the task attempt [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74347:
URL: https://github.com/apache/airflow/pull/74347#discussion_r4198202221
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/event_logs.py:
##########
@@ -16,14 +16,27 @@
# under the License.
from __future__ import annotations
-from sqlalchemy import inspect
+from sqlalchemy import case, inspect, select
from sqlalchemy.orm.attributes import set_committed_value
from airflow.api_fastapi.core_api.datamodels.event_logs import EventLogResponse
from airflow.models import Log
+from airflow.models.task_coordinates import public_map_index_expression
+from airflow.models.taskinstance import TaskInstance
-def event_log_to_response(event_log: Log) -> EventLogResponse:
+def event_log_public_map_index():
+ """Resolve an attributed execution's public index without guessing once its row has been purged."""
+ attributed = (
+ select(public_map_index_expression(TaskInstance))
Review Comment:
`event_log_public_map_index()` builds its subquery as an ORM select on `TaskInstance`, so `_restrict_to_current_attempts` adds `working_set IS TRUE` to it (the outer statement isn't a PK lookup, and `with_loader_criteria` reaches into scalar subqueries). Once an attempt is archived, which `prepare_db_for_next_try` does on every retry, the subquery finds nothing and the CASE yields NULL. So a plain task that failed try 1 and retried now shows `map_index: null` for try 1's audit rows, and `GET /eventLogs?...&map_index=-1` drops them from both the page and `total_entries`. That's the opposite of the "keeps doing so after that attempt is retired" behaviour in the description, and I'd expect the `archived=True` cases of `test_projects_public_mapping_index_for_exact_execution` to fail on it. Building the subquery against `TaskInstance.__table__` (which the hook leaves alone) should fix it. Setting `include_all_attempts` on the outer select alone wouldn't, because the count wraps the sta
tement in a new select that doesn't carry the option.
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1652,6 +1703,39 @@ def _extract_ol_info_from_asset_event(asset_event: AssetEvent) -> dict[str, str]
return None
+def _get_asset_event_source_regions(events: list[AssetEvent]) -> dict[UUID, UUID]:
+ if not AIRFLOW_V_3_4_PLUS:
+ return {}
+ source_ids = {
+ source_id
+ for event in events
+ if isinstance(source_id := getattr(event, "source_task_instance_id", None), UUID)
+ }
+ if not source_ids:
+ return {}
+
+ from sqlalchemy import select
+ from sqlalchemy.orm import object_session
+
+ from airflow.models.asset import AssetEvent
+ from airflow.utils.session import create_session
+
+ session = None
+ for event in events:
+ if isinstance(event, AssetEvent):
+ session = object_session(event)
+ if session is not None:
+ break
+ with nullcontext(session) if session is not None else create_session() as session:
+ return dict(
+ session.execute(
+ select(TaskInstance.id, TaskInstance.region_id).where(TaskInstance.id.in_(source_ids))
Review Comment:
I don't think this lookup can change the result. It's an ORM select on `TaskInstance` with an `in_()` bind, so the attempt hook filters it to `working_set IS TRUE` and a retried or cleared source never shows up here. For live sources the caller already joinedloads `source_task_instance`, and `get_regional_task_instance_run_id(ti)` returns the same `str(ti.id)`. So in production it's one extra session and query per asset-triggered run event with no effect on the output. `test_cleared_regional_asset_source_keeps_retiring_uuid` passes because the relationship fallback resolves the retired TI from the session identity map, not through this branch. Could we drop `_get_asset_event_source_regions` and the `source_regions` shortcut and rely on the TI branch?
##########
airflow-core/src/airflow/models/log.py:
##########
@@ -73,15 +73,16 @@ class Log(Base):
task_instance: Mapped[TaskInstance | None] = relationship(
"TaskInstance",
viewonly=True,
- foreign_keys=[dag_id, task_id, run_id, map_index],
- primaryjoin="and_(Log.dag_id == TaskInstance.dag_id, Log.task_id == TaskInstance.task_id, Log.run_id == TaskInstance.run_id, Log.map_index == TaskInstance.region_index, TaskInstance.working_set.is_(True))",
+ foreign_keys=[task_instance_id],
+ primaryjoin="Log.task_instance_id == TaskInstance.id",
Review Comment:
With the join now only on `task_instance_id`, every audit row written before the upgrade (the column came in 0142 with no backfill) loses `task_display_name` in `GET /eventLogs`, where 3.3 returned it through the coordinate join. Is that intended? The display name is task-level, so a coordinate fallback for `task_instance_id IS NULL` rows wouldn't run into the wrong-attempt problem the description is avoiding. If it is intended, a newsfragment line would help API users who read that field.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/event_logs.py:
##########
@@ -130,8 +136,14 @@ def get_event_logs(
# Exact match filters (for backward compatibility)
dag_id: Annotated[FilterParam[str | None], Depends(filter_param_factory(Log.dag_id, str | None))],
task_id: Annotated[FilterParam[str | None], Depends(filter_param_factory(Log.task_id, str | None))],
+ task_instance_id: Annotated[
Review Comment:
This adds a `task_instance_id` filter, but `EventLogResponse` doesn't return the field, and the same goes for `source_task_instance_id` on `AssetEventResponse`. Two loop passes of an unmapped task on try 1 come back with identical `task_id`, `map_index=-1` and `try_number=1`, so a client listing events can't tell which pass wrote which row without one request per attempt. `AssetStateStoreLastUpdatedBy` already exposes `task_instance_id` in this PR. Could the other two responses carry it as an optional field too?
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1466,6 +1495,14 @@ def is_dag_run_asset_triggered(
return dag_run.run_type == DagRunType.DATASET_TRIGGERED # type: ignore[attr-defined] # This attr is available on AF2, but mypy can't see it
+def get_regional_task_instance_run_id(task_instance: TaskInstance | RuntimeTaskInstance) -> str | None:
+ if AIRFLOW_V_3_4_PLUS:
+ region_id = getattr(task_instance, "region_id", None)
+ if isinstance(region_id, UUID) and region_id.int != 0:
Review Comment:
Since mapped expansions got their own region, this returns the TI UUID for every plain `.expand()` task too, not only loop work. Their OpenLineage run ids move from the deterministic `build_task_instance_run_id` value to `str(ti.id)` whenever the provider runs on a core that has regions, even though dag, task, try, logical date and map index already identified them uniquely. Is that intended? If so, it's a visible change for lineage consumers that reproduce run ids and probably deserves a provider changelog note. If not, the check would need to know whether the region sits inside a loop.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/event_logs.py:
##########
@@ -204,7 +216,9 @@ def get_event_logs(
# that bypass Log.__init__ (which always sets dttm = timezone.utcnow()).
# Making EventLogResponse.when nullable would be a breaking API contract change for
# clients that currently rely on `when` always being present.
- select(Log).where(Log.dttm.is_not(None)).options(*_eager_load_display_names())
+ select(Log, event_log_public_map_index())
Review Comment:
Putting `event_log_public_map_index()` in the paginated statement means the count query (`select(count()).select_from(stmt.subquery())`) carries the correlated subquery too. Postgres and SQLite should flatten it away, but MySQL won't merge a derived table that has a subquery in its select list, so it materializes it and runs the TI lookup for every attributed log row on each audit-log page load. Adding the column after `paginated_select` (`event_logs_select.add_columns(...)`) would keep the count as cheap as before. I haven't run EXPLAIN on MySQL, so worth a quick check.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Attribute log, asset-event and asset-state writes to the task attempt [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74347:
URL: https://github.com/apache/airflow/pull/74347#discussion_r4222267215
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/event_logs.py:
##########
@@ -16,14 +16,29 @@
# under the License.
from __future__ import annotations
-from sqlalchemy import inspect
+from sqlalchemy import case, inspect, select
from sqlalchemy.orm.attributes import set_committed_value
from airflow.api_fastapi.core_api.datamodels.event_logs import EventLogResponse
from airflow.models import Log
+from airflow.models.task_coordinates import public_map_index_expression
+from airflow.models.taskinstance import TaskInstance
-def event_log_to_response(event_log: Log) -> EventLogResponse:
+def event_log_public_map_index():
+ """Resolve an attributed execution's public index without guessing once its row has been purged."""
+ # The Table, unlike the ORM entity, is not narrowed to working_set rows, so archived tries still resolve.
+ task_instances = TaskInstance.__table__
+ attributed = (
+ select(public_map_index_expression(task_instances))
+ .where(task_instances.c.id == Log.task_instance_id)
+ .correlate(Log)
+ .scalar_subquery()
+ )
+ return case((Log.task_instance_id.is_(None), Log.map_index), else_=attributed).label("public_map_index")
Review Comment:
Asset events store the public map index when they're written (`assets/manager.py:344-352`), but event logs keep the stored index and resolve the public one here at read time. `db clean` purges `task_instance` by `start_date` and `log` by `dttm`, so a log row can outlive its attempt and then returns `map_index: null`, even for a plain unmapped task whose stored -1 was already right. It also puts this CASE in the `map_index` filter, where it can't use `idx_log_task_instance`. Would writing the public index onto `Log` when the row is created, the way asset events do, work here?
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/event_logs.py:
##########
@@ -130,8 +136,14 @@ def get_event_logs(
# Exact match filters (for backward compatibility)
dag_id: Annotated[FilterParam[str | None], Depends(filter_param_factory(Log.dag_id, str | None))],
task_id: Annotated[FilterParam[str | None], Depends(filter_param_factory(Log.task_id, str | None))],
+ task_instance_id: Annotated[
Review Comment:
The list endpoints filter on `task_instance_id` here and `source_task_instance_id` in `assets.py`, but `EventLogResponse` and `AssetEventResponse` still don't return either field. Two loop passes of an unmapped task on try 1 come back with the same `task_id`, `map_index=-1` and `try_number=1`, so a client listing events can't tell which pass wrote a row without one request per attempt. Could both carry it as an optional field, like `AssetStateStoreLastUpdatedBy` does?
##########
airflow-core/src/airflow/models/log.py:
##########
@@ -73,15 +73,16 @@ class Log(Base):
task_instance: Mapped[TaskInstance | None] = relationship(
"TaskInstance",
viewonly=True,
- foreign_keys=[dag_id, task_id, run_id, map_index],
- primaryjoin="and_(Log.dag_id == TaskInstance.dag_id, Log.task_id == TaskInstance.task_id, Log.run_id == TaskInstance.run_id, Log.map_index == TaskInstance.region_index, TaskInstance.working_set.is_(True))",
+ foreign_keys=[task_instance_id],
+ primaryjoin="Log.task_instance_id == TaskInstance.id",
Review Comment:
`task_display_name` still resolves only through `task_instance_id`, so any row without it comes back with no display name, where 3.3 filled it from the coordinates. That's not only rows written before 0142 (which adds the column without a backfill). `BaseExecutor.log_task_event` sets the id only for `TaskInstanceUuid` keys, and the released Celery and Kubernetes providers don't set `supports_task_instance_uuid`, so their executor events keep landing without it after the upgrade. Since the display name is task-level, a fallback for rows where `task_instance_id` is null that looks it up by `dag_id`/`task_id` would cover both cases.
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1466,6 +1495,14 @@ def is_dag_run_asset_triggered(
return dag_run.run_type == DagRunType.DATASET_TRIGGERED # type: ignore[attr-defined] # This attr is available on AF2, but mypy can't see it
+def get_regional_task_instance_run_id(task_instance: TaskInstance | RuntimeTaskInstance) -> str | None:
+ if AIRFLOW_V_3_4_PLUS:
+ region_id = getattr(task_instance, "region_id", None)
+ if isinstance(region_id, UUID) and region_id.int != 0:
Review Comment:
This still returns `str(task_instance.id)` for any non-sentinel region, and `DagRun` creates a region for every task that needs expansion (`dagrun.py:2198`), so plain `.expand()` tasks still move from the deterministic `build_task_instance_ol_run_id` value to the TI UUID. Lineage consumers that rebuild run ids from dag, task, try, logical date and map index will stop matching those runs, and the provider changelog doesn't mention it. If that isn't intended, a region whose `node_id` is the task itself and that has no `parent_region_id` is a plain expansion and could keep the deterministic id.
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -664,15 +666,40 @@ def get_dag_documentation(dag: DAG | SerializedDAG | None) -> tuple[str | None,
return None, None
+def get_task_instance_map_index(task_instance: TaskInstance | RuntimeTaskInstance) -> int:
+ if AIRFLOW_V_3_4_PLUS and isinstance(task_instance, TaskInstance):
+ if task_instance.region_id.int == 0:
+ return task_instance.region_index
+
+ from sqlalchemy import select
+ from sqlalchemy.orm import object_session
+
+ from airflow.models.dynamic_region import DynamicRegion
Review Comment:
This reimplements core's public map index rule (`public_map_index_expression` in `task_coordinates.py`) against `DynamicRegion`, which isn't part of the public interface, so the two can drift without either test suite noticing. It also runs a query for every regional ORM TI. Could core expose the public map index on the task instance (or as a small public helper) and have the provider call that?
##########
providers/openlineage/tests/unit/openlineage/utils/test_utils.py:
##########
@@ -3833,9 +3842,127 @@ def test_validate_uuid_invalid():
assert is_valid_uuid(uuid_str) is False
+@pytest.mark.db_test
+@pytest.mark.skipif(not AIRFLOW_V_3_4_PLUS, reason="Region projection requires Airflow 3.4")
+@pytest.mark.parametrize("mapped", [False, True])
+def test_orm_lineage_facets_project_public_index(dag_maker, session, mapped):
+ with dag_maker(serialized=True):
+ if mapped:
+ PythonOperator.partial(task_id="work", python_callable=list).expand(op_kwargs=[{}, {}, {}])
+ else:
+ EmptyOperator(task_id="work")
+ run = dag_maker.create_dagrun()
+ if mapped:
+ ti = next(ti for ti in run.task_instances if ti.region_index == 2)
+ else:
+ region = DynamicRegion(dag_id=run.dag_id, run_id=run.run_id, node_id="loop")
+ session.add(region)
+ session.flush()
+ ti = run.task_instances[0]
+ ti.region_id = region.id
+ ti.region_index = 2
+ session.flush()
+
+ with patch.object(DBDagBag, "get_dag", autospec=True) as get_dag:
+ info = dict(TaskInstanceInfo(ti))
+ facets = get_airflow_mapped_task_facet(ti)
+
+ get_dag.assert_not_called()
+ assert info["map_index"] == (2 if mapped else None)
+ if mapped:
+ assert facets["airflow_mappedTask"].mapIndex == 2
+ else:
+ assert facets == {}
+ assert ti in session
+
+
class TestExtractOlInfoFromAssetEvent:
"""Tests for _extract_ol_info_from_asset_event function."""
+ @pytest.mark.skipif(not AIRFLOW_V_3_4_PLUS, reason="Region attribution requires Airflow3.4")
+ @pytest.mark.parametrize("map_index", [-1, 3])
+ def test_regional_source_run_identity_uses_execution_uuid(self, map_index):
+ region_id = uuid4()
+ logical_date = datetime.datetime(2024, 1, 1, tzinfo=datetime.UTC)
+ ti = SimpleNamespace(
+ id=uuid4(),
+ dag_id="source",
+ task_id="task",
+ try_number=1,
+ region_id=region_id,
+ region_index=3,
+ map_index=map_index,
+ )
+ event = SimpleNamespace(
+ source_task_instance_id=ti.id,
+ source_task_instance=ti,
+ source_dag_id="source",
+ source_task_id="task",
+ source_map_index=map_index,
+ source_dag_run=SimpleNamespace(logical_date=logical_date),
+ )
+ result = _extract_ol_info_from_asset_event(event)
+
+ expected = {"job_name": "source.task", "job_namespace": namespace(), "run_id": str(ti.id)}
+ assert result == expected
+
+ @pytest.mark.db_test
+ @pytest.mark.skipif(not AIRFLOW_V_3_4_PLUS, reason="Region attribution requires Airflow3.4")
+ def test_live_regional_asset_sources_need_no_graph_lookup(self, dag_maker, session):
+ with dag_maker():
+ EmptyOperator(task_id="work")
+ run = dag_maker.create_dagrun()
+ root = DynamicRegion(dag_id=run.dag_id, run_id=run.run_id, node_id="mapped")
+ loop = DynamicRegion(dag_id=run.dag_id, run_id=run.run_id, node_id="loop")
+ session.add_all([root, loop])
+ session.flush()
+ nested = DynamicRegion(
+ dag_id=run.dag_id,
+ run_id=run.run_id,
+ node_id="nested",
+ parent_region_id=loop.id,
+ parent_region_index=2,
+ )
+ session.add(nested)
+ session.flush()
+ events = []
+ for region, task_id in [(root, "mapped"), (loop, "work"), (nested, "nested")]:
+ ti = SimpleNamespace(
+ id=uuid4(),
+ dag_id=run.dag_id,
+ task_id=task_id,
+ try_number=1,
+ region_id=region.id,
+ region_index=2,
+ )
+ for index in range(10):
+ events.append(
+ SimpleNamespace(
+ id=len(events),
+ asset_id=index,
+ uri=f"test://asset{index}",
+ extra={},
+ source_task_instance=ti,
+ source_dag_id=ti.dag_id,
+ source_task_id=ti.task_id,
+ source_run_id=run.run_id,
+ source_map_index=-1 if task_id == "work" else 2,
+ source_dag_run=SimpleNamespace(logical_date=run.run_after),
+ )
+ )
+ session.commit()
+
+ with assert_queries_count(0):
+ dependencies = _get_ol_job_dependencies_from_asset_events(events)
+
+ assert len(dependencies) == 3
+ assert {item["job_name"] for item in dependencies if "run_id" in item} == {
Review Comment:
The code under test only checks `region_id.int != 0` on the `SimpleNamespace` TIs, so the `DynamicRegion` rows aren't read, and this assertion only covers job names. A wrong `run_id` (the deterministic one, say) would still pass. Asserting the `(job_name, run_id)` pairs against each TI's `id` would catch that.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Attribute log, asset-event and asset-state writes to the task attempt [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb merged PR #74347:
URL: https://github.com/apache/airflow/pull/74347
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org