jedcunningham commented on code in PR #71621:
URL: https://github.com/apache/airflow/pull/71621#discussion_r3847395147
##########
airflow-core/tests/unit/dag_processing/test_collection.py:
##########
@@ -1437,6 +1438,105 @@ def
test_max_consecutive_failed_dag_runs_defaults_from_conf_when_none(
assert orm_dag.max_consecutive_failed_dag_runs == 6
[email protected]_test
+class TestUpdateImportErrors:
+ """Tests for the ``_update_import_errors`` helper."""
+
+ @pytest.fixture(autouse=True)
+ def clean_import_errors(self):
+ clear_db_import_errors()
+ yield
+ clear_db_import_errors()
+
+ @pytest.fixture
+ def import_error_statements(self, session):
+ """
+ Collect every SQL statement issued against the ``import_error`` table.
+
+ Matching on the bare table name would also catch statements naming
``dag.has_import_errors``,
+ so match the positions where the table itself can appear.
+ """
+ statements: list[str] = []
+
+ def _capture(conn, cursor, statement, parameters, context,
executemany):
+ lowered = statement.lower()
+ if any(f"{keyword} import_error" in lowered for keyword in
("from", "into", "update")):
+ statements.append(statement)
+
+ bind = session.get_bind()
+ event.listen(bind, "before_cursor_execute", _capture)
+ yield statements
+ event.remove(bind, "before_cursor_execute", _capture)
+
+ @staticmethod
+ def _selects(statements: list[str]) -> list[str]:
+ return [stmt for stmt in statements if
stmt.lower().lstrip().startswith("select")]
+
+ def test_no_lookup_when_there_are_no_import_errors(self, session,
import_error_statements):
+ session.add(ParseImportError(filename="broken.py",
bundle_name="testing", stacktrace="boom"))
+ session.flush()
+ import_error_statements.clear()
+
+ # files_parsed is empty so no DELETE runs either: on backends without
DELETE...RETURNING
+ # its synchronize_session fallback would emit a SELECT of its own and
muddy the assertion.
+ _update_import_errors(
+ files_parsed=set(),
+ import_errors={},
+ session=session,
+ )
+
+ assert self._selects(import_error_statements) == []
+
+ @patch.object(ParseImportError, "full_file_path", return_value="broken.py")
+ def test_existing_error_lookup_is_bounded(self, _mock_full_path, session,
import_error_statements):
+ session.add_all(
+ [
+ ParseImportError(filename="broken.py", bundle_name="testing",
stacktrace="old"),
+ ParseImportError(filename="untouched.py", bundle_name="other",
stacktrace="unrelated"),
+ ]
+ )
+ session.flush()
+ import_error_statements.clear()
+
+ _update_import_errors(
+ files_parsed={("testing", "broken.py")},
+ import_errors={("testing", "broken.py"): "new"},
+ session=session,
+ )
+
+ selects = self._selects(import_error_statements)
+ assert selects, "expected the existing-error lookup to run"
+ assert all("where" in stmt.lower() for stmt in selects), (
+ f"import_error must never be scanned unfiltered, got: {selects}"
+ )
+
+ # A list, not a dict: import_error has no unique constraint on
(bundle_name, filename), so
+ # keying by it would silently collapse a duplicate row the lookup
failed to match.
Review Comment:
This almost feels like a gap. Either way, probably dont need this comment.
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -394,16 +394,26 @@ def _update_dag_warnings(
def _update_import_errors(
files_parsed: set[tuple[str, str]],
- bundle_name: str,
import_errors: dict[tuple[str, str], str],
session: Session,
):
from airflow.listeners.listener import get_listener_manager
- # 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))
- )
+ # Check existing import errors BEFORE deleting, so we can determine if we
should update or create.
Review Comment:
We can probably simplify this comment a bit.
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -346,15 +346,23 @@ class SerializedDagModel(Base):
load_op_links = True
__table_args__ = (Index("idx_serialized_dag_dag_id_created_at", dag_id,
created_at),)
- def __init__(self, dag: LazyDeserializedDAG) -> None:
+ def __init__(self, dag: LazyDeserializedDAG, *, _dag_hash: str | None =
None) -> None:
+ """
+ Build a serialized Dag row.
+
+ :param _dag_hash: hash of ``dag.data`` when the caller has already
computed it, used by
+ ``write_dag`` to avoid hashing the same data twice. Hashing
re-sorts and re-encodes
+ the whole structure, so it is worth skipping. It must describe
``dag.data`` as it
+ stands now — recompute it first if the data was mutated after
hashing, or change
+ detection on the next parse silently misbehaves.
+ """
self.dag_id = dag.dag_id
dag_data = dag.data
- self.dag_hash = SerializedDagModel.hash(dag_data)
-
- # partially ordered json data
- dag_data_json = json.dumps(dag_data, sort_keys=True).encode("utf-8")
+ self.dag_hash = SerializedDagModel.hash(dag_data) if _dag_hash is None
else _dag_hash
Review Comment:
```suggestion
self.dag_hash = _dag_hash or SerializedDagModel.hash(dag_data)
```
Likely safe?
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -346,15 +346,23 @@ class SerializedDagModel(Base):
load_op_links = True
__table_args__ = (Index("idx_serialized_dag_dag_id_created_at", dag_id,
created_at),)
- def __init__(self, dag: LazyDeserializedDAG) -> None:
+ def __init__(self, dag: LazyDeserializedDAG, *, _dag_hash: str | None =
None) -> None:
+ """
+ Build a serialized Dag row.
+
+ :param _dag_hash: hash of ``dag.data`` when the caller has already
computed it, used by
Review Comment:
Here too, simpler is better.
--
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]
