Add FixedKeyMapper and SegmentWindow for categorical asset-partition rollup

[Lee-W]

Why

We already supports temporal asset-partition rollup by composing two primitives inside a RollupMapper: a mapper that routes upstream events to a downstream partition key (e.g. StartOfDayMapper) and a window that declares which upstream partitions must arrive before the downstream run is released (e.g. DayWindow). There was no equivalent for categorical rollup — folding a fixed set of named segments (such as us, eu, apac) into a single downstream run.

This PR adds the two primitives needed to express categorical rollup with the same composition model as the temporal family, rather than introducing a self-contained bespoke segment mapper:

RollupMapper(
upstream_mapper=FixedKeyMapper("all_regions"),
window=SegmentWindow(["us", "eu", "apac"]),
)

• FixedKeyMapper handles forward routing: every incoming upstream segment event (us / eu / apac) is routed to one downstream partition key ("all_regions"), so all 3 accumulate into the same run.
• SegmentWindow handles reverse gating: it declares the set of segments to wait for, and the scheduler releases the downstream run only once all of them have arrived (to_upstream + issubset).

Keeping the scheduler aware of only RollupMapper — and letting authors swap in different mapper/window pairs — means categorical and temporal rollup share one mental model instead of each shipping its own mapper.

The FixedKeyMapper downstream key is required and semantic (the example uses "all_regions") to avoid a silent default that would obscure where rolled-up events land.

SegmentWindow is also reusable as the gating side of a categorical scatter (1-to-n) when paired with the recently merged FanOutMapper.

What

• Add FixedKeyMapper: returns one fixed downstream key for any upstream key; the key is a required, validated non-empty str.

• Add SegmentWindow: enumerates a fixed categorical set of segments with expected_decoded_type=str.

• Extend the existing example_asset_partition Dag with a segment_region_stats_rollup section, and document categorical rollup in assets.rst.

• related: Segment-based partitioning #57695

• closes: Segment-based (categorical) asset partitions for static sets — rollup (n-1) and fan-out (1-n) #68238

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Lee-W]

Why

We already supports temporal asset-partition rollup by composing two primitives inside a RollupMapper: a mapper that routes upstream events to a downstream partition key (e.g. StartOfDayMapper) and a window that declares which upstream partitions must arrive before the downstream run is released (e.g. DayWindow). There was no equivalent for categorical rollup — folding a fixed set of named segments (such as us, eu, apac) into a single downstream run.

This PR adds the two primitives needed to express categorical rollup with the same composition model as the temporal family, rather than introducing a self-contained bespoke segment mapper:

RollupMapper(
upstream_mapper=FixedKeyMapper("all_regions"),
window=SegmentWindow(["us", "eu", "apac"]),
)

• FixedKeyMapper handles forward routing: every incoming upstream segment event (us / eu / apac) is routed to one downstream partition key ("all_regions"), so all 3 accumulate into the same run.
• SegmentWindow handles reverse gating: it declares the set of segments to wait for, and the scheduler releases the downstream run only once all of them have arrived (to_upstream + issubset).

Keeping the scheduler aware of only RollupMapper — and letting authors swap in different mapper/window pairs — means categorical and temporal rollup share one mental model instead of each shipping its own mapper.

The FixedKeyMapper downstream key is required and semantic (the example uses "all_regions") to avoid a silent default that would obscure where rolled-up events land.

SegmentWindow is also reusable as the gating side of a categorical scatter (1-to-n) when paired with the recently merged FanOutMapper.

What

• Add FixedKeyMapper: returns one fixed downstream key for any upstream key; the key is a required, validated non-empty str.

• Add SegmentWindow: enumerates a fixed categorical set of segments with expected_decoded_type=str.

• Extend the existing example_asset_partition Dag with a segment_region_stats_rollup section, and document categorical rollup in assets.rst.

• related: Segment-based partitioning #57695

• closes: Segment-based (categorical) asset partitions for static sets — rollup (n-1) and fan-out (1-n) #68238

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Lee-W]

Why

We already supports temporal asset-partition rollup by composing two primitives inside a RollupMapper: a mapper that routes upstream events to a downstream partition key (e.g. StartOfDayMapper) and a window that declares which upstream partitions must arrive before the downstream run is released (e.g. DayWindow). There was no equivalent for categorical rollup — folding a fixed set of named segments (such as us, eu, apac) into a single downstream run.

This PR adds the two primitives needed to express categorical rollup with the same composition model as the temporal family, rather than introducing a self-contained bespoke segment mapper:

RollupMapper(
upstream_mapper=FixedKeyMapper("all_regions"),
window=SegmentWindow(["us", "eu", "apac"]),
)

• FixedKeyMapper handles forward routing: every incoming upstream segment event (us / eu / apac) is routed to one downstream partition key ("all_regions"), so all 3 accumulate into the same run.
• SegmentWindow handles reverse gating: it declares the set of segments to wait for, and the scheduler releases the downstream run only once all of them have arrived (to_upstream + issubset).

Keeping the scheduler aware of only RollupMapper — and letting authors swap in different mapper/window pairs — means categorical and temporal rollup share one mental model instead of each shipping its own mapper.

The FixedKeyMapper downstream key is required and semantic (the example uses "all_regions") to avoid a silent default that would obscure where rolled-up events land.

SegmentWindow is also reusable as the gating side of a categorical scatter (1-to-n) when paired with the recently merged FanOutMapper.

What

• Add FixedKeyMapper: returns one fixed downstream key for any upstream key; the key is a required, validated non-empty str.

• Add SegmentWindow: enumerates a fixed categorical set of segments with expected_decoded_type=str.

• Extend the existing example_asset_partition Dag with a segment_region_stats_rollup section, and document categorical rollup in assets.rst.

• related: Segment-based partitioning #57695

• closes: Segment-based (categorical) asset partitions for static sets — rollup (n-1) and fan-out (1-n) #68238

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Lee-W]

Why

We already supports temporal asset-partition rollup by composing two primitives inside a RollupMapper: a mapper that routes upstream events to a downstream partition key (e.g. StartOfDayMapper) and a window that declares which upstream partitions must arrive before the downstream run is released (e.g. DayWindow). There was no equivalent for categorical rollup — folding a fixed set of named segments (such as us, eu, apac) into a single downstream run.

This PR adds the two primitives needed to express categorical rollup with the same composition model as the temporal family, rather than introducing a self-contained bespoke segment mapper:

RollupMapper(
upstream_mapper=FixedKeyMapper("all_regions"),
window=SegmentWindow(["us", "eu", "apac"]),
)

• FixedKeyMapper handles forward routing: every incoming upstream segment event (us / eu / apac) is routed to one downstream partition key ("all_regions"), so all 3 accumulate into the same run.
• SegmentWindow handles reverse gating: it declares the set of segments to wait for, and the scheduler releases the downstream run only once all of them have arrived (to_upstream + issubset).

Keeping the scheduler aware of only RollupMapper — and letting authors swap in different mapper/window pairs — means categorical and temporal rollup share one mental model instead of each shipping its own mapper.

The FixedKeyMapper downstream key is required and semantic (the example uses "all_regions") to avoid a silent default that would obscure where rolled-up events land.

SegmentWindow is also reusable as the gating side of a categorical scatter (1-to-n) when paired with the recently merged FanOutMapper.

What

• Add FixedKeyMapper: returns one fixed downstream key for any upstream key; the key is a required, validated non-empty str.

• Add SegmentWindow: enumerates a fixed categorical set of segments with expected_decoded_type=str.

• Extend the existing example_asset_partition Dag with a segment_region_stats_rollup section, and document categorical rollup in assets.rst.

• related: Segment-based partitioning #57695

• closes: Segment-based (categorical) asset partitions for static sets — rollup (n-1) and fan-out (1-n) #68238

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Lee-W]

Why

We already supports temporal asset-partition rollup by composing two primitives inside a RollupMapper: a mapper that routes upstream events to a downstream partition key (e.g. StartOfDayMapper) and a window that declares which upstream partitions must arrive before the downstream run is released (e.g. DayWindow). There was no equivalent for categorical rollup — folding a fixed set of named segments (such as us, eu, apac) into a single downstream run.

This PR adds the two primitives needed to express categorical rollup with the same composition model as the temporal family, rather than introducing a self-contained bespoke segment mapper:

RollupMapper(
upstream_mapper=FixedKeyMapper("all_regions"),
window=SegmentWindow(["us", "eu", "apac"]),
)

• FixedKeyMapper handles forward routing: every incoming upstream segment event (us / eu / apac) is routed to one downstream partition key ("all_regions"), so all 3 accumulate into the same run.
• SegmentWindow handles reverse gating: it declares the set of segments to wait for, and the scheduler releases the downstream run only once all of them have arrived (to_upstream + issubset).

Keeping the scheduler aware of only RollupMapper — and letting authors swap in different mapper/window pairs — means categorical and temporal rollup share one mental model instead of each shipping its own mapper.

The FixedKeyMapper downstream key is required and semantic (the example uses "all_regions") to avoid a silent default that would obscure where rolled-up events land.

SegmentWindow is also reusable as the gating side of a categorical scatter (1-to-n) when paired with the recently merged FanOutMapper.

What

• Add FixedKeyMapper: returns one fixed downstream key for any upstream key; the key is a required, validated non-empty str.

• Add SegmentWindow: enumerates a fixed categorical set of segments with expected_decoded_type=str.

• Extend the existing example_asset_partition Dag with a segment_region_stats_rollup section, and document categorical rollup in assets.rst.

• related: Segment-based partitioning #57695

• closes: Segment-based (categorical) asset partitions for static sets — rollup (n-1) and fan-out (1-n) #68238

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Lee-W]

Why

We already supports temporal asset-partition rollup by composing two primitives inside a RollupMapper: a mapper that routes upstream events to a downstream partition key (e.g. StartOfDayMapper) and a window that declares which upstream partitions must arrive before the downstream run is released (e.g. DayWindow). There was no equivalent for categorical rollup — folding a fixed set of named segments (such as us, eu, apac) into a single downstream run.

This PR adds the two primitives needed to express categorical rollup with the same composition model as the temporal family, rather than introducing a self-contained bespoke segment mapper:

RollupMapper(
upstream_mapper=FixedKeyMapper("all_regions"),
window=SegmentWindow(["us", "eu", "apac"]),
)

• FixedKeyMapper handles forward routing: every incoming upstream segment event (us / eu / apac) is routed to one downstream partition key ("all_regions"), so all 3 accumulate into the same run.
• SegmentWindow handles reverse gating: it declares the set of segments to wait for, and the scheduler releases the downstream run only once all of them have arrived (to_upstream + issubset).

Keeping the scheduler aware of only RollupMapper — and letting authors swap in different mapper/window pairs — means categorical and temporal rollup share one mental model instead of each shipping its own mapper.

The FixedKeyMapper downstream key is required and semantic (the example uses "all_regions") to avoid a silent default that would obscure where rolled-up events land.

SegmentWindow is also reusable as the gating side of a categorical scatter (1-to-n) when paired with the recently merged FanOutMapper.

What

• Add FixedKeyMapper: returns one fixed downstream key for any upstream key; the key is a required, validated non-empty str.

• Add SegmentWindow: enumerates a fixed categorical set of segments with expected_decoded_type=str.

• Extend the existing example_asset_partition Dag with a segment_region_stats_rollup section, and document categorical rollup in assets.rst.

• related: Segment-based partitioning #57695

• closes: Segment-based (categorical) asset partitions for static sets — rollup (n-1) and fan-out (1-n) #68238

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Lee-W]

Why

We already supports temporal asset-partition rollup by composing two primitives inside a RollupMapper: a mapper that routes upstream events to a downstream partition key (e.g. StartOfDayMapper) and a window that declares which upstream partitions must arrive before the downstream run is released (e.g. DayWindow). There was no equivalent for categorical rollup — folding a fixed set of named segments (such as us, eu, apac) into a single downstream run.

This PR adds the two primitives needed to express categorical rollup with the same composition model as the temporal family, rather than introducing a self-contained bespoke segment mapper:

RollupMapper(
upstream_mapper=FixedKeyMapper("all_regions"),
window=SegmentWindow(["us", "eu", "apac"]),
)

• FixedKeyMapper handles forward routing: every incoming upstream segment event (us / eu / apac) is routed to one downstream partition key ("all_regions"), so all 3 accumulate into the same run.
• SegmentWindow handles reverse gating: it declares the set of segments to wait for, and the scheduler releases the downstream run only once all of them have arrived (to_upstream + issubset).

Keeping the scheduler aware of only RollupMapper — and letting authors swap in different mapper/window pairs — means categorical and temporal rollup share one mental model instead of each shipping its own mapper.

The FixedKeyMapper downstream key is required and semantic (the example uses "all_regions") to avoid a silent default that would obscure where rolled-up events land.

SegmentWindow is also reusable as the gating side of a categorical scatter (1-to-n) when paired with the recently merged FanOutMapper.

What

• Add FixedKeyMapper: returns one fixed downstream key for any upstream key; the key is a required, validated non-empty str.

• Add SegmentWindow: enumerates a fixed categorical set of segments with expected_decoded_type=str.

• Extend the existing example_asset_partition Dag with a segment_region_stats_rollup section, and document categorical rollup in assets.rst.

• related: Segment-based partitioning #57695

• closes: Segment-based (categorical) asset partitions for static sets — rollup (n-1) and fan-out (1-n) #68238

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Lee-W]

Why

We already supports temporal asset-partition rollup by composing two primitives inside a RollupMapper: a mapper that routes upstream events to a downstream partition key (e.g. StartOfDayMapper) and a window that declares which upstream partitions must arrive before the downstream run is released (e.g. DayWindow). There was no equivalent for categorical rollup — folding a fixed set of named segments (such as us, eu, apac) into a single downstream run.

This PR adds the two primitives needed to express categorical rollup with the same composition model as the temporal family, rather than introducing a self-contained bespoke segment mapper:

RollupMapper(
upstream_mapper=FixedKeyMapper("all_regions"),
window=SegmentWindow(["us", "eu", "apac"]),
)

• FixedKeyMapper handles forward routing: every incoming upstream segment event (us / eu / apac) is routed to one downstream partition key ("all_regions"), so all 3 accumulate into the same run.
• SegmentWindow handles reverse gating: it declares the set of segments to wait for, and the scheduler releases the downstream run only once all of them have arrived (to_upstream + issubset).

Keeping the scheduler aware of only RollupMapper — and letting authors swap in different mapper/window pairs — means categorical and temporal rollup share one mental model instead of each shipping its own mapper.

The FixedKeyMapper downstream key is required and semantic (the example uses "all_regions") to avoid a silent default that would obscure where rolled-up events land.

SegmentWindow is also reusable as the gating side of a categorical scatter (1-to-n) when paired with the recently merged FanOutMapper.

What

• Add FixedKeyMapper: returns one fixed downstream key for any upstream key; the key is a required, validated non-empty str.

• Add SegmentWindow: enumerates a fixed categorical set of segments with expected_decoded_type=str.

• Extend the existing example_asset_partition Dag with a segment_region_stats_rollup section, and document categorical rollup in assets.rst.

• related: Segment-based partitioning #57695

• closes: Segment-based (categorical) asset partitions for static sets — rollup (n-1) and fan-out (1-n) #68238

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.