The logical key of a table tends to tell you what it means. I.e. what does a record represent.
Up to this point, the logical key of the dag run table was logical_date + dag_id.
(i will call it logical date not execution date just to keep it simple).
In other words, the combination of logical_date + dag_id is what defined a dag run, and what uniquely identified it.
What does the logical date represent? I think there's really no better way to describe it than the word "partition". So that is what I'll use here.
From the beginning, essentially the default assumption of Airflow was that the user was supporting a date-partition-driven workflow. The evidence for this is quite widespread.
In this way, the dag run table functioned as the store of data completeness. It told you in effect which partitions were fulfilled and which were not. Airflow could answer the question "which is the dag run for date X", "what's the dag run state for date X", "give me the run prior to run X". Indeed, Airflow needed to function in this way in order for backfill to function, and to prevent concurrent runs of the same partition. The grid view was also a data completeness view. Every vertical bar represented a single logical date.
I think that we can all recognize that, this canonical logical-date-driven dag pattern, not everyone uses it. I would guess that it's actually a minority of dags. But there isn't really great data on this.
Personally, in my former life as a data engineer, I almost never used this design pattern. I never worked at a hive shop. I mostly did things either incremental or full refresh. I had virtually no use for execution_date. Instead I wrote my own interface for storing task state and a watermark operator to make use of it for incremental loads.
That having been said, I know that other engineers and teams do use logical date, and do write their dags in that classic Airflow partition-driven style.
Recently, in AIP-83, we removed the uniqueness constraint on (dag_id, logical_date). This means that now there can now be many dag runs with the same logical date for the same dag. It also means that there's no longer a well-defined answer to "is this date fulfilled or not?" Or "give me the prior dag run". It introduces questions like, "if there are three runs for date X, and user backfills this range, do we create one new run, or clear and rerun all 3?" Or, how do we ascertain what is the state of date X (e.g. when user backfills with "rerun failed" or "rerun completed") when there are multiple runs for that date?
In other words, this blows up the traditional airflow dag run semantics.
Even though I personally almost never used this design pattern, I know it's a valid pattern. And fundamentally, since the beginning of time, Airflow has assumed all dags follow this pattern. And while most of us probably just ignore it or look the other way, I would guess some folks probably like the pattern or in any case are used to it.
Fundamentally my contention is that we should continue to support this pattern. I also think it should no longer be the default. I think the partition-driven style should be one available mode. But probably a more sensible default would be that the dag just runs when it is scheduled, and it's not connected to any partition or data interval.
The blanket removal of uniqueness constitutes a breaking change. The behavior is simply different.
Where previously, we could point to the uniqueness constraint of dag_run to understand what a dag run means, now we cannot.
Where previously there could only ever be one dag run per logical date, now there could be many, and that introduces some ambiguities with regard to inferring the state of the partition:
Data intervals are not a specific target of this proposal. But, in any of these solutions we have to think about what are the implications for data intervals and what to do about them. And, in some of the solutions considered, the proposal is to not have data intervals defined in some scenarios. Generally speaking, I think they are confusing and misleading and we would probably do well, as is planned for assets, to move away from them.
With that let's briefly discuss some of the issues.
One problem is that, they don't tend to tell the truth. If a dag's behavior is not driven by logical_date, then the data interval object has no meaning. In other words it is actively misleading. It tells you the data interval for this dag is A→B, but in reality it has nothing to do with what the dag did.
Data intervals were layered onto the existing (albeit implicit) partitioning concept represented by execution_date / logical_date. They assume more about what the user is doing than they really need to. So e.g. rather than saying this execution_date is simply the partition key for this run (whatever that may mean) they say this logical date represents an interval. But what if a M-F dag really does process only the M-F data, i.e. really only deals with one date at a time, and doesn't care about weekends?
It also creates trouble when trying to define what the data interval should be for a dataset-triggered run, or a manually-triggered run. With dataset-triggered, we take the min and max of the the dag runs that created the dataset events that triggered the run. In practice, I suspect this generally is not in agreement with the actual work that is being done.
We have a number of options.
Let's discuss each one
We could say, for now let's keep logical date unique and defer the removal of uniqueness constraint until 3.1+. With this approach, it's possible we could design a solution without the time pressure of airflow 3. The cost here is that, we would likely spend more time on the effort overall, and we would not have the option to break backcompat that we do with 3.0. Though, I don't think we really need to breakcompat – indeed that's sort of the point of this amendment proposal.
To me this is a very interesting approach. We could largely keep the dag run semantics unchanged. But allow users who want to trigger runs manually with no logical date to optionally set it null thereby avoiding any uniqueness constraint.
But there are multiple paths here. The simplest one is, keep everything the same but allow manual triggering of runs with null logical date and then these runs have no data interval. Another path is to add timetables for which logical_date is always null and there are never data intervals.
With this approach we sort of have our cake and eat it too.
If you don't care about logical date or data intervals, then maybe indicate that with the timetable you use:
DAG(dag_id="blah", schedule=MyNoLogicalDateNonDataIntervalTimetable(...)) |
Or, if we want to let users continue to use a cron expression, we could add a DAG level param.
DAG(dag_id="blah", schedule="0 0 * * *", logical_dates=False) |
We could allow users to declare whether the dag should use old semantics or new semantics.
This old vs new distinction is just shorthand; in reality we should not describe it as old vs new, since this implies a judgment about it, whereas I think we should recognize the old way as a valid way of designing pipelines, even as we make it just one option instead of the only option.
With this approach, we propose to add a partition param to the dag object. At this time, we would propose three available behaviors.
E.g. we could declare partition="implicit" to say that the dag should have partitions defined by the schedule.
DAG(dag_id="blah", partition="implicit") |
With implicit partitioning, we keep the old Airflow semantics. When the scheduler schedules a dag run, it creates a record in a dag_partition table. This table would ensure uniqueness, within the scope of the dag, of partition keys. Each partition key would be associated with only one dag run. So this would make well defined the questions "what is the state for this partition?", and "what is the run prior to this run?". Partition info would be made available in the execution context. Partition date would equal logical date always. Data interval would behave the same.
To disable partitioning for a dag, user can set partition=None.
DAG(dag_id="blah", partition=None) |
With partition=None, no partition records would be created in the dag_partition table. No uniqueness would be enforced w.r.t. logical date.
We have explored adding other configurability to the partition paramater. E.g. something to allow us to not have data intervals, or to change when the partition is run – e.g. on the partition date and not on the start of next partition.
Aside: I think making dag partitioning an explicit thing in airflow would be an overall good thing, beyond the whole uniqueness issue. It clarifies what exactly is going on, i.e. why execution X does not run until the next partition – cus it would not be complete otherwise. It also makes it possible for the user to say, every day rerun the last N partitions. This is essentially a partition-driven workflow with a configurable lookback.
One thing I don't love about this is how to handle the case where the user
This is essentially the do nothing approach. We more or less look the other way and don't worry about it. Leave it how it is. Backfill will look at the latest (by start_date) run for a given logical date to determine state. Similar elsewhere. No protection against concurrent runs. No well-defined view of data completeness.
Backfill always only adds new rows to the dag run table. It never "clears" an existing row to be rerun.
One suggestion that came up is best explained by example. Let's say there are 10 runs for date X that would be in the range of a backfill with "rerun failed". We could let the user choose at the time of backfill creation whether they mean "all failed" or "only latest failed" etc. And maybe we provide some mechanism for dag author, when setting depends on past, to state more precisely what they mean by that. To me this is all resolved much more simply and intuitively by just providing a flag or similar to preserve the old semantics. I don't see why we can't have both.
I would be in favor of 1, 2, or 3, with a preference for 3.
Additionally, if we do something like 3, we should make non-partition-driven dags the default, with no data interval.