Repository navigation
feat(core): add per-mapper max_fan_out override for partition fan-out cap - #67184
Merged
Merged
Conversation
boring-cyborg
Bot
added
area:API
Lee-W
changed the title
Lee-W
force-pushed
the
partition-fanout-max-keys
branch
14 times, most recently
from
May 28, 2026 10:11
d5d2ced to
af057ae
Compare
Lee-W
force-pushed
the
partition-fanout-max-keys
branch
3 times, most recently
from
May 29, 2026 14:13
9f03f8d to
efa8911
Compare
Lee-W
force-pushed
the
partition-fanout-max-keys
branch
from
June 8, 2026 13:03
c1521f6 to
81910a9
Compare
uranusjr
reviewed
Jun 9, 2026
uranusjr
reviewed
Jun 9, 2026
Lee-W
force-pushed
the
partition-fanout-max-keys
branch
4 times, most recently
from
June 9, 2026 13:17
aa011c3 to
03b76c8
Compare
Lee-W
force-pushed
the
partition-fanout-max-keys
branch
2 times, most recently
from
June 10, 2026 00:21
c15e65c to
a26e224
Compare
uranusjr
reviewed
Jun 10, 2026
uranusjr
approved these changes
Jun 10, 2026
Lee-W
force-pushed
the
partition-fanout-max-keys
branch
2 times, most recently
from
June 10, 2026 10:07
c634109 to
cc440c8
Compare
Lee-W
force-pushed
the
partition-fanout-max-keys
branch
4 times, most recently
from
June 10, 2026 15:30
1c8c418 to
9d28124
Compare
… fan-out cap Add a `max_downstream_keys: int | None = None` parameter to both the SDK and core `PartitionMapper` base classes, threaded through every subclass constructor that defines its own `__init__`. The validator rejects 0, negatives, floats, and strings. Serialization is updated in two parallel paths: - Per-class `serialize()` / `deserialize()` methods (custom-mapper path). - `encoders.py` singledispatch overrides (built-in-mapper path). Both paths write `max_downstream_keys` only when non-None, following the existing convention of omitting default-valued fields from the serialized form. In `manager.py`, the single-shot `max_downstream_keys` read is deleted; the mapper object is retained after the `to_downstream` call so its `max_downstream_keys` attribute is readable at the boundary check. The effective cap is computed per target-Dag as either the mapper's own cap or the global `[scheduler] partition_mapper_max_downstream_keys`. The `Log.extra` string uses `max_downstream_keys=N` when the per-mapper cap trips, and keeps the existing `[scheduler] partition_mapper_max_downstream_keys=N` wording when the global trips.
The scheduler error log for a partition fan-out overflow now carries a cap_source kwarg, so operators can tell whether the per-mapper max_downstream_keys override or the global scheduler cap was the limit that was hit.
…validator Convert the SDK PartitionMapper base into an attrs class so the max_downstream_keys bound is enforced through a field validator instead of a hand-written __init__ check, matching the attrs convention used across the rest of airflow.sdk.definitions. Behavior is unchanged. Also add the SDK-side validator tests that previously only existed on the core mapper hierarchy.
Make the whole SDK partition-mapper hierarchy use one mechanism so the attrs base no longer needs a slots exception. RollupMapper, AllowedKeyMapper, the temporal mappers, and FanOutMapper become attrs classes; ChainMapper and ProductMapper keep their variadic __init__ via init=False. Public constructor signatures and behavior are unchanged.
Lee-W
force-pushed
the
partition-fanout-max-keys
branch
from
June 11, 2026 00:53
9d28124 to
5f26df1
Compare
github-project-automation
Bot
moved this from In Review
to Done
in AIP-76 Asset Partitioning
imrichardwu
pushed a commit
to imrichardwu/airflow
that referenced
this pull request
Jun 16, 2026
dingo4dev
pushed a commit
to dingo4dev/airflow
that referenced
this pull request
Jun 16, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
a per-mapper cap so each
PartitionMapperinstance can override the global[scheduler] partition_fanout_max_keys(implemented in #66030)Lives on the mapper rather than the Dag because one Dag can bind several mappers
with very different fan-out profiles.
closes: #65760
What
max_fan_out: int | None = NonetoPartitionMapperbase (SDK + core). Validator rejects0, negatives, non-int, andbool.None→ global; positive int → override.__init__, and through both serialization paths (per-class.serialize()and theencoders.pysingledispatch overrides). Field emitted only when non-None — pre-change payloads round-trip byte-identical.assets/manager.py: read the cap per target Dag, swap acap_sourcefragment into the audit-logextra(max_fan_out=Nvs. existing[scheduler] partition_fanout_max_keys=N).Was generative AI tooling used to co-author this PR?
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.