o-nikolas commented on code in PR #70207:
URL: https://github.com/apache/airflow/pull/70207#discussion_r3647750987
##########
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:
Hey @jedcunningham , thanks for having a look!
> 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'm not sure I follow this. New DagVersions are only created if 1) the dag
bundle being used needs data storage 2) bundle versioning is enabled for that
bundle 3) the metadata has actually changed for that dag file 4) and that dag
has currently running task instances (which need the data persisted).
Specifically 3 and 4 above above means that not _every_ single dag in the
bundle will get a new version. The aperture for when new DagVersions would be
created, beyond which are already created today, is still quite small. Not to
mention S3 bucket deployments are a very intentional action by users, and I
think a possible chance that a subset of dags will mint a single record each
with a deployment to the bucket is not that much of an issue either, honestly.
With documentation and existing garbage collection tooling, I think it is not
such a crisis.
> Maybe an optional bundleversion table to allow (bundle_name,
version)->version_data mapping or something
This was proposed initially and vetoed by the community. Airflow as a
community is incredibly sensitive about adding new schema, never mind new
tables (the amount of hoops we're having to jump through in multi-team land
because of this would make your head spin). We also lose the FK graceful
cleanup that we get by colocating the version_data along with the DagVersion it
belongs to.
> I also don't love shoving version_data down to each dagrun either - that
has problems on a different axis.
I agree, and also considered this, it is probably the second least worst
option IMHO, behind the one I propose in this PR. Which again for the reasons
at the top of this comment (and in the PR description) I think have actually a
fairly small blast radius and is something a user can decide if they would like
to take on when choosing to enable versioning on the S3 bundle.
Let me know what you think with all that in mind!
--
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]
