1fanwang opened a new pull request, #71387:
URL: https://github.com/apache/airflow/pull/71387
<!-- DRAFT — not opened. Review before filing. -->
# Rationale for this change
The Iceberg provider ships a hook and nothing else, so there is no way to
schedule a DAG on "this table has new data". Anyone who wants it writes their
own trigger.
This adds `IcebergTableSnapshotTrigger`, which polls a table's branch head
through the existing `IcebergHook` and emits an event when it advances, so a
commit can drive an `AssetWatcher`:
```python
orders = Asset(
"orders",
watchers=[AssetWatcher(name="orders_commits",
trigger=IcebergTableSnapshotTrigger(table="sales.orders"))],
)
@dag(schedule=[orders])
def downstream(): ...
```
Polling is what Iceberg supports: a catalog has no commit notification to
subscribe to, and `BaseEventTrigger` names a polling REST API as an intended
case. Triggers sharing a catalog connection, branch and poll interval return
the same `shared_stream_key`, so watching many tables in one catalog is a
single poll rather than one per table.
The event carries `snapshot_id` and `previous_snapshot_id`, so a task can
scan the delta.
**Out of scope, deliberately:** an `iceberg://` asset URI scheme. `iceberg`
is missing from Airflow's registered schemes even though `hive`, `hdfs` and
`trino` are there, but every registered scheme maps to an OpenLineage naming
entry and Iceberg has none (OpenLineage/OpenLineage#3642 is open). A watcher
works with any `Asset` name, so this does not depend on that.
Two questions before adding more surface:
1. Does a trigger like this belong here, or would you rather wait for
catalog-side notifications?
2. `last_seen_snapshot_id` is a kwarg today. The `asset_state_store`
watermark looks like its right home across triggerer restarts. Is that the
direction you want?
# Are these changes tested?
Unit tests cover the cases that decide whether this is safe to schedule on:
cold start emits the current head; an unchanged table emits nothing; each
commit emits exactly one event carrying the snapshot it replaced; a branch that
does not exist yet is waited on rather than raising; `shared_stream_key` groups
by catalog; and the trigger round-trips serialization.
<details><summary>Unit tests</summary>
```
$ pytest
providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py -q
11 passed
```
Removing `triggers/iceberg.py` fails collection, so the tests cannot pass
vacuously.
</details>
I also ran the same polling logic against a **real Iceberg catalog and
table**, no mocks:
<details><summary>Against a live catalog</summary>
```
table created, head = 7966129066592937642
(a commit landed mid-poll)
1. cold start : 1 event(s) snapshot=7966129066592937642
2. already caught up : 0 event(s) (want 0)
3. new commit : fired 7966129066592937642 -> 1796904005523383578
4. shared poll loop : same catalog shares=True, different catalog
separate=True
5. serializable : True
```
Case 2 matters as much as case 3: a trigger that re-fires on an unchanged
table would schedule runs that should not happen.
</details>
An example DAG is included, following the Redis message-queue example.
# Are there any user-facing changes?
Additive: `IcebergTableSnapshotTrigger`, plus `AIRFLOW_V_3_0_PLUS` in the
provider's `version_compat` for the `BaseEventTrigger` import shim (the
provider supports Airflow 2.11+, where that class does not exist).
--
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]
