Dag parse-result persistence: follow-ups deferred from batching
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Description
### What this is
Follow-up work deferred from #71771, which changes the Dag processor to persist a sweep of parse
results a group at a time instead of one file at a time.
Each item below was found while reviewing that PR and deliberately scoped out of it. All of them are
described in the code they affect; none is linked to anything, which is what this issue is for.
**Timing:** the Primary item below is intended to land before the 3.4.0 release. That matters for
scope — while `persist_parsing_results` is unreleased, its payload contract can still change
without a deprecation cycle. Once 3.4.0 ships, widening `FileParseResult.parsing_result` to optional
becomes a breaking change and the same work needs a companion seam instead.
---
## Primary: make the batch persistence seam transport-neutral
`persist_parsing_results` replaces the per-file `persist_parsing_result`, which is now deprecated.
It is not yet a full replacement for `handle_parsing_result`, which is why that method is **not**
deprecated and remains supported.
Two gaps stop it being one:
1. **It is not reached without touching the metadata DB.** `_build_parse_result` resolves the team a
bundle belongs to before the hook, and in multi-team deployments that calls
`DagBundleModel.get_team_names` (once per bundle, not per file — it caches). A file whose lookup
fails is throttled and its processor closed without ever reaching the override, so an API-backed
deployment still depends on the metadata DB being reachable.
2. **It never sees callback-only completions or failed parses.** `_build_parse_result` returns
`None` for both, recording the stat itself. `handle_parsing_result` sees every finished file.
### Acceptance criteria
- [ ] The team a bundle belongs to reaches the hook without the hook's caller querying the metadata
database.
- [ ] Callback-only and no-result completions are delivered through the batch seam, or an explicit
companion seam, so nothing `handle_parsing_result` sees is lost.
- [ ] With both done, `handle_parsing_result` can be deprecated in favour of the batch seam, and the
newsfragment updated to say so.
- [ ] A test drives `_collect_results` into an override with the metadata database unavailable and
asserts every completion still arrives.
---
## Correctness, documented in the PR rather than fixed
- [ ] **A serialization failure can still stale a Dag a later file in the group wrote.** A write
reads what is already registered first — whether a `dag_id` is registered elsewhere, whether an
asset still has a live Dag scheduling it — and stales the Dags filed under a path last. The PR
ends a run at any file reporting *parse* errors, which closes that ordering. Errors from
`_serialize_dag_capturing_errors` are only known once the write is under way, so grouping
cannot anticipate them; the three fileloc guards cover only the subset where the blamed file is
in the same sweep. The duplicate-warning symptom self-corrects on the next parse. **The
watcher symptom does not** — an asset whose only scheduling Dag broke keeps watchers that
should have been removed, and the same pair keeps arriving together.
- [ ] **FAB commits inside a group's transaction.** `_sync_dag_perms` runs per Dag inside
`_serialize_dag_capturing_errors` on the group's session, and `create_permission` commits it.
Since `bulk_write_to_db` stages every Dag's rows first, a commit for one Dag flushes another's,
so a later failure can leave Dags written against a serialized form that never arrived, and the
per-file fallback can report a file as unpersisted while its data is committed. The retry
rewrites them, so it converges — and the same window exists per file on `main`, so batching
widens it rather than opening it. The fix is to make permission syncing non-committing, which is
provider-side work.
- [ ] **Listeners are told before a write commits.** `on_new_dag_import_error` and `on_asset_created`
fire inside the write, so a group that fails after announcing something has announced what was
then rolled back, and the retry announces it again. Pre-existing at-least-once behaviour that
batching adds one more replay path to: file A's event replays because file B in the same group
failed. Deferring hooks to commit, or an outbox, is the fix.
- [x] **Asset rows are locked in nondeterministic order.** `AssetModelOperation.collect` builds
`assets` from `dags.values()`, which under batching is files concatenated in completion order.
Assets are shared across files and bundles by design, so two concurrent writers (two Dag
processors, or one plus `airflow dags reserialize`) can take them in opposite orders. #71936
- [ ] **`persist_parsing_results` writes every group it is handed on one session.** The manager only
ever hands it one, which is what gives each group its own transaction. A caller passing a whole
sweep gets one transaction — documented on the method, not enforced.
- [x] **Unbounded scan in `_update_import_errors`.** `select(ParseImportError.bundle_name,
ParseImportError.filename)` has no `WHERE`, so every call scans the table. Batching amortises
it from per-file to per-group; it should still be filtered to the keys in play. #71621
---
## Test coverage the PR leaves open
- [ ] **Every equivalence shape starts from an empty database.** The differential test resets before
each half, so no shape has a Dag already registered at another fileloc, a pre-existing import
error, or an asset with watchers. That is structurally why it could not have caught the
staleness ordering the PR fixes, and it means `write_dag`'s "hash unchanged, bundle advanced"
branch is never reached.
- [ ] **Grouping's fourth rule has no isolating test.** Deleting the run's-Dag-locs against incoming
file-locs check leaves every test green, because the sixth rule splits the same pairs. Its
unique case is a Dag filed under another path where that path parses clean and defines no Dags.
- [ ] **No real-database test of a multi-file group rolling back** while an earlier group's writes
survive. The tests that claim it never form a group larger than one.
- [ ] **The snapshot does not read** `SerializedDagModel.data`/`dag_hash`, `DagTag`, alias-to-asset
associations, triggers and watchers, or `last_parse_duration` — the last being the one thing the
per-Dag duration mapping exists for.
---
## Open question
- [ ] `MAX_DAGS_PER_PERSISTENCE_GROUP` is a module constant with no config key, while its own
rationale names the deployment most likely to want it tuned. Decide config vs deliberately
fixed. It ships in 3.4.0 either way, so the name is committed once released.
---
## Not part of this issue
Two larger levers from the same investigation are separate work, not deferred scope of #71771:
hashing the serialized Dag in the parser subprocess (~37 ms/file on the Dags measured), and skipping
the manager-side pydantic validation of parse results. Both move per-Dag CPU off the Dag processor's
main process, which batching does not attempt.
Also not here: `on_asset_created` reporting the last definition of an asset rather than the first.
Coalescing repeated definitions is the batching, so preserving the old payload means not batching.
That is permanent by design and is documented as a behaviour change in the #71771 release note
rather than tracked as a fix.
### Use case/motivation
_No response_
### Related issues
_No response_
### Are you willing to submit a PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Assessment
This issue has not been assessed yet.