jedcunningham commented on code in PR #70207:
URL: https://github.com/apache/airflow/pull/70207#discussion_r3642880229
##########
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
Review Comment:
I worry that creating new dagversions for _every_ single dag in the bundle
when any file changes is going to get confusing/wasteful fast, especially at
scale.
I think we should explore other designs to keep the "dagversion changes when
we observe this specific dag changed" property. Maybe an optional bundleversion
table to allow (bundle_name, version)->version_data mapping or something? Given
the size, I also don't love shoving version_data down to each dagrun either -
that has problems on a different axis.
--
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]
