Serialize dag if it is not serialized even if it does not have task instances - #68947
Serialize dag if it is not serialized even if it does not have task instances#68947kgskgs wants to merge 3 commits into
Conversation
|
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
|
SameerMesiah97
left a comment
SameerMesiah97
left a comment
There was a problem hiding this comment.
This needs more explanation. Please see my comment on the test.
| session.execute(delete(SDM).where(SDM.dag_id == dag.dag_id)) | ||
| session.flush() | ||
| assert session.scalar(select(func.count()).select_from(SDM)) == 0 | ||
| assert session.scalar(select(func.count()).select_from(DagVersion)) == 1 |
There was a problem hiding this comment.
I am just not sure how this state you are simulating can arise under normal operation. The test simulates the issue by manually deleting the SerializedDagModel row while leaving the DagVersion row intact. Since DAGs are serialized during parsing, I would expect DagVersion and SerializedDagModel to be created together under normal operation.
Could you expand the issue/PR description to explain how this state can occur in practice? Is this intended to recover from metadata corruption, failed migrations, manual database modification, or some known failure during serialization?
Furthermore, have you ran into this specific failure mode yourself?
There was a problem hiding this comment.
Yes I've ran into this myself, but I am not sure what caused it, I am also running a custom application built on top of airflow so most likely it was not caused by airflow itself (I suspect it happened when upgrading from airflow 2 to 3 but there is lots of custom code, sometimes too tightly coupled to airflow and it would be a lot of effort to determine exactly how it happened)
I'd say that there being nothing preventing you from getting into this state + the user being able to manually break their dags by editing the db is reason enough to have this
There was a problem hiding this comment.
Yes I've ran into this myself, but I am not sure what caused it, I am also running a custom application built on top of airflow so most likely it was not caused by airflow itself (I suspect it happened when upgrading from airflow 2 to 3 but there is lots of custom code, sometimes too tightly coupled to airflow and it would be a lot of effort to determine exactly how it happened) I'd say that there being nothing preventing you from getting into this state + the user being able to manually break their dags by editing the db is reason enough to have this
I would add all the detail you can in both the issue and the PR description (please include the what, why and how as well as any implications for backwards compatibility).
eladkal
added
type:bug-fix
|
I had missed that there was already a test covering this, but that one asserted that write_dag returns |
potiuk
added
the
ready for maintainer review
f2a8147 to
76fe81d
Compare
| ) | ||
|
|
||
| if dag_version and not has_task_instances: | ||
| if dag_version and not has_task_instances and serialized_dag_hash is not None: |
There was a problem hiding this comment.
serialized_dag_hash is not None isn't the same predicate as "the latest dag_version has a serialized row". dag_hash is nullable=False and sd_subq in _prefetch_dag_write_metadata inner-joins serialized_dag to dag_version, so the hash comes from the highest version that still has a row, while dag_version here is the absolute latest. If only the latest version's row is missing, the hash is non-None, this guard passes, the UPDATE matches 0 rows and we return False at line 747 again.
Checked on this branch in breeze: v1 with a serialized row plus a bare latest v2 keeps returning False, with no row ever created for v2. If the incoming hash matches v1's, it returns even earlier, at line 710 on the unchanged-hash short-circuit. Since the only production creator of a dag_version row is line 769, in the same transaction as the insert at line 784, both shapes come out of the same accident class.
Pairing the prefetched hash with the version this branch actually updates covers both (outerjoin serialized_dag off the latest-version subquery so dag_hash is None whenever the latest version has no row). I tried that in a probe and both shapes then heal on the next parse. Letting rowcount == 0 fall through to DagVersion.write_dag is a good complement since it's the direct signal, but on its own it misses the unchanged-hash case, and it leaves min_update_interval reading the older row's last_updated. If the clause does stay here, a line of comment saying what it detects would help, since as written it reads as null-safety on a non-nullable column.
|
|
||
| assert result is False # Should return False when SerializedDagModel is missing | ||
| serialized_dag = session.scalar(select(SDM).where(SDM.dag_id == "test_missing_serdag").limit(1)) | ||
| assert serialized_dag is not None |
There was a problem hiding this comment.
This asserts that some row exists for the dag_id, but the property that makes the dag schedulable again is that the row belongs to the latest DagVersion. SDM.get("test_missing_serdag", session=session) already orders by version_number, so using it plus an assert on version_number would pin what the test name claims. As written the test still passes in the case where an older version holds the row and the latest one doesn't.
Worth keeping a case for the rowcount == 0 branch too. After this change nothing in the file exercises it, and if a later cleanup drops it as dead code that path would run the dag_version and dag_code updates and return True having written no serialized row.
| self, dag_maker, session | ||
| ): | ||
| """ | ||
| Test that dynamic DAG update gracefully handles case where SerializedDagModel doesn't exist. |
There was a problem hiding this comment.
Docstring still describes the old behavior. "gracefully handles" was the euphemism for returning False, and "dynamic DAG update" isn't the scenario (a missing serialized row is). Something like "A new DagVersion with a serialized row is created when the latest version has no serialized_dag row" would match the new assertions. Same for the comment on line 911, which reads "Try to update" then "should create".
76fe81d to
eb448d0
Compare
|
new unit tests generated with claude sonnet 4.6 |
6 participants
Prevent dags from not ever being serialized if they have no entry in
serialized_dagtable, but have one indag_versioncloses: #68944
Context:
I ran into this state while running a custom application on top of airflow. It is highly likely that it was caused by custom code and not airflow, but airflow doesn't seem to have any safeguards against entering the state, and it affected all instances of my application. I can't easily tell exactly how it happened exactly (we are in the middle of updating from airflow 2 to 3 and there are lots of changes)
I believe this fix is appropriate given that there should be multiple ways to reach this state e.g. by editing the db manually.
An alternative would be to try and add safeguards against getting into this state in the first place, but that would be way more complex than handling in this way + currently this is done at no additional cost as we already know whether the serialized dag exists or not by the old hash.
There should be no implications on backwards compatibility.
Was generative AI tooling used to co-author this PR?
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.