vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4376331032
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1531,8 +1531,26 @@ def _resort_file_queue(self):
# Put callback files at the front, then sorted regular files
self._file_queue = OrderedDict.fromkeys(callback_files +
sorted_regular_files)
+ def _stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+ """
+ Resolve a single stat per physical file, with the most recent parse
winning.
+
+ One file can hold several ``_file_stats`` entries: a callback pins
+ ``bundle_version`` on its ``DagFileInfo`` while the bundle scan leaves
it unset,
+ and both are tracked deliberately. Resolving by insertion order lets a
stale
+ callback entry mask a more recent scan parse, so compare parse times
instead.
+ """
+ resolved: dict[tuple[str, Path], DagFileStat] = {}
+ for file, stat in self._file_stats.items():
+ existing = resolved.get(file.presence_key)
+ if existing is None or (stat.last_finish_time or utc_epoch()) >= (
Review Comment:
Done: the fallback is gone, and both tests now assert the file is queued. I
renamed them to
`test_prepare_file_queue_queues_file_whose_versioned_stat_is_at_run_limit` and
`test_prepare_file_queue_queues_file_whose_versioned_stat_is_recent`.
--
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]
