Skip to content

Fix backfill completion race #68729

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

Open
PrithviBadiga wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix backfill completion race #68729

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cfc9794
Fix backfill completion race
PrithviBadiga Jun 18, 2026
9e605c4
Fix backfill inflight completion tracking
PrithviBadiga Jun 18, 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
37 changes: 33 additions & 4 deletions airflow-core/src/airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@
TaskOutletAssetReference,
)
from airflow.models.asset_state_store import AssetStateStoreModel
from airflow.models.backfill import Backfill, BackfillDagRun
from airflow.models.backfill import Backfill, BackfillDagRun, BackfillDagRunExceptionReason
from airflow.models.callback import Callback, CallbackKey, CallbackType, ExecutorCallback
from airflow.models.connection_test import (
ACTIVE_STATES as CONNECTION_TEST_ACTIVE_STATES,
Expand Down Expand Up @@ -2347,7 +2347,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
def _mark_backfills_complete(self, *, session: Session = NEW_SESSION) -> None:
"""Mark completed backfills as completed."""
self.log.debug("checking for completed backfills.")
unfinished_states = (DagRunState.RUNNING, DagRunState.QUEUED)
now = timezone.utcnow()
# todo: AIP-78 simplify this function to an update statement
initializing_cutoff = now - timedelta(minutes=2)
Expand All @@ -2362,8 +2361,38 @@ def _mark_backfills_complete(self, *, session: Session = NEW_SESSION) -> None:
Backfill.created_at < initializing_cutoff,
),
~exists(
select(DagRun.id).where(
and_(DagRun.backfill_id == Backfill.id, DagRun.state.in_(unfinished_states))
select(BackfillDagRun.id)

@potiuk potiuk Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The test covers the logical_date branch well, including the transition from "stays active" to "completes once the run succeeds". The partition_key branch — logical_date IS NULL with a non-null partition_key — has no coverage, and it's the harder of the two to reason about.

Since the partitioned path is the reason that branch exists, a companion test there would be worth adding.

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

.join(DagRun, DagRun.id == BackfillDagRun.dag_run_id, isouter=True)
.where(
BackfillDagRun.backfill_id == Backfill.id,
or_(
and_(
BackfillDagRun.dag_run_id.is_(None),
BackfillDagRun.exception_reason.is_(None),
),
DagRun.state.in_(State.unfinished_dr_states),
and_(
BackfillDagRun.dag_run_id.is_(None),
BackfillDagRun.exception_reason == BackfillDagRunExceptionReason.IN_FLIGHT,
exists(

@potiuk potiuk Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a correlated EXISTS containing a second correlated EXISTS, with the inner one referencing BackfillDagRun.logical_date / .partition_key from two levels out. SQLAlchemy's auto-correlation usually gets this right, but two-level correlation is a classic place for it to silently correlate against the wrong FROM and produce a subtly different predicate.

Could you paste the compiled SQL (print(stmt.compile(compile_kwargs={"literal_binds": True}))) into the PR description for at least Postgres? It'd let a reviewer confirm the correlation is what you intend without reconstructing it mentally.

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

select(DagRun.id).where(
DagRun.dag_id == Backfill.dag_id,
DagRun.state.in_(State.unfinished_dr_states),
or_(
and_(
BackfillDagRun.logical_date.is_not(None),
DagRun.logical_date == BackfillDagRun.logical_date,
),
and_(
BackfillDagRun.logical_date.is_(None),
BackfillDagRun.partition_key.is_not(None),
DagRun.partition_key == BackfillDagRun.partition_key,

@potiuk potiuk Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The outer correlation is fine — BackfillDagRun has a unique constraint on (backfill_id, dag_run_id), so filtering by backfill_id is indexed.

The inner one I'm less sure about: it scans DagRun filtered by dag_id + state + either logical_date or partition_key. Is DagRun.partition_key indexed? If not, this runs every 30 seconds against what is typically the largest table in the deployment. Worth checking the plan on an instance with a large dag_run table before this lands.

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

),
),
)
),
),
),
)
),
)
Expand Down
4 changes: 0 additions & 4 deletions airflow-core/src/airflow/models/backfill.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -383,7 +383,6 @@ def _create_backfill_dag_run_non_partitioned(
session.add(
BackfillDagRun(
backfill_id=backfill_id,
dag_run_id=None,
logical_date=info.logical_date,
partition_key=info.partition_key,
exception_reason=non_create_reason,
Expand Down Expand Up @@ -416,7 +415,6 @@ def _create_backfill_dag_run_non_partitioned(
session.add(
BackfillDagRun(
backfill_id=backfill_id,
dag_run_id=None,
logical_date=info.logical_date,
partition_key=info.partition_key,
exception_reason=BackfillDagRunExceptionReason.IN_FLIGHT,
Expand Down Expand Up @@ -464,7 +462,6 @@ def _create_backfill_dag_run_non_partitioned(
session.add(
BackfillDagRun(
backfill_id=backfill_id,
dag_run_id=None,
logical_date=info.logical_date,
partition_key=info.partition_key,
exception_reason=BackfillDagRunExceptionReason.IN_FLIGHT,
Expand Down Expand Up @@ -492,7 +489,6 @@ def _create_backfill_dag_run_partitioned(
session.add(
BackfillDagRun(
backfill_id=backfill_id,
dag_run_id=None,
logical_date=info.logical_date,
partition_key=info.partition_key,
exception_reason=non_create_reason,
Expand Down
84 changes: 83 additions & 1 deletion airflow-core/tests/unit/jobs/test_scheduler_job.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,13 @@
AssetPartitionDagRun,
PartitionedAssetKeyLog,
)
from airflow.models.backfill import Backfill, BackfillDagRun, ReprocessBehavior, _create_backfill
from airflow.models.backfill import (
Backfill,
BackfillDagRun,
BackfillDagRunExceptionReason,
ReprocessBehavior,
_create_backfill,
)
from airflow.models.callback import Callback, ExecutorCallback
from airflow.models.connection_test import (
ConnectionTestKey,
Expand Down Expand Up @@ -10199,6 +10205,82 @@ def test_mark_backfills_complete_multiple_independent(dag_maker, session):
assert b_running.completed_at is None


def test_mark_backfills_complete_waits_for_inflight_association(dag_maker, session):
"""Backfill should stay active while an IN_FLIGHT association points at a queued DagRun."""
clear_db_backfills()
dag_id = "test_backfill_waits_for_inflight_association"
with dag_maker(serialized=True, dag_id=dag_id, schedule="@daily"):
BashOperator(task_id="hi", bash_command="echo hi")
b = Backfill(
dag_id=dag_id,
from_date=pendulum.parse("2021-01-01"),
to_date=pendulum.parse("2021-01-03"),
max_active_runs=10,
dag_run_conf={},
reprocess_behavior=ReprocessBehavior.NONE,
)
session.add(b)
session.commit()
backfill_id = b.id

successful_dr = DagRun(
dag_id=dag_id,
run_id="backfill__2021-01-01T00:00:00+00:00",
run_type=DagRunType.BACKFILL_JOB,
logical_date=pendulum.parse("2021-01-01"),
data_interval=(pendulum.parse("2021-01-01"), pendulum.parse("2021-01-02")),
run_after=pendulum.parse("2021-01-02"),
state=DagRunState.SUCCESS,
backfill_id=backfill_id,
)
inflight_dr = DagRun(
dag_id=dag_id,
run_id="scheduled__2021-01-02T00:00:00+00:00",
run_type=DagRunType.SCHEDULED,
logical_date=pendulum.parse("2021-01-02"),
data_interval=(pendulum.parse("2021-01-02"), pendulum.parse("2021-01-03")),
run_after=pendulum.parse("2021-01-03"),
state=DagRunState.QUEUED,
)
session.add_all([successful_dr, inflight_dr])
session.flush()
session.add_all(
[
BackfillDagRun(
backfill_id=backfill_id,
dag_run_id=successful_dr.id,
logical_date=pendulum.parse("2021-01-01"),
sort_ordinal=1,
),
BackfillDagRun(
backfill_id=backfill_id,
dag_run_id=inflight_dr.id,
logical_date=pendulum.parse("2021-01-02"),
exception_reason=BackfillDagRunExceptionReason.IN_FLIGHT,
sort_ordinal=2,
),
]
)
session.commit()
session.expunge_all()

runner = SchedulerJobRunner(
job=Job(job_type=SchedulerJobRunner.job_type), executors=[MockExecutor(do_update=False)]
)
runner._mark_backfills_complete()
b = session.get(Backfill, backfill_id)
assert b.completed_at is None

inflight_dr = session.get(DagRun, inflight_dr.id)
inflight_dr.state = DagRunState.SUCCESS
session.commit()
session.expunge_all()

runner._mark_backfills_complete()
b = session.get(Backfill, backfill_id)
assert b.completed_at is not None


class Key1Mapper(CorePartitionMapper):
"""Partition Mapper that returns only key-1 as downstream key"""

Expand Down
Loading