This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8b401f28f03 Fix backfill params not overriding existing DAG run conf
(#64939)
8b401f28f03 is described below
commit 8b401f28f03ce99dc304d924d67a8b04befb2e7b
Author: Shivam Rastogi <[email protected]>
AuthorDate: Tue Apr 21 06:11:09 2026 -0400
Fix backfill params not overriding existing DAG run conf (#64939)
* Fix backfill params not overriding existing DAG run conf
When reprocessing an existing DAG run during backfill, the
dag_run_conf was not being applied to the cleared run. This adds
the dag_run_conf parameter to _handle_clear_run() and conditionally
updates conf in the DagRun UPDATE statement.
Also adds conf validation in _create_backfill() using
dag.params.deep_merge().validate() to match the validation
done in create_dagrun() for new runs, ensuring invalid conf
is rejected before any runs are created or cleared.
closes: #59043
* UI: Add override params checkbox to backfill form
Add an "Override parameters on existing runs" checkbox to the
backfill form, unchecked by default. When unchecked, dag_run_conf
is sent as null, preserving existing run params during reprocessing.
When checked, the form values are sent and override existing conf.
Also changes BackfillPostBody.dag_run_conf from dict={} to
dict|None=None so the API can distinguish between "no conf
provided" (null, preserve existing) and "empty conf" ({}, override).
related: #59043
* Address code review: pass null through and regenerate openapi artifacts
* Address review: InvalidBackfillConf exception and partitioned path comment
* UI: Disable max-lines in RunBackfillForm with justification
* Apply suggestion from @uranusjr
Co-authored-by: Tzu-ping Chung <[email protected]>
* Apply suggestion from @uranusjr: chained .values()
---------
Co-authored-by: Tzu-ping Chung <[email protected]>
---
.../api_fastapi/core_api/datamodels/backfills.py | 2 +-
.../core_api/openapi/v2-rest-api-generated.yaml | 7 +-
.../core_api/routes/public/backfills.py | 3 +
airflow-core/src/airflow/models/backfill.py | 29 +++-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 14 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 4 +-
.../ui/public/i18n/locales/en/components.json | 1 +
.../src/components/DagActions/RunBackfillForm.tsx | 13 +-
.../airflow/ui/src/queries/useCreateBackfill.ts | 2 +-
airflow-core/tests/unit/models/test_backfill.py | 150 +++++++++++++++++++++
.../src/airflowctl/api/datamodels/generated.py | 2 +-
11 files changed, 211 insertions(+), 16 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
index 87d2677028e..64ed8bb05e9 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
@@ -33,7 +33,7 @@ class BackfillPostBody(StrictBaseModel):
from_date: datetime
to_date: datetime
run_backwards: bool = False
- dag_run_conf: dict = {}
+ dag_run_conf: dict | None = None
reprocess_behavior: ReprocessBehavior = ReprocessBehavior.NONE
max_active_runs: int = 10
run_on_latest_version: bool = True
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index dbd84221ac6..8f892d0c657 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -9281,10 +9281,11 @@ components:
title: Run Backwards
default: false
dag_run_conf:
- additionalProperties: true
- type: object
+ anyOf:
+ - additionalProperties: true
+ type: object
+ - type: 'null'
title: Dag Run Conf
- default: {}
reprocess_behavior:
$ref: '#/components/schemas/ReprocessBehavior'
default: none
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
index a520d243e18..e9c28d4fba9 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
@@ -50,6 +50,7 @@ from airflow.models.backfill import (
Backfill,
BackfillDagRun,
DagNonPeriodicScheduleException,
+ InvalidBackfillConf,
InvalidBackfillDate,
InvalidBackfillDirection,
InvalidReprocessBehavior,
@@ -264,6 +265,7 @@ def create_backfill(
InvalidBackfillDirection,
DagNonPeriodicScheduleException,
InvalidBackfillDate,
+ InvalidBackfillConf,
) as e:
raise RequestValidationError(str(e))
@@ -316,5 +318,6 @@ def create_backfill_dry_run(
InvalidBackfillDirection,
DagNonPeriodicScheduleException,
InvalidBackfillDate,
+ InvalidBackfillConf,
) as e:
raise RequestValidationError(str(e))
diff --git a/airflow-core/src/airflow/models/backfill.py
b/airflow-core/src/airflow/models/backfill.py
index 86dbe1713f7..febd22df461 100644
--- a/airflow-core/src/airflow/models/backfill.py
+++ b/airflow-core/src/airflow/models/backfill.py
@@ -107,6 +107,14 @@ class InvalidBackfillDate(AirflowException):
"""
+class InvalidBackfillConf(AirflowException):
+ """
+ Raised when the provided ``dag_run_conf`` fails validation against the
DAG's params.
+
+ :meta private:
+ """
+
+
class UnknownActiveBackfills(AirflowException):
"""
Raised when the quantity of active backfills cannot be determined.
@@ -256,6 +264,7 @@ def _validate_backfill_params(
from_date: datetime,
to_date: datetime,
reprocess_behavior: ReprocessBehavior | None,
+ dag_run_conf: dict | None = None,
) -> None:
depends_on_past = any(x.depends_on_past for x in dag.tasks)
if depends_on_past:
@@ -271,6 +280,11 @@ def _validate_backfill_params(
current_time = timezone.utcnow()
if from_date >= current_time and to_date >= current_time:
raise InvalidBackfillDate("Backfill cannot be executed for future
dates.")
+ if dag_run_conf is not None:
+ try:
+ dag.params.deep_merge(dag_run_conf).validate()
+ except ValueError as e:
+ raise InvalidBackfillConf(str(e)) from e
def _do_dry_run(
@@ -367,6 +381,7 @@ def _create_backfill_dag_run_non_partitioned(
backfill_id=backfill_id,
sort_ordinal=backfill_sort_ordinal,
run_on_latest=run_on_latest_version,
+ dag_run_conf=dag_run_conf,
)
else:
session.add(
@@ -440,6 +455,10 @@ def _create_backfill_dag_run_partitioned(
triggering_user_name: str | None,
session: Session,
) -> None:
+ # Partitioned backfills don't currently reprocess existing runs — if a run
exists
+ # for this partition, it's recorded as skipped via exception_reason rather
than
+ # cleared and re-queued. As a result, this function never calls
``_handle_clear_run``
+ # and therefore doesn't need to forward ``dag_run_conf`` for the reprocess
path.
stmt = _get_latest_dag_run_row_query(dag_id=dag.dag_id, info=info)
dr = session.scalar(stmt)
if dr:
@@ -516,6 +535,7 @@ def _handle_clear_run(
backfill_id: int,
sort_ordinal: int,
run_on_latest: bool = False,
+ dag_run_conf: dict | None = None,
) -> None:
"""Clear the existing Dag run and update backfill metadata."""
from sqlalchemy.sql import update
@@ -532,8 +552,8 @@ def _handle_clear_run(
run_on_latest_version=run_on_latest,
)
- # Update backfill_id and run_type in DagRun table
- session.execute(
+ # Update backfill_id, run_type, and optionally conf in DagRun table
+ stmt = (
update(DagRun)
.where(DagRun.logical_date == info.logical_date, DagRun.dag_id ==
dag.dag_id)
.values(
@@ -542,6 +562,9 @@ def _handle_clear_run(
triggered_by=DagRunTriggeredByType.BACKFILL,
)
)
+ if dag_run_conf is not None:
+ stmt = stmt.values(conf=dag_run_conf)
+ session.execute(stmt)
session.add(
BackfillDagRun(
backfill_id=backfill_id,
@@ -600,7 +623,7 @@ def _create_backfill(
f"There can be only one running backfill per Dag."
)
- _validate_backfill_params(dag, reverse, from_date, to_date,
reprocess_behavior)
+ _validate_backfill_params(dag, reverse, from_date, to_date,
reprocess_behavior, dag_run_conf)
br = Backfill(
dag_id=dag_id,
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 3317a52593c..35589676fb4 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -445,10 +445,16 @@ export const $BackfillPostBody = {
default: false
},
dag_run_conf: {
- additionalProperties: true,
- type: 'object',
- title: 'Dag Run Conf',
- default: {}
+ anyOf: [
+ {
+ additionalProperties: true,
+ type: 'object'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Dag Run Conf'
},
reprocess_behavior: {
'$ref': '#/components/schemas/ReprocessBehavior',
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index cfb5b37bebb..4b3efcaabeb 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -122,8 +122,8 @@ export type BackfillPostBody = {
to_date: string;
run_backwards?: boolean;
dag_run_conf?: {
- [key: string]: unknown;
- };
+ [key: string]: unknown;
+} | null;
reprocess_behavior?: ReprocessBehavior;
max_active_runs?: number;
run_on_latest_version?: boolean;
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json
b/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json
index 28ed7c7a15a..64a78bee04b 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json
@@ -10,6 +10,7 @@
"maxRuns": "Max Active Runs",
"missingAndErroredRuns": "Missing and Errored Runs",
"missingRuns": "Missing Runs",
+ "overrideExistingParams": "Override parameters on existing runs",
"permissionDenied": "Dry Run Failed: User does not have permission to
create backfills.",
"reprocessBehavior": "Reprocess Behavior",
"run": "Run Backfill",
diff --git
a/airflow-core/src/airflow/ui/src/components/DagActions/RunBackfillForm.tsx
b/airflow-core/src/airflow/ui/src/components/DagActions/RunBackfillForm.tsx
index 089c914707d..c1873b9d352 100644
--- a/airflow-core/src/airflow/ui/src/components/DagActions/RunBackfillForm.tsx
+++ b/airflow-core/src/airflow/ui/src/components/DagActions/RunBackfillForm.tsx
@@ -1,3 +1,5 @@
+/* eslint-disable max-lines -- form aggregates date range, reprocess behavior,
and param controls in a single UX flow */
+
/*!
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -50,6 +52,7 @@ const RunBackfillForm = ({ dag, onClose }:
RunBackfillFormProps) => {
const { t: translate } = useTranslation(["components", "common"]);
const [errors, setErrors] = useState<{ conf?: string; date?: unknown }>({});
const [unpause, setUnpause] = useState(true);
+ const [overrideParams, setOverrideParams] = useState(false);
const [formError, setFormError] = useState(false);
const initialParamsDict = useDagParams(dag.dag_id, true);
const { conf } = useParamStore();
@@ -119,7 +122,7 @@ const RunBackfillForm = ({ dag, onClose }:
RunBackfillFormProps) => {
createBackfill({
requestBody: {
...fdata,
- dag_run_conf: JSON.parse(fdata.conf) as Record<string, unknown>,
+ dag_run_conf: overrideParams ? (JSON.parse(fdata.conf) as
Record<string, unknown>) : null,
},
});
};
@@ -250,6 +253,14 @@ const RunBackfillForm = ({ dag, onClose }:
RunBackfillFormProps) => {
<Spacer />
</>
) : undefined}
+ <Checkbox
+ checked={overrideParams}
+ colorPalette="brand"
+ onChange={() => setOverrideParams(!overrideParams)}
+ >
+ {translate("backfill.overrideExistingParams")}
+ </Checkbox>
+ <Spacer />
<ConfigForm
control={control}
errors={errors}
diff --git a/airflow-core/src/airflow/ui/src/queries/useCreateBackfill.ts
b/airflow-core/src/airflow/ui/src/queries/useCreateBackfill.ts
index 673f7d5a82d..c99430e8cb6 100644
--- a/airflow-core/src/airflow/ui/src/queries/useCreateBackfill.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useCreateBackfill.ts
@@ -79,7 +79,7 @@ export const useCreateBackfill = ({ onSuccessConfirm }: {
onSuccessConfirm: () =
mutate({
requestBody: {
dag_id: dagId,
- dag_run_conf: data.requestBody.dag_run_conf ?? {},
+ dag_run_conf: data.requestBody.dag_run_conf,
from_date: formattedDataIntervalStart,
max_active_runs: data.requestBody.max_active_runs,
reprocess_behavior: data.requestBody.reprocess_behavior,
diff --git a/airflow-core/tests/unit/models/test_backfill.py
b/airflow-core/tests/unit/models/test_backfill.py
index d3323aeab35..a8d7eb70d98 100644
--- a/airflow-core/tests/unit/models/test_backfill.py
+++ b/airflow-core/tests/unit/models/test_backfill.py
@@ -33,6 +33,7 @@ from airflow.models.backfill import (
BackfillDagRun,
BackfillDagRunExceptionReason,
DagNonPeriodicScheduleException,
+ InvalidBackfillConf,
InvalidBackfillDirection,
InvalidReprocessBehavior,
ReprocessBehavior,
@@ -389,6 +390,155 @@ def test_reprocess_behavior(reprocess_behavior, num_in_b,
exc_reasons, dag_maker
assert all(x.state == DagRunState.QUEUED for x in dag_runs_in_b)
[email protected](
+ "reprocess_behavior",
+ [ReprocessBehavior.FAILED, ReprocessBehavior.COMPLETED],
+)
+def test_backfill_conf_overrides_existing_dag_run(reprocess_behavior,
dag_maker, session):
+ """When reprocessing an existing DagRun, the backfill's dag_run_conf
should override the existing conf."""
+ with dag_maker(schedule="@daily") as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+
+ existing_date = "2021-01-03"
+ dag_maker.create_dagrun(
+ run_id=f"scheduled_{existing_date}",
+ logical_date=timezone.parse(existing_date),
+ session=session,
+ state=DagRunState.FAILED,
+ conf={"old_key": "old_value"},
+ )
+ session.commit()
+
+ new_conf = {"new_key": "new_value"}
+ b = _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2021-01-01"),
+ to_date=pendulum.parse("2021-01-05"),
+ max_active_runs=10,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf=new_conf,
+ reprocess_behavior=reprocess_behavior,
+ )
+
+ session.expunge_all()
+
+ # The reprocessed existing run should have the new conf
+ reprocessed_dr = session.scalar(
+ select(DagRun).where(
+ DagRun.dag_id == dag.dag_id,
+ DagRun.logical_date == timezone.parse(existing_date),
+ )
+ )
+ assert reprocessed_dr.conf == new_conf
+
+ # New runs created by the backfill should also have the new conf
+ all_backfill_runs = session.scalars(
+
select(DagRun).join(BackfillDagRun.dag_run).where(BackfillDagRun.backfill_id ==
b.id)
+ ).all()
+ assert all(x.conf == new_conf for x in all_backfill_runs)
+
+
+def test_backfill_none_conf_preserves_existing_dag_run_conf(dag_maker,
session):
+ """When backfill dag_run_conf is None, existing DagRun conf should be
preserved."""
+ with dag_maker(schedule="@daily") as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+
+ existing_date = "2021-01-03"
+ original_conf = {"keep": "this"}
+ dag_maker.create_dagrun(
+ run_id=f"scheduled_{existing_date}",
+ logical_date=timezone.parse(existing_date),
+ session=session,
+ state=DagRunState.FAILED,
+ conf=original_conf,
+ )
+ session.commit()
+
+ _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2021-01-01"),
+ to_date=pendulum.parse("2021-01-05"),
+ max_active_runs=10,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf=None,
+ reprocess_behavior=ReprocessBehavior.FAILED,
+ )
+
+ session.expunge_all()
+
+ reprocessed_dr = session.scalar(
+ select(DagRun).where(
+ DagRun.dag_id == dag.dag_id,
+ DagRun.logical_date == timezone.parse(existing_date),
+ )
+ )
+ assert reprocessed_dr.conf == original_conf
+
+
+def test_backfill_empty_conf_overrides_existing_dag_run(dag_maker, session):
+ """When backfill dag_run_conf is {}, existing DagRun conf should be
updated to {}."""
+ with dag_maker(schedule="@daily") as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+
+ existing_date = "2021-01-03"
+ dag_maker.create_dagrun(
+ run_id=f"scheduled_{existing_date}",
+ logical_date=timezone.parse(existing_date),
+ session=session,
+ state=DagRunState.FAILED,
+ conf={"old_key": "old_value"},
+ )
+ session.commit()
+
+ _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2021-01-01"),
+ to_date=pendulum.parse("2021-01-05"),
+ max_active_runs=10,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf={},
+ reprocess_behavior=ReprocessBehavior.FAILED,
+ )
+
+ session.expunge_all()
+
+ reprocessed_dr = session.scalar(
+ select(DagRun).where(
+ DagRun.dag_id == dag.dag_id,
+ DagRun.logical_date == timezone.parse(existing_date),
+ )
+ )
+ assert reprocessed_dr.conf == {}
+
+
+def test_backfill_rejects_invalid_conf(dag_maker, session):
+ """Backfill with invalid conf should fail validation before creating any
runs."""
+ from airflow.sdk import Param
+
+ with dag_maker(
+ schedule="@daily",
+ params={"validated_number": Param(1, type="integer", minimum=1,
maximum=10)},
+ ) as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+
+ with pytest.raises(InvalidBackfillConf, match="Invalid input for param
validated_number"):
+ _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2021-01-01"),
+ to_date=pendulum.parse("2021-01-05"),
+ max_active_runs=10,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf={"validated_number": 99},
+ )
+
+ # No runs should have been created
+ assert session.scalar(select(DagRun).where(DagRun.dag_id == dag.dag_id))
is None
+
+
def test_params_stored_correctly(dag_maker, session):
with dag_maker(schedule="@daily") as dag:
PythonOperator(task_id="hi", python_callable=print)
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index e10462fa553..50fdf66d40a 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -1132,7 +1132,7 @@ class BackfillPostBody(BaseModel):
from_date: Annotated[datetime, Field(title="From Date")]
to_date: Annotated[datetime, Field(title="To Date")]
run_backwards: Annotated[bool | None, Field(title="Run Backwards")] = False
- dag_run_conf: Annotated[dict[str, Any] | None, Field(title="Dag Run
Conf")] = {}
+ dag_run_conf: Annotated[dict[str, Any] | None, Field(title="Dag Run
Conf")] = None
reprocess_behavior: ReprocessBehavior | None = "none"
max_active_runs: Annotated[int | None, Field(title="Max Active Runs")] = 10
run_on_latest_version: Annotated[bool | None, Field(title="Run On Latest
Version")] = True
