AIP-65: Add DAG versioning support #42913

ephraimbuddy · 2024-10-10T18:18:21Z

This commit introduces versioning for DAGs.

Changes:

Introduced DagVersion model to handle versioning of DAGs.
Added version_name field to DAG for use in tracking the dagversion by users
Added support for version retrieval in the get_dag_source API endpoint
Modified DAG execution logic to reference dag_version_id instead of the dag_hash to ensure DAG runs are linked to specific versions.

jedcunningham · 2024-10-14T02:14:54Z

airflow/models/dag_version.py

+            version_number=version_number,
+            dag_code=dag_code,
+            serialized_dag=serialized_dag,
+            version_name=version_name or cls._generate_unique_random_string(session),


I think we should augment the version_name we store when a DAG author provides is to us. Otherwise, we'd have the same version_name for all versions until they change it, which isn't great.

e.g. if I have version_name=foo in my DAG, and I add a new task but don't update the version. So maybe something like foo-{version_number} or something? Or maybe we toss on an int to signify when it hasn't changed, but the DAG has, like foo-2.

We also only stamp a new version when the serdag actually changes. I wonder if we'd be better off combing the user provided version and the serdag hash to determine when to create a new version?

I think we should augment the version_name we store when a DAG author provides is to us. Otherwise, we'd have the same version_name for all versions until they change it, which isn't great.

e.g. if I have version_name=foo in my DAG, and I add a new task but don't update the version. So maybe something like foo-{version_number} or something? Or maybe we toss on an int to signify when it hasn't changed, but the DAG has, like foo-2.

That's my plan, but I want to have it in a method. The version_number increments, while the version_name remains the same until changed. However, a method named version would return version_name-version_number (foo-2), which would be used on the UI. I think having them separate is much better than using Python logic, which might go wrong.

We also only stamp a new version when the serdag actually changes. I wonder if we'd be better off combing the user provided version and the serdag hash to determine when to create a new version?

The serdag is 1:1 to dag_version. When it changes, the version also changes and there's a link between the two tables which allows us to access the hash e.g :

version = session.query(DagVersion).get(1) print(version.serialized_dag.dag_hash)

I feel we shouldn't combine the hash and version since a change in the user-provided version name would trigger a new serialized dag version, which also means a new dagversion.

The dag_hash can remain an internal property that we use to know that the version has changed

jscheffl · 2024-10-16T16:58:30Z

airflow/models/dag.py

@@ -2463,7 +2471,7 @@ def create_dagrun(
        conf: dict | None = None,
        run_type: DagRunType | None = None,
        session: Session = NEW_SESSION,
-        dag_hash: str | None = None,
+        dag_version_id: int | None = None,


Are we sure that we want to have (only) an integer value for the DAG version? Who woul dbe responsible to increment?

I am thinking about a backup/restore, how can we be sure durch such cases the IDs will not overwrite each other?

Besides the question about int: If it is an integer then it is not really an ID, we should rather call it dag_version if it represents a numeric value (I think)

dag_version_id is a foreignkey from dag_version table to dagrun and auto increments based on how sqlalchemy works. However, it will eventually be refactored to be a UUID.

I am thinking about a backup/restore, how can we be sure durch such cases the IDs will not overwrite each other?

Since it's a foreign key, I don't see how it will overwrite. The above code is just for creating the dag run. Maybe we should change it to the dag_version object itself and not the ID?

jscheffl · 2024-10-16T17:00:14Z

airflow/migrations/versions/0039_3_0_0_add_dag_versioning.py

+        sa.Column("dag_code_id", sa.Integer(), nullable=True),
+        sa.Column("serialized_dag_id", sa.Integer(), nullable=True),
+        sa.ForeignKeyConstraint(("dag_id",), ["dag.dag_id"], name=op.f("dag_version_dag_id_fkey")),
+        sa.PrimaryKeyConstraint("id", name=op.f("dag_version_pkey")),


Is the ID the only primary key w/o considering the dag_id contained in primary key? So the ID is a global ID irrespective of the DAG linked?
Which code part/logic is responsible to auto-increment this?

I think to second question I found the answer... which relates to the other (follow-up) comment)

Is the ID the only primary key w/o considering the dag_id contained in primary key? So the ID is a global ID irrespective of the DAG linked?

Yes. I don't see any issues with that, and it's more straightforward. Do you have a case where using only the ID would be an issue?

Not directly ... but in cases if (for whatever reason) data from DBs need to be merged INT values easily can generate clashes. So maybe as in the other comments... I'd feel a bit more comfortable with a UUID... but might be a matter of taste.

jscheffl · 2024-10-16T17:02:20Z

airflow/models/dag_version.py

+        version_number = 1
+
+        if existing_dag_version:
+            version_number = existing_dag_version.version_number + 1


This code smells to be non-thread safe. If multiple instances run concurrently it could generate a clash in app logic for ID generation. If this should represent an auto-increment... this should be rather pushed-down to database engine (e.g. identity column)

I will look into this

This commit introduces versioning for DAGs Changes: - Introduced DagVersion model to handle versioning of DAGs. - Added version_name field to DAG for use in tracking the dagversion by users - Added support for version retrieval in the get_dag_source API endpoint - Modified DAG execution logic to reference dag_version_id instead of the dag_hash to ensure DAG runs are linked to specific versions. Fix tests revert RESTAPI changes

ephraimbuddy force-pushed the versioned-dag2 branch from 8aab698 to aad084c Compare October 10, 2024 18:19

jedcunningham reviewed Oct 14, 2024

View reviewed changes

This was referenced Oct 15, 2024

AIP-65: Track the serialized DAG across DagRun & TaskInstance #42690

Closed

AIP-65: Add versioning to the SerializedDagModel #42547

Closed

ephraimbuddy force-pushed the versioned-dag2 branch 5 times, most recently from ad4f57c to dd272b0 Compare October 16, 2024 12:27

ephraimbuddy added the legacy api Whether legacy API changes should be allowed in PR label Oct 16, 2024

ephraimbuddy force-pushed the versioned-dag2 branch 3 times, most recently from eb13cdd to 1b81f2f Compare October 16, 2024 13:27

ephraimbuddy marked this pull request as ready for review October 16, 2024 13:27

ephraimbuddy requested review from potiuk, ashb, ryanahamilton, bbovenzi, pierrejeambrun, jscheffl and bolkedebruin as code owners October 16, 2024 13:27

ephraimbuddy requested a review from XD-DENG as a code owner October 16, 2024 13:27

ephraimbuddy force-pushed the versioned-dag2 branch from 1b81f2f to ef4120d Compare October 16, 2024 13:30

jscheffl reviewed Oct 16, 2024

View reviewed changes

ephraimbuddy force-pushed the versioned-dag2 branch 2 times, most recently from 21569f4 to 55e1257 Compare October 18, 2024 00:09

ephraimbuddy added 12 commits October 20, 2024 13:56

fixup! AIP-65: Add DAG versioning support

322b958

fixup! fixup! AIP-65: Add DAG versioning support

c491711

fix migration

fd3ba4a

fix test

5ba2965

more test fixes

688ac3f

update query count

809ce20

fix static checks

3d59777

Fix query and add created_at to dag_version table

4a8bf81

improve code

ecc022d

Change to using UUID for primary keys

b4c73e0

DagCode.bulk_write_code is no longer used

9dc8ec6

ephraimbuddy force-pushed the versioned-dag2 branch from fcf2e40 to 9dc8ec6 Compare October 20, 2024 13:12

fixup! Change to using UUID for primary keys

59f1a84

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

AIP-65: Add DAG versioning support #42913

AIP-65: Add DAG versioning support #42913

ephraimbuddy commented Oct 10, 2024 •

edited

Loading

jedcunningham Oct 14, 2024

ephraimbuddy Oct 14, 2024

jscheffl Oct 16, 2024

ephraimbuddy Oct 16, 2024

ephraimbuddy Oct 16, 2024 •

edited

Loading

jscheffl Oct 16, 2024

jscheffl Oct 16, 2024

ephraimbuddy Oct 16, 2024

jscheffl Oct 16, 2024

jscheffl Oct 16, 2024

ephraimbuddy Oct 16, 2024

AIP-65: Add DAG versioning support #42913

Are you sure you want to change the base?

AIP-65: Add DAG versioning support #42913

Conversation

ephraimbuddy commented Oct 10, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ephraimbuddy Oct 16, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ephraimbuddy commented Oct 10, 2024 •

edited

Loading

ephraimbuddy Oct 16, 2024 •

edited

Loading