Skip to content

feat(core): add per-mapper max_fan_out override for partition fan-out cap#67184

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Lee-W merged 4 commits into from
Jun 11, 2026
Merged

feat(core): add per-mapper max_fan_out override for partition fan-out cap #67184

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
126b80a
feat(core): add per-mapper max_downstream_keys override for partition…
Lee-W May 19, 2026
5ff0fed
feat(core): log which cap was exceeded on partition fan-out overflow
Lee-W Jun 9, 2026
efa797e
refactor(task-sdk): validate max_downstream_keys with an attrs field …
Lee-W Jun 9, 2026
5f26df1
refactor(task-sdk): define partition mapper subclasses with attrs
Lee-W Jun 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions airflow-core/newsfragments/67184.feature.rst
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
Add ``max_downstream_keys`` parameter to ``PartitionMapper`` to override ``[scheduler] partition_mapper_max_downstream_keys`` per mapper instance.
19 changes: 13 additions & 6 deletions airflow-core/src/airflow/assets/manager.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -552,8 +552,7 @@ def _queue_partitioned_dags(
)
return

max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")

global_cap = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
for target_dag in partition_dags:
if TYPE_CHECKING:
assert partition_key is not None
Expand All@@ -573,9 +572,8 @@ def _queue_partitioned_dags(

try:
# We'll need to catch every possible exception happen when mapping partition_key.
target_key = timetable.get_partition_mapper(
name=asset_model.name, uri=asset_model.uri
).to_downstream(partition_key)
mapper = timetable.get_partition_mapper(name=asset_model.name, uri=asset_model.uri)
target_key = mapper.to_downstream(partition_key)
except Exception as err:
log.exception(
"Could not map partition key for asset in target Dag. "
Expand DownExpand Up@@ -607,6 +605,14 @@ def _queue_partitioned_dags(
target_keys = [target_key]
del target_key

mapper_cap = mapper.max_downstream_keys
if mapper_cap is not None:
max_downstream_keys = mapper_cap
cap_source = f"max_downstream_keys={mapper_cap}"
else:
max_downstream_keys = global_cap
cap_source = f"[scheduler] partition_mapper_max_downstream_keys={global_cap}"

if len(target_keys) > max_downstream_keys:
log.error(
Comment thread
Lee-W marked this conversation as resolved.
"Partition mapper produced more downstream keys than allowed; skipping queue.",
Expand All@@ -615,6 +621,7 @@ def _queue_partitioned_dags(
target_dag=target_dag.dag_id,
produced_keys=len(target_keys),
max_downstream_keys=max_downstream_keys,
cap_source=cap_source,
)
session.add(
Log(
Expand All@@ -624,7 +631,7 @@ def _queue_partitioned_dags(
f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
f"produced {len(target_keys)} downstream keys from "
f"partition_key='{partition_key}', exceeding "
f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
f"{cap_source}. "
f"No Dag runs were queued for this event."
),
task_instance=task_instance,
Expand Down
15 changes: 8 additions & 7 deletions airflow-core/src/airflow/config_templates/config.yml
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -2690,13 +2690,14 @@ scheduler:
see_also: ":ref:`scheduler:ha:tunables`"
partition_mapper_max_downstream_keys:
description: |
Maximum number of downstream partition keys a single ``PartitionMapper``
invocation may produce. When any partition mapper (built-in or custom)
expands one upstream key into more keys than this limit, the scheduler
skips queuing the runs for that asset event and logs an error against
the source task instance. This guards against a misconfigured
``PartitionMapper`` from queuing an unbounded number of Dag runs per
upstream event.
Maximum number of downstream partition keys produced by a single
PartitionMapper invocation, applied to any PartitionMapper that returns
multiple keys (e.g. FanOutMapper). When a mapper instance sets a per-instance
``max_downstream_keys`` parameter, that value completely overrides this global
cap for that instance — including when the per-mapper value exceeds this
global. **Deployment managers cannot enforce this setting as a hard
cluster-wide ceiling**; treat this value as a default that user code may
override.
version_added: 3.3.0
type: integer
example: ~
Expand Down
10 changes: 7 additions & 3 deletions airflow-core/src/airflow/partition_mappers/allowed_key.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@
class AllowedKeyMapper(PartitionMapper):
"""Partition mapper that validates keys against a set of allowed keys."""

def __init__(self, allowed_keys: list[str]) -> None:
def __init__(self, allowed_keys: list[str], *, max_downstream_keys: int | None = None) -> None:
super().__init__(max_downstream_keys=max_downstream_keys)
self.allowed_keys = allowed_keys

def to_downstream(self, key: str) -> str:
Expand All@@ -34,8 +35,11 @@ def to_downstream(self, key: str) -> str:
return key

def serialize(self) -> dict[str, Any]:
return {"allowed_keys": self.allowed_keys}
data: dict[str, Any] = {"allowed_keys": self.allowed_keys}
if self.max_downstream_keys is not None:
data["max_downstream_keys"] = self.max_downstream_keys
return data

@classmethod
def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
return cls(allowed_keys=data["allowed_keys"])
return cls(allowed_keys=data["allowed_keys"], max_downstream_keys=data.get("max_downstream_keys"))
26 changes: 22 additions & 4 deletions airflow-core/src/airflow/partition_mappers/base.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,15 @@ class PartitionMapper(ABC):

is_rollup: ClassVar[bool] = False

def __init__(self, *, max_downstream_keys: int | None = None) -> None:
if max_downstream_keys is not None and (
not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
):
raise ValueError(
f"max_downstream_keys must be a positive integer or None, got {max_downstream_keys!r}"
)
self.max_downstream_keys = max_downstream_keys

def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
decode_overridden = cls.decode_downstream is not PartitionMapper.decode_downstream
Expand DownExpand Up@@ -107,11 +116,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
return None

def serialize(self) -> dict[str, Any]:
return {}
if self.max_downstream_keys is None:
return {}
return {"max_downstream_keys": self.max_downstream_keys}

@classmethod
def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
return cls()
return cls(max_downstream_keys=data.get("max_downstream_keys"))


class RollupMapper(PartitionMapper):
Expand All@@ -126,7 +137,9 @@ class RollupMapper(PartitionMapper):

is_rollup: ClassVar[bool] = True

def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
def __init__(
self, *, upstream_mapper: PartitionMapper, window: Window, max_downstream_keys: int | None = None
) -> None:
decode_overridden = type(upstream_mapper).decode_downstream is not PartitionMapper.decode_downstream
if not decode_overridden and window.expected_decoded_type is not str:
raise TypeError(
Expand All@@ -138,6 +151,7 @@ def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
f"{window.expected_decoded_type.__name__}, or use a window whose "
f"'expected_decoded_type' accepts str."
)
super().__init__(max_downstream_keys=max_downstream_keys)
self.upstream_mapper = upstream_mapper
self.window = window

Expand All@@ -160,10 +174,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
def serialize(self) -> dict[str, Any]:
from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
data: dict[str, Any] = {
"upstream_mapper": encode_partition_mapper(self.upstream_mapper),
"window": encode_window(self.window),
}
if self.max_downstream_keys is not None:
data["max_downstream_keys"] = self.max_downstream_keys
return data

@classmethod
def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -172,6 +189,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
return cls(
upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
window=decode_window(data["window"]),
max_downstream_keys=data.get("max_downstream_keys"),
)


Expand Down
9 changes: 7 additions & 2 deletions airflow-core/src/airflow/partition_mappers/chain.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -35,7 +35,9 @@ def __init__(
mapper1: PartitionMapper,
/,
*mappers: PartitionMapper,
max_downstream_keys: int | None = None,
) -> None:
super().__init__(max_downstream_keys=max_downstream_keys)
self.mappers = [mapper0, mapper1, *mappers]

def to_downstream(self, key: str) -> str | Iterable[str]:
Expand DownExpand Up@@ -70,11 +72,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
def serialize(self) -> dict[str, Any]:
from airflow.serialization.encoders import encode_partition_mapper

return {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
result: dict[str, Any] = {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
if self.max_downstream_keys is not None:
result["max_downstream_keys"] = self.max_downstream_keys
return result

@classmethod
def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
from airflow.serialization.decoders import decode_partition_mapper

mappers = [decode_partition_mapper(m) for m in data["mappers"]]
return cls(*mappers)
return cls(*mappers, max_downstream_keys=data.get("max_downstream_keys"))
13 changes: 11 additions & 2 deletions airflow-core/src/airflow/partition_mappers/product.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,9 @@ def __init__(
/,
*mappers: PartitionMapper,
delimiter: str = "|",
max_downstream_keys: int | None = None,
) -> None:
super().__init__(max_downstream_keys=max_downstream_keys)
self.mappers = [mapper0, mapper1, *mappers]
self.delimiter = delimiter

Expand All@@ -54,14 +56,21 @@ def to_downstream(self, key: str) -> str:
def serialize(self) -> dict[str, Any]:
from airflow.serialization.encoders import encode_partition_mapper

return {
result: dict[str, Any] = {
"delimiter": self.delimiter,
"mappers": [encode_partition_mapper(m) for m in self.mappers],
}
if self.max_downstream_keys is not None:
result["max_downstream_keys"] = self.max_downstream_keys
return result

@classmethod
def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
from airflow.serialization.decoders import decode_partition_mapper

mappers = [decode_partition_mapper(m) for m in data["mappers"]]
return cls(*mappers, delimiter=data.get("delimiter", "|"))
return cls(
*mappers,
delimiter=data.get("delimiter", "|"),
max_downstream_keys=data.get("max_downstream_keys"),
)
32 changes: 28 additions & 4 deletions airflow-core/src/airflow/partition_mappers/temporal.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -163,7 +163,9 @@ def __init__(
timezone: str | Timezone | FixedTimezone = "UTC",
input_format: str = "%Y-%m-%dT%H:%M:%S",
output_format: str | None = None,
max_downstream_keys: int | None = None,
):
super().__init__(max_downstream_keys=max_downstream_keys)
self.input_format = input_format
self.output_format = output_format or self.default_output_format
if isinstance(timezone, str):
Expand DownExpand Up@@ -230,18 +232,22 @@ def encode_upstream(self, dt: datetime) -> str:
def serialize(self) -> dict[str, Any]:
from airflow.serialization.encoders import encode_timezone

return {
result: dict[str, Any] = {
"timezone": encode_timezone(self._timezone),
"input_format": self.input_format,
"output_format": self.output_format,
}
if self.max_downstream_keys is not None:
result["max_downstream_keys"] = self.max_downstream_keys
return result

@classmethod
def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
return cls(
timezone=parse_timezone(data.get("timezone", "UTC")),
input_format=data["input_format"],
output_format=data["output_format"],
max_downstream_keys=data.get("max_downstream_keys"),
)


Expand DownExpand Up@@ -286,6 +292,7 @@ def __init__(
timezone: str | Timezone | FixedTimezone = "UTC",
input_format: str = "%Y-%m-%dT%H:%M:%S",
output_format: str | None = None,
max_downstream_keys: int | None = None,
) -> None:
"""
Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -302,7 +309,12 @@ def __init__(
**must** include ``%Y``, ``%m``, and ``%d`` so the week-start date
can be recovered for ``to_upstream``.
"""
super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
super().__init__(
timezone=timezone,
input_format=input_format,
output_format=output_format,
max_downstream_keys=max_downstream_keys,
)
# %V (ISO week) cannot be round-tripped through strptime without %G+%u,
# so derive a named-group regex from output_format and pull out %Y/%m/%d.
# Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -358,6 +370,7 @@ def __init__(
timezone: str | Timezone | FixedTimezone = "UTC",
input_format: str = "%Y-%m-%dT%H:%M:%S",
output_format: str | None = None,
max_downstream_keys: int | None = None,
) -> None:
"""
Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -376,7 +389,12 @@ def __init__(
and ``{quarter}`` so the quarter-start date can be recovered for
``to_upstream``.
"""
super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
super().__init__(
timezone=timezone,
input_format=input_format,
output_format=output_format,
max_downstream_keys=max_downstream_keys,
)
# ``{quarter}`` is a Python-format placeholder, not a strftime directive,
# so derive a named-group regex from output_format that handles both.
# Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -511,7 +529,9 @@ def __init__(
upstream_mapper: PartitionMapper,
window: Window,
downstream_mapper: PartitionMapper | None = None,
max_downstream_keys: int | None = None,
) -> None:
super().__init__(max_downstream_keys=max_downstream_keys)
self.upstream_mapper = upstream_mapper
self.window = window
self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
Expand All@@ -537,11 +557,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
def serialize(self) -> dict[str, Any]:
from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
result: dict[str, Any] = {
"upstream_mapper": encode_partition_mapper(self.upstream_mapper),
"window": encode_window(self.window),
"downstream_mapper": encode_partition_mapper(self.downstream_mapper),
}
if self.max_downstream_keys is not None:
result["max_downstream_keys"] = self.max_downstream_keys
return result

@classmethod
def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -551,6 +574,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
window=decode_window(data["window"]),
downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
max_downstream_keys=data.get("max_downstream_keys"),
)


Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
feat(core): add per-mapper max_fan_out override for partition fan-out cap by Lee-W · Pull Request #67184 · apache/airflow · GitHub
Skip to content
feat(core): add per-mapper max_fan_out override for partition fan-out cap#67184

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 4 commits into

from

Jun 11, 2026
 Merged


feat(core): add per-mapper max_fan_out override for partition fan-out cap
#67184
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
126b80a

feat(core): add per-mapper max_downstream_keys override for partition…


Lee-W May 19, 2026
5ff0fed

feat(core): log which cap was exceeded on partition fan-out overflow


Lee-W Jun 9, 2026
efa797e

refactor(task-sdk): validate max_downstream_keys with an attrs field …


Lee-W Jun 9, 2026
5f26df1

refactor(task-sdk): define partition mapper subclasses with attrs


Lee-W Jun 10, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
airflow-core/newsfragments/67184.feature.rst

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
Add ``max_downstream_keys`` parameter to ``PartitionMapper`` to override ``[scheduler] partition_mapper_max_downstream_keys`` per mapper instance.

19 changes: 13 additions & 6 deletions
airflow-core/src/airflow/assets/manager.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -552,8 +552,7 @@ def _queue_partitioned_dags(
 )
 return

 max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")

 global_cap = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
 for target_dag in partition_dags:
 if TYPE_CHECKING:
 assert partition_key is not None
Expand All@@ -573,9 +572,8 @@ def _queue_partitioned_dags(

 try:
 # We'll need to catch every possible exception happen when mapping partition_key.
 target_key = timetable.get_partition_mapper(
 name=asset_model.name, uri=asset_model.uri
 ).to_downstream(partition_key)
 mapper = timetable.get_partition_mapper(name=asset_model.name, uri=asset_model.uri)
 target_key = mapper.to_downstream(partition_key)
 except Exception as err:
 log.exception(
 "Could not map partition key for asset in target Dag. "
Expand DownExpand Up@@ -607,6 +605,14 @@ def _queue_partitioned_dags(
 target_keys = [target_key]
 del target_key

 mapper_cap = mapper.max_downstream_keys
 if mapper_cap is not None:
 max_downstream_keys = mapper_cap
 cap_source = f"max_downstream_keys={mapper_cap}"
 else:
 max_downstream_keys = global_cap
 cap_source = f"[scheduler] partition_mapper_max_downstream_keys={global_cap}"

 if len(target_keys) > max_downstream_keys:
 log.error(
Comment thread
Lee-W marked this conversation as resolved.

 "Partition mapper produced more downstream keys than allowed; skipping queue.",
Expand All@@ -615,6 +621,7 @@ def _queue_partitioned_dags(
 target_dag=target_dag.dag_id,
 produced_keys=len(target_keys),
 max_downstream_keys=max_downstream_keys,
 cap_source=cap_source,
 )
 session.add(
 Log(
Expand All@@ -624,7 +631,7 @@ def _queue_partitioned_dags(
 f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
 f"produced {len(target_keys)} downstream keys from "
 f"partition_key='{partition_key}', exceeding "
 f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
 f"{cap_source}. "
 f"No Dag runs were queued for this event."
 ),
 task_instance=task_instance,
Expand Down

15 changes: 8 additions & 7 deletions
airflow-core/src/airflow/config_templates/config.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -2690,13 +2690,14 @@ scheduler:
 see_also: ":ref:`scheduler:ha:tunables`"
 partition_mapper_max_downstream_keys:
 description: |
 Maximum number of downstream partition keys a single ``PartitionMapper``
 invocation may produce. When any partition mapper (built-in or custom)
 expands one upstream key into more keys than this limit, the scheduler
 skips queuing the runs for that asset event and logs an error against
 the source task instance. This guards against a misconfigured
 ``PartitionMapper`` from queuing an unbounded number of Dag runs per
 upstream event.
 Maximum number of downstream partition keys produced by a single
 PartitionMapper invocation, applied to any PartitionMapper that returns
 multiple keys (e.g. FanOutMapper). When a mapper instance sets a per-instance
 ``max_downstream_keys`` parameter, that value completely overrides this global
 cap for that instance — including when the per-mapper value exceeds this
 global. **Deployment managers cannot enforce this setting as a hard
 cluster-wide ceiling**; treat this value as a default that user code may
 override.
 version_added: 3.3.0
 type: integer
 example: ~
Expand Down

10 changes: 7 additions & 3 deletions
airflow-core/src/airflow/partition_mappers/allowed_key.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@
class AllowedKeyMapper(PartitionMapper):
 """Partition mapper that validates keys against a set of allowed keys."""

 def __init__(self, allowed_keys: list[str]) -> None:
 def __init__(self, allowed_keys: list[str], *, max_downstream_keys: int | None = None) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.allowed_keys = allowed_keys

 def to_downstream(self, key: str) -> str:
Expand All@@ -34,8 +35,11 @@ def to_downstream(self, key: str) -> str:
 return key

 def serialize(self) -> dict[str, Any]:
 return {"allowed_keys": self.allowed_keys}
 data: dict[str, Any] = {"allowed_keys": self.allowed_keys}
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(allowed_keys=data["allowed_keys"])
 return cls(allowed_keys=data["allowed_keys"], max_downstream_keys=data.get("max_downstream_keys"))

26 changes: 22 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/base.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,15 @@ class PartitionMapper(ABC):

 is_rollup: ClassVar[bool] = False

 def __init__(self, *, max_downstream_keys: int | None = None) -> None:
 if max_downstream_keys is not None and (
 not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
 ):
 raise ValueError(
 f"max_downstream_keys must be a positive integer or None, got {max_downstream_keys!r}"
 )
 self.max_downstream_keys = max_downstream_keys

 def __init_subclass__(cls, **kwargs: Any) -> None:
 super().__init_subclass__(**kwargs)
 decode_overridden = cls.decode_downstream is not PartitionMapper.decode_downstream
Expand DownExpand Up@@ -107,11 +116,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 return None

 def serialize(self) -> dict[str, Any]:
 return {}
 if self.max_downstream_keys is None:
 return {}
 return {"max_downstream_keys": self.max_downstream_keys}

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls()
 return cls(max_downstream_keys=data.get("max_downstream_keys"))


class RollupMapper(PartitionMapper):
Expand All@@ -126,7 +137,9 @@ class RollupMapper(PartitionMapper):

 is_rollup: ClassVar[bool] = True

 def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 def __init__(
 self, *, upstream_mapper: PartitionMapper, window: Window, max_downstream_keys: int | None = None
 ) -> None:
 decode_overridden = type(upstream_mapper).decode_downstream is not PartitionMapper.decode_downstream
 if not decode_overridden and window.expected_decoded_type is not str:
 raise TypeError(
Expand All@@ -138,6 +151,7 @@ def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 f"{window.expected_decoded_type.__name__}, or use a window whose "
 f"'expected_decoded_type' accepts str."
 )
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window

Expand All@@ -160,10 +174,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
data: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 }
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -172,6 +189,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down

9 changes: 7 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/chain.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -35,7 +35,9 @@ def __init__(
 mapper1: PartitionMapper,
 /,
 *mappers: PartitionMapper,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]

 def to_downstream(self, key: str) -> str | Iterable[str]:
Expand DownExpand Up@@ -70,11 +72,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

 return {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 result: dict[str, Any] = {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers)
 return cls(*mappers, max_downstream_keys=data.get("max_downstream_keys"))

13 changes: 11 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/product.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,9 @@ def __init__(
 /,
 *mappers: PartitionMapper,
 delimiter: str = "|",
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]
 self.delimiter = delimiter

Expand All@@ -54,14 +56,21 @@ def to_downstream(self, key: str) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

return {
result: dict[str, Any] = {
 "delimiter": self.delimiter,
 "mappers": [encode_partition_mapper(m) for m in self.mappers],
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers, delimiter=data.get("delimiter", "|"))
 return cls(
 *mappers,
 delimiter=data.get("delimiter", "|"),
 max_downstream_keys=data.get("max_downstream_keys"),
 )

32 changes: 28 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/temporal.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -163,7 +163,9 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ):
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.input_format = input_format
 self.output_format = output_format or self.default_output_format
 if isinstance(timezone, str):
Expand DownExpand Up@@ -230,18 +232,22 @@ def encode_upstream(self, dt: datetime) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_timezone

return {
result: dict[str, Any] = {
 "timezone": encode_timezone(self._timezone),
 "input_format": self.input_format,
 "output_format": self.output_format,
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 timezone=parse_timezone(data.get("timezone", "UTC")),
 input_format=data["input_format"],
 output_format=data["output_format"],
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand DownExpand Up@@ -286,6 +292,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -302,7 +309,12 @@ def __init__(
 **must** include ``%Y``, ``%m``, and ``%d`` so the week-start date
 can be recovered for ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # %V (ISO week) cannot be round-tripped through strptime without %G+%u,
 # so derive a named-group regex from output_format and pull out %Y/%m/%d.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -358,6 +370,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -376,7 +389,12 @@ def __init__(
 and ``{quarter}`` so the quarter-start date can be recovered for
 ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # ``{quarter}`` is a Python-format placeholder, not a strftime directive,
 # so derive a named-group regex from output_format that handles both.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -511,7 +529,9 @@ def __init__(
 upstream_mapper: PartitionMapper,
 window: Window,
 downstream_mapper: PartitionMapper | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window
 self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
Expand All@@ -537,11 +557,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
result: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -551,6 +574,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
feat(core): add per-mapper max_fan_out override for partition fan-out cap by Lee-W · Pull Request #67184 · apache/airflow · GitHub
Skip to content
feat(core): add per-mapper max_fan_out override for partition fan-out cap#67184

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 4 commits into

from

Jun 11, 2026
 Merged


feat(core): add per-mapper max_fan_out override for partition fan-out cap
#67184
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
126b80a

feat(core): add per-mapper max_downstream_keys override for partition…


Lee-W May 19, 2026
5ff0fed

feat(core): log which cap was exceeded on partition fan-out overflow


Lee-W Jun 9, 2026
efa797e

refactor(task-sdk): validate max_downstream_keys with an attrs field …


Lee-W Jun 9, 2026
5f26df1

refactor(task-sdk): define partition mapper subclasses with attrs


Lee-W Jun 10, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
airflow-core/newsfragments/67184.feature.rst

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
Add ``max_downstream_keys`` parameter to ``PartitionMapper`` to override ``[scheduler] partition_mapper_max_downstream_keys`` per mapper instance.

19 changes: 13 additions & 6 deletions
airflow-core/src/airflow/assets/manager.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -552,8 +552,7 @@ def _queue_partitioned_dags(
 )
 return

 max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")

 global_cap = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
 for target_dag in partition_dags:
 if TYPE_CHECKING:
 assert partition_key is not None
Expand All@@ -573,9 +572,8 @@ def _queue_partitioned_dags(

 try:
 # We'll need to catch every possible exception happen when mapping partition_key.
 target_key = timetable.get_partition_mapper(
 name=asset_model.name, uri=asset_model.uri
 ).to_downstream(partition_key)
 mapper = timetable.get_partition_mapper(name=asset_model.name, uri=asset_model.uri)
 target_key = mapper.to_downstream(partition_key)
 except Exception as err:
 log.exception(
 "Could not map partition key for asset in target Dag. "
Expand DownExpand Up@@ -607,6 +605,14 @@ def _queue_partitioned_dags(
 target_keys = [target_key]
 del target_key

 mapper_cap = mapper.max_downstream_keys
 if mapper_cap is not None:
 max_downstream_keys = mapper_cap
 cap_source = f"max_downstream_keys={mapper_cap}"
 else:
 max_downstream_keys = global_cap
 cap_source = f"[scheduler] partition_mapper_max_downstream_keys={global_cap}"

 if len(target_keys) > max_downstream_keys:
 log.error(
Comment thread
Lee-W marked this conversation as resolved.

 "Partition mapper produced more downstream keys than allowed; skipping queue.",
Expand All@@ -615,6 +621,7 @@ def _queue_partitioned_dags(
 target_dag=target_dag.dag_id,
 produced_keys=len(target_keys),
 max_downstream_keys=max_downstream_keys,
 cap_source=cap_source,
 )
 session.add(
 Log(
Expand All@@ -624,7 +631,7 @@ def _queue_partitioned_dags(
 f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
 f"produced {len(target_keys)} downstream keys from "
 f"partition_key='{partition_key}', exceeding "
 f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
 f"{cap_source}. "
 f"No Dag runs were queued for this event."
 ),
 task_instance=task_instance,
Expand Down

15 changes: 8 additions & 7 deletions
airflow-core/src/airflow/config_templates/config.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -2690,13 +2690,14 @@ scheduler:
 see_also: ":ref:`scheduler:ha:tunables`"
 partition_mapper_max_downstream_keys:
 description: |
 Maximum number of downstream partition keys a single ``PartitionMapper``
 invocation may produce. When any partition mapper (built-in or custom)
 expands one upstream key into more keys than this limit, the scheduler
 skips queuing the runs for that asset event and logs an error against
 the source task instance. This guards against a misconfigured
 ``PartitionMapper`` from queuing an unbounded number of Dag runs per
 upstream event.
 Maximum number of downstream partition keys produced by a single
 PartitionMapper invocation, applied to any PartitionMapper that returns
 multiple keys (e.g. FanOutMapper). When a mapper instance sets a per-instance
 ``max_downstream_keys`` parameter, that value completely overrides this global
 cap for that instance — including when the per-mapper value exceeds this
 global. **Deployment managers cannot enforce this setting as a hard
 cluster-wide ceiling**; treat this value as a default that user code may
 override.
 version_added: 3.3.0
 type: integer
 example: ~
Expand Down

10 changes: 7 additions & 3 deletions
airflow-core/src/airflow/partition_mappers/allowed_key.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@
class AllowedKeyMapper(PartitionMapper):
 """Partition mapper that validates keys against a set of allowed keys."""

 def __init__(self, allowed_keys: list[str]) -> None:
 def __init__(self, allowed_keys: list[str], *, max_downstream_keys: int | None = None) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.allowed_keys = allowed_keys

 def to_downstream(self, key: str) -> str:
Expand All@@ -34,8 +35,11 @@ def to_downstream(self, key: str) -> str:
 return key

 def serialize(self) -> dict[str, Any]:
 return {"allowed_keys": self.allowed_keys}
 data: dict[str, Any] = {"allowed_keys": self.allowed_keys}
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(allowed_keys=data["allowed_keys"])
 return cls(allowed_keys=data["allowed_keys"], max_downstream_keys=data.get("max_downstream_keys"))

26 changes: 22 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/base.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,15 @@ class PartitionMapper(ABC):

 is_rollup: ClassVar[bool] = False

 def __init__(self, *, max_downstream_keys: int | None = None) -> None:
 if max_downstream_keys is not None and (
 not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
 ):
 raise ValueError(
 f"max_downstream_keys must be a positive integer or None, got {max_downstream_keys!r}"
 )
 self.max_downstream_keys = max_downstream_keys

 def __init_subclass__(cls, **kwargs: Any) -> None:
 super().__init_subclass__(**kwargs)
 decode_overridden = cls.decode_downstream is not PartitionMapper.decode_downstream
Expand DownExpand Up@@ -107,11 +116,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 return None

 def serialize(self) -> dict[str, Any]:
 return {}
 if self.max_downstream_keys is None:
 return {}
 return {"max_downstream_keys": self.max_downstream_keys}

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls()
 return cls(max_downstream_keys=data.get("max_downstream_keys"))


class RollupMapper(PartitionMapper):
Expand All@@ -126,7 +137,9 @@ class RollupMapper(PartitionMapper):

 is_rollup: ClassVar[bool] = True

 def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 def __init__(
 self, *, upstream_mapper: PartitionMapper, window: Window, max_downstream_keys: int | None = None
 ) -> None:
 decode_overridden = type(upstream_mapper).decode_downstream is not PartitionMapper.decode_downstream
 if not decode_overridden and window.expected_decoded_type is not str:
 raise TypeError(
Expand All@@ -138,6 +151,7 @@ def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 f"{window.expected_decoded_type.__name__}, or use a window whose "
 f"'expected_decoded_type' accepts str."
 )
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window

Expand All@@ -160,10 +174,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
data: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 }
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -172,6 +189,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down

9 changes: 7 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/chain.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -35,7 +35,9 @@ def __init__(
 mapper1: PartitionMapper,
 /,
 *mappers: PartitionMapper,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]

 def to_downstream(self, key: str) -> str | Iterable[str]:
Expand DownExpand Up@@ -70,11 +72,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

 return {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 result: dict[str, Any] = {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers)
 return cls(*mappers, max_downstream_keys=data.get("max_downstream_keys"))

13 changes: 11 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/product.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,9 @@ def __init__(
 /,
 *mappers: PartitionMapper,
 delimiter: str = "|",
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]
 self.delimiter = delimiter

Expand All@@ -54,14 +56,21 @@ def to_downstream(self, key: str) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

return {
result: dict[str, Any] = {
 "delimiter": self.delimiter,
 "mappers": [encode_partition_mapper(m) for m in self.mappers],
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers, delimiter=data.get("delimiter", "|"))
 return cls(
 *mappers,
 delimiter=data.get("delimiter", "|"),
 max_downstream_keys=data.get("max_downstream_keys"),
 )

32 changes: 28 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/temporal.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -163,7 +163,9 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ):
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.input_format = input_format
 self.output_format = output_format or self.default_output_format
 if isinstance(timezone, str):
Expand DownExpand Up@@ -230,18 +232,22 @@ def encode_upstream(self, dt: datetime) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_timezone

return {
result: dict[str, Any] = {
 "timezone": encode_timezone(self._timezone),
 "input_format": self.input_format,
 "output_format": self.output_format,
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 timezone=parse_timezone(data.get("timezone", "UTC")),
 input_format=data["input_format"],
 output_format=data["output_format"],
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand DownExpand Up@@ -286,6 +292,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -302,7 +309,12 @@ def __init__(
 **must** include ``%Y``, ``%m``, and ``%d`` so the week-start date
 can be recovered for ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # %V (ISO week) cannot be round-tripped through strptime without %G+%u,
 # so derive a named-group regex from output_format and pull out %Y/%m/%d.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -358,6 +370,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -376,7 +389,12 @@ def __init__(
 and ``{quarter}`` so the quarter-start date can be recovered for
 ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # ``{quarter}`` is a Python-format placeholder, not a strftime directive,
 # so derive a named-group regex from output_format that handles both.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -511,7 +529,9 @@ def __init__(
 upstream_mapper: PartitionMapper,
 window: Window,
 downstream_mapper: PartitionMapper | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window
 self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
Expand All@@ -537,11 +557,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
result: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -551,6 +574,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
feat(core): add per-mapper max_fan_out override for partition fan-out cap by Lee-W · Pull Request #67184 · apache/airflow · GitHub
Skip to content
feat(core): add per-mapper max_fan_out override for partition fan-out cap#67184

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 4 commits into

from

Jun 11, 2026
 Merged


feat(core): add per-mapper max_fan_out override for partition fan-out cap
#67184
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
126b80a

feat(core): add per-mapper max_downstream_keys override for partition…


Lee-W May 19, 2026
5ff0fed

feat(core): log which cap was exceeded on partition fan-out overflow


Lee-W Jun 9, 2026
efa797e

refactor(task-sdk): validate max_downstream_keys with an attrs field …


Lee-W Jun 9, 2026
5f26df1

refactor(task-sdk): define partition mapper subclasses with attrs


Lee-W Jun 10, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
airflow-core/newsfragments/67184.feature.rst

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
Add ``max_downstream_keys`` parameter to ``PartitionMapper`` to override ``[scheduler] partition_mapper_max_downstream_keys`` per mapper instance.

19 changes: 13 additions & 6 deletions
airflow-core/src/airflow/assets/manager.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -552,8 +552,7 @@ def _queue_partitioned_dags(
 )
 return

 max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")

 global_cap = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
 for target_dag in partition_dags:
 if TYPE_CHECKING:
 assert partition_key is not None
Expand All@@ -573,9 +572,8 @@ def _queue_partitioned_dags(

 try:
 # We'll need to catch every possible exception happen when mapping partition_key.
 target_key = timetable.get_partition_mapper(
 name=asset_model.name, uri=asset_model.uri
 ).to_downstream(partition_key)
 mapper = timetable.get_partition_mapper(name=asset_model.name, uri=asset_model.uri)
 target_key = mapper.to_downstream(partition_key)
 except Exception as err:
 log.exception(
 "Could not map partition key for asset in target Dag. "
Expand DownExpand Up@@ -607,6 +605,14 @@ def _queue_partitioned_dags(
 target_keys = [target_key]
 del target_key

 mapper_cap = mapper.max_downstream_keys
 if mapper_cap is not None:
 max_downstream_keys = mapper_cap
 cap_source = f"max_downstream_keys={mapper_cap}"
 else:
 max_downstream_keys = global_cap
 cap_source = f"[scheduler] partition_mapper_max_downstream_keys={global_cap}"

 if len(target_keys) > max_downstream_keys:
 log.error(
Comment thread
Lee-W marked this conversation as resolved.

 "Partition mapper produced more downstream keys than allowed; skipping queue.",
Expand All@@ -615,6 +621,7 @@ def _queue_partitioned_dags(
 target_dag=target_dag.dag_id,
 produced_keys=len(target_keys),
 max_downstream_keys=max_downstream_keys,
 cap_source=cap_source,
 )
 session.add(
 Log(
Expand All@@ -624,7 +631,7 @@ def _queue_partitioned_dags(
 f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
 f"produced {len(target_keys)} downstream keys from "
 f"partition_key='{partition_key}', exceeding "
 f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
 f"{cap_source}. "
 f"No Dag runs were queued for this event."
 ),
 task_instance=task_instance,
Expand Down

15 changes: 8 additions & 7 deletions
airflow-core/src/airflow/config_templates/config.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -2690,13 +2690,14 @@ scheduler:
 see_also: ":ref:`scheduler:ha:tunables`"
 partition_mapper_max_downstream_keys:
 description: |
 Maximum number of downstream partition keys a single ``PartitionMapper``
 invocation may produce. When any partition mapper (built-in or custom)
 expands one upstream key into more keys than this limit, the scheduler
 skips queuing the runs for that asset event and logs an error against
 the source task instance. This guards against a misconfigured
 ``PartitionMapper`` from queuing an unbounded number of Dag runs per
 upstream event.
 Maximum number of downstream partition keys produced by a single
 PartitionMapper invocation, applied to any PartitionMapper that returns
 multiple keys (e.g. FanOutMapper). When a mapper instance sets a per-instance
 ``max_downstream_keys`` parameter, that value completely overrides this global
 cap for that instance — including when the per-mapper value exceeds this
 global. **Deployment managers cannot enforce this setting as a hard
 cluster-wide ceiling**; treat this value as a default that user code may
 override.
 version_added: 3.3.0
 type: integer
 example: ~
Expand Down

10 changes: 7 additions & 3 deletions
airflow-core/src/airflow/partition_mappers/allowed_key.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@
class AllowedKeyMapper(PartitionMapper):
 """Partition mapper that validates keys against a set of allowed keys."""

 def __init__(self, allowed_keys: list[str]) -> None:
 def __init__(self, allowed_keys: list[str], *, max_downstream_keys: int | None = None) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.allowed_keys = allowed_keys

 def to_downstream(self, key: str) -> str:
Expand All@@ -34,8 +35,11 @@ def to_downstream(self, key: str) -> str:
 return key

 def serialize(self) -> dict[str, Any]:
 return {"allowed_keys": self.allowed_keys}
 data: dict[str, Any] = {"allowed_keys": self.allowed_keys}
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(allowed_keys=data["allowed_keys"])
 return cls(allowed_keys=data["allowed_keys"], max_downstream_keys=data.get("max_downstream_keys"))

26 changes: 22 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/base.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,15 @@ class PartitionMapper(ABC):

 is_rollup: ClassVar[bool] = False

 def __init__(self, *, max_downstream_keys: int | None = None) -> None:
 if max_downstream_keys is not None and (
 not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
 ):
 raise ValueError(
 f"max_downstream_keys must be a positive integer or None, got {max_downstream_keys!r}"
 )
 self.max_downstream_keys = max_downstream_keys

 def __init_subclass__(cls, **kwargs: Any) -> None:
 super().__init_subclass__(**kwargs)
 decode_overridden = cls.decode_downstream is not PartitionMapper.decode_downstream
Expand DownExpand Up@@ -107,11 +116,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 return None

 def serialize(self) -> dict[str, Any]:
 return {}
 if self.max_downstream_keys is None:
 return {}
 return {"max_downstream_keys": self.max_downstream_keys}

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls()
 return cls(max_downstream_keys=data.get("max_downstream_keys"))


class RollupMapper(PartitionMapper):
Expand All@@ -126,7 +137,9 @@ class RollupMapper(PartitionMapper):

 is_rollup: ClassVar[bool] = True

 def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 def __init__(
 self, *, upstream_mapper: PartitionMapper, window: Window, max_downstream_keys: int | None = None
 ) -> None:
 decode_overridden = type(upstream_mapper).decode_downstream is not PartitionMapper.decode_downstream
 if not decode_overridden and window.expected_decoded_type is not str:
 raise TypeError(
Expand All@@ -138,6 +151,7 @@ def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 f"{window.expected_decoded_type.__name__}, or use a window whose "
 f"'expected_decoded_type' accepts str."
 )
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window

Expand All@@ -160,10 +174,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
data: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 }
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -172,6 +189,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down

9 changes: 7 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/chain.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -35,7 +35,9 @@ def __init__(
 mapper1: PartitionMapper,
 /,
 *mappers: PartitionMapper,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]

 def to_downstream(self, key: str) -> str | Iterable[str]:
Expand DownExpand Up@@ -70,11 +72,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

 return {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 result: dict[str, Any] = {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers)
 return cls(*mappers, max_downstream_keys=data.get("max_downstream_keys"))

13 changes: 11 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/product.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,9 @@ def __init__(
 /,
 *mappers: PartitionMapper,
 delimiter: str = "|",
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]
 self.delimiter = delimiter

Expand All@@ -54,14 +56,21 @@ def to_downstream(self, key: str) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

return {
result: dict[str, Any] = {
 "delimiter": self.delimiter,
 "mappers": [encode_partition_mapper(m) for m in self.mappers],
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers, delimiter=data.get("delimiter", "|"))
 return cls(
 *mappers,
 delimiter=data.get("delimiter", "|"),
 max_downstream_keys=data.get("max_downstream_keys"),
 )

32 changes: 28 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/temporal.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -163,7 +163,9 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ):
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.input_format = input_format
 self.output_format = output_format or self.default_output_format
 if isinstance(timezone, str):
Expand DownExpand Up@@ -230,18 +232,22 @@ def encode_upstream(self, dt: datetime) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_timezone

return {
result: dict[str, Any] = {
 "timezone": encode_timezone(self._timezone),
 "input_format": self.input_format,
 "output_format": self.output_format,
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 timezone=parse_timezone(data.get("timezone", "UTC")),
 input_format=data["input_format"],
 output_format=data["output_format"],
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand DownExpand Up@@ -286,6 +292,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -302,7 +309,12 @@ def __init__(
 **must** include ``%Y``, ``%m``, and ``%d`` so the week-start date
 can be recovered for ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # %V (ISO week) cannot be round-tripped through strptime without %G+%u,
 # so derive a named-group regex from output_format and pull out %Y/%m/%d.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -358,6 +370,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -376,7 +389,12 @@ def __init__(
 and ``{quarter}`` so the quarter-start date can be recovered for
 ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # ``{quarter}`` is a Python-format placeholder, not a strftime directive,
 # so derive a named-group regex from output_format that handles both.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -511,7 +529,9 @@ def __init__(
 upstream_mapper: PartitionMapper,
 window: Window,
 downstream_mapper: PartitionMapper | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window
 self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
Expand All@@ -537,11 +557,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
result: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -551,6 +574,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
feat(core): add per-mapper max_fan_out override for partition fan-out cap by Lee-W · Pull Request #67184 · apache/airflow · GitHub
Skip to content
feat(core): add per-mapper max_fan_out override for partition fan-out cap#67184

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 4 commits into

from

Jun 11, 2026
 Merged


feat(core): add per-mapper max_fan_out override for partition fan-out cap
#67184
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
126b80a

feat(core): add per-mapper max_downstream_keys override for partition…


Lee-W May 19, 2026
5ff0fed

feat(core): log which cap was exceeded on partition fan-out overflow


Lee-W Jun 9, 2026
efa797e

refactor(task-sdk): validate max_downstream_keys with an attrs field …


Lee-W Jun 9, 2026
5f26df1

refactor(task-sdk): define partition mapper subclasses with attrs


Lee-W Jun 10, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
airflow-core/newsfragments/67184.feature.rst

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
Add ``max_downstream_keys`` parameter to ``PartitionMapper`` to override ``[scheduler] partition_mapper_max_downstream_keys`` per mapper instance.

19 changes: 13 additions & 6 deletions
airflow-core/src/airflow/assets/manager.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -552,8 +552,7 @@ def _queue_partitioned_dags(
 )
 return

 max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")

 global_cap = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
 for target_dag in partition_dags:
 if TYPE_CHECKING:
 assert partition_key is not None
Expand All@@ -573,9 +572,8 @@ def _queue_partitioned_dags(

 try:
 # We'll need to catch every possible exception happen when mapping partition_key.
 target_key = timetable.get_partition_mapper(
 name=asset_model.name, uri=asset_model.uri
 ).to_downstream(partition_key)
 mapper = timetable.get_partition_mapper(name=asset_model.name, uri=asset_model.uri)
 target_key = mapper.to_downstream(partition_key)
 except Exception as err:
 log.exception(
 "Could not map partition key for asset in target Dag. "
Expand DownExpand Up@@ -607,6 +605,14 @@ def _queue_partitioned_dags(
 target_keys = [target_key]
 del target_key

 mapper_cap = mapper.max_downstream_keys
 if mapper_cap is not None:
 max_downstream_keys = mapper_cap
 cap_source = f"max_downstream_keys={mapper_cap}"
 else:
 max_downstream_keys = global_cap
 cap_source = f"[scheduler] partition_mapper_max_downstream_keys={global_cap}"

 if len(target_keys) > max_downstream_keys:
 log.error(
Comment thread
Lee-W marked this conversation as resolved.

 "Partition mapper produced more downstream keys than allowed; skipping queue.",
Expand All@@ -615,6 +621,7 @@ def _queue_partitioned_dags(
 target_dag=target_dag.dag_id,
 produced_keys=len(target_keys),
 max_downstream_keys=max_downstream_keys,
 cap_source=cap_source,
 )
 session.add(
 Log(
Expand All@@ -624,7 +631,7 @@ def _queue_partitioned_dags(
 f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
 f"produced {len(target_keys)} downstream keys from "
 f"partition_key='{partition_key}', exceeding "
 f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
 f"{cap_source}. "
 f"No Dag runs were queued for this event."
 ),
 task_instance=task_instance,
Expand Down

15 changes: 8 additions & 7 deletions
airflow-core/src/airflow/config_templates/config.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -2690,13 +2690,14 @@ scheduler:
 see_also: ":ref:`scheduler:ha:tunables`"
 partition_mapper_max_downstream_keys:
 description: |
 Maximum number of downstream partition keys a single ``PartitionMapper``
 invocation may produce. When any partition mapper (built-in or custom)
 expands one upstream key into more keys than this limit, the scheduler
 skips queuing the runs for that asset event and logs an error against
 the source task instance. This guards against a misconfigured
 ``PartitionMapper`` from queuing an unbounded number of Dag runs per
 upstream event.
 Maximum number of downstream partition keys produced by a single
 PartitionMapper invocation, applied to any PartitionMapper that returns
 multiple keys (e.g. FanOutMapper). When a mapper instance sets a per-instance
 ``max_downstream_keys`` parameter, that value completely overrides this global
 cap for that instance — including when the per-mapper value exceeds this
 global. **Deployment managers cannot enforce this setting as a hard
 cluster-wide ceiling**; treat this value as a default that user code may
 override.
 version_added: 3.3.0
 type: integer
 example: ~
Expand Down

10 changes: 7 additions & 3 deletions
airflow-core/src/airflow/partition_mappers/allowed_key.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@
class AllowedKeyMapper(PartitionMapper):
 """Partition mapper that validates keys against a set of allowed keys."""

 def __init__(self, allowed_keys: list[str]) -> None:
 def __init__(self, allowed_keys: list[str], *, max_downstream_keys: int | None = None) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.allowed_keys = allowed_keys

 def to_downstream(self, key: str) -> str:
Expand All@@ -34,8 +35,11 @@ def to_downstream(self, key: str) -> str:
 return key

 def serialize(self) -> dict[str, Any]:
 return {"allowed_keys": self.allowed_keys}
 data: dict[str, Any] = {"allowed_keys": self.allowed_keys}
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(allowed_keys=data["allowed_keys"])
 return cls(allowed_keys=data["allowed_keys"], max_downstream_keys=data.get("max_downstream_keys"))

26 changes: 22 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/base.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,15 @@ class PartitionMapper(ABC):

 is_rollup: ClassVar[bool] = False

 def __init__(self, *, max_downstream_keys: int | None = None) -> None:
 if max_downstream_keys is not None and (
 not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
 ):
 raise ValueError(
 f"max_downstream_keys must be a positive integer or None, got {max_downstream_keys!r}"
 )
 self.max_downstream_keys = max_downstream_keys

 def __init_subclass__(cls, **kwargs: Any) -> None:
 super().__init_subclass__(**kwargs)
 decode_overridden = cls.decode_downstream is not PartitionMapper.decode_downstream
Expand DownExpand Up@@ -107,11 +116,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 return None

 def serialize(self) -> dict[str, Any]:
 return {}
 if self.max_downstream_keys is None:
 return {}
 return {"max_downstream_keys": self.max_downstream_keys}

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls()
 return cls(max_downstream_keys=data.get("max_downstream_keys"))


class RollupMapper(PartitionMapper):
Expand All@@ -126,7 +137,9 @@ class RollupMapper(PartitionMapper):

 is_rollup: ClassVar[bool] = True

 def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 def __init__(
 self, *, upstream_mapper: PartitionMapper, window: Window, max_downstream_keys: int | None = None
 ) -> None:
 decode_overridden = type(upstream_mapper).decode_downstream is not PartitionMapper.decode_downstream
 if not decode_overridden and window.expected_decoded_type is not str:
 raise TypeError(
Expand All@@ -138,6 +151,7 @@ def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 f"{window.expected_decoded_type.__name__}, or use a window whose "
 f"'expected_decoded_type' accepts str."
 )
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window

Expand All@@ -160,10 +174,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
data: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 }
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -172,6 +189,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down

9 changes: 7 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/chain.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -35,7 +35,9 @@ def __init__(
 mapper1: PartitionMapper,
 /,
 *mappers: PartitionMapper,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]

 def to_downstream(self, key: str) -> str | Iterable[str]:
Expand DownExpand Up@@ -70,11 +72,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

 return {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 result: dict[str, Any] = {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers)
 return cls(*mappers, max_downstream_keys=data.get("max_downstream_keys"))

13 changes: 11 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/product.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,9 @@ def __init__(
 /,
 *mappers: PartitionMapper,
 delimiter: str = "|",
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]
 self.delimiter = delimiter

Expand All@@ -54,14 +56,21 @@ def to_downstream(self, key: str) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

return {
result: dict[str, Any] = {
 "delimiter": self.delimiter,
 "mappers": [encode_partition_mapper(m) for m in self.mappers],
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers, delimiter=data.get("delimiter", "|"))
 return cls(
 *mappers,
 delimiter=data.get("delimiter", "|"),
 max_downstream_keys=data.get("max_downstream_keys"),
 )

32 changes: 28 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/temporal.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -163,7 +163,9 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ):
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.input_format = input_format
 self.output_format = output_format or self.default_output_format
 if isinstance(timezone, str):
Expand DownExpand Up@@ -230,18 +232,22 @@ def encode_upstream(self, dt: datetime) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_timezone

return {
result: dict[str, Any] = {
 "timezone": encode_timezone(self._timezone),
 "input_format": self.input_format,
 "output_format": self.output_format,
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 timezone=parse_timezone(data.get("timezone", "UTC")),
 input_format=data["input_format"],
 output_format=data["output_format"],
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand DownExpand Up@@ -286,6 +292,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -302,7 +309,12 @@ def __init__(
 **must** include ``%Y``, ``%m``, and ``%d`` so the week-start date
 can be recovered for ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # %V (ISO week) cannot be round-tripped through strptime without %G+%u,
 # so derive a named-group regex from output_format and pull out %Y/%m/%d.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -358,6 +370,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -376,7 +389,12 @@ def __init__(
 and ``{quarter}`` so the quarter-start date can be recovered for
 ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # ``{quarter}`` is a Python-format placeholder, not a strftime directive,
 # so derive a named-group regex from output_format that handles both.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -511,7 +529,9 @@ def __init__(
 upstream_mapper: PartitionMapper,
 window: Window,
 downstream_mapper: PartitionMapper | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window
 self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
Expand All@@ -537,11 +557,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
result: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -551,6 +574,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
feat(core): add per-mapper max_fan_out override for partition fan-out cap by Lee-W · Pull Request #67184 · apache/airflow · GitHub
Skip to content
feat(core): add per-mapper max_fan_out override for partition fan-out cap#67184

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 4 commits into

from

Jun 11, 2026
 Merged


feat(core): add per-mapper max_fan_out override for partition fan-out cap
#67184
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
126b80a

feat(core): add per-mapper max_downstream_keys override for partition…


Lee-W May 19, 2026
5ff0fed

feat(core): log which cap was exceeded on partition fan-out overflow


Lee-W Jun 9, 2026
efa797e

refactor(task-sdk): validate max_downstream_keys with an attrs field …


Lee-W Jun 9, 2026
5f26df1

refactor(task-sdk): define partition mapper subclasses with attrs


Lee-W Jun 10, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
airflow-core/newsfragments/67184.feature.rst

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
Add ``max_downstream_keys`` parameter to ``PartitionMapper`` to override ``[scheduler] partition_mapper_max_downstream_keys`` per mapper instance.

19 changes: 13 additions & 6 deletions
airflow-core/src/airflow/assets/manager.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -552,8 +552,7 @@ def _queue_partitioned_dags(
 )
 return

 max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")

 global_cap = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
 for target_dag in partition_dags:
 if TYPE_CHECKING:
 assert partition_key is not None
Expand All@@ -573,9 +572,8 @@ def _queue_partitioned_dags(

 try:
 # We'll need to catch every possible exception happen when mapping partition_key.
 target_key = timetable.get_partition_mapper(
 name=asset_model.name, uri=asset_model.uri
 ).to_downstream(partition_key)
 mapper = timetable.get_partition_mapper(name=asset_model.name, uri=asset_model.uri)
 target_key = mapper.to_downstream(partition_key)
 except Exception as err:
 log.exception(
 "Could not map partition key for asset in target Dag. "
Expand DownExpand Up@@ -607,6 +605,14 @@ def _queue_partitioned_dags(
 target_keys = [target_key]
 del target_key

 mapper_cap = mapper.max_downstream_keys
 if mapper_cap is not None:
 max_downstream_keys = mapper_cap
 cap_source = f"max_downstream_keys={mapper_cap}"
 else:
 max_downstream_keys = global_cap
 cap_source = f"[scheduler] partition_mapper_max_downstream_keys={global_cap}"

 if len(target_keys) > max_downstream_keys:
 log.error(
Comment thread
Lee-W marked this conversation as resolved.

 "Partition mapper produced more downstream keys than allowed; skipping queue.",
Expand All@@ -615,6 +621,7 @@ def _queue_partitioned_dags(
 target_dag=target_dag.dag_id,
 produced_keys=len(target_keys),
 max_downstream_keys=max_downstream_keys,
 cap_source=cap_source,
 )
 session.add(
 Log(
Expand All@@ -624,7 +631,7 @@ def _queue_partitioned_dags(
 f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
 f"produced {len(target_keys)} downstream keys from "
 f"partition_key='{partition_key}', exceeding "
 f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
 f"{cap_source}. "
 f"No Dag runs were queued for this event."
 ),
 task_instance=task_instance,
Expand Down

15 changes: 8 additions & 7 deletions
airflow-core/src/airflow/config_templates/config.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -2690,13 +2690,14 @@ scheduler:
 see_also: ":ref:`scheduler:ha:tunables`"
 partition_mapper_max_downstream_keys:
 description: |
 Maximum number of downstream partition keys a single ``PartitionMapper``
 invocation may produce. When any partition mapper (built-in or custom)
 expands one upstream key into more keys than this limit, the scheduler
 skips queuing the runs for that asset event and logs an error against
 the source task instance. This guards against a misconfigured
 ``PartitionMapper`` from queuing an unbounded number of Dag runs per
 upstream event.
 Maximum number of downstream partition keys produced by a single
 PartitionMapper invocation, applied to any PartitionMapper that returns
 multiple keys (e.g. FanOutMapper). When a mapper instance sets a per-instance
 ``max_downstream_keys`` parameter, that value completely overrides this global
 cap for that instance — including when the per-mapper value exceeds this
 global. **Deployment managers cannot enforce this setting as a hard
 cluster-wide ceiling**; treat this value as a default that user code may
 override.
 version_added: 3.3.0
 type: integer
 example: ~
Expand Down

10 changes: 7 additions & 3 deletions
airflow-core/src/airflow/partition_mappers/allowed_key.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@
class AllowedKeyMapper(PartitionMapper):
 """Partition mapper that validates keys against a set of allowed keys."""

 def __init__(self, allowed_keys: list[str]) -> None:
 def __init__(self, allowed_keys: list[str], *, max_downstream_keys: int | None = None) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.allowed_keys = allowed_keys

 def to_downstream(self, key: str) -> str:
Expand All@@ -34,8 +35,11 @@ def to_downstream(self, key: str) -> str:
 return key

 def serialize(self) -> dict[str, Any]:
 return {"allowed_keys": self.allowed_keys}
 data: dict[str, Any] = {"allowed_keys": self.allowed_keys}
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(allowed_keys=data["allowed_keys"])
 return cls(allowed_keys=data["allowed_keys"], max_downstream_keys=data.get("max_downstream_keys"))

26 changes: 22 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/base.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,15 @@ class PartitionMapper(ABC):

 is_rollup: ClassVar[bool] = False

 def __init__(self, *, max_downstream_keys: int | None = None) -> None:
 if max_downstream_keys is not None and (
 not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
 ):
 raise ValueError(
 f"max_downstream_keys must be a positive integer or None, got {max_downstream_keys!r}"
 )
 self.max_downstream_keys = max_downstream_keys

 def __init_subclass__(cls, **kwargs: Any) -> None:
 super().__init_subclass__(**kwargs)
 decode_overridden = cls.decode_downstream is not PartitionMapper.decode_downstream
Expand DownExpand Up@@ -107,11 +116,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 return None

 def serialize(self) -> dict[str, Any]:
 return {}
 if self.max_downstream_keys is None:
 return {}
 return {"max_downstream_keys": self.max_downstream_keys}

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls()
 return cls(max_downstream_keys=data.get("max_downstream_keys"))


class RollupMapper(PartitionMapper):
Expand All@@ -126,7 +137,9 @@ class RollupMapper(PartitionMapper):

 is_rollup: ClassVar[bool] = True

 def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 def __init__(
 self, *, upstream_mapper: PartitionMapper, window: Window, max_downstream_keys: int | None = None
 ) -> None:
 decode_overridden = type(upstream_mapper).decode_downstream is not PartitionMapper.decode_downstream
 if not decode_overridden and window.expected_decoded_type is not str:
 raise TypeError(
Expand All@@ -138,6 +151,7 @@ def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 f"{window.expected_decoded_type.__name__}, or use a window whose "
 f"'expected_decoded_type' accepts str."
 )
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window

Expand All@@ -160,10 +174,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
data: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 }
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -172,6 +189,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down

9 changes: 7 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/chain.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -35,7 +35,9 @@ def __init__(
 mapper1: PartitionMapper,
 /,
 *mappers: PartitionMapper,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]

 def to_downstream(self, key: str) -> str | Iterable[str]:
Expand DownExpand Up@@ -70,11 +72,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

 return {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 result: dict[str, Any] = {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers)
 return cls(*mappers, max_downstream_keys=data.get("max_downstream_keys"))

13 changes: 11 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/product.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,9 @@ def __init__(
 /,
 *mappers: PartitionMapper,
 delimiter: str = "|",
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]
 self.delimiter = delimiter

Expand All@@ -54,14 +56,21 @@ def to_downstream(self, key: str) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

return {
result: dict[str, Any] = {
 "delimiter": self.delimiter,
 "mappers": [encode_partition_mapper(m) for m in self.mappers],
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers, delimiter=data.get("delimiter", "|"))
 return cls(
 *mappers,
 delimiter=data.get("delimiter", "|"),
 max_downstream_keys=data.get("max_downstream_keys"),
 )

32 changes: 28 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/temporal.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -163,7 +163,9 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ):
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.input_format = input_format
 self.output_format = output_format or self.default_output_format
 if isinstance(timezone, str):
Expand DownExpand Up@@ -230,18 +232,22 @@ def encode_upstream(self, dt: datetime) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_timezone

return {
result: dict[str, Any] = {
 "timezone": encode_timezone(self._timezone),
 "input_format": self.input_format,
 "output_format": self.output_format,
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 timezone=parse_timezone(data.get("timezone", "UTC")),
 input_format=data["input_format"],
 output_format=data["output_format"],
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand DownExpand Up@@ -286,6 +292,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -302,7 +309,12 @@ def __init__(
 **must** include ``%Y``, ``%m``, and ``%d`` so the week-start date
 can be recovered for ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # %V (ISO week) cannot be round-tripped through strptime without %G+%u,
 # so derive a named-group regex from output_format and pull out %Y/%m/%d.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -358,6 +370,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -376,7 +389,12 @@ def __init__(
 and ``{quarter}`` so the quarter-start date can be recovered for
 ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # ``{quarter}`` is a Python-format placeholder, not a strftime directive,
 # so derive a named-group regex from output_format that handles both.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -511,7 +529,9 @@ def __init__(
 upstream_mapper: PartitionMapper,
 window: Window,
 downstream_mapper: PartitionMapper | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window
 self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
Expand All@@ -537,11 +557,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
result: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -551,6 +574,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
feat(core): add per-mapper max_fan_out override for partition fan-out cap by Lee-W · Pull Request #67184 · apache/airflow · GitHub
Skip to content
feat(core): add per-mapper max_fan_out override for partition fan-out cap#67184

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 4 commits into

from

Jun 11, 2026
 Merged


feat(core): add per-mapper max_fan_out override for partition fan-out cap
#67184
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
126b80a

feat(core): add per-mapper max_downstream_keys override for partition…


Lee-W May 19, 2026
5ff0fed

feat(core): log which cap was exceeded on partition fan-out overflow


Lee-W Jun 9, 2026
efa797e

refactor(task-sdk): validate max_downstream_keys with an attrs field …


Lee-W Jun 9, 2026
5f26df1

refactor(task-sdk): define partition mapper subclasses with attrs


Lee-W Jun 10, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
airflow-core/newsfragments/67184.feature.rst

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
Add ``max_downstream_keys`` parameter to ``PartitionMapper`` to override ``[scheduler] partition_mapper_max_downstream_keys`` per mapper instance.

19 changes: 13 additions & 6 deletions
airflow-core/src/airflow/assets/manager.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -552,8 +552,7 @@ def _queue_partitioned_dags(
 )
 return

 max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")

 global_cap = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
 for target_dag in partition_dags:
 if TYPE_CHECKING:
 assert partition_key is not None
Expand All@@ -573,9 +572,8 @@ def _queue_partitioned_dags(

 try:
 # We'll need to catch every possible exception happen when mapping partition_key.
 target_key = timetable.get_partition_mapper(
 name=asset_model.name, uri=asset_model.uri
 ).to_downstream(partition_key)
 mapper = timetable.get_partition_mapper(name=asset_model.name, uri=asset_model.uri)
 target_key = mapper.to_downstream(partition_key)
 except Exception as err:
 log.exception(
 "Could not map partition key for asset in target Dag. "
Expand DownExpand Up@@ -607,6 +605,14 @@ def _queue_partitioned_dags(
 target_keys = [target_key]
 del target_key

 mapper_cap = mapper.max_downstream_keys
 if mapper_cap is not None:
 max_downstream_keys = mapper_cap
 cap_source = f"max_downstream_keys={mapper_cap}"
 else:
 max_downstream_keys = global_cap
 cap_source = f"[scheduler] partition_mapper_max_downstream_keys={global_cap}"

 if len(target_keys) > max_downstream_keys:
 log.error(
Comment thread
Lee-W marked this conversation as resolved.

 "Partition mapper produced more downstream keys than allowed; skipping queue.",
Expand All@@ -615,6 +621,7 @@ def _queue_partitioned_dags(
 target_dag=target_dag.dag_id,
 produced_keys=len(target_keys),
 max_downstream_keys=max_downstream_keys,
 cap_source=cap_source,
 )
 session.add(
 Log(
Expand All@@ -624,7 +631,7 @@ def _queue_partitioned_dags(
 f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
 f"produced {len(target_keys)} downstream keys from "
 f"partition_key='{partition_key}', exceeding "
 f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
 f"{cap_source}. "
 f"No Dag runs were queued for this event."
 ),
 task_instance=task_instance,
Expand Down

15 changes: 8 additions & 7 deletions
airflow-core/src/airflow/config_templates/config.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -2690,13 +2690,14 @@ scheduler:
 see_also: ":ref:`scheduler:ha:tunables`"
 partition_mapper_max_downstream_keys:
 description: |
 Maximum number of downstream partition keys a single ``PartitionMapper``
 invocation may produce. When any partition mapper (built-in or custom)
 expands one upstream key into more keys than this limit, the scheduler
 skips queuing the runs for that asset event and logs an error against
 the source task instance. This guards against a misconfigured
 ``PartitionMapper`` from queuing an unbounded number of Dag runs per
 upstream event.
 Maximum number of downstream partition keys produced by a single
 PartitionMapper invocation, applied to any PartitionMapper that returns
 multiple keys (e.g. FanOutMapper). When a mapper instance sets a per-instance
 ``max_downstream_keys`` parameter, that value completely overrides this global
 cap for that instance — including when the per-mapper value exceeds this
 global. **Deployment managers cannot enforce this setting as a hard
 cluster-wide ceiling**; treat this value as a default that user code may
 override.
 version_added: 3.3.0
 type: integer
 example: ~
Expand Down

10 changes: 7 additions & 3 deletions
airflow-core/src/airflow/partition_mappers/allowed_key.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@
class AllowedKeyMapper(PartitionMapper):
 """Partition mapper that validates keys against a set of allowed keys."""

 def __init__(self, allowed_keys: list[str]) -> None:
 def __init__(self, allowed_keys: list[str], *, max_downstream_keys: int | None = None) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.allowed_keys = allowed_keys

 def to_downstream(self, key: str) -> str:
Expand All@@ -34,8 +35,11 @@ def to_downstream(self, key: str) -> str:
 return key

 def serialize(self) -> dict[str, Any]:
 return {"allowed_keys": self.allowed_keys}
 data: dict[str, Any] = {"allowed_keys": self.allowed_keys}
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(allowed_keys=data["allowed_keys"])
 return cls(allowed_keys=data["allowed_keys"], max_downstream_keys=data.get("max_downstream_keys"))

26 changes: 22 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/base.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,15 @@ class PartitionMapper(ABC):

 is_rollup: ClassVar[bool] = False

 def __init__(self, *, max_downstream_keys: int | None = None) -> None:
 if max_downstream_keys is not None and (
 not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
 ):
 raise ValueError(
 f"max_downstream_keys must be a positive integer or None, got {max_downstream_keys!r}"
 )
 self.max_downstream_keys = max_downstream_keys

 def __init_subclass__(cls, **kwargs: Any) -> None:
 super().__init_subclass__(**kwargs)
 decode_overridden = cls.decode_downstream is not PartitionMapper.decode_downstream
Expand DownExpand Up@@ -107,11 +116,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 return None

 def serialize(self) -> dict[str, Any]:
 return {}
 if self.max_downstream_keys is None:
 return {}
 return {"max_downstream_keys": self.max_downstream_keys}

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls()
 return cls(max_downstream_keys=data.get("max_downstream_keys"))


class RollupMapper(PartitionMapper):
Expand All@@ -126,7 +137,9 @@ class RollupMapper(PartitionMapper):

 is_rollup: ClassVar[bool] = True

 def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 def __init__(
 self, *, upstream_mapper: PartitionMapper, window: Window, max_downstream_keys: int | None = None
 ) -> None:
 decode_overridden = type(upstream_mapper).decode_downstream is not PartitionMapper.decode_downstream
 if not decode_overridden and window.expected_decoded_type is not str:
 raise TypeError(
Expand All@@ -138,6 +151,7 @@ def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 f"{window.expected_decoded_type.__name__}, or use a window whose "
 f"'expected_decoded_type' accepts str."
 )
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window

Expand All@@ -160,10 +174,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
data: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 }
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -172,6 +189,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down

9 changes: 7 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/chain.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -35,7 +35,9 @@ def __init__(
 mapper1: PartitionMapper,
 /,
 *mappers: PartitionMapper,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]

 def to_downstream(self, key: str) -> str | Iterable[str]:
Expand DownExpand Up@@ -70,11 +72,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

 return {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 result: dict[str, Any] = {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers)
 return cls(*mappers, max_downstream_keys=data.get("max_downstream_keys"))

13 changes: 11 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/product.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,9 @@ def __init__(
 /,
 *mappers: PartitionMapper,
 delimiter: str = "|",
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]
 self.delimiter = delimiter

Expand All@@ -54,14 +56,21 @@ def to_downstream(self, key: str) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

return {
result: dict[str, Any] = {
 "delimiter": self.delimiter,
 "mappers": [encode_partition_mapper(m) for m in self.mappers],
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers, delimiter=data.get("delimiter", "|"))
 return cls(
 *mappers,
 delimiter=data.get("delimiter", "|"),
 max_downstream_keys=data.get("max_downstream_keys"),
 )

32 changes: 28 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/temporal.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -163,7 +163,9 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ):
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.input_format = input_format
 self.output_format = output_format or self.default_output_format
 if isinstance(timezone, str):
Expand DownExpand Up@@ -230,18 +232,22 @@ def encode_upstream(self, dt: datetime) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_timezone

return {
result: dict[str, Any] = {
 "timezone": encode_timezone(self._timezone),
 "input_format": self.input_format,
 "output_format": self.output_format,
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 timezone=parse_timezone(data.get("timezone", "UTC")),
 input_format=data["input_format"],
 output_format=data["output_format"],
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand DownExpand Up@@ -286,6 +292,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -302,7 +309,12 @@ def __init__(
 **must** include ``%Y``, ``%m``, and ``%d`` so the week-start date
 can be recovered for ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # %V (ISO week) cannot be round-tripped through strptime without %G+%u,
 # so derive a named-group regex from output_format and pull out %Y/%m/%d.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -358,6 +370,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -376,7 +389,12 @@ def __init__(
 and ``{quarter}`` so the quarter-start date can be recovered for
 ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # ``{quarter}`` is a Python-format placeholder, not a strftime directive,
 # so derive a named-group regex from output_format that handles both.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -511,7 +529,9 @@ def __init__(
 upstream_mapper: PartitionMapper,
 window: Window,
 downstream_mapper: PartitionMapper | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window
 self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
Expand All@@ -537,11 +557,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
result: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -551,6 +574,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();
feat(core): add per-mapper max_fan_out override for partition fan-out cap by Lee-W · Pull Request #67184 · apache/airflow · GitHub
Skip to content
feat(core): add per-mapper max_fan_out override for partition fan-out cap#67184

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 4 commits into

from

Jun 11, 2026
 Merged


feat(core): add per-mapper max_fan_out override for partition fan-out cap
#67184
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
126b80a

feat(core): add per-mapper max_downstream_keys override for partition…


Lee-W May 19, 2026
5ff0fed

feat(core): log which cap was exceeded on partition fan-out overflow


Lee-W Jun 9, 2026
efa797e

refactor(task-sdk): validate max_downstream_keys with an attrs field …


Lee-W Jun 9, 2026
5f26df1

refactor(task-sdk): define partition mapper subclasses with attrs


Lee-W Jun 10, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


1 change: 1 addition & 0 deletions
airflow-core/newsfragments/67184.feature.rst

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
Add ``max_downstream_keys`` parameter to ``PartitionMapper`` to override ``[scheduler] partition_mapper_max_downstream_keys`` per mapper instance.

19 changes: 13 additions & 6 deletions
airflow-core/src/airflow/assets/manager.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -552,8 +552,7 @@ def _queue_partitioned_dags(
 )
 return

 max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")

 global_cap = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
 for target_dag in partition_dags:
 if TYPE_CHECKING:
 assert partition_key is not None
Expand All@@ -573,9 +572,8 @@ def _queue_partitioned_dags(

 try:
 # We'll need to catch every possible exception happen when mapping partition_key.
 target_key = timetable.get_partition_mapper(
 name=asset_model.name, uri=asset_model.uri
 ).to_downstream(partition_key)
 mapper = timetable.get_partition_mapper(name=asset_model.name, uri=asset_model.uri)
 target_key = mapper.to_downstream(partition_key)
 except Exception as err:
 log.exception(
 "Could not map partition key for asset in target Dag. "
Expand DownExpand Up@@ -607,6 +605,14 @@ def _queue_partitioned_dags(
 target_keys = [target_key]
 del target_key

 mapper_cap = mapper.max_downstream_keys
 if mapper_cap is not None:
 max_downstream_keys = mapper_cap
 cap_source = f"max_downstream_keys={mapper_cap}"
 else:
 max_downstream_keys = global_cap
 cap_source = f"[scheduler] partition_mapper_max_downstream_keys={global_cap}"

 if len(target_keys) > max_downstream_keys:
 log.error(
Comment thread
Lee-W marked this conversation as resolved.

 "Partition mapper produced more downstream keys than allowed; skipping queue.",
Expand All@@ -615,6 +621,7 @@ def _queue_partitioned_dags(
 target_dag=target_dag.dag_id,
 produced_keys=len(target_keys),
 max_downstream_keys=max_downstream_keys,
 cap_source=cap_source,
 )
 session.add(
 Log(
Expand All@@ -624,7 +631,7 @@ def _queue_partitioned_dags(
 f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
 f"produced {len(target_keys)} downstream keys from "
 f"partition_key='{partition_key}', exceeding "
 f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
 f"{cap_source}. "
 f"No Dag runs were queued for this event."
 ),
 task_instance=task_instance,
Expand Down

15 changes: 8 additions & 7 deletions
airflow-core/src/airflow/config_templates/config.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -2690,13 +2690,14 @@ scheduler:
 see_also: ":ref:`scheduler:ha:tunables`"
 partition_mapper_max_downstream_keys:
 description: |
 Maximum number of downstream partition keys a single ``PartitionMapper``
 invocation may produce. When any partition mapper (built-in or custom)
 expands one upstream key into more keys than this limit, the scheduler
 skips queuing the runs for that asset event and logs an error against
 the source task instance. This guards against a misconfigured
 ``PartitionMapper`` from queuing an unbounded number of Dag runs per
 upstream event.
 Maximum number of downstream partition keys produced by a single
 PartitionMapper invocation, applied to any PartitionMapper that returns
 multiple keys (e.g. FanOutMapper). When a mapper instance sets a per-instance
 ``max_downstream_keys`` parameter, that value completely overrides this global
 cap for that instance — including when the per-mapper value exceeds this
 global. **Deployment managers cannot enforce this setting as a hard
 cluster-wide ceiling**; treat this value as a default that user code may
 override.
 version_added: 3.3.0
 type: integer
 example: ~
Expand Down

10 changes: 7 additions & 3 deletions
airflow-core/src/airflow/partition_mappers/allowed_key.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@
class AllowedKeyMapper(PartitionMapper):
 """Partition mapper that validates keys against a set of allowed keys."""

 def __init__(self, allowed_keys: list[str]) -> None:
 def __init__(self, allowed_keys: list[str], *, max_downstream_keys: int | None = None) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.allowed_keys = allowed_keys

 def to_downstream(self, key: str) -> str:
Expand All@@ -34,8 +35,11 @@ def to_downstream(self, key: str) -> str:
 return key

 def serialize(self) -> dict[str, Any]:
 return {"allowed_keys": self.allowed_keys}
 data: dict[str, Any] = {"allowed_keys": self.allowed_keys}
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(allowed_keys=data["allowed_keys"])
 return cls(allowed_keys=data["allowed_keys"], max_downstream_keys=data.get("max_downstream_keys"))

26 changes: 22 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/base.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -36,6 +36,15 @@ class PartitionMapper(ABC):

 is_rollup: ClassVar[bool] = False

 def __init__(self, *, max_downstream_keys: int | None = None) -> None:
 if max_downstream_keys is not None and (
 not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
 ):
 raise ValueError(
 f"max_downstream_keys must be a positive integer or None, got {max_downstream_keys!r}"
 )
 self.max_downstream_keys = max_downstream_keys

 def __init_subclass__(cls, **kwargs: Any) -> None:
 super().__init_subclass__(**kwargs)
 decode_overridden = cls.decode_downstream is not PartitionMapper.decode_downstream
Expand DownExpand Up@@ -107,11 +116,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 return None

 def serialize(self) -> dict[str, Any]:
 return {}
 if self.max_downstream_keys is None:
 return {}
 return {"max_downstream_keys": self.max_downstream_keys}

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls()
 return cls(max_downstream_keys=data.get("max_downstream_keys"))


class RollupMapper(PartitionMapper):
Expand All@@ -126,7 +137,9 @@ class RollupMapper(PartitionMapper):

 is_rollup: ClassVar[bool] = True

 def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 def __init__(
 self, *, upstream_mapper: PartitionMapper, window: Window, max_downstream_keys: int | None = None
 ) -> None:
 decode_overridden = type(upstream_mapper).decode_downstream is not PartitionMapper.decode_downstream
 if not decode_overridden and window.expected_decoded_type is not str:
 raise TypeError(
Expand All@@ -138,6 +151,7 @@ def __init__(self, *, upstream_mapper: PartitionMapper, window: Window) -> None:
 f"{window.expected_decoded_type.__name__}, or use a window whose "
 f"'expected_decoded_type' accepts str."
 )
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window

Expand All@@ -160,10 +174,13 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
data: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 }
 if self.max_downstream_keys is not None:
 data["max_downstream_keys"] = self.max_downstream_keys
 return data

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -172,6 +189,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down

9 changes: 7 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/chain.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -35,7 +35,9 @@ def __init__(
 mapper1: PartitionMapper,
 /,
 *mappers: PartitionMapper,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]

 def to_downstream(self, key: str) -> str | Iterable[str]:
Expand DownExpand Up@@ -70,11 +72,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

 return {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 result: dict[str, Any] = {"mappers": [encode_partition_mapper(m) for m in self.mappers]}
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers)
 return cls(*mappers, max_downstream_keys=data.get("max_downstream_keys"))

13 changes: 11 additions & 2 deletions
airflow-core/src/airflow/partition_mappers/product.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -32,7 +32,9 @@ def __init__(
 /,
 *mappers: PartitionMapper,
 delimiter: str = "|",
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.mappers = [mapper0, mapper1, *mappers]
 self.delimiter = delimiter

Expand All@@ -54,14 +56,21 @@ def to_downstream(self, key: str) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper

return {
result: dict[str, Any] = {
 "delimiter": self.delimiter,
 "mappers": [encode_partition_mapper(m) for m in self.mappers],
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 from airflow.serialization.decoders import decode_partition_mapper

 mappers = [decode_partition_mapper(m) for m in data["mappers"]]
 return cls(*mappers, delimiter=data.get("delimiter", "|"))
 return cls(
 *mappers,
 delimiter=data.get("delimiter", "|"),
 max_downstream_keys=data.get("max_downstream_keys"),
 )

32 changes: 28 additions & 4 deletions
airflow-core/src/airflow/partition_mappers/temporal.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -163,7 +163,9 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ):
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.input_format = input_format
 self.output_format = output_format or self.default_output_format
 if isinstance(timezone, str):
Expand DownExpand Up@@ -230,18 +232,22 @@ def encode_upstream(self, dt: datetime) -> str:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_timezone

return {
result: dict[str, Any] = {
 "timezone": encode_timezone(self._timezone),
 "input_format": self.input_format,
 "output_format": self.output_format,
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 return cls(
 timezone=parse_timezone(data.get("timezone", "UTC")),
 input_format=data["input_format"],
 output_format=data["output_format"],
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand DownExpand Up@@ -286,6 +292,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -302,7 +309,12 @@ def __init__(
 **must** include ``%Y``, ``%m``, and ``%d`` so the week-start date
 can be recovered for ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # %V (ISO week) cannot be round-tripped through strptime without %G+%u,
 # so derive a named-group regex from output_format and pull out %Y/%m/%d.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -358,6 +370,7 @@ def __init__(
 timezone: str | Timezone | FixedTimezone = "UTC",
 input_format: str = "%Y-%m-%dT%H:%M:%S",
 output_format: str | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 """
 Compile *output_format* eagerly so malformed patterns raise here.
Expand All@@ -376,7 +389,12 @@ def __init__(
 and ``{quarter}`` so the quarter-start date can be recovered for
 ``to_upstream``.
 """
 super().__init__(timezone=timezone, input_format=input_format, output_format=output_format)
 super().__init__(
 timezone=timezone,
 input_format=input_format,
 output_format=output_format,
 max_downstream_keys=max_downstream_keys,
 )
 # ``{quarter}`` is a Python-format placeholder, not a strftime directive,
 # so derive a named-group regex from output_format that handles both.
 # Compile eagerly so a malformed output_format raises ValueError here
Expand DownExpand Up@@ -511,7 +529,9 @@ def __init__(
 upstream_mapper: PartitionMapper,
 window: Window,
 downstream_mapper: PartitionMapper | None = None,
 max_downstream_keys: int | None = None,
 ) -> None:
 super().__init__(max_downstream_keys=max_downstream_keys)
 self.upstream_mapper = upstream_mapper
 self.window = window
 self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
Expand All@@ -537,11 +557,14 @@ def to_partition_date(self, downstream_key: str) -> datetime | None:
 def serialize(self) -> dict[str, Any]:
 from airflow.serialization.encoders import encode_partition_mapper, encode_window

return {
result: dict[str, Any] = {
 "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
 "window": encode_window(self.window),
 "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
 }
 if self.max_downstream_keys is not None:
 result["max_downstream_keys"] = self.max_downstream_keys
 return result

 @classmethod
 def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
Expand All@@ -551,6 +574,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
 upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
 window=decode_window(data["window"]),
 downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
 max_downstream_keys=data.get("max_downstream_keys"),
 )


Expand Down
Loading
Loading