Skip to content

Add ADR for Dag importer process model - #73457

Open
dilnazanlid wants to merge 2 commits into
apache:mainfrom
dilnazanlid:aip-85-adr
Open

dilnazanlid wants to merge 2 commits into
apache:mainfrom
dilnazanlid:aip-85-adr

Conversation

@dilnazanlid

@dilnazanlid dilnazanlid commented Sep 21, 2026 •

Copy link
Copy Markdown
Contributor

This PR introduces ADR-0001 to document the process model and supervision interfaces for AIP-85: Dag Importer.

Key decisions:

  1. The importer owns the process. The manager calls a non-blocking submit_import() and supervises the returned ImportHandle. By default the import forks a child (so existing importers keep working unchanged); importing inline is an opt-in for importers that run no author code; language runtimes can use a process they keep warm. This replaces the manager's current special-casing of coordinator importers.
  2. The unit of work is the Dag definition, not the file - e.g. each zip member gets its own queue entry and child. Callbacks resolve against their requested bundle version, independent of discovery.
  3. Fork and GC handling belong to the framework. A final fork_import() keeps forked children from copying the manager's heap; importer authors never touch the collector.
  4. Importers live for the whole Dag processor run, with warm_up() / shut_down() hooks for long-lived state such as a warm runtime.
  5. Definitions cross process boundaries as coordinates (a self-describing envelope), not file contents.
  6. The handle's result is already serialised (the DagFileParsingResult shape), so no Task SDK objects reach the manager and non-Python runtimes can produce it.
  7. Callbacks stay Python-only for now.

The ADR is aligned with the native Lang-SDK Dag parsing (lang-sdk ADR-0010/0012, #74035).

Related to: AIP-85


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

Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated

@ephraimbuddy ephraimbuddy left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for writing this down. The alternatives and explicit tradeoffs make the reasoning behind each decision easier to review.

I support giving importers control over their runtime. I’d like to clarify how that fits with an AIP I’m drafting on running Dag parsing through executors. It hasn’t reached the dev list yet, so please treat it as an early proposal.

Decision 1’s rejected alternative appears to rule out that direction, but I think it addresses a different design. In my proposal, executors would manage placement and capacity, while importers would retain control over the runtime inside each worker. My inline comments focus on making that boundary explicit and identifying the contracts the two proposals would need to agree on.

Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
@jason810496
jason810496 self-requested a review September 23, 2026 06:10
@jason810496 jason810496 added type:doc-only Changelog: Doc Only area:dag-processor area:dag-bundles AIP-66 AIP-85: Dag Importer https://cwiki.apache.org/confluence/spaces/AIRFLOW/pages/315494137/AIP-85+DAG+importer area:DAG-processing labels Sep 23, 2026
@jason810496 jason810496 moved this from Todo to In Progress in AIP-85 Dag Importer Sep 23, 2026

@pierrejeambrun pierrejeambrun left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the PR

Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated

@jason810496 jason810496 left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks a lot for drafting the design!

Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated

```
ImportHandle what the manager does with it
is_ready ──────────────────► polled every loop pass — must never block

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

From my perspative, the is_ready should be renamed to is_finished. Since we're using "not proc.is_ready" as "finished".

def _collect_results(self):
finished = []
for file, proc in self._processors.items():
if not proc.is_ready:
# This processor hasn't finished yet, or we haven't read all the output from it yet
continue
finished.append(file)
self.handle_parsing_result(file, proc)
for file in finished:
processor = self._processors.pop(file)
processor.close()

@property
def is_ready(self) -> bool:
if self._check_subprocess_exit() is None:
# Process still alive, def can't be finished yet
return False
return not self._open_sockets

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I considered both, but still is_ready seems more appropriate. It's the name the existing parse processes already use. DagFileProcessorProcess and LangSDKDagFileProcessorProcess both expose is_ready, and the manager's _collect_results loop polls exactly that property. ImportHandle is a structural protocol, so keeping the name means today's process classes already satisfy it without any renaming or adapter, and the manager loop doesn't change.

WDYT?

Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment on lines +165 to +181
### 3. Fork mechanics and collector discipline belong to the base class

Forking out of a long-lived process carries a memory contract that is easy to miss: the child shares
the parent's pages until something writes to them, and a garbage collection writes to every object
it inspects. One collection in a freshly forked child therefore copies most of the parent's heap to
achieve nothing.

```
manager process parse child
────────────────────────────────────────────── ────────────────────────
prepare bundles
warm_importers() every importer realised, every warm_up() run
gc.collect() nothing collectable is left to be frozen
gc.freeze() the shared heap becomes permanent
│
├─ per import, steady state ─────────────────
│ fork_import(): gc.disable()

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • section title: "Fork mechanics and collector discipline belong to the DagImporter.start_import"
  • "long-lived process" -> "ephemeral processor subprocesses"

Additionally, should we show the hierarchy of manager -> start_import -- (processor subprocess) --> fork_import instead of just fork_import?

(Please note that there's another comment for suggestion better naming for the "start_import" method)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, I added the call chain to the diagram, so it now shows submit_import() → fork_import() → spawn_child() → parse child.

I kept "long-lived" because there's no processor subprocess in between anymore. The importer runs inside the manager process, so the fork happens from the manager itself, which lives for the whole Dag processor run.

I also kept submit_import out of the title. The GC handling lives in fork_import, which is final, while submit_import is the method importers can override.

Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
Comment thread airflow-core/adr/dag-processing/0001-dag-importer-process-model.md Outdated
- Make submit_import non-blocking and fork by default, so importers
  written against import_definition never run in the manager; inline
  import becomes an opt-in for importers that run no author code.
- Rename start_import/kill to submit_import/abort, drop pid, and record
  the naming rationale.
- Add shut_down() to the importer lifecycle.
- Make queue entries per definition, scope isolation to definitions,
  and resolve callbacks at their requested bundle version.
- Have the parse child freeze its inherited heap before re-enabling
  the collector.
- Fix the serialised result as the manager boundary and list its
  fields; keep callbacks Python-only for now.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

AIP-85: Dag Importer https://cwiki.apache.org/confluence/spaces/AIRFLOW/pages/315494137/AIP-85+DAG+importer area:dag-bundles AIP-66 area:DAG-processing area:dag-processor type:doc-only Changelog: Doc Only

Projects

Status: In Progress

Development

Successfully merging this pull request may close these issues.

7 participants