You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ep...@apache.org on 2025/12/16 09:53:10 UTC
(airflow) branch backport-796b46e-v3-1-test created (now 6cada4fc9fe)
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a change to branch backport-796b46e-v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
at 6cada4fc9fe [v3-1-test] Fix import errors not cleared for files without DAGs (#58242)
This branch includes the following new commits:
new 6cada4fc9fe [v3-1-test] Fix import errors not cleared for files without DAGs (#58242)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(airflow) 01/01: [v3-1-test] Fix import errors not cleared for files without DAGs (#58242)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch backport-796b46e-v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6cada4fc9fe39d480af56d2f8fd383eb557f02ab
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Tue Nov 25 13:13:46 2025 +0100
[v3-1-test] Fix import errors not cleared for files without DAGs (#58242)
* Fix import errors not cleared for files without DAGs
Previously, import errors persisted in the database for files that were
successfully parsed but no longer contained any DAGs. This happened
because we only tracked files that had DAGs, missing files that
were parsed successfully but had their DAGs removed.
Now, when files are parsed, all parsed files are tracked (not just those
with DAGs), ensuring import errors are properly cleared when a file is
successfully parsed without errors, even if it no longer contains DAGs.
closes: #57621
* Apply suggestions from code review
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
* fixup! Fix import errors not cleared for files without DAGs
* Remove fallback for files_parsed
* fixup! fixup! Fix import errors not cleared for files without DAGs
* fixup! Remove fallback for files_parsed
* fixup! fixup! fixup! Fix import errors not cleared for files without DAGs
---------
(cherry picked from commit 796b46ee8aba6ebaef767ec5592b826a636174e3)
Co-authored-by: Ephraim Anierobi <sp...@gmail.com>
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
---
.../src/airflow/dag_processing/collection.py | 40 ++++++++--------
airflow-core/src/airflow/dag_processing/manager.py | 11 +++++
airflow-core/src/airflow/models/dagbag.py | 11 +++++
.../tests/unit/dag_processing/test_collection.py | 54 +++++++++++++++++++++-
4 files changed, 94 insertions(+), 22 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py
index a4e6af03373..324232c2986 100644
--- a/airflow-core/src/airflow/dag_processing/collection.py
+++ b/airflow-core/src/airflow/dag_processing/collection.py
@@ -285,21 +285,22 @@ def _update_import_errors(
):
from airflow.listeners.listener import get_listener_manager
- # We can remove anything from files parsed in this batch that doesn't have an error. We need to remove old
- # errors (i.e. from files that are removed) separately
-
- session.execute(
- delete(ParseImportError).where(
- tuple_(ParseImportError.bundle_name, ParseImportError.filename).in_(files_parsed)
- )
- )
-
- # the below query has to match (bundle_name, filename) tuple in that order since the
- # import_errors list is a dict with keys as (bundle_name, relative_fileloc)
+ # Check existing import errors BEFORE deleting, so we can determine if we should update or create
existing_import_error_files = set(
session.execute(select(ParseImportError.bundle_name, ParseImportError.filename))
)
- # Add the errors of the processed files
+
+ # Delete errors for files that were parsed but don't have errors in import_errors
+ # (i.e., files that were successfully parsed without errors)
+ files_to_clear = files_parsed.difference(import_errors)
+ if files_to_clear:
+ session.execute(
+ delete(ParseImportError).where(
+ tuple_(ParseImportError.bundle_name, ParseImportError.filename).in_(files_to_clear)
+ )
+ )
+
+ # Add or update the errors of the processed files
for key, stacktrace in import_errors.items():
bundle_name_, relative_fileloc = key
@@ -370,6 +371,7 @@ def update_dag_parsing_results_in_db(
session: Session,
*,
warning_types: tuple[DagWarningType] = (DagWarningType.NONEXISTENT_POOL,),
+ files_parsed: set[tuple[str, str]] | None = None,
):
"""
Update everything to do with DAG parsing in the DB.
@@ -387,6 +389,10 @@ def update_dag_parsing_results_in_db(
then all warnings and errors related to this file will be removed.
``import_errors`` will be updated in place with an new errors
+
+ :param files_parsed: Set of (bundle_name, relative_fileloc) tuples for all files that were parsed.
+ If None, will be inferred from dags and import_errors. Passing this explicitly ensures that
+ import errors are cleared for files that were parsed but no longer contain DAGs.
"""
# Retry 'DAG.bulk_write_to_db' & 'SerializedDagModel.bulk_sync_to_db' in case
# of any Operational Errors
@@ -422,16 +428,8 @@ def update_dag_parsing_results_in_db(
import_errors.update(serialize_errors)
# Record import errors into the ORM - we don't retry on this one as it's not as critical that it works
try:
- # TODO: This won't clear errors for files that exist that no longer contain DAGs. Do we need to pass
- # in the list of file parsed?
-
- good_dag_filelocs = {
- (bundle_name, dag.relative_fileloc)
- for dag in dags
- if dag.relative_fileloc is not None and (bundle_name, dag.relative_fileloc) not in import_errors
- }
_update_import_errors(
- files_parsed=good_dag_filelocs,
+ files_parsed=files_parsed if files_parsed is not None else set(),
bundle_name=bundle_name,
import_errors=import_errors,
session=session,
diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py
index 2a37a1bbf6f..c362112646d 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -848,6 +848,7 @@ class DagFileProcessorManager(LoggingMixin):
parsing_result=proc.parsing_result,
session=session,
is_callback_only=is_callback_only,
+ relative_fileloc=str(file.rel_path),
)
for file in finished:
@@ -1136,6 +1137,7 @@ def process_parse_results(
session: Session,
*,
is_callback_only: bool = False,
+ relative_fileloc: str | None = None,
) -> DagFileStat:
"""Take the parsing result and stats about the parser process and convert it into a DagFileStat."""
if is_callback_only:
@@ -1170,6 +1172,14 @@ def process_parse_results(
import_errors = {
(bundle_name, rel_path): error for rel_path, error in parsing_result.import_errors.items()
}
+
+ # Build the set of files that were parsed. This includes the file that was parsed,
+ # even if it no longer contains DAGs, so we can clear old import errors.
+ files_parsed: set[tuple[str, str]] | None = None
+ if relative_fileloc is not None:
+ files_parsed = {(bundle_name, relative_fileloc)}
+ files_parsed.update(import_errors.keys())
+
update_dag_parsing_results_in_db(
bundle_name=bundle_name,
bundle_version=bundle_version,
@@ -1178,6 +1188,7 @@ def process_parse_results(
parse_duration=run_duration,
warnings=set(parsing_result.warnings or []),
session=session,
+ files_parsed=files_parsed,
)
stat.num_dags = len(parsing_result.serialized_dags)
if parsing_result.import_errors:
diff --git a/airflow-core/src/airflow/models/dagbag.py b/airflow-core/src/airflow/models/dagbag.py
index fc29cf99155..5ee277d949d 100644
--- a/airflow-core/src/airflow/models/dagbag.py
+++ b/airflow-core/src/airflow/models/dagbag.py
@@ -662,6 +662,16 @@ def sync_bag_to_db(
from airflow.dag_processing.collection import update_dag_parsing_results_in_db
import_errors = {(bundle_name, rel_path): error for rel_path, error in dagbag.import_errors.items()}
+
+ # Build the set of all files that were parsed and include files with import errors
+ # in case they are not in file_last_changed
+ files_parsed = set(import_errors)
+ if dagbag.bundle_path:
+ files_parsed.update(
+ (bundle_name, dagbag._get_relative_fileloc(abs_filepath))
+ for abs_filepath in dagbag.file_last_changed
+ )
+
update_dag_parsing_results_in_db(
bundle_name,
bundle_version,
@@ -670,6 +680,7 @@ def sync_bag_to_db(
None, # file parsing duration is not well defined when parsing multiple files / multiple DAGs.
dagbag.dag_warnings,
session=session,
+ files_parsed=files_parsed,
)
diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py b/airflow-core/tests/unit/dag_processing/test_collection.py
index d4d10ffd510..d38e613d2bd 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -492,7 +492,9 @@ class TestUpdateDagParsingResults:
mock_full_path.return_value = "abc.py"
import_errors = {}
- update_dag_parsing_results_in_db("testing", None, [dag], import_errors, None, set(), session)
+ update_dag_parsing_results_in_db(
+ "testing", None, [dag], import_errors, None, set(), session, files_parsed={("testing", "abc.py")}
+ )
assert "SerializationError" in caplog.text
# Should have been edited in place
@@ -656,6 +658,7 @@ class TestUpdateDagParsingResults:
parse_duration=None,
warnings=set(),
session=session,
+ files_parsed={("testing", "abc.py")},
)
import_error = (
@@ -714,6 +717,7 @@ class TestUpdateDagParsingResults:
parse_duration=None,
warnings=set(),
session=session,
+ files_parsed={(bundle_name, "abc.py")},
)
dag_model: DagModel = session.get(DagModel, (dag.dag_id,))
assert dag_model.has_import_errors is False
@@ -758,6 +762,7 @@ class TestUpdateDagParsingResults:
parse_duration=None,
warnings=set(),
session=session,
+ files_parsed={(bundle_name, "abc.py")},
)
dag_model = session.get(DagModel, (dag.dag_id,))
assert dag_model.has_import_errors is True
@@ -774,6 +779,53 @@ class TestUpdateDagParsingResults:
)
assert dag_model.has_import_errors is False
+ @pytest.mark.usefixtures("clean_db")
+ def test_clear_import_error_for_file_without_dags(self, testing_dag_bundle, session):
+ """
+ Test that import errors are cleared for files that were parsed but no longer contain DAGs.
+ """
+ bundle_name = "testing"
+ filename = "no_dags.py"
+
+ prev_error = ParseImportError(
+ filename=filename,
+ bundle_name=bundle_name,
+ timestamp=tz.utcnow(),
+ stacktrace="Previous import error",
+ )
+ session.add(prev_error)
+
+ # And import error for another file we haven't parsed (this shouldn't be deleted)
+ other_file_error = ParseImportError(
+ filename="other.py",
+ bundle_name=bundle_name,
+ timestamp=tz.utcnow(),
+ stacktrace="Some error",
+ )
+ session.add(other_file_error)
+ session.flush()
+
+ import_errors = set(session.execute(select(ParseImportError.filename, ParseImportError.bundle_name)))
+ assert import_errors == {("no_dags.py", bundle_name), ("other.py", bundle_name)}
+
+ # Simulate parsing the file: it was parsed successfully (no import errors),
+ # but it no longer contains any DAGs. By passing files_parsed, we ensure
+ # the import error is cleared even though there are no DAGs.
+ files_parsed = {(bundle_name, filename)}
+ update_dag_parsing_results_in_db(
+ bundle_name=bundle_name,
+ bundle_version=None,
+ dags=[], # No DAGs in this file
+ import_errors={}, # No import errors
+ parse_duration=None,
+ warnings=set(),
+ session=session,
+ files_parsed=files_parsed,
+ )
+
+ import_errors = set(session.execute(select(ParseImportError.filename, ParseImportError.bundle_name)))
+ assert import_errors == {("other.py", bundle_name)}, "Import error for parsed file should be cleared"
+
@pytest.mark.need_serialized_dag(False)
@pytest.mark.parametrize(
("attrs", "expected"),