Manual run with same execution date as scheduled run causes the scheduled run to be skipped.

Apache Airflow version

main (development)

If "Other Airflow 2 version" selected, which one?

No response

What happened?

When there is a manual dagrun, it is created with the logical date as data interval end. In case of scheduled runs the logical date is the data interval start. In this case when a future run is scheduled there is a check for existing dagruns in the loop with respect to execution date (logical date) which causes the dagrun to be skipped since it exists from the manual run with a different data interval.

This looks like a known issue in 2.x and can to be changed in 3.x . Related issue #41014

airflow/airflow/jobs/scheduler_job_runner.py

Lines 1314 to 1348 in 4d849e8

    
           existing_dagruns = ( 
        
               session.execute( 
        
                   select(DagRun.dag_id, DagRun.execution_date).where( 
        
                       tuple_in_condition( 
        
                           (DagRun.dag_id, DagRun.execution_date), 
        
                           ((dm.dag_id, dm.next_dagrun) for dm in dag_models), 
        
                       ), 
        
                   ) 
        
               ) 
        
               .unique() 
        
               .all() 
        
           ) 
        
           active_runs_of_dags = Counter( 
        
               DagRun.active_runs_of_dags(dag_ids=(dm.dag_id for dm in dag_models), session=session), 
        
           ) 
        
           for dag_model in dag_models: 
        
               dag = self.dagbag.get_dag(dag_model.dag_id, session=session) 
        
               if not dag: 
        
                   self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id) 
        
                   continue 
        
               dag_hash = self.dagbag.dags_hash.get(dag.dag_id) 
        
               data_interval = dag.get_next_data_interval(dag_model) 
        
               # Explicitly check if the DagRun already exists. This is an edge case 
        
               # where a Dag Run is created but `DagModel.next_dagrun` and `DagModel.next_dagrun_create_after` 
        
               # are not updated. 
        
               # We opted to check DagRun existence instead 
        
               # of catching an Integrity error and rolling back the session i.e 
        
               # we need to set dag.next_dagrun_info if the Dag Run already exists or if we 
        
               # create a new one. This is so that in the next Scheduling loop we try to create new runs 
        
               # instead of falling in a loop of Integrity Error. 
        
               if (dag.dag_id, dag_model.next_dagrun) not in existing_dagruns:

What you think should happen instead?

Scheduled dag run should be executed though a manual run with same execution date exists.

How to reproduce

Place the below dag and parse it.
Before unpausing the dag trigger a manual run with logical date as "2024-07-01T00:00:00+00".
A manual run is triggered with run_id as manual__2024-07-01T00:00:00+00:00 and exeuction_date as 2024-07-01 00:00:00+00:00. Data interval is (2024-06-01, 00:00:00 UTC, 2024-07-01, 00:00:00 UTC)
The run with data interval (2024-07-01T00:00:00+00:00, 2024-08-01T00:00:00+00:00) is not scheduled.

from datetime import datetime, timedelta

from airflow import DAG
from airflow.decorators import task


with DAG(
    dag_id="execution_date_debug",
    start_date=datetime(2024, 1, 1),
    catchup=True,
    schedule_interval="0 0 1 * *",
) as dag:

    @task
    def empty():
        pass

    empty()

Operating System

Ubuntu

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

@shahar1

hi @shahar1, I'd like to work on this issue .Could you please assign it to me ?

@shahar1

hi @shahar1, I'd like to work on this issue .Could you please assign it to me ?

Sure, thank you and good luck!

This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.

@Prab-27

@Prab-27 We are unassigning you from this issue as part of our updated assignment policy.

This is not meant to discourage your contribution — quite the opposite! You are still very welcome to work on this issue and submit a PR for it. Simply comment that you are working on it and open a PR when ready.

We found that formal assignments were not working well, as they often prevented others from contributing when the assignee was not actively working on the issue.

Navigation Menu

Repository navigation

Manual run with same execution date as scheduled run causes the scheduled run to be skipped. #41208

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	existing_dagruns = (
	session.execute(
	select(DagRun.dag_id, DagRun.execution_date).where(
	tuple_in_condition(
	(DagRun.dag_id, DagRun.execution_date),
	((dm.dag_id, dm.next_dagrun) for dm in dag_models),
	),
	)
	)
	.unique()
	.all()
	)

	active_runs_of_dags = Counter(
	DagRun.active_runs_of_dags(dag_ids=(dm.dag_id for dm in dag_models), session=session),
	)

	for dag_model in dag_models:
	dag = self.dagbag.get_dag(dag_model.dag_id, session=session)
	if not dag:
	self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
	continue

	dag_hash = self.dagbag.dags_hash.get(dag.dag_id)

	data_interval = dag.get_next_data_interval(dag_model)
	# Explicitly check if the DagRun already exists. This is an edge case
	# where a Dag Run is created but `DagModel.next_dagrun` and `DagModel.next_dagrun_create_after`
	# are not updated.
	# We opted to check DagRun existence instead
	# of catching an Integrity error and rolling back the session i.e
	# we need to set dag.next_dagrun_info if the Dag Run already exists or if we
	# create a new one. This is so that in the next Scheduling loop we try to create new runs
	# instead of falling in a loop of Integrity Error.
	if (dag.dag_id, dag_model.next_dagrun) not in existing_dagruns:

Manual run with same execution date as scheduled run causes the scheduled run to be skipped. #41208

Description

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Activity

Prab-27 commented on Sep 6, 2024

shahar1 commented on Sep 6, 2024

github-actions commented on Sep 7, 2025

potiuk commented on Mar 2, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions