jason810496 commented on code in PR #66854:
URL: https://github.com/apache/airflow/pull/66854#discussion_r3610077828
##########
airflow-core/src/airflow/models/asset.py:
##########
@@ -788,6 +789,45 @@ def __repr__(self):
return f"{self.__class__.__name__}({', '.join(args)})"
+class AssetEventQueue(Base):
+ """
+ Pending asset-event registrations awaiting processing by the scheduler.
+
+ A row is enqueued on the task-success path in the execution API when
synchronous
+ asset-event registration is skipped or fails (for example under database
lock
+ contention). The scheduler drains this table, (re)runs
+
:meth:`~airflow.models.taskinstance.TaskInstance.register_asset_changes_in_db`
to
+ create the ``AssetEvent`` and ``AssetDagRunQueue`` rows, and deletes the
queue row
+ once that write commits. This keeps the long-running asset-registration
work off the
+ synchronous ``ti_update_state`` request and guarantees the events are
eventually
+ delivered instead of being dropped when the write cannot complete inline.
+
+ ``ti_id`` is the primary key: at most one pending registration exists per
task
+ instance, and the row is cascade-deleted if the task instance is removed.
+ """
+
+ ti_id: Mapped[UUID] = mapped_column(sa.Uuid(), primary_key=True,
nullable=False)
+ task_outlets: Mapped[list] = mapped_column(sa.JSON(), nullable=False,
default=list)
+ outlet_events: Mapped[list] = mapped_column(sa.JSON(), nullable=False,
default=list)
Review Comment:
Additionally, the `AssetEventQueue` table will only be used for the further
async Asset Event records ingestion purpose. I don't need we need the
`task_outlets` and `outlet_events` columns at all.
I'd like to collapse the both `task_outlets` and `outlet_events` into a
single `JSON` type `payload` or `assets_profiles` column (or other better
naming) to improve the write speed. Or even just `text` column? Need double
check the write path for the append-only purpose among the DB backend we
supported then use the appropriate column type for the durable buffer purpose.
The time we need to access the `AssetEventQueue` records, we will need all
the `task_outlets` and `outlet_events` anyway and the scheduler side also
validate the integrity again, so IMO it's fine to collapse the columns.
--
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]
