yakovlef commented on code in PR #76827:
URL: https://github.com/apache/airflow/pull/76827#discussion_r4549672798
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -112,16 +113,32 @@ def _add_asset_alias_association(
*,
session: Session,
) -> None:
- already_related = {m.name for m in asset_model.aliases}
- existing_aliases = {
- m.name: m
- for m in
session.scalars(select(AssetAliasModel).where(AssetAliasModel.name.in_(alias_names)))
- }
- asset_model.aliases.extend(
- existing_aliases.get(name, AssetAliasModel(name=name))
- for name in alias_names
- if name not in already_related
- )
+ # Another task can link the same alias between our read of
asset_model.aliases and the
+ # insert. The savepoint keeps that conflict from failing the whole
transaction; after a
+ # rollback the aliases are reloaded and only the still-missing links
are added.
+ for attempt in range(1, _ALIAS_ASSOCIATION_ATTEMPTS + 1):
+ already_related = {m.name for m in asset_model.aliases}
+ missing = [name for name in alias_names if name not in
already_related]
+ if not missing:
+ return
+ existing_aliases = {
+ m.name: m
+ for m in
session.scalars(select(AssetAliasModel).where(AssetAliasModel.name.in_(missing)))
+ }
+ try:
+ with session.begin_nested():
+ asset_model.aliases.extend(
+ existing_aliases.get(name, AssetAliasModel(name=name))
for name in missing
Review Comment:
yes, the unique index on name. it goes through the same retry: the
IntegrityError is caught inside the savepoint and the next attempt's select
picks up the alias the other writer committed. added a test for it
--
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]
