Hi All, I'd like to discuss a gap in partitioned-asset scheduling around late-arriving or corrected data, and a small opt-in mechanism to address it. This is my first proposal to the project, so feedback on both the idea and whether an AIP is the right vehicle is very welcome.
*Problem* Airflow 3's partitioned-asset scheduling fires a DAG once per partition key once its asset_condition is satisfied (e.g. a & b for a given snapshot date). That works well for the *first* completion of a partition. It has no declarative way to re-process a partition after it has already fired, when a late or corrected event arrives for that same partition later which is the norm in regulated/financial data platforms assembling one dataset from many independently arriving sources. Digging into airflow/assets/manager.py and airflow/jobs/scheduler_job_runner.py (verified against 3.3.1), the root cause is precise: AssetPartitionDagRun bookkeeping is append-only by design (multiple APDRs per (dag, partition_key) are expected), but when a late/corrected event arrives after a partition has fired, the *new APDR only carries the arriving asset's key* sibling assets' earlier contributions aren't carried forward into it. So the asset_condition can never re-satisfy on that new APDR, and it sits pending forever. This is not "Airflow can only fire once" the data model already supports multiple runs per partition it's specifically that a partial late/corrected event orphans an unsatisfiable provisional run. Today the only workaround is imperative and outside asset semantics (TriggerDagRunOperator(reset_dag_run=True, ...)), which doesn't compose with the asset graph and pushes dedup/versioning onto the user. *Proposal* An opt-in refire_policy on PartitionedAssetTimetable, sibling to the existing wait_policy, default None = fire-once (no behavior change for existing DAGs). When enabled, a new event for an already-fired partition seeds the new APDR with the sibling assets' key-log rows carried forward from the last completed APDR, letting the condition re-satisfy from a single corrected feed. Two selectable modes: - RerunExisting: clear + re-run the *same* DagRun (task attempts increment), matching the common "re-run this date" mental model. - NewRun: create a new DagRun per re-fire, for teams that want immutable per-restatement lineage. I have a working prototype against vanilla apache/airflow:3.3.1 demonstrating both modes end-to-end, plus a control case reproducing the current gap. I'd like feedback on: 1. Whether this is better scoped as its own AIP or folded under an existing effort (I see overlap with AIP-93 Asset Watermarks on the "persisted per-key state" side, and AIP-73 Expanded Data Awareness as the asset model this builds on). 2. Whether RerunExisting should be the implicit default when a user opts in, or an explicit choice should always be required. 3. Any prior discussion/rejected proposals on this I should be aware of before writing this up formally. Happy to share the full write-up (with the exact gap analysis, code pointers, and PoC results) as a Google Doc or start the formal AIP page if there's interest. Thanks, Ramzi Alashabi Chapter Lead Data & Streaming @ABNAMRO
