Populate partition_date when manually triggering partitioned Dags#68458
Conversation
a827aca to
74042bd
Compare
dcfc99c to
d708916
Compare
58c6b18 to
6469fa4
Compare
ff2f3f1 to
8fa93ed
Compare
8fa93ed to
dc848e2
Compare
When a Dag with a partitioned timetable (e.g. CronPartitionTimetable) is manually triggered with a partition_key, partition_date was left NULL because _trigger_dag and the core API route only forwarded partition_key without deriving the corresponding date. Add Timetable.resolve_partition_date(key) — overridden by CronPartitionTimetable (strptime inverse of _format_key) and PartitionedAssetTimetable (consensus across asset mappers) — and call it in both _trigger_dag and validate_context so partition_date is populated on all manually triggered runs. Invalid keys raise InvalidPartitionKeyError (new ValueError subclass) which translates to HTTP 400 at FastAPI route boundaries.
PartitionedAssetTimetable._decode_partition_date called the temporal mappers' to_partition_date() unguarded, so a malformed partition key raised a plain ValueError and surfaced as HTTP 500. Wrap both mapper loops to raise InvalidPartitionKeyError, which the execution API already translates to 400 invalid_partition_key.
dc848e2 to
8f8b933
Compare
Cover the happy path where a temporal mapper populates partition_date from a valid key, and the execution API returning 400 invalid_partition_key when a partitioned Dag is triggered with a malformed partition key.
8f8b933 to
5bc155a
Compare
|
Hi maintainer, this PR was merged without a milestone set.
|
Backport failed to create: v3-3-test. View the failure log Run detailsNote: As of Merging PRs targeted for Airflow 3.X In matter of doubt please ask in #release-management Slack channel.
You can attempt to backport this manually by running: cherry_picker db9fa42 v3-3-testThis should apply the commit to the v3-3-test branch and leave the commit in conflict state marking After you have resolved the conflicts, you can continue the backport process by running: cherry_picker --continueIf you don't have cherry-picker installed, see the installation guide. |
…ned Dags (apache#68458) (cherry picked from commit db9fa42) Co-authored-by: Wei Lee <weilee.rx@gmail.com>
Why
When a Dag with a partitioned timetable (e.g.
CronPartitionTimetable) is manually triggered with apartition_key, the run'spartition_datewas leftNULL. Both the manual-trigger code path (_trigger_dag) and the core API route only forwarded thepartition_keywithout deriving the corresponding date, so manually triggered partitioned runs ended up missing the date that scheduled runs carry.What
Timetable.resolve_partition_date(partition_key)to the timetable protocol (timetables/base.py). It guards onpartitioned/partitioned_at_runtimeand delegates to_decode_partition_date, whose default returnsNone._decode_partition_datein the partitioned timetables:CronPartitionTimetable(timetables/trigger.py): inverts_key_formatviastrptimeand localizes the result withmake_aware; raisesInvalidPartitionKeyErroron parse failure.PartitionedAssetTimetable(timetables/simple.py): collectsto_partition_date(key)from each asset mapper and returns the value only if all temporal mappers agree; otherwise,None.InvalidPartitionKeyError(ValueError)(exceptions.py) for invalid partition keys.partition_dateis populated on all manually triggered runs:api/common/trigger_dag.pyresolves the date and passes it toDagRun.create_or_update_dagrunaspartition_date=.api_fastapi/core_api/datamodels/dag_run.py(TriggerDAGRunPostBody.to_trigger_params) resolves it and includespartition_datein thereturned params; the public route (
routes/public/dag_run.py) forwardsparams["partition_date"]toDagModel.create_dagrun.execution_api/routes/dag_runs.pycatchesInvalidPartitionKeyErrorand returns HTTP 400.Was generative AI tooling used to co-author this PR?
Generated-by: [Claude] following the guidelines
{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.