mirror of
https://github.com/apache/airflow.git
synced 2026-11-04 07:12:48 +00:00
Ash Berlin-Taylor
87be002687
This is a data-model change precursor for the Task Loop implementation.
The subject sounds simple, but unfortunately this change is far more complicated
than I'd like, but for a couple of good reasons.
First though, why do we want this: to support Task Loops (AIP-111) which lets
the number of loops of a task group be decided, and created just-in-time at
runtime. All good so far, and no changes needed far.
However things change when you get to clearing TIs. Task Loops are most useful
when the output of one loop feeds in to the next. So the only sensible default
is that when you clear mid-way through a loop, say you clear iterations 2, 3, 4,
and 5, we restart and run 2' onwards. Since it is a runtime decision is is no
guarantee how many it will run, and most crucially, we don't want to make the
previous TIs invisible.
Visually, this looks like (or [png][fig1])
```
+--------+ +--------+ + - - - -+ + - - - -+ + - - - -+
| iter 1 |------| iter 2 |- - - : iter 3 :- - - : iter 4 :- - - : iter 5 : old attempts
+--------+ +--------+ + - - - -+ + - - - -+ + - - - -+ keep them + their XComs
|
clear at 3 |
|
| +--------+ +--------+ +--------+
'----->| iter 3'|------| iter 4'|------| iter 5'| re-run
+--------+ +--------+ +--------+ new attempts, new outputs
```
[fig1]: https://github.com/user-attachments/assets/cbbf3ea6-499b-42fe-8cd8-955c3f5310ff
(where each box there is a set of tasks from a looped task group for AIP-111)
We could choose to just keep the retired/superseded TI in the existing
task_instance_history and delete/loose the XCom and other data.
But that struck me as very sub-optimal, and I'm using this to also fix a long term
pain point: that having TI and TIHistory tables separate causes some other odd quirks.
```
+---------------------------------+
| task_instance |
+------>| one row per task = latest try |<--------------+
| +---------------------------------+ |
| FK on |
| copied on retry (dag_id, task_id, |
| run_id, map_index) |
+- - - - - - - - - - - - - -+ +------------------------+
: task_instance_history : | xcom · rtif |
: older tries, copied out : | no try_number column |
+- - - - - - - - - - - - - -+ +------------------------+
```
> XCom points at "whoever is current at these coordinates". When the task retries or is cleared, the old attempt's outputs have nowhere to live.
[fig2]: https://github.com/user-attachments/assets/018d0611-8bef-48d0-8467-8568d7832622
So why the `is_workingset` column? This is to keep the scheduler performant. The
current main reason to have TI and TIHistory separate is so that the scheduler
can
Now to the gnarliest point of this change: xcom_v2 et al. Why do this? Simply
put, it's because of the time to migrate the data. With the new FK on TI, we
would need to update _every single tuple_ on XCom table. I ran some tests on a
copy of a production database with 309 million XCom rows on how long it would
have taken to migrate XCom to store the creating task_instance_id on it: I gave
up and cancelled it after 14.5 hours. 309M is larger that most people will have,
but more than 14.5 hours is simply not worth it.
So instead I've got for the "generational table" approach, where we write to a
new table, and read from both. It's slightly more complicate than that due to a
few shadowing cases etc, but that's the gist of the approach.
Now, I hear you asking: "Doesn't this mean we have to carry this legacy code
path until Airflow 4 at an unknown time in the future?" Nope, I've got this
covered, just not included in this PR. In a future PR (likely landing for 3.4.x
or 3.5.0) I plan on having the scheduler "slowly" (but not too slowly) migrate
chunks of the data from the v1 table to v2, and then in 3.6 we add a migration
that does the final "move all data form xcom_v1 to xcom_v2" and document "You
need to upgrade to 3.4.x/3.5.0, and wait for X time, before upgrading to 3.6,
else expect a very slow migration" (some status in the admin screens to be added
to make this visible.)
A few notes:
- On Postgres, the migration creates a new unique constraint in the "NOT
VALID" state. This is safe because that migration was already covered
by an existing constraint we dropped. NOT VALID means that Postgres
hasn't checked all the current rows, but that it will apply it do an
insert or update going forward. This is safe for us, and saves ~20mins
on the large DB I was testing on.
- The "legacy-owner" table is created mostly so that rows in xcom_v1 et
al. still have a FK that will be caught by ON DELETE CASCADE.
```
+----------------------------+
| task_instance |
+--------->| one UUID per attempt |<--------------+
| +----------------------------+ |
| |
| task_instance_id | task_instance_id
| |
+---------------------------------+ +------------------------------------+
| legacy_task_data_owner | | xcom_v2 · rtif_v2 |
| task coords → UUID, 8.2M row | | all new writes, empty at upgrade |
+---------------------------------+ +------------------------------------+
^
| FK on task coords, NOT VALID
|
+---------------------------------+
| xcom_v1 · rtif_v1 |
| ~300M old rows, untouched |
+---------------------------------+
```
[fig3]: https://github.com/user-attachments/assets/164861a5-cf6b-44f4-a850-1eadcbc06203
10 lines
647 B
Plaintext
10 lines
647 B
Plaintext
f"DELETE {source_table} FROM { ', '.join(_from_name(tbl) for tbl in stmt.froms) }"
|
|
for frm in source_query.selectable.froms:
|
|
roles = relationship("Role", secondary=assoc_user_role, backref="user", lazy="selectin")
|
|
The platform supports **C**reate, **R**ead, **U**pdate, and **D**elete operations on most resources.
|
|
<pre><code>Code block\ndoes not\nrespect\nnewlines\n</code></pre>
|
|
"trough",
|
|
assert "task_instance_id" in route.dependant.path_param_names, (
|
|
assert "connection_test_id" in route.dependant.path_param_names, (
|
|
for element in joined - set(linter.froms)
|