Persist a sweep of Dag parse results in one pass by ephraimbuddy · Pull Request #71771 · apache/airflow

ephraimbuddy · 2026-08-18T14:03:42Z

Persist a sweep of Dag parse results in one pass, instead of one file at a time.

Credit to @seanmuth, whose investigation of the Airflow 2-to-3 dag-processor CPU
regression identified this. Full public repro:
https://github.com/seanmuth/af2-af3-scheduler-cpu-repro

This is stacked on the statement-budget branch (parse-result-query-budget); the budget harness
it measures against lands there, and this branch is rebased onto its tip.

The problem

DagFileProcessorManager._collect_results() handled each finished parse separately, and
each call opened its own session and its own update_dag_parsing_results_in_db call. Most
of what that call costs is fixed per call rather than per Dag — building duplicate-dag_id
warnings, finding the ORM Dags, prefetching write metadata, reading and clearing import
errors, syncing warnings — so a sweep of single-Dag files paid that fixed price once per
file.

update_dag_parsing_results_in_db was already multi-file: sync_bag_to_db (used by
airflow dags reserialize) passes every Dag of a bundle in one call. The Dag processor was
the odd caller out.

What changed

A sweep is collected, split into groups, and each group persisted in a single call and a
single transaction.

Groups are formed so that:

one bundle per group — the bundle determines the version and version data the write
needs;
each dag_id claimed at most once — writing two files that define the same dag_id
together would merge them into one Dag and lose the duplicate warning, which is raised by
comparing an incoming Dag against the file already recorded in the DB. Only the files
actually in conflict are held back, and a file is only ever held back to a group after
the one holding the file it duplicates, so the same file wins a duplicated dag_id as
when each was written on its own;
groups are capped by Dag count (32 Dags) — grouping saves a fixed cost once per file, but
what one transaction holds locked grows with the Dags in it rather than the files, and a few
files generating Dags dynamically can be thousands. A file is never split, since the record of
it having been parsed and its import errors belong to it as a whole, so a file defining more
Dags than the budget is still written on its own.

Error isolation is preserved: each group is its own transaction, and a group that fails is
retried one file at a time, so one unwritable file cannot discard the results of the others.
Finished processors are now closed whether or not the sweep succeeded.

parse_duration is per file, so it can no longer be one value for a call; it is carried per
Dag. Single-file callers still pass a single value.

Measured results

Real manager, 60 generated files × 200 tasks each, 32 parsing processes, two passes,
alternating between main and this branch on the same machine and database:

run	persistence calls / file	statements / file	ms/file inside the write
baseline 1	1.00	15.00	211.99
branch 1	0.16	4.83	183.73
baseline 2	1.00	15.00	155.29
branch 2	0.17	5.20	145.81

120 persistence calls become 19–21 (sweeps averaged six files), and statements fall 15.00 to
4.83–5.20 per file, a 2.9–3.1x reduction. The baseline reproduces to the hundredth across runs;
the branch varies with how many sweeps a run happens to produce.

The statement counts match the budget model in
airflow-core/tests/unit/dag_processing/test_parse_result_query_budget.py. That model prices a
call at FIXED_PER_CALL = 10 plus a per-Dag cost that depends on whether the serialized Dag is
rewritten — UNCHANGED_PER_DAG = 3 or REWRITE_PER_DAG = 5. The measurement above ran warm with
the default min_serialized_dag_update_interval, so the unchanged price applies: a 21-call run
predicts 21×10 + 120×3 = 570 against 624 measured, and the baseline predicts 120×10 + 120×3 = 1560 against 1800. The remainder in each is _collect_results overhead outside the write.

Two honest caveats

Wall clock is not resolvable on this workload. The branch was faster in both time-adjacent
pairs, by 13% and 6%, but the two baselines differ from each other by 36% — far more than either
gap — so this is directionally consistent with a small gain rather than a measurement of one. The
Dags measured carry 200 tasks each, so per-Dag work (hashing and serializing) dominates per-file
cost and the fixed portion this change removes is a small slice of it. Files defining smaller Dags
pay proportionally more of the fixed cost and gain more of it back.

The win is gated on processor width. At the default parsing_processes = 2, every sweep
observed held exactly one file (120 sweeps, 120 files), so there is nothing to batch and the
change is a no-op. Deployments running a wide Dag processor are the ones this is for.

Backward compatibility

handle_parsing_result and persist_parsing_result shipped in 3.3.0 as the documented seams
for deployments that forward parse results somewhere other than the metadata DB (AIP-92).
Both keep their released contract — same signature, same session ownership, same per-file
call — and both are deprecated in favour of the new persist_parsing_results, which is handed
a whole group.

Overriding a deprecated seam costs that deployment its batching and nothing else. A subclass
that adopts persist_parsing_results while still carrying persist_parsing_result for 3.3
compatibility keeps batching. Detection reads the instance as well as the class, so replacing
a hook on a single manager counts.

A DeprecationWarning naming the replaced method is emitted once at Dag processor startup.

Reproducing

The harness used for the numbers above is not part of this PR. It generates uniform Dag files,
runs the real manager, and reports sweep width, persistence calls, statements, and time — with
instrumentation that deliberately avoids the public seams, since overriding them switches
batching off.

Was generative AI tooling used to co-author this PR?

Yes — Claude Code (Opus 5)

Generated-by: Claude Code (Opus 5) following the guidelines

The Dag processor's manager issues these statements once per parsed file, and how many was written down nowhere, so a change could add round trips to that path unnoticed. Measure it: a call costs 10 statements per file plus 3 per Dag the file defines, and a sweep pays that fixed price once per persistence call -- currently one call per file. Those counts are calibrated against Postgres, and the tests carrying them are marked for it, since statement counts differ by dialect. The sweep cost is measured through _collect_results and derives both prices from its own measurements, so it carries no dialect-specific number, runs on any backend, and tracks whatever strategy the manager uses. Timing this proved unreliable: on a throttled single-core container, repeat runs of identical code vary by more than a third, far larger than the savings under consideration. Counting statements is exact.

@seanmuth

The Dag processor's manager wrote parse results one file at a time. Most of what that costs is fixed per call rather than per Dag, so a sweep of single-Dag files paid the fixed cost once per file. Persisting a sweep together pays it once. Measured over 60 files at 32 parsing processes, where sweeps held six files on average: 120 persistence calls become 21, and statements fall from 15.0 to 5.1 per file. What that saves in wall clock depends on how much of a file's cost is fixed. The Dags measured carry 200 tasks each, so the per-Dag work of hashing and serializing them dominates, and the saving sits inside the run-to-run variance of the machine it was measured on. Files defining smaller Dags pay proportionally more of the fixed cost and gain more of it back. The saving is also worth nothing at the default two parsing processes, where a sweep only ever holds one file: deployments running a wide Dag processor are the ones this is for. Parse duration is per file, so batching means it can no longer be a single value for the call; it is carried per Dag instead. Callers persisting one file still pass one value. A sweep is split into groups: one bundle each, since the bundle determines the version and version data the write needs, and each dag_id claimed at most once, since writing two files that define the same dag_id together would merge them into one Dag and lose the duplicate warning. Groups are written in the order they are returned, and a file is only held back to a group after the one holding the file it duplicates, so the same file wins a duplicated dag_id as when each was written on its own. A group is also capped, because the saving flattens out quickly while the rows one transaction holds locked do not. Each group is persisted by its own call, and so in its own transaction: update_dag_parsing_results_in_db rolls the session back before retrying a database error, which would otherwise discard a group already written alongside it while its files were still recorded as persisted. A group that fails is retried one file at a time, so one unwritable file does not discard the results of the others, and finished processors are closed whether or not the sweep succeeded. Persistence moves out of handle_parsing_result, which keeps the stat and metric handling. That method and persist_parsing_result were released in 3.3 as the seams for deployments forwarding results somewhere other than the metadata DB. Both keep their released contract and are still called once per file, and both are now deprecated in favour of persist_parsing_results, which is handed a whole group at once. Overriding a deprecated seam costs that deployment its batching and nothing else. From @seanmuth's investigation of the Airflow 2-to-3 dag-processor CPU regression: github.com/seanmuth/af2-af3-scheduler-cpu-repro. This is the per-sweep result batching that investigation identified as the largest lever.

Replacing one of the persistence hooks on a single manager left the detection reading the class, which still reported the default. The manager went on writing to the metadata DB while looking, to whoever replaced the hook, as though it no longer did. Typing the Dags a group merges lets the checker see what reaches the write, which is the one place merging several files could go wrong. The sweep budget derives the number of calls it expects from the group cap rather than assuming a sweep fits in one, and measures a sweep past that cap, so splitting a sweep into groups is exercised rather than only asserted. Naming a local set of Dag warnings `warnings` shadowed the module of the same name, which the deprecation warning now needs.

Overriding the batch seam was checked by calling it directly, which only shows that Python dispatches to an override; the manager was never involved, so nothing said it routes a sweep through the seam at all. It is now driven through a sweep, with a default manager doing the same as its negative. A file can parse to no Dags and is still a file the sweep parsed, so the error it left behind last time has to be cleared with the rest. Removing the line that records such a file left every test passing. A group carries one bundle's version and version data, and nothing wrote a sweep spanning two bundles to check neither was crossed over. The stale-error test asserts about the file it is checking rather than about the whole table, which other tests leave rows in.

Grouping saves a fixed cost once per file, but what one transaction holds locked grows with the Dags in it. Counting files bounds neither: sixteen files that generate their Dags dynamically can be thousands of rows locked for the length of the write, which is the case the bound exists for. Counting Dags does not bind on files defining one or two of them, which is the shape a sweep usually has, so this costs nothing in the common case. A file is still never split, since the record of it having been parsed and the import errors keyed to it belong to it as a whole.

The reasoning was spread over paragraphs where a sentence carries it, repeated between the two deprecated seams, and in places narrated what the next line already says. What stays is what the code cannot say for itself: why each group needs its own transaction, why files sharing a dag_id are ordered rather than only separated, and why a hook is looked for on the instance as well as the class.

ephraimbuddy added 4 commits August 17, 2026 17:16

ephraimbuddy force-pushed the batch-parse-result-persistence branch from 19b8afa to 847d2b4 Compare August 18, 2026 17:01

ephraimbuddy added 2 commits August 18, 2026 18:24

Navigation Menu

Uh oh!

Persist a sweep of Dag parse results in one pass - #71771

Persist a sweep of Dag parse results in one pass#71771
ephraimbuddy wants to merge 6 commits into
apache:mainfrom
astronomer:batch-parse-result-persistence

ephraimbuddy commented Aug 18, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

ephraimbuddy commented Aug 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

The problem

What changed

Measured results

Two honest caveats

Backward compatibility

Reproducing

Was generative AI tooling used to co-author this PR?

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

ephraimbuddy commented Aug 18, 2026 •

edited

Loading