You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2026/06/19 23:47:11 UTC
[PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
FrankYang0529 opened a new pull request, #68778:
URL: https://github.com/apache/airflow/pull/68778
When an upstream partition that a rollup's downstream window already consumed is cleared and re-run, the framework had no defined behavior. The de-facto outcome silently depended on the rollup's wait policy: WaitForAll left a provisional run stuck waiting for keys that never re-arrive, while MinimumCount re-fired the downstream run on partial data. Neither is something a Dag author can rely on.
Give RollupMapper an explicit rerun_policy so the author chooses what happens. The default preserves the historical behavior, so existing Dags are unchanged; re-firing with the corrected data is opt-in.
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
closes: #65923
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [X] Yes - Claude Code with Opus 4.8
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3504726344
##########
airflow-core/src/airflow/partition_mappers/base.py:
##########
@@ -37,6 +38,12 @@ class PartitionMapper(ABC):
"""
is_rollup: ClassVar[bool] = False
+ #: Rerun handling does not apply to a non-rollup mapper, so the base default
+ #: is ``RerunPolicy.NONE``. :class:`RollupMapper` sets it per instance, so the
+ #: scheduler can read ``mapper.rerun_policy`` uniformly without a rollup
+ #: type-check. (An instance attribute with a class default, not a ClassVar, so
+ #: RollupMapper can assign to it.)
Review Comment:
IIRC, this is used by Sphinx. It's used in a few places. In core, `airflow/sdk/execution_time/context.py` NEVER_EXPIRE and `expected_decoded_type` in partition mapper. I feel this is something new. We can remove it altogether if not desired 🤔
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3465369626
##########
airflow-core/src/airflow/partition_mappers/rerun_policy.py:
##########
Review Comment:
Add pre-commit check for rerun_policy.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449018323
##########
airflow-core/src/airflow/partition_mappers/base.py:
##########
@@ -215,6 +224,9 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
window=decode_window(data["window"]),
wait_policy=decode_wait_policy(data["wait_policy"]),
+ # Default for serialized Dags written before rerun_policy existed —
+ # HOLD reproduces their pre-feature behavior (wait for the full window).
+ rerun_policy=data.get("rerun_policy", RerunPolicy.HOLD),
Review Comment:
Comment is obvious. Remove it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3509135824
##########
task-sdk/src/airflow/sdk/definitions/partition_mappers/rerun_policy.py:
##########
@@ -0,0 +1,60 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from enum import Enum
+
+
+class RerunPolicy(str, Enum):
+ """
+ How a rollup reacts when an upstream partition is re-emitted after its downstream window already fired.
+
+ A rollup fires its downstream Dag run once the window is satisfied (e.g. a
+ monthly rollup fires once all of March's daily partitions arrive). If an
+ upstream partition that the fired window already consumed is later cleared
+ and re-run, a fresh asset event arrives for an already-materialized window.
+ This policy decides what the scheduler does with it.
+
+ ``HOLD`` (default): queue a provisional run that waits for the *entire*
+ window to re-materialize before firing again. A single re-run does not
+ re-fire; only a full recompute of the window does. This is the historical
+ behavior of a rollup before this policy existed, so it is the default to keep
+ existing Dags unchanged.
Review Comment:
This is only true for WaitForAll. If the wait policy is `MinimumCount`, a single re-run can re-fire before the whole window re-materializes. A reword to something like _re-evaluates the window through the wait policy_ would be nice.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3504664935
##########
airflow-core/src/airflow/partition_mappers/base.py:
##########
@@ -37,6 +38,12 @@ class PartitionMapper(ABC):
"""
is_rollup: ClassVar[bool] = False
+ #: Rerun handling does not apply to a non-rollup mapper, so the base default
+ #: is ``RerunPolicy.NONE``. :class:`RollupMapper` sets it per instance, so the
+ #: scheduler can read ``mapper.rerun_policy`` uniformly without a rollup
+ #: type-check. (An instance attribute with a class default, not a ClassVar, so
+ #: RollupMapper can assign to it.)
Review Comment:
What is special about the `#:` comment prefix? I don’t think we use this pattern anywhere else.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3465368065
##########
airflow-core/src/airflow/partition_mappers/base.py:
##########
@@ -147,6 +153,7 @@ def __init__(
upstream_mapper: PartitionMapper,
window: Window,
wait_policy: WaitPolicy | None = None,
+ rerun_policy: RerunPolicy | str = RerunPolicy.HOLD,
Review Comment:
Updated it. Thank you.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3491065376
##########
airflow-core/src/airflow/partition_mappers/rerun_policy.py:
##########
@@ -0,0 +1,55 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from enum import Enum
+
+
+class RerunPolicy(str, Enum):
+ """
+ Core-side mirror of the SDK :class:`airflow.sdk.RerunPolicy`.
+
+ Decides what the scheduler does when an upstream partition is cleared and
+ re-run after a rollup's downstream window has already fired. See the SDK
+ class for the authoring-facing documentation; the two are serialized by
+ string value so they round-trip across the Dag-parse / scheduler boundary.
+
+ ``HOLD`` (default): queue a provisional run that waits for the entire window
+ to re-materialize before firing again. This is the historical behavior, so it
+ is the default to keep existing Dags unchanged.
+
+ ``REFRESH``: re-fire the downstream Dag run immediately so it reprocesses with
+ the corrected upstream data.
+
+ ``IGNORE``: drop the late upstream event; do not re-fire.
+
+ ``NONE``: rerun handling does not apply — reported by non-rollup mappers and
+ stamped on a window's first materialization (which is not a rerun).
+ """
+
+ REFRESH = "refresh"
+ HOLD = "hold"
+ IGNORE = "ignore"
+ NONE = "none"
+
+ def fires_immediately(self) -> bool:
Review Comment:
looks like property to me
```suggestion
@property
def fires_immediately(self) -> bool:
```
##########
airflow-core/docs/migrations-ref.rst:
##########
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description |
+=========================+==================+===================+==============================================================+
-| ``d2f4e1b3c5a7`` (head) | ``9ff64e1c35d3`` | ``3.3.0`` | Add partition_date to asset_partition_dag_run. |
+| ``623bce373cdf`` (head) | ``d2f4e1b3c5a7`` | ``3.3.0`` | Add rerun_policy to AssetPartitionDagRun. |
Review Comment:
I don' think we'll make it in 3.3.0. so that's make it 3.4.0
##########
task-sdk/src/airflow/sdk/definitions/partition_mappers/rerun_policy.py:
##########
@@ -0,0 +1,63 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from enum import Enum
+
+
+class RerunPolicy(str, Enum):
+ """
+ How a rollup reacts when an upstream partition is re-emitted after its downstream window already fired.
+
+ A rollup fires its downstream Dag run once the window is satisfied (e.g. a
+ monthly rollup fires once all of March's daily partitions arrive). If an
+ upstream partition that the fired window already consumed is later cleared
+ and re-run, a fresh asset event arrives for an already-materialized window.
+ This policy decides what the scheduler does with it.
+
+ ``HOLD`` (default): queue a provisional run that waits for the *entire*
+ window to re-materialize before firing again. A single re-run does not
+ re-fire; only a full recompute of the window does. This is the historical
+ behavior of a rollup before this policy existed, so it is the default to keep
+ existing Dags unchanged.
+
+ ``REFRESH``: re-fire the downstream Dag run so it reprocesses with the
+ corrected upstream data. The rest of the window is still materialized, so the
+ refresh run fires immediately rather than waiting for the whole window to
+ re-arrive. This mirrors how a non-partitioned asset-triggered Dag re-runs on
+ every new asset event.
+
+ ``IGNORE``: drop the late upstream event. The downstream Dag run is not
+ re-fired and no provisional run is queued.
+
+ ``NONE``: rerun handling does not apply. This is the value reported by
+ non-rollup mappers and stamped on a window's first materialization (which is
+ not a rerun); it is not a meaningful choice for a rollup.
+ """
+
+ REFRESH = "refresh"
+ HOLD = "hold"
+ IGNORE = "ignore"
+ NONE = "none"
+
+ def fires_immediately(self) -> bool:
Review Comment:
same. a good idea to make them properties
##########
task-sdk/tests/task_sdk/definitions/test_partition_mappers.py:
##########
@@ -67,6 +68,77 @@ class _AlphaWindow(Window):
RollupMapper(upstream_mapper=_StringOnlyMapper(), window=_AlphaWindow())
+class TestSdkRollupMapperRerunPolicy:
+ """``rerun_policy`` defaults to HOLD, coerces strings, and rejects junk."""
+
+ def _mapper(self, **kwargs):
Review Comment:
```suggestion
def _make_mapper(self, **kwargs):
```
##########
airflow-core/src/airflow/migrations/versions/0124_3_3_0_add_rerun_policy_to_apdr.py:
##########
@@ -0,0 +1,60 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add rerun_policy to AssetPartitionDagRun.
+
+The ``rerun_policy`` column stores the RerunPolicy (as its string value) that
+governs how the scheduler fires this provisional partition Dag run. A run stamped
+``"refresh"`` fires immediately after an upstream partition was cleared and
+re-run; everything else is stamped ``"none"`` and goes through normal evaluation.
+
+Revision ID: 623bce373cdf
+Revises: d2f4e1b3c5a7
+Create Date: 2026-06-17 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+from airflow.migrations.utils import disable_sqlite_fkeys
+
+revision = "623bce373cdf"
+down_revision = "d2f4e1b3c5a7"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
Review Comment:
```suggestion
airflow_version = "3.4.0"
```
##########
airflow-core/tests/unit/timetables/test_base_timetable.py:
##########
@@ -114,6 +114,58 @@ def test_compute_rollup_fingerprint_window_change_produces_different_fingerprint
assert fp_hour != fp_day
+def test_compute_rollup_fingerprint_excludes_rerun_policy():
+ """
+ Changing only ``rerun_policy`` must NOT change the fingerprint.
+
+ ``rerun_policy`` governs post-fire behavior, not which upstream keys a window
+ requires. If it fed the fingerprint, a policy-only edit would discard in-flight
+ APDRs, and upgrading from a version that pre-dates the key would invalidate
+ every pending partition run (stored fingerprints lack the key).
+ """
+ from airflow.sdk import Asset, HourWindow, RerunPolicy, RollupMapper, StartOfHourMapper
Review Comment:
I think we can move most of these inline imports to the top. (just tested it locally)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #68778:
URL: https://github.com/apache/airflow/pull/68778#issuecomment-4852599709
How is NONE different from HOLD? I did a brief mental check and from what I can tell internals seem to treat them the same.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3504879121
##########
airflow-core/src/airflow/partition_mappers/base.py:
##########
@@ -37,6 +38,12 @@ class PartitionMapper(ABC):
"""
is_rollup: ClassVar[bool] = False
+ #: Rerun handling does not apply to a non-rollup mapper, so the base default
+ #: is ``RerunPolicy.NONE``. :class:`RollupMapper` sets it per instance, so the
+ #: scheduler can read ``mapper.rerun_policy`` uniformly without a rollup
+ #: type-check. (An instance attribute with a class default, not a ClassVar, so
+ #: RollupMapper can assign to it.)
Review Comment:
What does Sphinx do with this?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449005589
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -643,14 +645,22 @@ def _queue_partitioned_dags(
)
continue
+ # rerun_policy only applies to rollups (it decides what happens when an
+ # upstream key re-arrives after the window already fired). Non-rollup
+ # mappers have no such field; ``None`` keeps their legacy behavior.
+ rerun_policy = mapper.rerun_policy if is_rollup(mapper) else None
Review Comment:
I don’t like this. The mapper should instead decide on its own whether the rerun policy applies to it or not. So `rerun_policy` should simply always be *None* on any mapper that is not a rollup (and emit an error if the user tries to set it to something else).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449020992
##########
airflow-core/src/airflow/partition_mappers/rerun_policy.py:
##########
Review Comment:
A pre-commit check should be added to ensure the enum classes here and in sdk have the same cases.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3465361675
##########
airflow-core/src/airflow/partition_mappers/base.py:
##########
@@ -215,6 +224,9 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
window=decode_window(data["window"]),
wait_policy=decode_wait_policy(data["wait_policy"]),
+ # Default for serialized Dags written before rerun_policy existed —
+ # HOLD reproduces their pre-feature behavior (wait for the full window).
+ rerun_policy=data.get("rerun_policy", RerunPolicy.HOLD),
Review Comment:
Removed it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449019551
##########
airflow-core/src/airflow/partition_mappers/base.py:
##########
@@ -147,6 +153,7 @@ def __init__(
upstream_mapper: PartitionMapper,
window: Window,
wait_policy: WaitPolicy | None = None,
+ rerun_policy: RerunPolicy | str = RerunPolicy.HOLD,
Review Comment:
```suggestion
rerun_policy: RerunPolicy = RerunPolicy.HOLD,
```
No need to consider the str case since this is a core class, not user-facing. (See `wait_policy`.)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449006099
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -643,14 +645,22 @@ def _queue_partitioned_dags(
)
continue
+ # rerun_policy only applies to rollups (it decides what happens when an
+ # upstream key re-arrives after the window already fired). Non-rollup
+ # mappers have no such field; ``None`` keeps their legacy behavior.
+ rerun_policy = mapper.rerun_policy if is_rollup(mapper) else None
for target_key in target_keys:
apdr = cls._get_or_create_apdr(
target_key=target_key,
target_dag=target_dag,
rollup_fingerprint=fingerprint,
asset_id=asset_id,
+ rerun_policy=rerun_policy,
session=session,
)
+ # IGNORE dropped this late event for an already-fired window.
Review Comment:
This comment should be expanded somewhat.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3505246276
##########
airflow-core/tests/unit/timetables/test_base_timetable.py:
##########
@@ -114,6 +114,58 @@ def test_compute_rollup_fingerprint_window_change_produces_different_fingerprint
assert fp_hour != fp_day
+def test_compute_rollup_fingerprint_excludes_rerun_policy():
+ """
+ Changing only ``rerun_policy`` must NOT change the fingerprint.
+
+ ``rerun_policy`` governs post-fire behavior, not which upstream keys a window
+ requires. If it fed the fingerprint, a policy-only edit would discard in-flight
+ APDRs, and upgrading from a version that pre-dates the key would invalidate
+ every pending partition run (stored fingerprints lack the key).
+ """
+ from airflow.sdk import Asset, HourWindow, RerunPolicy, RollupMapper, StartOfHourMapper
Review Comment:
Thank you. Updated it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3506240847
##########
airflow-core/src/airflow/partition_mappers/base.py:
##########
@@ -37,6 +38,12 @@ class PartitionMapper(ABC):
"""
is_rollup: ClassVar[bool] = False
+ #: Rerun handling does not apply to a non-rollup mapper, so the base default
+ #: is ``RerunPolicy.NONE``. :class:`RollupMapper` sets it per instance, so the
+ #: scheduler can read ``mapper.rerun_policy`` uniformly without a rollup
+ #: type-check. (An instance attribute with a class default, not a ClassVar, so
+ #: RollupMapper can assign to it.)
Review Comment:
Sorry, I think we don't need `Sphinx` here. Removed it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449016861
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2261,23 +2261,35 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
source_key_by_asset = source_key_by_asset_per_apdr[apdr.id]
timetable = dag.timetable
- statuses: dict[SerializedAssetUniqueKey, bool] = {}
- for asset_id, (name, uri) in asset_info_per_apdr[apdr.id].items():
- key = SerializedAssetUniqueKey(name=name, uri=uri)
- if timetable.partitioned:
- statuses[key] = self._resolve_asset_partition_status(
- session=session,
- asset_id=asset_id,
- name=name,
- uri=uri,
- apdr=apdr,
- timetable=timetable,
- actual_by_asset=source_key_by_asset,
- )
- else:
- statuses[key] = True
- if not evaluator.run(timetable.asset_condition, statuses=statuses):
- continue
+ contributing_assets = asset_info_per_apdr[apdr.id]
+ if apdr.is_refresh:
Review Comment:
Instead of storing a simple boolean, it may be a good idea to store the retun policy directly on the model (as a string), and resolve the stored value to a full RerunPolicy object. That way, this block can be extracted into a function on RerunPolicy. Making the logic live on RerunPolicy is nicer because it is easier to keep track of possible enum values on the same class, and if we potentially add new values to the enum, it is much easier to remember to add a branch if the logic is close.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3465366822
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2261,23 +2261,35 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
source_key_by_asset = source_key_by_asset_per_apdr[apdr.id]
timetable = dag.timetable
- statuses: dict[SerializedAssetUniqueKey, bool] = {}
- for asset_id, (name, uri) in asset_info_per_apdr[apdr.id].items():
- key = SerializedAssetUniqueKey(name=name, uri=uri)
- if timetable.partitioned:
- statuses[key] = self._resolve_asset_partition_status(
- session=session,
- asset_id=asset_id,
- name=name,
- uri=uri,
- apdr=apdr,
- timetable=timetable,
- actual_by_asset=source_key_by_asset,
- )
- else:
- statuses[key] = True
- if not evaluator.run(timetable.asset_condition, statuses=statuses):
- continue
+ contributing_assets = asset_info_per_apdr[apdr.id]
+ if apdr.is_refresh:
Review Comment:
Changed to store the rerun policy in apdr and move the check to RerunPolicy.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3465353561
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -643,14 +645,22 @@ def _queue_partitioned_dags(
)
continue
+ # rerun_policy only applies to rollups (it decides what happens when an
+ # upstream key re-arrives after the window already fired). Non-rollup
+ # mappers have no such field; ``None`` keeps their legacy behavior.
+ rerun_policy = mapper.rerun_policy if is_rollup(mapper) else None
for target_key in target_keys:
apdr = cls._get_or_create_apdr(
target_key=target_key,
target_dag=target_dag,
rollup_fingerprint=fingerprint,
asset_id=asset_id,
+ rerun_policy=rerun_policy,
session=session,
)
+ # IGNORE dropped this late event for an already-fired window.
Review Comment:
Updated comment.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3465355308
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -669,8 +679,9 @@ def _get_or_create_apdr(
target_dag: DagModel,
rollup_fingerprint: dict,
asset_id: int,
+ rerun_policy: RerunPolicy | None = None,
Review Comment:
No, we don't need the default value here. Updated it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #68778:
URL: https://github.com/apache/airflow/pull/68778#issuecomment-4855628153
> How is NONE different from HOLD? I did a brief mental check and from what I can tell internals seem to treat them the same. This is particularly a problem since there’s zero documentation available here. Maybe we should add some?
Originally, I used `NONE` for non-RollupMapper and `HOLD` for the RollupMapper. However, both have same behavior. They all follow `WaitPolicy` to decide whether a partitioned Dag run should fire. I remove the `NONE`. The non-rollup mappers and a window's first materialization now use `HOLD` as the default value.
For document, we have it in both `airflow-core/src/airflow/partition_mappers/rerun_policy.py` and `task-sdk/src/airflow/sdk/definitions/partition_mappers/rerun_policy.py`. Let me know if you'd like the docs expanded anywhere else. Thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3509160785
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2277,23 +2278,32 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
source_key_by_asset = source_key_by_asset_per_apdr[apdr.id]
timetable = dag.timetable
- statuses: dict[SerializedAssetUniqueKey, bool] = {}
- for asset_id, (name, uri) in asset_info_per_apdr[apdr.id].items():
- key = SerializedAssetUniqueKey(name=name, uri=uri)
- if timetable.partitioned:
- statuses[key] = self._resolve_asset_partition_status(
- session=session,
- asset_id=asset_id,
- name=name,
- uri=uri,
- apdr=apdr,
- timetable=timetable,
- actual_by_asset=source_key_by_asset,
- )
- else:
- statuses[key] = True
- if not evaluator.run(timetable.asset_condition, statuses=statuses):
- continue
+ contributing_assets = asset_info_per_apdr[apdr.id]
+ if RerunPolicy(apdr.rerun_policy).fires_immediately:
+ # A REFRESH-stamped APDR supersedes an already-fired window: the
+ # rest of the window is still materialized, so fire immediately
+ # with the re-arrived events rather than re-satisfying the wait
+ # policy. Still require at least one active contributing asset.
+ if not contributing_assets:
+ continue
Review Comment:
Does this need to evaluate the downstream’s asset_condition is satisfied? This would unconditionally re-fire everything. Maybe a note in docs if this is intended?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449012639
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -703,11 +727,25 @@ def _get_or_create_apdr(
)
return latest_apdr
+ is_refresh = False
+ if latest_apdr is not None:
+ # The latest APDR already fired, so this event re-arrives for an
+ # already-materialized window. Apply the rollup's rerun policy.
+ if rerun_policy is RerunPolicy.IGNORE:
+ cls.logger().debug(
+ "Dropping re-arrived event for fired window key %s dag_id %s (IGNORE)",
+ target_key,
+ target_dag.dag_id,
+ )
+ return None
+ is_refresh = rerun_policy is RerunPolicy.REFRESH
Review Comment:
It’s probably better to make this a function on `RerunPolicy` instead, like this
```python
if rerun_policy is None:
is_refresh = False
else:
is_refresh = rerun_policy.is_refresh(target_dag, target_key)
```
Furthermore, it is worth considering whether instead of *None*, we can add a dedicated `RerunPolicy.NONE` instead. This avoids needing to special-case *None* every time the value is used.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449008694
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -669,8 +679,9 @@ def _get_or_create_apdr(
target_dag: DagModel,
rollup_fingerprint: dict,
asset_id: int,
+ rerun_policy: RerunPolicy | None = None,
Review Comment:
Is a default needed here? This is a private function so we should control all its call sites and can change them directly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3465360127
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -703,11 +727,25 @@ def _get_or_create_apdr(
)
return latest_apdr
+ is_refresh = False
+ if latest_apdr is not None:
+ # The latest APDR already fired, so this event re-arrives for an
+ # already-materialized window. Apply the rollup's rerun policy.
+ if rerun_policy is RerunPolicy.IGNORE:
+ cls.logger().debug(
+ "Dropping re-arrived event for fired window key %s dag_id %s (IGNORE)",
+ target_key,
+ target_dag.dag_id,
+ )
+ return None
+ is_refresh = rerun_policy is RerunPolicy.REFRESH
Review Comment:
Since we changed to use rerun_policy directly, we don't need is_refresh anymore. Thank you.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add rerun policy for rollups when an upstream partition is re-run [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3465352017
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -643,14 +645,22 @@ def _queue_partitioned_dags(
)
continue
+ # rerun_policy only applies to rollups (it decides what happens when an
+ # upstream key re-arrives after the window already fired). Non-rollup
+ # mappers have no such field; ``None`` keeps their legacy behavior.
+ rerun_policy = mapper.rerun_policy if is_rollup(mapper) else None
Review Comment:
Thanks for the review. Yes, leave this check for mapper is more clear. Updated it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org