ephraimbuddy opened a new issue, #71911:
URL: https://github.com/apache/airflow/issues/71911
### 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.
- [ ] **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. Sorting by
`(name, uri)` closes it.
- [ ] **`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.
- [ ] **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.
---
## 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)
--
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]
