Lee-W commented on code in PR #70972:
URL: https://github.com/apache/airflow/pull/70972#discussion_r3711138731
##########
airflow-core/docs/howto/usage-cli.rst:
##########
@@ -223,6 +223,12 @@ The ``db clean`` command works by deleting from each table
the records older tha
You can optionally provide a list of tables to perform deletes on. If no list
of tables is supplied, all tables will be included.
+.. note::
+
+ Cleaning the ``asset_event`` table also drops any
queued-but-not-yet-consumed asset events. This mainly
+ affects DAGs waiting on a multi-asset condition, where a pending event can
be purged before the condition
+ is met, meaning the DAG will not be triggered by it.
Review Comment:
```suggestion
affects Dags waiting on a multi-asset condition, where a pending event can
be purged before the condition
is met, meaning the Dag will not be triggered by it.
```
##########
airflow-core/tests/unit/assets/test_manager.py:
##########
@@ -212,23 +212,43 @@ def test_register_asset_change_no_downstreams(self,
session, mock_task_instance)
)
assert
session.scalar(select(func.count()).select_from(AssetDagRunQueue)) == 0
- @pytest.mark.parametrize(
- ("dialect_name", "expected_helper"),
- [
- ("postgresql", "_queue_dagruns_nonpartitioned_conflict_update"),
- ("mysql", "_queue_dagruns_nonpartitioned_mysql"),
- ("sqlite", "_queue_dagruns_nonpartitioned_conflict_update"),
- ],
- )
- def test_queue_dagruns_routes_by_dialect(self, dialect_name,
expected_helper):
- """Test that _queue_dagruns routes to the dialect-appropriate queue
helper."""
+ def test_register_asset_change_is_atomic_on_caller_session(self, session,
mock_task_instance):
+ """The AssetEvent is written on the caller's session: visible before
commit, gone after rollback.
+
+ Under the old side-session behaviour the event was committed
independently and would be
+ orphaned if the caller's transaction rolled back. Registration is now
atomic with the caller.
+ """
+ asset_manager = AssetManager()
+
+ asset = Asset(uri="test://atomic1", name="atomic_asset")
+ asm = AssetModel(uri="test://atomic1/", name="atomic_asset",
group="asset")
+ session.add(asm)
+ session.execute(delete(AssetDagRunQueue))
Review Comment:
why do we need to delete ADRQ here? should we make a
`clear_adrq`/`reset_adrq` function instead if there's ADRQ leftover?
##########
airflow-core/newsfragments/70972.significant.rst:
##########
@@ -0,0 +1,13 @@
+Changes to how pending asset events trigger DAG runs
+
+Deleting asset event records with ``airflow db clean`` now also removes any
+asset events that are still pending for a downstream DAG. This mainly affects
+DAGs waiting on a multi-asset condition, where a pending event can be purged
+before the condition is met. Deleting an asset event now consistently means it
+will not trigger a run; previously such a pending trigger could still fire
after
+the event had been purged.
+
+Relatedly, an asset event that was queued for a DAG while it was scheduled on
+that asset is still used to trigger the run, and appears among the run's
+triggering events, even if the asset is later removed from the DAG's schedule.
+Previously, they were silently dropped.
Review Comment:
```suggestion
asset events that are still pending for a downstream Dag. This mainly affects
Dags waiting on a multi-asset condition, where a pending event can be purged
before the condition is met. Deleting an asset event now consistently means
it
will not trigger a run; previously such a pending trigger could still fire
after
the event had been purged.
Relatedly, an asset event that was queued for a Dag while it was scheduled on
that asset is still used to trigger the run, and appears among the run's
triggering events, even if the asset is later removed from the Dag's
schedule.
Previously, they were silently dropped.
```
##########
airflow-core/newsfragments/70972.significant.rst:
##########
@@ -0,0 +1,13 @@
+Changes to how pending asset events trigger DAG runs
Review Comment:
```suggestion
Changes to how pending asset events trigger Dag runs
```
--
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]
