diff --git a/airflow-core/adr/lang-sdk/0010-native-dag-processing.md b/airflow-core/adr/lang-sdk/0010-native-dag-processing.md new file mode 100644 index 0000000000000..ff2acb006031c --- /dev/null +++ b/airflow-core/adr/lang-sdk/0010-native-dag-processing.md @@ -0,0 +1,212 @@ + + +# ADR-0010: Native Dag Processing — DagImporter Registration and Routing + +## Status + +Proposed + +## Context + +`AbstractDagImporter` (AIP-85, `task-sdk/src/airflow/sdk/importers/`) makes the Dag processor aware of source formats beyond `.py`. It exposes `can_handle`, `list_dag_definitions`, +`import_definition` and `get_source_code`, and returns `DagImportResult.dags: list[DAG]` — `airflow.sdk.DAG` objects. + +A Lang-SDK artifact can be one of those sources: a Dag authored entirely in Go or Java, with `Dag(spec)` on the SDK side. Parsing it means launching a runtime, which is what a +coordinator does. This ADR settles where that importer comes from, which coordinator instance backs it, and how the two avoid being configured twice. + +`JavaDagImporter` and `JavaCoordinator` are the examples throughout. `ExecutableDagImporter` (Go) and `NodeDagImporter` (TypeScript) follow the same shape. + +## Decision + +### The coordinator hands out its importer + +``` +BaseCoordinator.get_dag_importer() -> AbstractDagImporter + abstract — a coordinator names the importer for the artifacts it parses + JavaCoordinator.get_dag_importer() -> JavaDagImporter(coordinator=self) +``` + +The importer comes back already bound to the coordinator, so an operator never configures which coordinator an importer uses. `[sdk] coordinators` stays the one place a runtime is +declared. + +The return is not optional. A coordinator that never contributes an importer to any bundle already says so by the artifact source it was configured with — `for_bundle` below +answers that, and answering it a second time with a `None` return would let the two disagree. What a coordinator can parse and where it gets registered are separate questions. + +### Registration order + +``` +DagImporterRegistry.from_config(bundle_name) + ├── defaults PythonDagImporter, ZipImporter + ├── CoordinatorManager.for_bundle(bundle_name) (new) + │ └── register(coordinator.get_dag_importer()) + ├── [dag_processor] dag_importer_configs (global, unchanged) + └── that bundle's own `importers` list (unchanged) +``` + +`dag_importer_configs` remains the door for importers with no runtime behind them — a YAML importer, say. A Lang SDK never arrives that way. + +### `CoordinatorManager.for_bundle` + +``` +CoordinatorManager + ├── for_queue(queue) → one coordinator (shipped — task execution) + └── for_bundle(name) → the coordinators serving that bundle (new) +``` + +`for_queue` answers "who runs this task". `for_bundle` answers "who can parse Dags in this bundle", which is what the registry tier above needs. It reads the same `[sdk] +coordinators` specs, selecting by the artifact source below. + +### One coordinator instance owns a DagBundle + +``` +EXPLICIT_ROOT jars_root / executables_root / bundles_root + → no DagBundle at all → not returned by for_bundle +NAMED_BUNDLE dag_bundle_name + → that bundle, at the version current when work starts +TASK_BUNDLE neither set + → the bundle the delegating Dag lives in, at the run's version +``` + +Importers are keyed by file extension, one per extension, so two `JavaCoordinator` instances on different JDKs would both claim `.jar`. Only `NAMED_BUNDLE` names a bundle, which +makes it the mode a deployment running two runtimes of the same language has to use. `EXPLICIT_ROOT` has no DagBundle, so nothing scans its artifacts and it cannot produce a Dag to +persist; it serves mixed-language work only. Appendix A covers the three modes in full. + +### The integration point is `import_definition` + +An importer runs inside the Dag-parsing child, so a native Lang-SDK Dag is parsed by a process the importer itself starts. That process is +`LangSDKDagFileProcessorProcess`, which differs from the one the manager started only in the target callable it runs +([ADR-0012](0012-lang-sdk-parse-protocol.md)). + +``` +DagFileProcessorProcess(analytics.jar) ← manager spawns, as for any file + └── _parse_file_entrypoint → _parse_file + └── BundleDagBag → DagImporterRegistry.get_importer(".jar") → JavaDagImporter + │ + └── JavaDagImporter.import_definition(definition, bundle=...) + │ + ├── LangSDKDagFileProcessorProcess.start( + │ target=_parse_lang_sdk_dag_entrypoint, + │ coordinator=JavaCoordinator(...), path=analytics.jar) + │ │ + │ ├── in the child: _build_parse_dag_command() → (command, schema_version) + │ │ coordinator.parse_dag() — spawn JVM, fd 0 ⇄ comm socket + │ │ + │ │ ──DagFileParseRequest───▶ JVM (ToDagProcessor) + │ │ ◀─DagFileParsingResult─── JVM (ToManager) + │ │ ┌────────────────────────────────────────────────────┐ + │ │ │ serialized_dags: ["java_report"] (@Builder.Dag) │ + │ │ └────────────────────────────────────────────────────┘ + │ │ TaskHandler registrations have no Dag to serialize — + │ │ no "etl" entry exists to be discarded. + │ │ + │ └── Get* from the JVM relayed up to the manager unchanged + │ + ├── LazyDeserializedDAG(data=...) → airflow.sdk.DAG + └── DagImportResult(dags=[DAG("java_report")]) + ▼ + _serialize_dags(bag) → DagFileParsingResult → manager + ▼ + DagModelOperation → PERSIST "java_report" only +``` + +The coordinator is reached through the importer, never through the manager's file-to-process routing. ADR-0004's `_resolve_processor_target` scan, which picks a coordinator by +asking each one `can_handle_dag_file`, is superseded here: extension-keyed importer registration has already decided that `.jar` belongs to this coordinator, and two mechanisms +claiming the same file would have to agree. + +For comparison, a pure Python file with no stub tasks: + +``` +PythonDagImporter.import_definition(definition, bundle=...) + │ + ├── Parse → DAG objects + ├── serialize_dag(dag) → no stub tasks, nothing to cross-validate + ├── Return DagImportResult(dags=[dag]) + ▼ +DagModelOperation → PERSIST +``` + +`DagImportResult.dags` is `list[DAG]`, so the importer wraps each serialized entry as a `LazyDeserializedDAG` and transforms it into an `airflow.sdk.DAG`. + +## Consequences + +- A Lang-SDK importer is never configured by hand. The runtime is declared once, in `[sdk] coordinators`, and the importer follows from it. +- Routing a Lang-SDK artifact to its runtime becomes the importer registry's job. ADR-0004's `can_handle_dag_file` / `_resolve_processor_target` scan no longer decides which + process parses a file, and the coordinator method it drove is replaced by `parse_dag` ([ADR-0012](0012-lang-sdk-parse-protocol.md)). +- One coordinator instance per DagBundle becomes a deployment constraint: two JDKs mean two `dag_bundle_name` values and two bundle-scoped registries. This is what keeps + extension-keyed registration unambiguous. +- A coordinator in `EXPLICIT_ROOT` mode cannot back a Dag importer. Its artifacts live outside any DagBundle, so nothing scans them. It still implements `get_dag_importer`; the + importer is simply never asked for, because `for_bundle` does not return that coordinator. +- `CoordinatorManager` gains `for_bundle`, a second lookup axis beside `for_queue`. +- A packed Go bundle claims the empty extension, which three call sites currently treat as absent rather than as a key. Appendix B lists them. +- `get_source_code` is abstract, so every Lang-SDK importer must implement it, and a native Lang-SDK Dag has no Python source to return. What it should return, and how that squares + with [ADR-0006](0006-no-lang-sdk-source-display.md), is not settled here. +- A native Dag crosses serialization twice over: the runtime serializes it, the importer deserializes it into an `airflow.sdk.DAG`, and the enclosing parse serializes it again. + That round trip is the price of `DagImportResult.dags` being `list[DAG]` — the same cost AIP-85's `list[DAG]` / `list[LazyDeserializedDAG]` question is about. +- Parsing a native Dag costs two processes, not one: the Dag-processing child the manager already starts, plus the runtime the importer starts inside it. The inner one inherits its + request, result and logging from `DagFileProcessorProcess`, so the only Lang-SDK-specific code is the target callable. + +## References + +- [ADR-0012](0012-lang-sdk-parse-protocol.md) — `parse_dag` and the coordinator interface it belongs to +- [ADR-0011](0011-mixed-language-dag-processing.md) — why `TaskHandler` registrations never reach a `DagImporter` +- [ADR-0003](0003-pure-java-dags.md) — `BundleScanner` / `BuilderProcessor`, build-time artifact inventory +- [ADR-0004](0004-dag-parsing.md) — `can_handle_dag_file`, the subprocess bridge +- [ADR-0006](0006-no-lang-sdk-source-display.md) — no Lang-SDK source display +- `task-sdk/src/airflow/sdk/importers/` — `AbstractDagImporter`, `DagImportResult`, `DagImporterRegistry` +- `airflow-core/src/airflow/dag_processing/processor.py` — `DagFileProcessorProcess`, the class `LangSDKDagFileProcessorProcess` extends +- [AIP-85](https://cwiki.apache.org/confluence/x/_Q7OEg) — DagImporter +- [AIP-108](https://cwiki.apache.org/confluence/x/pY4mGQ) — Language SDKs + +## Appendix + +### Appendix A — Artifact sources and what `for_bundle` returns + +`SubprocessCoordinator` classifies artifact ownership at construction. The explicit root and `dag_bundle_name` are mutually exclusive, and both that conflict and a +`dag_bundle_name` naming an unconfigured bundle are rejected there. + +`NAMED_BUNDLE` is the unambiguous case. `for_bundle(name)` returns it when its `dag_bundle_name` matches, so its importer lands in exactly one bundle-scoped registry. Two +`dag_bundle_name` values give two registries, and `.jar` is claimed once in each. + +`TASK_BUNDLE` has no fixed bundle — its artifacts ride along with whichever Dag delegates to them — so `for_bundle` returns it for every bundle and its importer registers +everywhere. That is sound only while it is the sole claimant of its extension, which is the co-located single-runtime deployment. + +`EXPLICIT_ROOT` points at a filesystem path outside any DagBundle. The Dag processor never scans it, so there is no file for an importer to claim and `for_bundle` never returns it. +Such a coordinator is reachable only by queue, for mixed-language work. + +`get_importer_registry(bundle_name)` is already cached per bundle, and `CoordinatorManager` caches instances separately, so the two caches have to be reset together. + +### Appendix B — Extensionless artifacts + +A packed Go bundle has no suffix. `ExecutableDagImporter` claims the empty extension as a first-class key rather than depending on a `can_handle` scan, whose winner varies with +registration order because `_ordered_importers` is scanned in reverse. + +Three places assume a non-empty suffix today: + +- `_normalize_extensions` rewrites `""` to `"."`. +- `get_importer` and `can_handle` guard on `if suffix:`, which skips the extension map entirely for an extensionless file. +- `find_file_dag_definitions` filters on `path.suffix.lower()`. + +Empty has to pass through all three, with the guards testing `suffix is not None`. + +### Appendix C — Resolving artifact roots at parse time + +`_init_root_source` already resolves roots for all three modes, but publishes them through `_get_scan_roots()`, which is scoped to an active task and raises outside one. Both +parse-side commands need the same roots with no `TaskInstance` in hand: `EXPLICIT_ROOT` and `NAMED_BUNDLE` resolve from the coordinator's own configuration, and `TASK_BUNDLE` +resolves against the bundle the Dag processor is parsing. The scope that publishes the roots has to open for a parse as well as for a task. diff --git a/airflow-core/adr/lang-sdk/0011-mixed-language-dag-processing.md b/airflow-core/adr/lang-sdk/0011-mixed-language-dag-processing.md new file mode 100644 index 0000000000000..c2acaadbedf2a --- /dev/null +++ b/airflow-core/adr/lang-sdk/0011-mixed-language-dag-processing.md @@ -0,0 +1,245 @@ + + +# ADR-0011: Mixed-Language Dag Processing — Task Handlers Are Not Dags + +## Status + +Proposed + +## Context + +A Lang-SDK artifact can serve either of the two authoring features the Language SDK spec fixes, or both at once: + +| Feature | Who owns the graph | Author writes | Artifact contributes | +|---------------------------------|----------------------------|------------------------------|----------------------| +| **Mixed Language Task Handler** | Python, via `@task.stub` | `TaskHandler(dag, task, fn)` | Only task bodies | +| **Native Dag** | the Lang-SDK source itself | `Dag(spec)` | The entire Dag | + +In the mixed-language role the artifact used to author a Dag too, under the same `dag_id` the Python file already owns. Two conflicting definitions reached persistence and +something downstream had to choose between them. This ADR removes the conflict at the authoring interface, and defines where a stub task is validated against the handler that +implements it. + +Terms follow the Language SDK spec (`task-sdk/docs/lang-sdk-spec.rst`, spec version `1.0`). Note that the spec's `bundle` is the SDK-side registration container, not Airflow's +`DagBundle` — both appear below. + +## Decision + +### Two registration kinds, one bundle + +``` +┌──────────────────────────────────────────────────────────────────────────┐ +│ Per-task (Python): is_stub │ +│ AbstractOperator.is_stub = False (default) │ +│ _StubOperator.is_stub = True │ +│ │ +│ Per-registration (Lang-SDK): which interface the author wrote │ +│ Dag(spec) → DagRef │ +│ TaskHandler(dag, task, fn) → TaskHandlerRef │ +│ │ +│ Both kinds go into one bundle, through one verb: │ +│ bundle.register(dag, handler, ...) → bundle.serve() │ +└──────────────────────────────────────────────────────────────────────────┘ +``` + +A `TaskHandlerRef` has no schedule, no task graph, and no `dag_id` of its own to persist. A `DagRef` has all three. Dag parsing draws on the `Dag` registrations +([ADR-0010](0010-native-dag-processing.md)); validation draws on the `TaskHandler` registrations. Because both kinds share one bundle and one `register` verb, the bundle cannot be +the discriminator — the registration kind is. Appendix A gives the rejected alternative and why. + +### One artifact, both kinds + +``` +analytics.jar +├── EtlTasks @Builder.TaskHandler(dagId = "etl", taskId = "extract") +│ @Builder.TaskHandler(dagId = "etl", taskId = "transform") +│ backs etl.py's stub tasks — no Dag on the Java side +└── JavaReportPipeline @Builder.Dag(id = "java_report") + native, persisted as "java_report" + +bundle.register(javaReport, etlExtract, etlTransform); // one verb, both kinds +bundle.serve(args); +``` + +The split is per registration, not per file and not per bundle. + +### Validation runs in the Dag-file parse, after the Dags exist + +Comparing a stub against its handler needs two things at once: the `airflow.sdk.DAG` objects the Python file produced, and the `arg_bindings` that only appear once those Dags are +serialized. `_parse_file` holds both, between `_serialize_dags` and the `DagFileParsingResult` it returns. That is where the handler query is issued. + +``` +DagFileProcessorProcess(etl.py) ← manager spawns, as for any file + └── _parse_file_entrypoint → _parse_file + ├── BundleDagBag → PythonDagImporter → airflow.sdk.DAG objects ← the Dags now exist + ├── _serialize_dags(bag) → is_stub tasks carry arg_bindings (ADR-0007) + │ + │ ┌─────────────────────────────────────────────────────────────────────┐ + │ │ Step 1: Collect the dag_ids to ask about — every Dag in this │ + │ │ file with at least one is_stub task → ["etl"] │ + │ └─────────────────────────────────────────────────────────────────────┘ + │ + │ ┌─────────────────────────────────────────────────────────────────────┐ + │ │ Step 2: Resolve stub tasks → Coordinator instances via │ + │ │ CoordinatorManager.for_queue(queue) │ + │ │ ([sdk] queue_to_coordinator) │ + │ │ │ + │ │ stub task queue Coordinator instance │ + │ │ ────────────────────────────────────────────────── │ + │ │ extract → "jdk-11" → JavaCoordinator(name="jdk-11") │ + │ │ transform → "jdk-17" → JavaCoordinator(name="jdk-17") │ + │ │ load → "jdk-11" → JavaCoordinator(name="jdk-11") │ + │ │ │ + │ │ Deduplicate → 2 distinct coordinator instances │ + │ └─────────────────────────────────────────────────────────────────────┘ + │ + │ ┌─────────────────────────────────────────────────────────────────────┐ + │ │ Step 3: Locate the artifact backing each dag_id, no JVM launch — │ + │ │ reuse BundleScanner (ADR-0003), whose build-time inventory │ + │ │ indexes registered (dagId, taskId) handler pairs alongside │ + │ │ the artifact's native Dag ids │ + │ │ │ + │ │ JavaCoordinator(name="jdk-11") │ + │ │ ├── resolve artifact roots for its mode │ + │ │ │ EXPLICIT_ROOT → jars_root │ + │ │ │ NAMED_BUNDLE → dag_bundle_name, pinned │ + │ │ │ TASK_BUNDLE → the bundle being parsed │ + │ │ │ │ + │ │ └── BundleScanner.scanBundles(roots) │ + │ │ → "etl" → ResolvedBundle(analytics.jar, mainClass, ...) │ + │ │ │ + │ │ Group the dag_ids by (coordinator, artifact) — one group, one │ + │ │ process, one request │ + │ └─────────────────────────────────────────────────────────────────────┘ + │ + │ ┌─────────────────────────────────────────────────────────────────────┐ + │ │ Step 4: Query each group — one request, one response │ + │ │ │ + │ │ SDKTaskHandlerProcessorProcess.start( │ + │ │ target=_parse_task_handler_entrypoint, │ + │ │ coordinator=JavaCoordinator("jdk-11"), │ + │ │ path=analytics.jar) │ + │ │ │ │ + │ │ ├── in the child: _build_parse_task_handler_command() │ + │ │ │ coordinator.parse_task_handler() — spawn JVM │ + │ │ │ │ + │ │ │ ──TaskHandlerParseRequest(file=analytics.jar, │ + │ │ │ dag_ids=["etl"])─────▶ JVM │ + │ │ │ (ToSDKTaskHandlerProcessor) │ + │ │ │ │ + │ │ │ JVM answers from its own TaskHandler registrations │ + │ │ │ whose dagId is one of the requested ids │ + │ │ │ │ + │ │ │ ◀─TaskHandlerParsingResult(task_handlers={ │ + │ │ │ "etl": [extract, transform, load]})─────── JVM │ + │ │ │ (ToManager) │ + │ │ │ │ + │ │ └── Get* from the JVM relayed up to the manager unchanged │ + │ │ │ + │ │ JavaCoordinator(name="jdk-17") │ + │ │ └── its own process, its own resolved artifact and handler set │ + │ └─────────────────────────────────────────────────────────────────────┘ + │ + │ ┌─────────────────────────────────────────────────────────────────────┐ + │ │ Step 5: Compare per dag_id against the Dag just serialized — │ + │ │ union task_handlers[dag_id] across every process first │ + │ │ │ + │ │ Python Dag "etl" (stub tasks) TaskHandlerDeclaration │ + │ │ ────────────────────────────── ────────────────────────────── │ + │ │ task_id ↔ task_id (sets must match) │ + │ │ arg_bindings[*].name ↔ params[*].name (in order) │ + │ │ arg_bindings[*].value_schema ↔ params[*].value_schema │ + │ │ compared only where neither side is null │ + │ │ │ + │ │ On mismatch → import_errors["etl.py"] │ + │ └─────────────────────────────────────────────────────────────────────┘ + ▼ + DagFileParsingResult(serialized_dags=["etl"], import_errors={...}) + ▼ + manager → DagModelOperation → PERSIST (the Python Dag is the sole DB record) +``` + +The parse owns validation, not an importer. `PythonDagImporter` returns `airflow.sdk.DAG` objects and knows nothing about coordinators or queues, so `@task.stub` keeps working for +any importer that can produce a Dag carrying stub tasks. `_parse_file` is also the only place where the whole file's Dags are visible at once, which is what lets one request cover +every `dag_id` that resolved to the same artifact ([ADR-0012](0012-lang-sdk-parse-protocol.md)). + +Resolution goes through the coordinator registry, not the filesystem, so the Python Dag and the Lang-SDK artifact **do not need to be in the same DagBundle**. Nothing here needs an +`airflow.sdk.DAG` round-trip either — validation compares against the Dag the Python parser already built. Appendix B states exactly what is compared. + +### Decision matrix + +| Caller | Coordinator call | What comes back | Action | +|--------------------------------------------|------------------------------------------------------|------------------------------------------|--------------------------------------------------| +| `_parse_file` → `PythonDagImporter` | — (the Python file is parsed in process) | its own parsed Dags | PERSIST | +| `_parse_file`, per (coordinator, artifact) | `parse_task_handler`, scoped to that group's dag_ids | `TaskHandlerParsingResult` | VALIDATE only — not a Dag, so nothing to persist | +| `_parse_file` → `JavaDagImporter` | `parse_dag` | `DagFileParsingResult`, native Dags only | PERSIST | + +There is no fourth row. A `TaskHandlerRef` has no Dag, so no `DagImporter` — and nothing reading a `DagImporter`'s results — ever sees one. + +## Consequences + +- Python leads, Lang-SDK follows. The Dag-file parse persists the Dag and drives validation via `queue → Coordinator → parse_task_handler`. +- No importer knows about coordinators. `PythonDagImporter` is unchanged by this ADR; the stub-to-handler comparison sits in `_parse_file`, above every importer. +- A mixed-language `dag_id` never appears in Dag processing results. No `Dag` registration exists for a `dag_id` a Python file already owns, so everything downstream sees exactly + one record per `dag_id`, with no flag to interpret. +- Stub/implementation mismatches — missing handler, extra handler, parameter name or order, incompatible schema — surface as import errors against the Python file at parse time, + alongside the errors the parse already reports. An unannotated stub argument is checked by name and position only. +- The Python Dag and Lang-SDK artifact can live in different DagBundles. +- A single Dag can have stubs targeting different queues, some Java, some Go. Each resolves to its own coordinator instance, and validation unions their declarations per `dag_id` + before comparing task ids. +- Validating a file costs one extra process per (coordinator, artifact) pair its stubs resolve to — one for the common case of a file whose stubs all target a single runtime, and + none at all for a file with no stub tasks. +- Mixed-language is Python-primary only. Lang-SDK runtimes cannot define stub operators; a native Dag cannot delegate tasks to Python. +- No per-Dag flag, no schema migration, no new `DagModel` column, no REST/UI change. +- Terms track Language SDK spec `1.0`. A spec rename of `TaskHandler`, or of the `register` / `serve` verbs, lands here too. + +## References + +- [ADR-0012](0012-lang-sdk-parse-protocol.md) — `parse_task_handler` and the `TaskHandlerParsingResult` shape this ADR compares against +- [ADR-0010](0010-native-dag-processing.md) — the `Dag`-registration half, and the importer that persists it +- [ADR-0003](0003-pure-java-dags.md) — `BundleScanner` / `BuilderProcessor`, build-time artifact inventory +- [ADR-0006](0006-no-lang-sdk-source-display.md) — no Lang-SDK source display for mixed-language Dags +- [ADR-0007](0007-taskflow-across-language-boundary.md) — `arg_bindings` / `TaskArgBinding` / `ArgValueSchema` +- Language SDK spec (`task-sdk/docs/lang-sdk-spec.rst`) +- [AIP-108](https://cwiki.apache.org/confluence/x/pY4mGQ) — Language SDKs +- [AIP-85](https://cwiki.apache.org/confluence/x/_Q7OEg) — DagImporter +- `airflow-core/src/airflow/dag_processing/processor.py` — `_parse_file`, `_serialize_dags`, `DagFileParsingResult` + +## Appendix + +### Appendix A — Why the split lives on the interface + +The alternative was to keep authoring the Dag on both sides and mark the Lang-SDK copy with a per-Dag flag, `is_mixed_language_dag`, so importers knew which copy to drop. That +keeps producing the definition it then has to discard, and it pushes the question "is this Dag real?" onto every consumer of a serialized Dag — the importer, the persistence layer, +anything later reading the record. Taking the Dag out of the authoring interface answers the question once, at the only point where the answer is known for free: the author already +chose which interface to write against. The only marker left in serialization is `is_stub`, and it is per-task. + +The bundle cannot carry the distinction either. The Language SDK spec puts both kinds in one `bundle`, takes them through one `register` verb in any mixture, and serves the process +with one `bundle.serve()`. One artifact can hold both, so neither the file nor the bundle tells a `DagImporter` what it is holding. + +### Appendix B — What is compared + +Handler declarations from every process spawned in Step 4 are unioned per `dag_id` before comparison, since one Dag's stubs can target several queues. + +- `task_id` sets must match exactly. A missing or extra handler is an error. +- `arg_bindings[*].name` against `params[*].name`, in order — both sides bind positionally. +- `arg_bindings[*].value_schema` against `params[*].value_schema`, compared only where neither side is null. An unannotated `@task.stub` parameter produces `null` today, so a + strict comparison would make every untyped stub argument a parse error. + +Any mismatch is reported against the Python file, which is the definition the author can act on, and travels back on `DagFileParsingResult.import_errors` with everything else the +parse found. diff --git a/airflow-core/adr/lang-sdk/0012-lang-sdk-parse-protocol.md b/airflow-core/adr/lang-sdk/0012-lang-sdk-parse-protocol.md new file mode 100644 index 0000000000000..e144b87445b58 --- /dev/null +++ b/airflow-core/adr/lang-sdk/0012-lang-sdk-parse-protocol.md @@ -0,0 +1,233 @@ + + +# ADR-0012: Lang-SDK Parse Protocol — Handler Messages and Coordinator Verbs + +## Status + +Proposed + +## Context + +The Dag processor asks a Lang-SDK runtime two different questions. "Which Dags does this artifact define?" is answered over the messages [ADR-0004](0004-dag-parsing.md) already +defines. "Which task handlers does this artifact register for a `dag_id` Python already owns?" has no answer in those messages, because a `TaskHandler` registration carries no Dag +([ADR-0011](0011-mixed-language-dag-processing.md)). + +This ADR defines the request that carries the second question, the subprocess classes that carry both, and the two parse-side entry points on the coordinator. + +Terms follow the Language SDK spec (`task-sdk/docs/lang-sdk-spec.rst`, spec version `1.0`). + +## Decision + +### One channel shape, two request types + +``` +parse_dag parse_task_handler + + parent process parent process + │ DagFileParseRequest │ TaskHandlerParseRequest + ▼ (ToDagProcessor) ▼ (ToSDKTaskHandlerProcessor) + coordinator (raw byte forward) coordinator (raw byte forward) + ▼ ▼ + runtime runtime + │ DagFileParsingResult │ TaskHandlerParsingResult + ▼ (ToManager) ▼ (ToManager) + parent process parent process +``` + +Both verbs are byte forwarders: the coordinator spawns the runtime, wires `fd 0` to the comm socket, and never decodes the payload. The process that spawned the parse decodes the +reply. They stay two methods, not one `parse(request)`, because a coordinator can serve handlers without serving native Dag parsing. Appendix A has the longer argument. + +### The reply travels on `ToManager` + +``` +_ParseSideResponses = shared tail — same members, same `type` discriminator + ConnectionResult | VariableResult | VariableKeysResult | TaskStatesResult + | PreviousDagRunResult | PreviousTIResult | PrevSuccessfulDagRunResult + | ErrorResponse | OKResponse | XComCountResponse | XComResult + | XComSequenceIndexResult | XComSequenceSliceResult + +ToDagProcessor = DagFileParseRequest | _ParseSideResponses parent → child +ToSDKTaskHandlerProcessor = TaskHandlerParseRequest | _ParseSideResponses parent → child (new) + +ToManager = DagFileParsingResult | TaskHandlerParsingResult child → parent + | GetConnection | GetVariable | … | MaskSecret +``` + +`ToSDKTaskHandlerProcessor` is the only new union; `ToManager` gains one member. The two parent → child unions differ in exactly one member, because the child's questions about +connections, variables and XComs do not depend on which parse it was asked for. + +`ToManager` is named for the process that usually holds the other end, but the role it describes is "whoever spawned this parse". The Dag processor manager fills it for +`DagFileProcessorProcess`; a Dag-parsing child fills it for the two processes below, relaying anything that is not a parsing result up its own `ToManager` channel unchanged. That +relay is only type-safe because both hops speak the same pair, which is the reason not to mint a separate `ToCoordinator`. + +### Message shapes + +``` +class TaskHandlerParseRequest: + file: str # the artifact resolved for this coordinator + dag_ids: list[str] # every Dag in the parsed file with stub tasks that resolved here + bundle_path: Path + bundle_name: str + type: Literal["TaskHandlerParseRequest"] + +class TaskHandlerParsingResult: + fileloc: str + task_handlers: dict[str, list[TaskHandlerDeclaration]] # dag_id → its declarations + import_errors: dict[str, str] | None = None + warnings: list | None = None + type: Literal["TaskHandlerParsingResult"] + +class TaskHandlerDeclaration: + task_id: str + params: list[TaskHandlerParam] # ordered — arg_bindings are positional + +class TaskHandlerParam: + name: str + value_schema: ArgValueSchema | None = None + required: bool # the handler declares no default +``` + +One request carries every `dag_id` that resolved to the same artifact under the same coordinator, so a file whose stubs all target one runtime costs one process. A `dag_id` the +artifact registers nothing for is **omitted** from `task_handlers` rather than returned empty: the key set is not required to match `dag_ids`, because it is the union across +coordinators that has to cover the stubs ([ADR-0011](0011-mixed-language-dag-processing.md)). + +`value_schema` reuses the `ArgValueSchema` definition `arg_bindings` already carries ([ADR-0007](0007-taskflow-across-language-boundary.md)), so both sides of a comparison are the +same type. Two properties matter to validation: the field is nullable on both sides, and `params` is ordered. Appendix B says what that forces. + +`task_handlers` is the counterpart to `DagFileParsingResult.serialized_dags`, but fully typed. `serialized_dags` is `list[LazyDeserializedDAG]`, which is an opaque object in the +schema snapshot. A handler declaration carries no Dag, so it code-generates and schema-validates in every SDK, and nothing on this path needs a DagSerialization implementation. + +### Parse processes + +``` +WatchedSubprocess + └── BaseParsingProcess socket lifecycle · ToManager decoding · Get* handling + │ · log forwarding under dag_processor.* + ├── DagFileProcessorProcess (shipped, now a subclass) + │ │ target = _parse_file_entrypoint + │ │ DagFileParseRequest → DagFileParsingResult + │ │ + │ └── LangSDKDagFileProcessorProcess (new — ADR-0010) + │ target = _parse_lang_sdk_dag_entrypoint + │ └── coordinator.parse_dag() — spawn runtime, forward fd 0 ⇄ comm socket + │ same request and result types as its base class + │ + └── SDKTaskHandlerProcessorProcess (new — ADR-0011) + target = _parse_task_handler_entrypoint + └── coordinator.parse_task_handler() — same forwarding + TaskHandlerParseRequest → TaskHandlerParsingResult +``` + +`BaseParsingProcess` is `DagFileProcessorProcess` minus the Dag-specific request and result: the comm socket, the `ToManager` decode, the `Get*` dispatch, and the +`task.` → `dag_processor.` log-forwarder rename. The subclasses supply the first message they send, the result they collect, and the target the child runs. + +`LangSDKDagFileProcessorProcess` differs from its base in the target callable alone. Everything else — the request, the result, the socket, the logging — is inherited, because a +native Lang-SDK Dag answers the same question a Python file does. + +Answering `Get*` needs a `Client`, which only the manager holds. `BaseParsingProcess` therefore resolves a request one of two ways: directly against `self.client` when the manager +is the parent, or by relaying it up `SUPERVISOR_COMMS` when a Dag-parsing child is. + +### Coordinator interface + +``` +BaseCoordinator execution_time/coordinator.py + ├── execute_task (shipped) + ├── parse_dag (new — native Dags, ADR-0010) + └── parse_task_handler (new — handlers, ADR-0011) + │ +SubprocessCoordinator coordinators/_subprocess.py + implements all three; each resolves (command, subprocess_schema_version) + from a hook and owns the socket lifecycle: + ├── _build_execute_task_command (shipped) + ├── _build_parse_dag_command (new) + └── _build_parse_task_handler_command (new) + │ +JavaCoordinator · ExecutableCoordinator · NodeCoordinator + supply the three commands; no socket or protocol code +``` + +Names follow the shipped `execute_task` / `_build_execute_task_command` pair and supersede ADR-0004's `run_dag_parsing` / `dag_parsing_cmd`. Each hook returns its own +`subprocess_schema_version`, so handler parsing negotiates the schema the same way task execution does. + +## Consequences + +- The handler channel is typed; the Dag channel is opaque. Every SDK gets generated models for the declaration shape. +- Each runtime implements a second request type instead of a new field on the existing one. A `DagRef` and a `TaskHandlerRef` have different shapes, so one request/result pair + could not carry both. +- `ToSDKTaskHandlerProcessor` extends the set of unions the supervisor schema package introspects, from four to five. It adds two union members — `TaskHandlerParseRequest` and + `TaskHandlerParsingResult`, the latter carrying `TaskHandlerDeclaration` and `TaskHandlerParam` as nested definitions — because the shared responses are classes the registry + already holds. +- Nothing in the protocol distinguishes a coordinator-backed parse from a Python one. A runtime's `Get*` request is answered by the same handlers that answer a Python parser's, + through however many relay hops lie between it and the manager. +- `DagFileProcessorProcess` becomes a subclass. Its public surface does not move, but the shipped `_handle_request` and socket code shifts to `BaseParsingProcess`. +- Neither verb is reached through ADR-0004's `can_handle_dag_file` scan. `parse_dag` is reached through the importer registered for the artifact's extension + ([ADR-0010](0010-native-dag-processing.md)); `parse_task_handler` through `queue → coordinator` ([ADR-0011](0011-mixed-language-dag-processing.md)). +- Terms track Language SDK spec `1.0`. A spec rename of `TaskHandler` lands here too. + +## References + +- [ADR-0004](0004-dag-parsing.md) — coordinator subprocess bridge, `DagFileParseRequest` / `DagFileParsingResult`, `can_handle_dag_file` +- [ADR-0006](0006-no-lang-sdk-source-display.md) — no Lang-SDK source display +- [ADR-0007](0007-taskflow-across-language-boundary.md) — `arg_bindings` / `TaskArgBinding` / `ArgValueSchema` +- [ADR-0010](0010-native-dag-processing.md) — who calls `parse_dag` +- [ADR-0011](0011-mixed-language-dag-processing.md) — who calls `parse_task_handler`, and what it compares the reply against +- Language SDK spec (`task-sdk/docs/lang-sdk-spec.rst`) +- `airflow-core/src/airflow/dag_processing/processor.py` — `DagFileProcessorProcess`, `ToManager` / `ToDagProcessor` +- `task-sdk/src/airflow/sdk/execution_time/schema/` — supervisor schema version bundle, union registry, generated snapshot +- [AIP-108](https://cwiki.apache.org/confluence/x/pY4mGQ) — Language SDKs + +## Appendix + +### Appendix A — Why two verbs, and why one reply union + +The two verbs differ in what they ask for and in which command starts the runtime, not in how they talk. Collapsing them into one `parse(request)` would hide a real deployment +distinction: a coordinator can serve mixed-language handlers with no interest in native Dag parsing, and an absent method states that better than a runtime rejection does. + +The reply direction does not need the same split. An earlier draft gave handler parsing its own `ToRuntime` / `ToCoordinator` pair on the theory that the runtime was answering the +coordinator rather than the manager. It is not: the coordinator forwards bytes in both directions and decodes nothing, so the peer at the far end of the socket is whichever process +spawned the parse. Giving that peer two unions to decode would mean two `CommsDecoder` configurations, two relay paths for the identical `Get*` traffic, and two registry entries for +bodies that never differ. Reusing `ToManager` leaves one reply union with one new member. + +A boolean on `DagFileParseRequest` was the other alternative. It cannot work: a `DagRef` and a `TaskHandlerRef` are different payloads, not two subsets of one, so the flag would +select between shapes the result type cannot both hold. + +### Appendix B — What the nullable, ordered parameter list forces + +`value_schema` is nullable on both sides. An unannotated `@task.stub` parameter produces `value_schema: null` today, so validation compares schemas only where neither side is null, +and falls back to name-and-arity otherwise. A strict comparison would turn every untyped stub argument into a parse error. + +`params` is ordered because `LiteralArgBinding` and `XComArgBinding` are each documented as "one positional stub-task argument". Position is part of the contract, not incidental, +and both sides bind positionally. + +A declaration carries no class, method, or source location. [ADR-0006](0006-no-lang-sdk-source-display.md) rules out Lang-SDK source display, and putting it on the wire would +invite a consumer to render it. It carries no `dag_id` either — the `task_handlers` key supplies it, so a declaration cannot disagree with the bucket it arrived in. + +### Appendix C — Schema registration mechanics + +`registered_models_by_name()` in `task-sdk/src/airflow/sdk/execution_time/schema/` introspects a fixed set of unions — `ToTask` / `ToSupervisor` and `ToManager` / `ToDagProcessor` +— so adding `ToSDKTaskHandlerProcessor` extends that set to five. The shared responses appear in two unions now; the registry keys by class name and rejects two *distinct* classes +under one name, so a member reached twice is not a clash. + +Two prek hooks guard the generated snapshot. Their interaction on a first-introduction body needs checking against the hooks rather than against that package's `AGENTS.md`: the doc +says no `VersionChange` is required for a new body, while `check-supervisor-schemas-versions` fails when the snapshot moves and nothing under `versions/` was touched. + +Artifact roots are resolved per mode by `_init_root_source`, but published through `_get_scan_roots()`, which is scoped to an active task and raises outside one. Both parse-side +commands need those roots with no `TaskInstance` in hand, so the scope that publishes them has to open for a parse as well as for a task. [ADR-0010](0010-native-dag-processing.md) +covers the modes themselves. diff --git a/airflow-core/adr/lang-sdk/README.md b/airflow-core/adr/lang-sdk/README.md index 359275ac2696b..ec80aa587326c 100644 --- a/airflow-core/adr/lang-sdk/README.md +++ b/airflow-core/adr/lang-sdk/README.md @@ -35,6 +35,9 @@ bind core interfaces and apply to every language SDK, not just the Java SDK. - [ADR-0007](0007-taskflow-across-language-boundary.md): TaskFlow across the language boundary — argument binding for Lang-SDK tasks. - [ADR-0008](0008-control-flow-constructs.md): control-flow constructs — grouping, conditional skipping, branching, and triggering a Dag run. - [ADR-0009](0009-provider-operators-as-generated-dsl.md): provider operators as generated, serialization-only DSL in every Lang SDK. +- [ADR-0010](0010-native-dag-processing.md): native Dag processing — DagImporter registration and routing. +- [ADR-0011](0011-mixed-language-dag-processing.md): mixed-language Dag processing — task handlers are not Dags. +- [ADR-0012](0012-lang-sdk-parse-protocol.md): Lang-SDK parse protocol — task handler messages and coordinator verbs. Decisions specific to a single SDK stay next to that SDK — for example, the Go SDK's bundle-format decisions live in [`go-sdk/adr/`](../../../go-sdk/adr). Java-SDK-only interface-design decisions —