[AIRFLOW-1156] BugFix: Unpausing a DAG with catchup=False creates an extra DAG run

[kaxil]

Fixes the following issues:

• https://issues.apache.org/jira/browse/AIRFLOW-1056
• https://issues.apache.org/jira/browse/AIRFLOW-1156
• https://issues.apache.org/jira/browse/AIRFLOW-3369
• https://issues.apache.org/jira/browse/AIRFLOW-6577

If you create a DAG with catchup=False, when it is un-paused, it creates 2 dag runs. One for the most recent scheduled interval (expected) and one for the interval before that (unexpected).

Sample DAG:

from airflow import DAG
from datetime import datetime
from airflow.operators.dummy_operator import DummyOperator

dag = DAG(
    dag_id='DummyTest',
    start_date=datetime(2018,1,1),
    catchup=False
)

do = DummyOperator(
    task_id='dummy_task',
    dag=dag
)

Result:

2 DAG runs are created. 2018-11-18 and 108-11-17

Expected Result:

Only 1 DAG run should have been created (2018-11-18)

Cause:
This only occurs when using a timedelta as Schedule Interval (not when using Cron expression or Cron preset).

This is caused by the following block of code:

airflow/airflow/jobs/scheduler_job.py

Lines 578 to 595 in 6e4f5fa

	# don't do scheduler catchup for dag's that don't have dag.catchup = True
	if not (dag.catchup or dag.schedule_interval == '@once'):
	# The logic is that we move start_date up until
	# one period before, so that timezone.utcnow() is AFTER
	# the period end, and the job can be created...
	now = timezone.utcnow()
	next_start = dag.following_schedule(now)
	last_start = dag.previous_schedule(now)
	if next_start <= now:
	new_start = last_start
	else:
	new_start = dag.previous_schedule(last_start)
	if dag.start_date:
	if new_start >= dag.start_date:
	dag.start_date = new_start
	else:
	dag.start_date = new_start

---

Make sure to mark the boxes below before creating PR: [x]

• Description above provides context of the change
• Unit tests coverage for changes (not needed for documentation changes)
• Target Github ISSUE in description if exists
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions.
• I will engage committers as explained in Contribution Workflow Example.

---

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

[ashb]

Lol, one line fixes are the Best*

Tests please :) (I'm sure you'd get around to it)

* The worst.

[kaxil]

Lol, one line fixes are the Best*

Tests please :) (I'm sure you'd get around to it)

• The worst.

😄 Yeah - Looks like this bug was for a long time (possibly from the very start) !!! Found some old JIRAs.

[marclamberti]

Oh that's funny because I struggled a lot on this to understand why there was a difference between the cron expressions and timdelta. That's why I started to use CRON instead. Didn't think it was a bug and very happy to know it is solved :)

[tambulkar]

Is it possible to also prevent the run for the most recent scheduled interval?

[JeffryMAC]

Oh that's funny because I struggled a lot on this to understand why there was a difference between the cron expressions and timdelta. That's why I started to use CRON instead. Didn't think it was a bug and very happy to know it solved :)

There is a difference between cron exp and time delta.It's not solved and I don't know if it's a bug. See discussion here:
#8649

[kaxil]

Is it possible to also prevent the run for the most recent scheduled interval?

No, not yet

[kaxil]

Oh that's funny because I struggled a lot on this to understand why there was a difference between the cron expressions and timdelta. That's why I started to use CRON instead. Didn't think it was a bug and very happy to know it solved :)

There is a difference between cron exp and time delta.It's not solved and I don't know if it's a bug. See discussion here:
#8649

The issue you are referring to is a different one. This OR brought parity between using timedelta and cron expression for "the number of DagRuns that are created at the very start"

[kaxil]

Oh that's funny because I struggled a lot on this to understand why there was a difference between the cron expressions and timdelta. That's why I started to use CRON instead. Didn't think it was a bug and very happy to know it solved :)

There is a difference between cron exp and time delta.It's not solved and I don't know if it's a bug. See discussion here:
#8649

The issue you are referring to is a different one. This OR brought parity between using timedelta and cron expression for "the number of DagRuns that are created at the very start"

[tambulkar]

Is it possible to also prevent the run for the most recent scheduled interval?

No, not yet

Is there an issue for this/can we create one?

[kaxil]

Is it possible to also prevent the run for the most recent scheduled interval?

No, not yet

Is there an issue for this/can we create one?

Yeah please check https://github.com/apache/airflow/issues if the issue exists, if not create one :) Thanks

[mpeteuil]

The problem in AIRFLOW-1156 still seems to be an issue (at least in 1.10.x, and by the looks of the code, 2.x as well). Using a timedelta as the schedule_interval with catchup=False causes the start_date (particularly the time component) to still not be honored. This appears to be an issue with the Dag's following_schedule and previous_schedule methods from what I can tell.

[ashb]

@mpeteuil could you create a new issue for that please?

[kaxil]

@mpeteuil yeah like Ash suggested please create a new issue with steps to reproduce. We fixed it in 1.10.11