vuonghoangbntt commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4378573994
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1573,8 +1573,21 @@ 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 _resolve_stats_by_presence_key(self) -> dict[tuple[str, Path],
DagFileStat]:
+ """
+ Resolve one stat per physical file from the bundle scan's entries.
+
+ A callback pins ``bundle_version``, so its entry is separate from the
scan's and does
+ not track parses of the current file: a callback-only result has no
``last_finish_time``,
+ and a timed-out callback processor is stamped although nothing parsed
the current version.
+ A file with no scan entry has no stat, so it is queued.
+ """
+ return {
+ file.presence_key: stat for file, stat in self._file_stats.items()
if file.bundle_version is None
Review Comment:
Good catch, thanks. It needs `_resolve_stats_by_presence_key`, so I'll fix
it in a follow-up once this one is merged: #75666.
--
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]
