Page History

Status

Motivation

Incremental processing is a very one of the most, it not the most common pattern in Airflow use cases. In some pipelines, this is strictly necessaryimplement with Airflow. Although external event driven scheduling is supported today in Airflow through via AIP-82, incremental polling is not natively supported. In some scenariossomething that has been easy to implement. In more traditional DAG authoring (Sensors, Operators, etc.), users are forced to use XCom or Variables handling the watermark for storing values (watermarks) for incremental processing, or inventing and implementing their own solution. In the case of event-driven triggering, it’s even more challenging for users to build their custom incremental processing implantationimplementation, due to the lack of XCom support and limitations of Variable support s. 

There have been several attempts to store state within a child of the BaseEventTrigger to  to manage incremental processes, while none of these have proven to be effective or robust. This has seriously hampered the ability for the community to build logic to monitor Assets such as object stores, SQL databases, and other non-queue/stream-based Assets. With AIP-103, this state store will be developed and available to use as part of this AIP.

For Triggers built for "Asset “watching”watching", it is helpful, if not essential to to persist some state. This AIP will provide an interface for using the state store built by AIP-103 for "Asset watching".

Considerations

What change do you propose to make?

At a high level, we propose to make the following changes:

•  Pass through context including the Asset being "watched" to the BaseEventTrigger  to make the BaseEventTrigger  Asset-aware.
•  (Stretch Goal) Provide a more intuitive, decorator-based approach for authoring logic used for "Asset watching".

What problem does it solve?

This solves one of the most glaring problems with building Triggers compatible with "Asset-watching", or event-driven scheduling; the inability to persist and retrieve state for an Asset. Without a model like watermarks way to retrieve the state for an Asset , it's nearly impossible to do things like monitor an S3 bucket for new files landing or handle the addition/removal of a new row to a SQL database. Despite being one of the most touted features of Airflow 3.0, the community has been very slow to develop and distribute Event Triggers to be used for "Asset-watching". This AIP aims to address this problem in an elegant and "Airflow-onic" way, using a "batteries included" model, along with a pluggable backendusing the work laid in AIP-103.

Technical Details

Building the model

Building the watermarks model would require minimal lift. Like with the Variable model, this would be “unrelated” to other models, making the creation a bit easier. This model would include four fields; an Asset name, namespace, a key, and a value. The namespace would allow for user-specified identification between two Assets with the same name. Below is the pseudo-code for the watermarks model. 

The important thing about this change is this; this is not a breaking change to any existing processes. The same interface will be used to define AssetWatcher 's; a BaseEventTrigger  is defined and passed to the AssetWatcher . Then, the AssetWatcher  is passed to the Asset  via the watchers  parameter. At runtime, a reference to the Asset  will be passed through to the BaseEventTrigger , allowing for the Trigger to be Asset-aware.

Defining Assets and AssetWatchers

from airflow.sdk import Asset, AssetWatcher
from plugins.triggers import GenericEventTrigger

...  # Other imports

generic_asset_watcher = AssetWatcher(
    name="generic_asset_watcher",
    trigger=GenericEventTrigger(

class AssetWatermark:
    """
    Watermark class allows pluggable db backend.
    """
    def __init__(self, asset_name: str, namespace: str, key: str):
        self.asset_name = asset_name
        self.namespace = namespace
		self.key = key

    def set_value(self, value):
        db_backend.set_value(asset_name=self.asset_name, namespace=self.namespace, watermark_key=self.key, watermark_value=value)

    def get_value(self):
        return db_backend.get_value(asset_name=self.asset_name, namespace=self.namespace, watermark_key=self.key)

Building the pluggable backend

For users who plan on using watermarks extensively, a pluggable backend might provide them more control over the actual storage resources used to persist state. This backend would closely mimic the XComObjectStorageBackend, which includes an out-of-the-box model with the ability to configure a “pluggable” backend. 

Implementing “get/set” logic

It would require some sort of get and set logic to be implemented in a class that inherits from BaseEventTrigger in order to store and retrieve watermark value. This would most likely be in the form of a method for each as part of the BaseEventTrigger class, as shown in below pseudo-code.

)
)

generic_asset = Asset(
    name="generic_asset",
    watchers=[generic_asset_watcher]
)

with DAG(
    dag_id="my_dag",
    start_date=datetime(2026, 1, 1),
    schedule=[generic_asset]
) as dag:
    ...

BaseEventTrigger

Examples

“Watching” an S3 Bucket

One of the most common use-cases for event-driven scheduling will most likely be “watching” an object store for changes. Ideally, each time that an AssetWatcher’s AssetWatcher ’s Trigger runs, it willshould not re-scan the entire bucket. The workflow would look something like this:

1. After the first Trigger run, a timestamp of this “scan”
for data
2. of the Asset is recorded.
3. On the second run, the Trigger retrieves the timestamp (watermark) from the completion of the last run for that Asset and only scans the object store for
files
4. files after that watermark.
5. Trigger writes the
watermark
6. state for the Asset again.
7. Process repeats for N Trigger runs following.

Incremental Data in SQL Databases

Another common use-case for event-driven scheduling is “watching” tables in relational databases; we’ll use Postgres as an example. Outside of traditional CDC, it’s common for data teams to use an updated column in a Postgres database to upsert data. Using the asset_watermark approach, they can use the following workflow. This  This workflow should feel quite similar to the workflow for “watching” an S3 Bucket; that’s intentional, as we’re trying to implement a repeatable pattern.

1. After the first Trigger run, a timestamp of this “scan”
for data
2. of the Asset is recorded and persisted.
3. On the second run, the Trigger retrieves the
watermark
4. state for that Asset from the completion of the last run and only scans the updated column in the desired table for records with a timestamp greater than the watermark.
5. Trigger
writes the watermark
6. updates the state of the Asset again.
7. Process repeats for N Trigger runs following
.

Below is an example of a Trigger that executes a SQL query to incrementally read data, with a watermark.

1. .

Why is it needed?

Allow for an Airflow-supported methodology for persisting watermarks and unblocking Being able to persist and access the state  of an Asset unblocks the development of Event Triggers for Airflow users interested in "Asset-watching".

Are there any downsides to this change?

There is one theorized downside; Airflow users leveraging the watermarks  model as a sort of key-value store outside of an Event Trigger. However, this is not something that is overly concerning. It would be much more likely for Airflow users to abuse the Variable  model instead. Along with this are the general downsides to added complexity, but those are minimal with this proposalNo, there are no significant downsides to this change. This will not be a breaking change.

Which users are affected by the change?

DAG Authors: these users will now have access to a tool that makes authoring Triggers, Sensors, and Tasks used to orchestrate incremental processes more accessible and unlocks the ability to further build out event-driven logic.

Deployment Managers: introducing a pattern with a pluggable backend may provide deployment managers an additional component of their Airflow “stack” to stand up and manage. This will also require a DB migration, which is outlined below.

How are users affected by the change? (e.g. DB upgrade required?)

Current users who don't use incremental event triggers are unaffected by this change. However, this new feature requires a DB migration, since a new model is being added. The only changes needed in this case would be the addition of a new tableUsers who are "Asset-watching" will now have the ability to make these process Asset-ware and persist state for that Asset.

What is the level of migration effort (manual and automated) needed for the users to adapt to the breaking changes? (especially in context of Airflow 3)

There are not breaking changes included as part of this AIP. Added the watermarks model will require a DB migration, but forces no breaking changes upon users.

Other considerations?

• This would be compatible with Asset partitions
.For now, this sort of “state store” would be limited to Asset “watching”
• .

What defines this AIP as "done"?

This AIP will be considered “done” when the PR creates the model and the pluggable backend is merged passes through the needed context to make BaseEventTrigger  Asset-aware and the solution is well-docmenteddocumented.