jedcunningham commented on code in PR #71936:
URL: https://github.com/apache/airflow/pull/71936#discussion_r3857228581
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -923,8 +941,21 @@ def activate_assets_if_possible(self, models:
Iterable[AssetModel], *, session:
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
stmt = sqlite_insert(AssetActive).on_conflict_do_nothing()
- if values := [{"name": m.name, "uri": m.uri} for m in models]:
- session.execute(stmt, values)
+ # Of two assets sharing a name or a uri only the one offered first is
activated. Choosing
+ # that here rather than leaving it to the insert means the rows can
then be sorted, which
+ # they must be: ``asset_active`` is unique on both columns, so two
writers inserting them
+ # in opposite orders deadlock on the index.
+ claimed_names: set[str] = set()
+ claimed_uris: set[str] = set()
+ values = []
+ for model in models:
+ if model.name in claimed_names or model.uri in claimed_uris:
+ continue
Review Comment:
These two sets only know about the current batch, so an asset that's going
to get rejected by the insert anyway can still grab the claim and knock out one
that would've made it in.
Say asset_active already has `("n9", "s3://u1")`, and this batch offers
`("n1", "s3://u1")` then `("n1", "s3://u2")`. Before, the second one went in as
its name and uri were both free. Now the first grabs n1, the second gets
skipped, and then the first's insert conflicts anyway, so we end up with
neither.
The scheduler's `_activate_referenced_assets` does it the other way round -
it looks at what's already active and skips before recording the claim. Seeding
these two sets from the existing asset_active rows would match that, and the
sorted insert still works.
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -850,6 +855,10 @@ def collect(cls, dags: dict[str, LazyDeserializedDAG]) ->
Self:
dag_id: list(_get_dag_assets(dag, SerializedAsset,
inlets=False, outlets=True))
for dag_id, dag in dags.items()
},
+ # Left in the order the Dags define them: that order decides which
of two assets
Review Comment:
The read-back in sync_assets isn't ordered, so this only holds when all the
assets are new. Anything already in asset from a previous parse comes back in
DB order and gets offered first. Same as before this PR, so not a regression,
but the comment and the new test both read as though collection order always
wins, and the test clears assets each time so it only ever hits the all-new
case.
If you want the claim to actually hold, rebuilding the dict in self.assets
order at the end would do it (`return {key: orm_assets[key] for key in
self.assets if key in orm_assets}`), and then the test could cover the
existing-asset half too. Happy to leave it if you'd rather keep the scope
tight, in that case maybe just narrow the comment.
--
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]
