AIP-39: Handle DAG scheduling with timetables

[uranusjr]

This creates a skeleton for the AIP-39 implementation to build on, and refactor the logic handling @once schedules out of the DAG class.

The current implementation contains a lot of duplicated code, and the PR probably shouldn’t be merged until I can pick most of next_dagrun_after_date apart (and ultimately eliminate it entirely and rewrite all the tests against DAG.next_dagrun_info() or the various TimeTable classes instead).

Update: I’ve since did that, so this PR is clean now.

I want to post this WIP since the trivial @once implementation already exposes things missing from the interface outlined in AIP-39, and I had to invent an argument (between) to cover it. I want to know how ya’ll think of it 🙂

(Note: The current TimeTable.next_dagrun_info() interface is missing the session argument from AIP-39 because I don’t need it yet. It will be added when it needs to be.)

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
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.

[jhtimmins]

@uranusjr Can you explain why the between argument is necessary? just from looking at the code I can't tell

[uranusjr]

betwewn: TimeRestriction is a time interval that the next DAG run must be scheduled between, reflecting start_date and end_date of the DAG and tasks in it. A run cannot be scheduled before the latest start_date (must be postponed until then), nor after the earliest end_date (must be cancelled). This logic is implemented in TimeRestriction.restrict() to adjust the schedule “preferred” by the time table.

I’ve actually removed restrict() and put the logic inside next_dagrun_info() instead (will push shortly) since the restriction logic needs to be slightly different depending on whether the time table is based on a data interval or not.

[uranusjr]

Alright, I’ve pushed my implementation to all the possible schedule_interval variants in use, and this should be ready to be merged on its own. The next step would be to gradually refactor things calling into the DAG class to use the time table API, instead of accessing schedule_interval and methods related to it.

I’ve noticed a difficulty during my initial investigation though. There are various places using DAG.previous_schedule(), which calculates the DAG’s previous run schedule (logically; no guarantee the run actually happened). This is difficult to represent with the time table design, since the time table only schedules the next run (which can be used to replace DAG.following_schedule(), in case someone wonders).

The simple solution would be to introduce something like TimeTable.prev_dagrun_info() to do it, but I’m wondering, is this really a good way to calculate the “previous run”? I’d imagine I could build something like a linked list of DAG runs, so I can just query when the last scheduled DAG run was from the database, instead of calculating a logical value that does not actually guarantee to reflect reality…

[ashb]

Is your PR description (or the bit about "should not be merged") still accurate?

[uranusjr]

Is your PR description (or the bit about "should not be merged") still accurate?

No it’s not, thanks for catching this. I’ve edited the top message to reflect the current status.

[uranusjr]

Grr, not sure what’s going on the the Static Checks job, it just says “This check failed” for me and nothing else.

The Postgres check failed with

Cannot start service trino: driver failed programming external connectivity on endpoint trino (87de12f0f1304fe6056905791286b37b88245be569e5be81c8f993c29560fd28):
Error starting userland proxy: listen tcp4 0.0.0.0:38080: bind: address already in use

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[kaxil]

👏 Nice work 🥳 🎉

[kaxil]

Suggested change

	(its execution date), and when it can be scheduled, , according to the
	(its execution date), and when it can be scheduled, according to the

[uranusjr]

I’m going to pull the trigger on this soon-ish if the checks pass; this one modifies way too many files and is causing merge conflicts left and right.

[uranusjr]

Man it’s so difficult to see all checks pass together. I think I’ve seen each of them passed at least once during my multiple attempts though, so this is going in.

[potiuk]

Man it’s so difficult to see all checks pass together. I think I’ve seen each of them passed at least once during my multiple attempts though, so this is going in.

Should be now more stable with #16689 and #16682 merged. It's a constant struggle :)

[ashb]

🎉

[potiuk]

Woohoo!

[kaxil]

🎉