kaxil commented on code in PR #73433:
URL: https://github.com/apache/airflow/pull/73433#discussion_r4351788860
##########
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:
I suggested the fallback, but you're right that it adds nothing. A versioned
stat is only ever empty or a timeout stamp, so when the fallback fires (a file
with no scan entry yet) it either does nothing or lets another version's
timed-out callback hold back the file's first parse, which is the case this PR
fixes. I'd drop it and flip
`test_prepare_file_queue_skips_file_when_versioned_stat_is_at_run_limit` and
`test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats` to
assert the file is queued.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1560,14 +1578,7 @@ def _sort_by_mtime(self, files: Iterable[DagFileInfo]):
return file_infos, changed_recently
def processed_recently(self, now, file):
Review Comment:
Separate PR, please. That keeps this one minimal for the backport.
--
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]
