Skip to content

Avoid duplicate Iceberg asset runs after a triggerer restart - #71751

Closed
1fanwang wants to merge 2 commits into
apache:mainfrom
1fanwang:iceberg-trigger-multi-asset-watermark
Closed

Avoid duplicate Iceberg asset runs after a triggerer restart#71751
1fanwang wants to merge 2 commits into
apache:mainfrom
1fanwang:iceberg-trigger-multi-asset-watermark

Conversation

@1fanwang

@1fanwang 1fanwang commented Aug 18, 2026

Copy link
Copy Markdown
Contributor

Rationale for this change

Two assets watch one Iceberg table through one trigger. Restart the triggerer for a deploy and both fire again for a commit already reported, so every DAG watching that table gets a duplicate run.

The watermark stops that, and the trigger drops it here. It also detects the case by matching an exception's text against "concrete inlets and outlets", wording from a private helper that nobody promised, so a reword turns the guard into a re-raise. #71460 makes the accessors iterable, so the trigger writes to each asset.

What changes are included in this PR?

One cursor per asset, not one or none:

stored = [a.get(WATERMARK_KEY) for a in accessors]   # oldest wins, so a partial
...                        # write repeats a snapshot rather than skipping one
for accessor in accessors:
    accessor.set(WATERMARK_KEY, head)

A state store failure now propagates, no longer read as the several-assets case. Before #71460 the accessors are neither iterable nor countable, so the trigger reads once to see if it can address them, polling unwatermarked if not.

Are these changes tested?

Live, on Postgres with a real Iceberg catalog and two assets on one trigger row.

Restart with no new commit, before and after
# append a row to sales.orders via pyiceberg, then:
$ kill $(pgrep -f "airflow triggerer") && airflow triggerer &  # no new commit
$ psql -c "SELECT e.timestamp, e.extra::json->'payload'->>'snapshot_id' snap,
  e.extra::json->'payload'->>'previous_snapshot_id' prev FROM asset_event e
  JOIN asset a ON a.id=e.asset_id WHERE a.name='orders_raw' ORDER BY 1"

--- on main ---   (orders_reporting identical)
 00:08:58 | 9203870129439515238 | 5620644195193149555  # the commit
 00:10:08 | 9203870129439515238 |                      # RESTART: again

--- with this change ---
 00:11:56 | 9203870129439515238 |     # RESTART after this: no event
 00:14:18 | 9209950183446620906 | 9203870129439515238

$ pytest providers/apache/iceberg/tests/unit/apache/iceberg/triggers/ -q
5 failed, 18 passed  # on main
23 passed            # with this change

That last prev proves the watermark survived. A stand-in for the pre-#71460 accessors covers the older path.

Are there any user-facing changes?

A table watched by several assets keeps its watermark, so restarting the triggerer no longer fires a duplicate run for a handled commit. One asset behaves as before. Stacked on #71460.

Triggers are deduplicated by hash(classpath, kwargs) while asset_watcher is
many-to-many, so several assets can watch one trigger. The trigger detected
that by matching the text of a ValueError raised by a private helper, then gave
up on the watermark, leaving a restart free to re-emit the current head.

apache#71460 makes the accessors iterable, so address each asset
directly: write the snapshot to all of them and resume from the oldest, which
makes a write that reached only some assets repeat a snapshot rather than skip
one. A state store failure now propagates instead of being read as the
several-assets case.

Signed-off-by: 1fanwang <1fannnw@gmail.com>
@1fanwang

Copy link
Copy Markdown
Contributor Author

cc @vincbeck as a follow up to #71387 and #71460, thank you!

The fallback treated the whole accessors object as one accessor, so on an
Airflow that predates the iterable accessors a trigger watched by several
assets raised ValueError out of run() rather than polling on without a
watermark. That version cannot report how many assets watch the trigger,
so read once and drop the watermark if it refuses to guess.

Signed-off-by: 1fanwang <1fannnw@gmail.com>
@1fanwang 1fanwang changed the title Keep a snapshot watermark for every asset watching an Iceberg trigger Keep the Iceberg snapshot watermark when several assets watch a table Aug 18, 2026
@1fanwang 1fanwang changed the title Keep the Iceberg snapshot watermark when several assets watch a table Avoid duplicate Iceberg asset runs after a triggerer restart Aug 21, 2026
@1fanwang 1fanwang closed this Aug 24, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant