Add processor_subdir to import_error table to handle multiple dag processors

airflow/dag_processing/manager.py

@@ -743,7 +743,9 @@ def _refresh_dag_dir(self) -> bool:
 
             try:
                 self.log.debug("Removing old import errors")
-                DagFileProcessorManager.clear_nonexistent_import_errors(file_paths=self._file_paths)
+                DagFileProcessorManager.clear_nonexistent_import_errors(
+                    file_paths=self._file_paths, processor_subdir=self.get_dag_directory()
+                )
             except Exception:
                 self.log.exception("Error removing old import errors")
 
@@ -793,7 +795,9 @@ def _print_stat(self):
     @staticmethod
     @internal_api_call
     @provide_session
-    def clear_nonexistent_import_errors(file_paths: list[str] | None, session=NEW_SESSION):
+    def clear_nonexistent_import_errors(
+        file_paths: list[str] | None, processor_subdir: str | None, session=NEW_SESSION
+    ):
         """
         Clear import errors for files that no longer exist.
 
@@ -803,7 +807,10 @@ def clear_nonexistent_import_errors(file_paths: list[str] | None, session=NEW_SE
         query = delete(errors.ImportError)
 
         if file_paths:
-            query = query.where(~errors.ImportError.filename.in_(file_paths))
+            query = query.where(
+                ~errors.ImportError.filename.in_(file_paths),
+                errors.ImportError.processor_subdir == processor_subdir,
+            )
 
         session.execute(query.execution_options(synchronize_session="fetch"))
         session.commit()

airflow/dag_processing/processor.py

@@ -592,7 +592,10 @@ def manage_slas(cls, dag_folder, dag_id: str, session: Session = NEW_SESSION) ->
     @internal_api_call
     @provide_session
     def update_import_errors(
-        file_last_changed: dict[str, datetime], import_errors: dict[str, str], session: Session = NEW_SESSION
+        file_last_changed: dict[str, datetime],
+        import_errors: dict[str, str],
+        processor_subdir: str | None,
+        session: Session = NEW_SESSION,
     ) -> None:
         """
         Update any import errors to be displayed in the UI.
@@ -627,7 +630,12 @@ def update_import_errors(
                 )
             else:
                 session.add(
-                    errors.ImportError(filename=filename, timestamp=timezone.utcnow(), stacktrace=stacktrace)
+                    errors.ImportError(
+                        filename=filename,
+                        timestamp=timezone.utcnow(),
+                        stacktrace=stacktrace,
+                        processor_subdir=processor_subdir,
+                    )
                 )
             (
                 session.query(DagModel)
@@ -835,6 +843,7 @@ def process_file(
             DagFileProcessor.update_import_errors(
                 file_last_changed=dagbag.file_last_changed,
                 import_errors=dagbag.import_errors,
+                processor_subdir=self._dag_directory,
                 session=session,
             )
             if callback_requests:
@@ -860,6 +869,7 @@ def process_file(
             DagFileProcessor.update_import_errors(
                 file_last_changed=dagbag.file_last_changed,
                 import_errors=dagbag.import_errors,
+                processor_subdir=self._dag_directory,
                 session=session,
             )
         except Exception:

airflow/migrations/versions/0133_2_8_0_add_processor_subdir_import_error.py

@@ -0,0 +1,54 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Add processor_subdir to ImportError.
+
+Revision ID: 10b52ebd31f7
+Revises: 624ecf3b6a5e
+Create Date: 2023-11-29 16:54:48.101834
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = "10b52ebd31f7"
+down_revision = "624ecf3b6a5e"
+branch_labels = None
+depends_on = None
+airflow_version = "2.8.0"
+
+
+def upgrade():
+    """Apply Add processor_subdir to ImportError."""
+    conn = op.get_bind()
+
+    with op.batch_alter_table("import_error") as batch_op:
+        if conn.dialect.name == "mysql":
+            batch_op.add_column(sa.Column("processor_subdir", sa.Text(length=2000), nullable=True))
+        else:
+            batch_op.add_column(sa.Column("processor_subdir", sa.String(length=2000), nullable=True))
+
+
+def downgrade():
+    """Unapply Add processor_subdir to ImportError."""
+    conn = op.get_bind()
+    with op.batch_alter_table("import_error", schema=None) as batch_op:
+        batch_op.drop_column("processor_subdir")

airflow/models/errors.py

@@ -31,3 +31,4 @@ class ImportError(Base):
     timestamp = Column(UtcDateTime)
     filename = Column(String(1024))
     stacktrace = Column(Text)
+    processor_subdir = Column(String(2000), nullable=True)

airflow/utils/db.py

@@ -88,7 +88,7 @@
     "2.6.0": "98ae134e6fff",
     "2.6.2": "c804e5c76e3e",
     "2.7.0": "405de8318b3a",
-    "2.8.0": "624ecf3b6a5e",
+    "2.8.0": "10b52ebd31f7",
 }
 
 

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-4739d87664d779f93e39b09ca6e5e662d72f1fa88857d8b6e44d2f2557656753
+5d80302f775a966cffc5ed6c452c56a9e181afce92dc4ccf9b29b21171091f38