Unexpected behaviour when manually triggering DAG with config using a custom timetable

[Jamie3213]

Apache Airflow version

2.5.2

What happened

When using a custom timetable, triggering a DAG manually via the Airflow UI using the "Trigger DAG w/ config" option and specifying a logical date in the past causes Airflow to schedule all subsequent DAG runs up to the current time (which have not yet been run) in addition to the manual run.

For context, I'm writing a timetable to support uneven intervals, so the behaviour should be similar to the behaviour when providing a cron schedule. All behaviour is as expected, other than when triggering a DAG run with a logical date in the past. I'd assumed this was an issue with my implementation, but I can see that the AfterWorkdayTimetable example in the docs also exhibits the same behaviour. I know that cron strings are scheduled using the CronDataIntervalTimetable, but it's not clear to me from the source code why this behaves differently to my own timetable or the example timetable in this respect, and I can't see anything in the docs which mentions this behaviour.

What you think should happen instead

Airflow should create a manual DAG run for the specified logical date, but should not schedule any subsequent DAG runs as a result of this manual DAG run. The behaviour should be equivalent to the behaviour observed when running a DAG whose schedule is given using a cron string.

How to reproduce

• Create a DAG using the AfterWorkdayTimetable example in the docs and set catchup to False.
• From the Airflow UI, use the "Trigger DAG w/ config" feature and specify a logical date in the past which falls on a weekday - Airflow will create a manual run for the specified weekday, but will then schedule all DAG runs subsequent to this up to today.

Operating System

macOS Ventural Version 13.2.1

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

Standard Airflow Docker deployment exactly as outlined in the docs.

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[jscheffl]

I tested this bug report with the provided example on Airflow 2.7.0 and latest main and can confirm this "strange" behavior. Made some regression and can confirm that even when catchup=False is set, an execution of a manual run from a DAG with a custom TimeTable adds scheduled jobs filling the gap. In total with my tests the scheduling behavior looks inconsistent like below:

Scheduling Expression	Initial Run of Last Time Period	Filling Gaps	Notes
Custom TImetable as example AfterWorkdayTimetable	No. Current time period is not started automatically	Yes. When a manual run with a logical date in the past is executed, occurrences after this are started and fill the gaps.	This is actually unexpected and inconsistent to other default timetables and feels like a partly "catchup" after first logic date execution is implicitly made.
Classic cron expression "0 0 * * *"	Yes. An initial run for the current period is made	No.	This is how I expected it as well for a custom timetable.
Usage of CronTriggerTimetable	Yes. An initial run for the current period is made	No.	This is how I expected it as well for a custom timetable.
Usage of EventsTimetable	Ups, all events defined in the past were executed.	n/a	This was a surprise. Whereas I would see this also as a feature but all defined events in the past were run even when catchup=False was set. Might be another bug? Or explicity feature? Could not find any mentioning in the docs, so rather documentation flaw of bug.
Usage of datetime.timedelta()	Yes. An initial run for the current period is made	? Another ups. A second run was scheduled after the manual trigger.	This was also rather unexpected. Can be repeated. Whenever a run with a logical date in the past is triggered, one additional scheduled run based on current time is started as well.

I tried to walk through the code in airflow/jobs/scheduler_job_runner.py which I assume is the core part of logic where this is implemented. But after drilling into the code I was lost :-( and not able to pin down the root error. I fear an scheduler expert is needed here for investigation. (@ashb ?)
I assume it is an artifact related to the AIP-39 integration.

The following DAG code was used for the regression test, just for reference

import datetime
import pendulum

from airflow import DAG
from airflow.decorators import task
from airflow.models.taskinstance import TaskInstance
from airflow.timetables.events import EventsTimetable
from airflow.timetables.trigger import CronTriggerTimetable
from workday import AfterWorkdayTimetable
# Note: to make timetable available from examples start webserver and scheduler with
# AIRFLOW__CORE__PLUGINS_FOLTER=/opt/airflow/airflow/example_dags/plugins
# as prefix in breeze

# Have some parameters to force trigger form to show to be able to pick a logical date
DUMMY_PARAM={"test": 123}

with DAG(
    dag_id="after_workday_timetable_dag",
    start_date=pendulum.datetime(2023, 8, 1, tz="UTC"),
    catchup=False,
    schedule=AfterWorkdayTimetable(),
    params=DUMMY_PARAM
):
    @task
    def test(ti: TaskInstance=None):
        print(ti.execution_date)
    test()

with DAG(
    dag_id="after_workday_cron_regression",
    start_date=pendulum.datetime(2023, 8, 1, tz="UTC"),
    catchup=False,
    schedule="0 0 * * *",
    params=DUMMY_PARAM
):
    @task
    def test(ti: TaskInstance=None):
        print(ti.execution_date)
    test()

with DAG(
    dag_id="after_workday_crontt_regression",
    start_date=pendulum.datetime(2023, 8, 1, tz="UTC"),
    catchup=False,
    schedule=CronTriggerTimetable("0 0 * * *", timezone="UTC"),
    params=DUMMY_PARAM
):
    @task
    def test(ti: TaskInstance=None):
        print(ti.execution_date)
    test()


with DAG(
    dag_id="after_workday_events_regression",
    start_date=pendulum.datetime(2023, 8, 1, tz="UTC"),
    catchup=False,
    schedule=EventsTimetable(
        event_dates=[
            pendulum.datetime(2023, 8, 1, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 2, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 3, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 4, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 5, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 6, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 7, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 8, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 9, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 10, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 11, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 12, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 13, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 14, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 15, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 16, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 17, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 18, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 19, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 20, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 21, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 22, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 23, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 24, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 25, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 26, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 27, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 28, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 29, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 30, 1, 0, tz="America/Chicago"),
            pendulum.datetime(2023, 8, 31, 1, 0, tz="America/Chicago"),
        ],
        description="My Team's Baseball Games",
        restrict_to_events=False,
    ),
    params=DUMMY_PARAM
):
    @task
    def test(ti: TaskInstance=None):
        print(ti.execution_date)
    test()

with DAG(
    dag_id="after_workday_delta_regression",
    start_date=pendulum.datetime(2023, 8, 1, tz="UTC"),
    catchup=False,
    schedule=datetime.timedelta(days=1),
    params=DUMMY_PARAM
):
    @task
    def test(ti: TaskInstance=None):
        print(ti.execution_date)
    test()

[github-actions]

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.

[github-actions]

This issue has been closed because it has not received response from the issue author.