codeant-ai-for-open-source[bot] commented on code in PR #42919:
URL: https://github.com/apache/superset/pull/42919#discussion_r3820909182
##########
superset/commands/importers/v1/utils.py:
##########
@@ -318,28 +318,29 @@ def import_tag(
for tag_name in target_tag_names:
try:
- tag = existing_tags.get(tag_name)
-
- # If tag does not exist, create it
- if tag is None:
- description = tag_descriptions.get(tag_name, None)
- tag = Tag(name=tag_name, description=description,
type="custom")
- db_session.add(tag)
- existing_tags[tag_name] = tag # Update the existing_tags
dictionary
-
- # Ensure the association with the object
- tagged_object = (
- db_session.query(TaggedObject)
- .filter_by(object_id=object_id, object_type=object_type,
tag_id=tag.id)
- .first()
- )
- if not tagged_object:
- new_tagged_object = TaggedObject(
- tag_id=tag.id, object_id=object_id, object_type=object_type
+ with db_session.begin_nested():
+ tag = existing_tags.get(tag_name)
+
+ # If tag does not exist, create it
+ if tag is None:
+ description = tag_descriptions.get(tag_name, None)
+ tag = Tag(name=tag_name, description=description,
type="custom")
+ db_session.add(tag)
+ existing_tags[tag_name] = tag # Update the existing_tags
dict
Review Comment:
Yes, this is a valid race. After the savepoint rollback,
`existing_tags[tag_name]` still references the rolled-back `Tag` instance, so
the next iteration can skip the association.
On a uniqueness error, discard the stale instance, re-query the committed
`Tag`, update `existing_tags`, and retry the association in a new savepoint.
This should be limited to an integrity/unique-violation error rather than every
`SQLAlchemyError`.
For example:
```python
except IntegrityError as err:
logger.error(
"Error processing tag '%s' for %s ID %d: %s",
tag_name,
object_type,
object_id,
err,
)
# The failed savepoint may have rolled back a
concurrently-created Tag.
existing_tags.pop(tag_name, None)
tag = (
db_session.query(Tag)
.filter(Tag.name == tag_name)
.one_or_none()
)
if tag is None:
continue
existing_tags[tag_name] = tag
try:
with db_session.begin_nested():
if not (
db_session.query(TaggedObject)
.filter_by(
object_id=object_id,
object_type=object_type,
tag_id=tag.id,
)
.first()
):
db_session.add(
TaggedObject(
tag_id=tag.id,
object_id=object_id,
object_type=object_type,
)
)
new_tag_ids.append(tag.id)
except SQLAlchemyError:
continue
```
The exact exception check should match Superset’s existing database-dialect
utilities for unique violations. This preserves the savepoint recovery behavior
while ensuring the losing import reuses the winning transaction’s `Tag` row and
creates the requested association.
##########
superset/commands/importers/v1/utils.py:
##########
@@ -318,28 +318,29 @@ def import_tag(
for tag_name in target_tag_names:
try:
- tag = existing_tags.get(tag_name)
-
- # If tag does not exist, create it
- if tag is None:
- description = tag_descriptions.get(tag_name, None)
- tag = Tag(name=tag_name, description=description,
type="custom")
- db_session.add(tag)
- existing_tags[tag_name] = tag # Update the existing_tags
dictionary
-
- # Ensure the association with the object
- tagged_object = (
- db_session.query(TaggedObject)
- .filter_by(object_id=object_id, object_type=object_type,
tag_id=tag.id)
- .first()
- )
- if not tagged_object:
- new_tagged_object = TaggedObject(
- tag_id=tag.id, object_id=object_id, object_type=object_type
+ with db_session.begin_nested():
+ tag = existing_tags.get(tag_name)
+
+ # If tag does not exist, create it
+ if tag is None:
+ description = tag_descriptions.get(tag_name, None)
+ tag = Tag(name=tag_name, description=description,
type="custom")
+ db_session.add(tag)
+ existing_tags[tag_name] = tag # Update the existing_tags
dict
+
+ # Ensure the association with the object
+ tagged_object = (
+ db_session.query(TaggedObject)
+ .filter_by(object_id=object_id, object_type=object_type,
tag_id=tag.id)
+ .first()
)
- db_session.add(new_tagged_object)
+ if not tagged_object:
+ new_tagged_object = TaggedObject(
+ tag_id=tag.id, object_id=object_id,
object_type=object_type
+ )
+ db_session.add(new_tagged_object)
- new_tag_ids.append(tag.id)
+ new_tag_ids.append(tag.id)
Review Comment:
Yes. The append should occur after the `begin_nested()` context exits
successfully, so any flush or SAVEPOINT release failure is handled before
recording the tag as imported:
```python
try:
with db_session.begin_nested():
tag = existing_tags.get(tag_name)
if tag is None:
description = tag_descriptions.get(tag_name)
tag = Tag(name=tag_name, description=description,
type="custom")
db_session.add(tag)
existing_tags[tag_name] = tag
tagged_object = (
db_session.query(TaggedObject)
.filter_by(
object_id=object_id,
object_type=object_type,
tag_id=tag.id,
)
.first()
)
if not tagged_object:
db_session.add(
TaggedObject(
tag_id=tag.id,
object_id=object_id,
object_type=object_type,
)
)
# Only record success after the SAVEPOINT was released.
new_tag_ids.append(tag.id)
except SQLAlchemyError as err:
logger.error(...)
continue
```
This ensures a failed flush or SAVEPOINT release cannot leave a stale ID in
`new_tag_ids`, so both the return value and subsequent cleanup reflect only
successfully processed tags.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
