Add IcebergTableSnapshotTrigger for event-driven scheduling

[1fanwang]

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.

IcebergTableSnapshotTrigger polls a table's branch head through the existing IcebergHook and emits an event when it advances, so a commit can drive an AssetWatcher:

orders = Asset(
    "orders",
    watchers=[AssetWatcher(name="orders_commits", trigger=IcebergTableSnapshotTrigger(table="sales.orders"))],
)

@dag(schedule=[orders])
def downstream(): ...

The event carries snapshot_id and previous_snapshot_id, so a task can scan the delta instead of the whole table.

Polling is what Iceberg supports: the REST catalog spec defines no subscribe, webhook or event endpoint, and none is proposed.

Two behaviors worth calling out, both found by running it on a real Airflow rather than in tests:

• The cursor lives in the asset_state_store watermark, not on the instance. serialize() is captured once when the trigger row is written, so a value mutated during run() is lost on restart and the trigger re-reports a commit it already emitted. Read through getattr, since that attribute postdates the oldest Airflow this provider supports.
• An absent table is waited on, like an absent branch. A watcher outlives the table it watches, and raising kills the trigger, which the triggerer then restarts about once a second until the table appears.

A trigger watched by more than one asset gets one accessor per asset, where the watermark shorthand raises. It degrades to no watermark; #71460 adds what is needed to keep a cursor there.

Are these changes tested?

Unit tests — 19 passed

Cold start emits the current head; an unchanged table emits nothing; each commit emits one event carrying the snapshot it replaced; an absent branch or table is waited on; the watermark is restored on start and written per event; a state store failure is not mistaken for the several-assets case; the triggerer's dispatch reaches the poll loop; serialization round-trips.

$ pytest providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py -q
19 passed

Removing triggers/iceberg.py fails collection, so the tests cannot pass vacuously.

Then on a running Airflow: Postgres metadata database, api-server, dag-processor, scheduler and triggerer as separate processes, a real Iceberg catalog behind a real iceberg_default connection, and a DAG scheduled on the watched asset. Nothing stubbed.

On a running Airflow

$ airflow db migrate && airflow connections add iceberg_default --conn-type iceberg \
    --conn-extra '{"type":"sql","uri":"sqlite:////tmp/ice-wh/catalog.db","warehouse":"file:///tmp/ice-wh"}'
$ airflow api-server & airflow dag-processor & airflow scheduler & airflow triggerer &
$ airflow dags unpause iceberg_downstream

-- the trigger is registered and wired to the asset
 id |                                   classpath
----+-------------------------------------------------------------------------------
  1 | airflow.providers.apache.iceberg.triggers.iceberg.IcebergTableSnapshotTrigger
 trigger_id |  name
------------+--------
          1 | orders

1. table absent          starts=1  restarts=0  crashes=0  events=0
2. table created         Trigger fired event  TriggerEvent<{'table': 'sales.orders', ...}>
3. dag_run                iceberg_downstream | asset_triggered__2026-08-11T21:47:04 | asset_triggered
4. asset_state_store      asset_id=1  key=snapshot_id  value=8102133228104772796  last_updated_by_kind=watcher
5. triggerer killed and restarted, no new commit
   dag_runs=1 (unchanged)   events fired since restart=0

Step 4 is the real asset_state_store row, written by the triggerer over the execution API rather than by a test double. Step 5 is why the watermark is not just a kwarg: killing the triggerer and bringing it back produced no second run for a commit already reported.

An example DAG is included, following the Redis message-queue one.

[amoghrajesh]

Instead of reinventing the async behaviour in every trigger, we should migrate this to aget, aset etc once #72127 lands.

[amoghrajesh]

We should avoid if/else based on version as much as possible, gets hard to manage conditional code over time. An improvement: #72140

[vincbeck]

@1fanwang can you work on these comments in a separate PR please?

[1fanwang]

@1fanwang can you work on these comments in a separate PR please?

Ack @vincbeck checking now