1fanwang opened a new pull request, #71751: URL: https://github.com/apache/airflow/pull/71751
# Rationale for this change Follow-up to https://github.com/apache/airflow/pull/71387, where I said I would replace this once the accessor API landed. Triggers are deduplicated by `hash(classpath, kwargs)` while `asset_watcher` is many-to-many, so several assets can watch one trigger and the triggerer hands it one accessor per asset. The merged trigger detects that by matching the text of a `ValueError` raised by a private helper: ```python if _MULTI_ASSET_ERROR not in str(err): raise ``` Two problems. It couples to a private method's message, and having detected the case it gives up on the watermark entirely, so a triggerer restart is free to re-emit the current head and schedule a duplicate DAG run for a commit already reported. https://github.com/apache/airflow/pull/71460 makes the accessors iterable, so the trigger can address each asset instead of guessing from an exception. # What changes are included in this PR? Write the snapshot to every accessor, and resume from the oldest of their watermarks so a write that reached only some assets repeats a snapshot rather than skips one. The error-string match is gone, so a state store failure propagates rather than being read as the several-assets case. `list(store)` falls back to the single-accessor form on an Airflow that predates #71460, so the provider keeps working across the versions it supports. # Are these changes tested? Unit tests cover a single asset, several assets each getting their own cursor, resuming from the oldest watermark, and a backend failure propagating. All five watermark tests fail against the current `main`: <details><summary>Unit tests</summary> ``` $ pytest providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py -q 20 passed $ git checkout origin/main -- providers/apache/iceberg/src/airflow/providers/apache/iceberg/triggers/iceberg.py $ pytest providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py -q 5 failed, 15 passed ``` </details> I also ran the trigger against the real `AssetStateStoreAccessors` rather than a stand-in, restarting it the way the triggerer does by rebuilding from the serialized row, and separately against an Airflow that has no iterable accessors: <details><summary>Against the real accessors, and against the older API</summary> ``` == two assets sharing one trigger, real accessors == len(store) = 2 1. cold start : 1 event(s) snapshot=111 rows written : {('orders_raw', 'snapshot_id'): 111, ('orders_curated', 'snapshot_id'): 111} 2. after restart : 0 event(s) (want 0, was 1 before this change) 3. partial write : 1 event(s) (repeats rather than skipping) == single asset on an Airflow without the new API == 1. cold start (old Airflow): 1 event(s), watermark=111 2. after restart : 0 event(s) (want 0) 3. no store at all : 1 event(s) (want 1) ``` Line 2 of the first block is the fix: the restart used to emit a duplicate. Line 3 shows a cursor that reached only one asset repeating a snapshot instead of skipping one. The second block is the same trigger on an Airflow without the new API, where it still keeps a single-asset watermark. </details> -- 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]
