1fanwang opened a new pull request, #71460: URL: https://github.com/apache/airflow/pull/71460
# Rationale for this change The documented watcher-trigger example raises when two assets watch the same trigger. Triggers are deduplicated by `hash(classpath, kwargs)` while `asset_watcher` is many-to-many, so two assets watching the same source with the same arguments share one trigger row. The triggerer then hands that trigger an accessor per asset, and the shorthand the docs tell it to use refuses to guess between them: ``` ValueError: Task has 2 concrete inlets and outlets — use context['asset_state_store'][MY_ASSET] to specify which ``` A trigger cannot take that advice. It never names its assets, and `AssetStateStoreAccessors` keeps them privately, so there is no way to subscript and no way to ask how many there are. The only signal is the `ValueError` itself, raised by a private helper, and a state store backend raises `ValueError` from `deserialize_asset_state_store_from_ref` on the same call, so catching it can swallow a real failure and silently drop the state the trigger was keeping. `__len__` answers the question directly. `__iter__` lets the trigger hold its cursor on every asset it serves and stay correct across a restart, instead of giving up on per asset state. Iteration yields the accessors, since the class keeps names and uris rather than the `Asset` objects `__getitem__` takes. The docs promised the shorthand always applies to a watcher, so this corrects that and shows the shared form. related: https://github.com/apache/airflow/pull/71354 makes these accessors async and leaves `_single_accessor` in place, so it is unaffected either way. It edits the same watcher section, and I am happy to rebase behind it. The first consumer is a snapshot trigger in the Iceberg provider, https://github.com/apache/airflow/pull/71387. # Are these changes tested? Six unit tests beside the existing accessor tests: `len` over one asset, two, none, and a uri ref beside a name; iteration yielding every accessor, repeatable, and empty with no assets. All six fail on `main`. <details><summary>Unit tests</summary> ``` $ pytest task-sdk/tests/task_sdk/execution_time/test_context.py -k AssetStateStoreAccessors -q 22 passed, 154 deselected $ git checkout origin/main -- task-sdk/src/airflow/sdk/execution_time/context.py $ pytest task-sdk/tests/task_sdk/execution_time/test_context.py -k AssetStateStoreAccessors -q 6 failed, 16 passed, 154 deselected ``` </details> I also ran the trigger from the docs, unchanged, against real accessors, faking only the supervisor round-trip. Cases 1 and 2 are the published example; 3 and 4 are the form this PR documents: <details><summary>The documented trigger, one asset and two</summary> ``` 1. one asset watching : emitted {'status': 'success', 'record_id': 'record-1'} 2. two assets watching: ValueError: Task has 2 concrete inlets and outlets — use context['asset_state_store'][MY_ASSET] to s 3. one asset, new form : emitted {'status': 'success', 'record_id': 'record-1'} 4. two assets, new form: emitted {'status': 'success', 'record_id': 'record-1'} rows written : {('orders_api', 'last_seen_id'): 'record-1', ('orders_raw', 'last_seen_id'): 'record-1', ('orders_curated', 'last_seen_id'): 'record-1'} ``` Case 4 writes the cursor to both assets, so a restart resumes rather than replaying. The read takes the oldest, so a write that reaches only some assets repeats a record instead of skipping one. </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]
