ephraimbuddy commented on code in PR #70207:
URL: https://github.com/apache/airflow/pull/70207#discussion_r3685773373
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -688,38 +703,44 @@ def write_dag(
new_dag_hash = cls.hash(dag.data)
+ preserve_pinned_version_data = False
if serialized_dag_hash == new_dag_hash and dag_version and
dag_version.bundle_name == bundle_name:
- # 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
+ # Serialized content is unchanged, so we normally don't create a
new DagVersion.
+ # 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
)
- if bundle_metadata_changed:
- dag_version.bundle_version = bundle_version
- dag_version.version_data = version_data
- session.merge(dag_version)
- DagCode.update_source_code(dag_id=dag.dag_id,
fileloc=dag.fileloc, session=session)
- if name_updated or bundle_metadata_changed:
- # A write occurred — a deadline alert name update and/or a
bundle
- # metadata refresh — so report True so callers know the DB
changed.
- return True
- log.debug("Serialized DAG (%s) is unchanged. Skipping writing to
DB", dag.dag_id)
- return False
+ # The in-place refresh is safe only for bundles whose
``bundle_version`` alone can recover the
+ # code (e.g. Git: the commit SHA resolves against the immutable
object store). When
+ # ``version_data`` is involved (e.g. For bundles like the S3
bundle a json manifest of S3 object
+ # versions), this row holds the only copy of the data which is
needed for runs pinned to the old
+ # bundle_version to reconstruct their code. Overwriting it would
break their reproducibility. In
+ # that case fall through and create a new DagVersion instead: the
old row keeps serving pinned
+ # runs, the new row becomes the latest for new runs.
+ preserve_pinned_version_data = (
+ bundle_metadata_changed
+ and (dag_version.version_data is not None or version_data is
not None)
+ and cls._has_task_instances(dag.dag_id, dag_version.id,
session=session)
+ )
+ if not preserve_pinned_version_data:
Review Comment:
This is where I disagree with point 3 from the list. This branch only runs
when the serialized hash did not change. So the Dag itself did not change at
all. What changed is version_data, and version_data covers the whole bucket. So
one changed file anywhere in the bucket makes this true for every Dag in the
bundle. All of them get a new DagVersion even though their own files were not
touched. To keep the aperture small, the code would need to ask "did this Dag's
own files change". The manifest is opaque here, so it cannot ask that.
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -594,6 +594,21 @@ def _prefetch_dag_write_metadata(
for dag_id in dag_id_list
}
+ @classmethod
+ def _has_task_instances(cls, dag_id: str, dag_version_id: UUID, *,
session: Session) -> bool:
+ """Check whether any task instance references the given DagVersion."""
+ return bool(
+ session.scalar(
+ select(
+ exists().where(
Review Comment:
This is point 4 from the list. There is no filter on task state here.
Finished tasks count the same as running ones. So after the first run of the
current version, this check is true. It stays true until db clean removes the
old task instances. In practice it means "the Dag has ever run", not "the Dag
has running tasks". Together with the other comment, every bucket deploy
creates a new DagVersion, SerializedDagModel and DagCode row for every Dag that
has ever run.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
