potiuk commented on code in PR #68729:
URL: https://github.com/apache/airflow/pull/68729#discussion_r3681668934
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -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)
+ .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(
Review Comment:
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
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -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)
Review Comment:
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
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -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)
+ .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(
+ 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,
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
