You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ephraimbuddy (via GitHub)" <gi...@apache.org> on 2026/06/10 12:35:15 UTC
[PR] Refresh bundle version in place when DAG serialization is unchanged [airflow]
ephraimbuddy opened a new pull request, #68336:
URL: https://github.com/apache/airflow/pull/68336
When a DAG bundle advances to a new version/commit but the DAG's serialized content is byte-identical, write_dag took the 'unchanged' fast path and left the latest DagVersion's bundle_version pinned to the old commit. Tasks resolve their code from ti.dag_version.bundle_version at run time, so new runs and reruns 'with the latest bundle version' executed an outdated commit.
Refresh the latest version's bundle_version/version_data (and dag code) in place instead. This does not create a new DagVersion, avoiding version inflation on every bundle commit, while keeping the recorded bundle pointer current.
closes: #67657
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refresh bundle version in place when DAG serialization is unchanged [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #68336:
URL: https://github.com/apache/airflow/pull/68336#discussion_r3412977518
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -687,9 +687,21 @@ def write_dag(
and dag_version
and dag_version.bundle_name == bundle_name
):
- if name_updated:
- # The serialized DAG itself is unchanged, but deadline alert name(s) were
- # updated in the DB, so report True so callers know a write did occur.
+ # Serialized content is unchanged, so we don't create a new DagVersion.
+ # But if the bundle advanced, refresh the latest version's pointer in place — tasks resolve
+ # their code from ``ti.dag_version.bundle_version`` at run time, so a stale
+ # pointer makes runs execute an outdated commit.
+ bundle_metadata_changed = (
+ dag_version.bundle_version != bundle_version or dag_version.version_data != version_data
Review Comment:
I think a hash should be a follow up. And since python `or` short-circuits, it will be rare for bundle_version to not be equal to bundle_version, hence only on rare occasion will the version_data be compared. I prefer to do it as a follow up as it would likely involve a new column to store the hash otherwise, it won't be beneficial doing it on the fly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refresh bundle version in place when DAG serialization is unchanged [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68336:
URL: https://github.com/apache/airflow/pull/68336#discussion_r3411818172
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -687,9 +687,21 @@ def write_dag(
and dag_version
and dag_version.bundle_name == bundle_name
):
- if name_updated:
- # The serialized DAG itself is unchanged, but deadline alert name(s) were
- # updated in the DB, so report True so callers know a write did occur.
+ # Serialized content is unchanged, so we don't create a new DagVersion.
+ # But if the bundle advanced, refresh the latest version's pointer in place — tasks resolve
+ # their code from ``ti.dag_version.bundle_version`` at run time, so a stale
+ # pointer makes runs execute an outdated commit.
+ bundle_metadata_changed = (
+ dag_version.bundle_version != bundle_version or dag_version.version_data != version_data
Review Comment:
I wonder if this may be a bit memory-expensive if `version_date` is large. Should we have a hash to check instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refresh bundle version in place when DAG serialization is unchanged [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #68336:
URL: https://github.com/apache/airflow/pull/68336#discussion_r3413065528
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -687,9 +687,21 @@ def write_dag(
and dag_version
and dag_version.bundle_name == bundle_name
):
- if name_updated:
- # The serialized DAG itself is unchanged, but deadline alert name(s) were
- # updated in the DB, so report True so callers know a write did occur.
+ # Serialized content is unchanged, so we don't create a new DagVersion.
+ # But if the bundle advanced, refresh the latest version's pointer in place — tasks resolve
+ # their code from ``ti.dag_version.bundle_version`` at run time, so a stale
+ # pointer makes runs execute an outdated commit.
+ bundle_metadata_changed = (
+ dag_version.bundle_version != bundle_version or dag_version.version_data != version_data
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refresh bundle version in place when DAG serialization is unchanged [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68336:
URL: https://github.com/apache/airflow/pull/68336#discussion_r3413011270
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -687,9 +687,21 @@ def write_dag(
and dag_version
and dag_version.bundle_name == bundle_name
):
- if name_updated:
- # The serialized DAG itself is unchanged, but deadline alert name(s) were
- # updated in the DB, so report True so callers know a write did occur.
+ # Serialized content is unchanged, so we don't create a new DagVersion.
+ # But if the bundle advanced, refresh the latest version's pointer in place — tasks resolve
+ # their code from ``ti.dag_version.bundle_version`` at run time, so a stale
+ # pointer makes runs execute an outdated commit.
+ bundle_metadata_changed = (
+ dag_version.bundle_version != bundle_version or dag_version.version_data != version_data
Review Comment:
Let’s log an issue for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refresh bundle version in place when DAG serialization is unchanged [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy merged PR #68336:
URL: https://github.com/apache/airflow/pull/68336
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refresh bundle version in place when DAG serialization is unchanged [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #68336:
URL: https://github.com/apache/airflow/pull/68336#discussion_r3388336736
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -585,6 +585,63 @@ def test_new_dag_version_created_when_bundle_name_changes_and_hash_unchanged(sel
# There should now be two versions of the DAG
assert session.scalar(select(func.count()).select_from(DagVersion)) == 2
+ def test_bundle_version_refreshed_in_place_when_hash_unchanged(self, dag_maker, session):
+ """When the bundle advances to a new version/commit but the DAG's serialized
+ content is unchanged, ``write_dag`` must refresh the latest DagVersion's
+ ``bundle_version`` in place (so tasks resolve the current commit) without
+ creating a new DagVersion (which would inflate versions on every commit).
+ """
+ with dag_maker("test_dag_bundle_version_refresh", bundle_name="bundleA") as dag:
+ EmptyOperator(task_id="task1")
+ # Pin the version with task instances so the in-place "no TI" branch is NOT taken.
+ dag_maker.create_dagrun(run_id="test_run")
+
+ did_write = SDM.write_dag(
+ LazyDeserializedDAG.from_dag(dag),
+ bundle_name="bundleA",
+ bundle_version="commit_A",
+ session=session,
+ )
+ assert did_write is True
+ assert session.scalar(select(func.count()).select_from(DagVersion)) == 1
+ assert DagVersion.get_latest_version(dag.dag_id, session=session).bundle_version == "commit_A"
+
+ # Same content, same bundle_name, but the bundle moved to a new commit.
+ did_write = SDM.write_dag(
+ LazyDeserializedDAG.from_dag(dag),
+ bundle_name="bundleA",
+ bundle_version="commit_B",
+ session=session,
+ )
+
+ # No new version was created, but the latest version's bundle pointer advanced.
+ assert did_write is True
+ assert session.scalar(select(func.count()).select_from(DagVersion)) == 1
+ assert DagVersion.get_latest_version(dag.dag_id, session=session).bundle_version == "commit_B"
Review Comment:
The in-place refresh logic updates both `bundle_version` and `version_data`, but this test only asserts the `bundle_version` change. Adding `version_data` assertions would better protect the new behavior from regressions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org