From 719a685079ccb9bf43b4080b1845ad8f24ac0d5b Mon Sep 17 00:00:00 2001 From: "Jason(Zhe-You) Liu" <68415893+jason810496@users.noreply.github.com> Date: Fri, 23 Oct 2026 22:28:33 +0800 Subject: [PATCH] Add consolidated Language SDK compatibility matrix (#71757) * Add consolidated Language SDK compatibility matrix Users comparing Language SDKs had to read each SDK's README, and the per-SDK manifests lagged behind what the SDKs actually ship. One generated matrix, split by authoring mode, shows what each SDK supports for a Dag definition and for a Python Dag with stub TaskHandler. * Add callbacks, listeners and provider operators to the SDK matrix Users comparing Language SDKs could not see where callbacks, Airflow's listener API and Python provider operators stand, because the matrix had no dimension for any of them. Callbacks ask different things of an SDK per authoring mode, so they are declared per mode like the other shared capabilities. * Address review on the Language SDK matrix generator The umbrella capability and the authoring mode both read `dag-definition`, which gives one name two meanings in the same manifest. A row's kind also belongs in its entry key: state and capability names happen not to collide today, so the lookup worked by accident rather than by design. * Rename the dynamic-task-mapping dimension to mapped-tasks The dimension is about whether a task can be a mapped task, which the shorter name says directly and keeps the matrix row readable next to the other task-level dimensions. * Split callbacks into Dag-level and task-level dimensions The callbacks of a Python Dag run whichever SDK executes its stub tasks, so ``dag-callbacks`` is a Dag-definition concern only, while ``task-callbacks`` stays declared per authoring mode. Drop ``provider-operators`` until ADR-0009 is accepted, and reuse the shared ``capabilities_doc`` fixture in the per-SDK README tests. * Fix the hook attribution in the matrix generation diagram ``index.rst`` is written by ``update-lang-sdk-central-matrix`` from every SDK in ``LANG_SDKS``, not by the per-SDK README hook. --------- --- .pre-commit-config.yaml | 15 + .../language-sdks/index.rst | 251 ++++++++++++++ contributing-docs/30_new_language_sdk.rst | 171 ++++++--- go-sdk/README.md | 92 +++-- go-sdk/capabilities.yaml | 171 ++++++--- java-sdk/README.md | 92 +++-- java-sdk/capabilities.yaml | 153 ++++++-- java-sdk/sdk/module.md | 92 +++-- scripts/ci/prek/lang_sdk_compat_matrix.py | 328 ++++++++++++------ .../ci/prek/update_go_sdk_readme_matrix.py | 6 +- .../ci/prek/update_java_sdk_readme_matrix.py | 6 +- .../ci/prek/update_lang_sdk_central_matrix.py | 152 ++++++++ .../ci/prek/update_ts_sdk_readme_matrix.py | 6 +- scripts/tests/ci/prek/conftest.py | 55 +++ .../ci/prek/test_lang_sdk_compat_matrix.py | 322 +++++++++++------ .../prek/test_update_go_sdk_readme_matrix.py | 25 +- .../test_update_java_sdk_readme_matrix.py | 27 +- .../test_update_lang_sdk_central_matrix.py | 148 ++++++++ .../prek/test_update_ts_sdk_readme_matrix.py | 27 +- ts-sdk/README.md | 92 +++-- ts-sdk/capabilities.yaml | 188 +++++++--- 21 files changed, 1822 insertions(+), 597 deletions(-) create mode 100755 scripts/ci/prek/update_lang_sdk_central_matrix.py create mode 100644 scripts/tests/ci/prek/test_update_lang_sdk_central_matrix.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 20721f3a85b..d200aac79a6 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -581,6 +581,21 @@ repos: additional_dependencies: ['PyYAML>=6.0', 'rich>=13.6.0'] pass_filenames: false require_serial: true + - id: update-lang-sdk-central-matrix + name: Update the consolidated Language SDK compatibility matrix + env: + PYTHONPYCACHEPREFIX: .build/pycache + entry: ./scripts/ci/prek/update_lang_sdk_central_matrix.py + language: python + files: > + (?x) + ^(go|java|ts)-sdk/capabilities\.yaml$| + ^airflow-core/docs/authoring-and-scheduling/language-sdks/index\.rst$| + ^scripts/ci/prek/lang_sdk_compat_matrix\.py$| + ^scripts/ci/prek/update_lang_sdk_central_matrix\.py$ + additional_dependencies: ['PyYAML>=6.0', 'rich>=13.6.0'] + pass_filenames: false + require_serial: true - id: check-go-example-mod-tidy name: Check lang-SDK Go example module is tidy against the Go SDK env: diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst index 8e6e725b999..ea8d4b8448d 100644 --- a/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst +++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst @@ -204,6 +204,257 @@ Both settings can be supplied as environment variables using the standard Airflo AIRFLOW__SDK__COORDINATORS='{"my-coordinator": {...}}' AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"jdk17": "my-coordinator"}' +Compatibility matrix +-------------------- + +Language SDKs mature independently, so they do not all support the same features yet. The tables below show +what each Language SDK supports: the features common to both authoring modes, the features of a +:ref:`Dag definition `, and the features of a +:ref:`Python Dag with stub TaskHandler `. A supported feature shows the SDK +release that shipped it. See the legend below the tables for what each mark means. + +.. BEGIN AUTO-GENERATED LANG-SDK COMPAT MATRIX + +.. list-table:: Language SDK versions + :header-rows: 1 + :widths: 30 15 15 15 + + * - Version + - Go + - Java + - TypeScript + * - Min. Airflow version + - 3.4.0 + - 3.4.0 + - 3.4.0 + * - Current Supervisor Schema Version + - 2026-10-30 + - 2026-10-30 + - 2026-10-30 + +.. list-table:: Common to both authoring modes + :header-rows: 1 + :widths: 30 15 15 15 + + * - Dimension + - Go + - Java + - TypeScript + * - **TaskInstance states** + - + - + - + * - ``success`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``failed`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``up_for_retry`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``skipped`` (SHOULD) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``deferred`` (MAY) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``up_for_reschedule`` (MAY) + - ✗ + - ✗ + - ✗ + * - ``awaiting_input`` (MAY) + - ✗ + - ✗ + - ✓ 1.0.0 + * - ``removed`` (MAY) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - **Task runtime** + - + - + - + * - ``task-logging`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``xcom-read-write`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``connection-read`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``variable-read-write`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``self-contained-bundle`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``retry-policy`` (MAY) + - ✗ + - ✗ + - ✗ + * - ``task-state-store`` (MAY) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``asset-state-store`` (MAY) + - ✗ + - ✗ + - ✗ + * - ``asset-event-emit`` (MAY) + - ✗ + - ✗ + - ✗ + * - ``asset-event-read`` (MAY) + - ✗ + - ✗ + - ✗ + * - ``object-store`` (MAY) + - ✗ + - ✗ + - ✗ + * - ``listeners`` (MAY) + - ✗ + - ✗ + - ✗ + +.. list-table:: Dag definition + :header-rows: 1 + :widths: 30 15 15 15 + + * - Dimension + - Go + - Java + - TypeScript + * - ``dag-definition-authoring`` (SHOULD) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``dag-args`` (MUST †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``task-args`` (MUST †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``dag-params`` (MUST †) + - ✗ + - ✗ + - ✗ + * - ``dag-callbacks`` (MAY †) + - ✗ + - ✗ + - ✗ + * - ``explicit-dependencies`` (MUST †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``edge-labels`` (MAY †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``taskflow-dependencies`` (MUST †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``task-group`` (MAY †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``task-loops`` (MAY †) + - ✗ + - ✗ + - ✗ + * - ``if-else-branching`` (SHOULD †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``switch-case-branching`` (SHOULD †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``trigger-dag-run`` (MAY †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``dynamic-dag-generation`` (MAY †) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``human-in-the-loop`` (MAY †) + - ✗ + - ✗ + - ✓ 1.0.0 + * - ``dag-test`` (SHOULD †) + - ✗ + - ✗ + - ✗ + * - ``asset-definition`` (MAY †) + - ✗ + - ✗ + - ✗ + * - ``asset-scheduling`` (MAY †) + - ✗ + - ✗ + - ✗ + * - ``mapped-tasks`` (MAY †) + - ✗ + - ✗ + - ✗ + * - ``asset-inlets-outlets`` (MAY †) + - ✗ + - ✗ + - ✗ + * - ``task-callbacks`` (MAY †) + - ✗ + - ✗ + - ✗ + +.. list-table:: Python Dag with stub TaskHandler + :header-rows: 1 + :widths: 30 15 15 15 + + * - Dimension + - Go + - Java + - TypeScript + * - ``mixed-lang-stub-target`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``taskflow-binding`` (MUST) + - ✓ 1.0.0 + - ✓ 1.0.0 + - ✓ 1.0.0 + * - ``mapped-tasks`` (MAY) + - ✗ + - ✗ + - ✗ + * - ``asset-inlets-outlets`` (MAY) + - ✗ + - ✗ + - ✗ + * - ``task-callbacks`` (MAY) + - ✗ + - ✗ + - ✗ + +*Marks:* ``✓`` supported (with the SDK release that shipped it, when known) · ``✗`` not supported · ``n/a`` not applicable · ``n/p`` not published. A tier marked ``†`` applies only when ``dag-definition-authoring`` is supported. + +.. END AUTO-GENERATED LANG-SDK COMPAT MATRIX + .. _language-sdks/bundle-spec: Implementing a new compiled language SDK diff --git a/contributing-docs/30_new_language_sdk.rst b/contributing-docs/30_new_language_sdk.rst index 3746c3e9b05..5bb761e60a0 100644 --- a/contributing-docs/30_new_language_sdk.rst +++ b/contributing-docs/30_new_language_sdk.rst @@ -416,18 +416,9 @@ Runtime capabilities ~~~~~~~~~~~~~~~~~~~~~~ Runtime capabilities describe what a task *body* can do while it executes, regardless of -whether the task was declared in a Python Dag (via ``@task.stub``) or in a native Dag +whether the task was declared in a Python Dag (via ``@task.stub``) or in a Dag definition authored in the target language. An SDK declares each one independently. -``mixed-lang-stub-target`` (MUST) - The SDK can execute a task declared in a Python Dag with the ``@task.stub`` - decorator. This is the primary execution path for every Language SDK. - -``taskflow-binding`` (MUST) - Arguments bound to a ``@task.stub`` target in the Python TaskFlow Dag — literal values - and upstream XCom results alike — are delivered to the task function's parameters at - runtime. - ``task-logging`` (MUST) The SDK forwards the task's log output — stdout / stderr and any structured log records — to the supervisor over the ``--logs`` socket, so it surfaces in the task @@ -454,14 +445,16 @@ authored in the target language. An SDK declares each one independently. ``retry-policy`` (MAY) The SDK lets task code inspect a failure and override whether the task retries or fails, - optionally with a custom retry delay. This is distinct from reporting ``up_for_retry``: - an SDK can support ordinary retries without exposing a task-facing retry-policy API. + optionally with a custom retry delay, the equivalent of Python's ``RetryPolicy``. This is + distinct from reporting ``up_for_retry``: an SDK can support ordinary retries without + exposing a task-facing retry-policy API. ``task-state-store`` (MAY) The task can read and write the per-task state store. ``asset-state-store`` (MAY) - The task can read and write the per-asset state store. + The task can read and write the state store keyed by an asset name or URI. It says + nothing about whether the SDK can *declare* an asset, which is ``asset-definition``. ``asset-event-emit`` (MAY) The task can emit asset events — outlet events, including alias events — as it runs. @@ -469,75 +462,169 @@ authored in the target language. An SDK declares each one independently. ``asset-event-read`` (MAY) The task can read prior asset events. -Native-Dag authoring -~~~~~~~~~~~~~~~~~~~~~~ +``object-store`` (MAY) + The SDK exposes an object-storage API (an ``ObjectStoragePath`` equivalent) usable from + task code. -Native-Dag authoring lets an entire Dag — not just individual task bodies — be written in +``listeners`` (MAY) + A task feeds Airflow's listener API, so a listener plugin sees it as it does a Python + task. OpenLineage is the main consumer: without this, a task of this SDK produces no + lineage facets of its own. + +Dag definition +~~~~~~~~~~~~~~ + +A Dag definition lets an entire Dag — not just individual task bodies — be written in the target language, without a Python Dag file. It is gated by a single umbrella capability: -``native-dag-authoring`` (SHOULD) +``dag-definition-authoring`` (SHOULD) The SDK can define a whole Dag in the target language. The project intends every Language SDK to reach this bar, but an SDK is not required to ship it before it is ready; an SDK that only executes ``@task.stub`` tasks from a Python Dag is still useful and remains conformant to the MUST tier. -The remaining native-Dag capabilities are **conditional**: their tier applies only when -``native-dag-authoring`` is supported (written ``†``). For an SDK that does not (yet) author -native Dags they are *not applicable* (``n/a``) rather than unsupported. +The remaining Dag-definition capabilities are **conditional**: their tier applies only when +``dag-definition-authoring`` is supported (written ``†``). For an SDK that does not (yet) author Dags +in its own language they are *not applicable* (``n/a``) rather than unsupported. + +``dag-args`` (MUST †) + Dag-level arguments (schedule, tags, default queue, and so on) can be set on a Dag definition. ``task-args`` (MUST †) - Task-level arguments can be bound to a native task. + Task-level arguments (retries, pool, queue, and so on) can be set on a task of a Dag definition. ``dag-params`` (MUST †) - Dag-level params can be declared on a native Dag. + Dag-level params can be declared on a Dag definition. + +``dag-callbacks`` (MAY †) + Dag-level callbacks (``on_failure_callback`` and the rest) can be declared on a Dag + definition as functions in the target language, and Airflow runs them. This is a + Dag-definition concern only: the callbacks of a Python Dag are Python functions that the + Python runtime runs whichever SDK executes the tasks. + +``explicit-dependencies`` (MUST †) + Order-only dependencies can be drawn between tasks, the equivalent of Python's ``>>`` and + ``<<``. + +``edge-labels`` (MAY †) + A dependency can carry a label, the equivalent of Python's + ``loaded >> Label("when empty") >> notify``, and the Dag graph shows it. ``taskflow-dependencies`` (MUST †) Task dependencies can be expressed in a TaskFlow style, including passing a task's return value to a downstream task. -``branching`` (SHOULD †) - The native DSL offers a branch construct that skips the non-selected branches. - -``dag-test`` (SHOULD †) - A native Dag can be exercised locally with ``airflow dags test``. - ``task-group`` (MAY †) Tasks can be organised into task groups. -``dynamic-task-mapping`` (MAY †) - Tasks can be expanded at runtime via dynamic task mapping. +``task-loops`` (MAY †) + A task group can be repeated at runtime, the equivalent of Python's ``@task_group`` + with ``.loop()``, bounded by ``max_iterations`` and an optional ``until`` condition. -``asset-inlets-outlets`` (MAY †) - Native tasks can declare asset inlets and outlets. +``if-else-branching`` (SHOULD †) + A task can decide between two downstream paths and skip the one not taken. + +``switch-case-branching`` (SHOULD †) + A task can pick one of several downstream paths and skip the others. + +``trigger-dag-run`` (MAY †) + A Dag definition can declare a task that triggers a run of another Dag, like + ``TriggerDagRunOperator``. + +``dynamic-dag-generation`` (MAY †) + One artifact can generate several Dags from code, for example in a loop. + +``human-in-the-loop`` (MAY †) + A Dag definition can declare a task that waits for human input. + +``dag-test`` (SHOULD †) + A Dag definition can be exercised locally with ``airflow dags test``. + +``asset-definition`` (MAY †) + Assets can be declared in the target language: the equivalents of Python's ``Asset``, + ``AssetAlias``, ``AssetAll`` / ``AssetAny``, and the partition mappers a partitioned + asset timetable consumes. ``asset-scheduling`` (MAY †) - A native Dag can be scheduled on asset events. + A Dag definition can be scheduled on asset events, including the ``AssetWatcher`` + equivalent that turns an external trigger into those events. -``object-store`` (MAY †) - The SDK exposes an object-storage API (an ``ObjectStoragePath`` equivalent) usable from - native Dag code. +Python Dag with stub TaskHandler +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +These capabilities cover tasks declared in a Python Dag with the ``@task.stub`` decorator and +implemented in the target language. + +``mixed-lang-stub-target`` (MUST) + The SDK can execute a task declared in a Python Dag with the ``@task.stub`` + decorator. This is the primary execution path for every Language SDK. + +``taskflow-binding`` (MUST) + Arguments bound to a ``@task.stub`` target in the Python TaskFlow Dag — literal values + and upstream XCom results alike — are delivered to the task function's parameters at + runtime. + +Shared capabilities +~~~~~~~~~~~~~~~~~~~ + +These capabilities exist in both authoring modes, and an SDK may support them in one mode but not +the other, so each one is declared once per mode. In a Dag definition the tier applies only when +``dag-definition-authoring`` is supported. + +``mapped-tasks`` (MAY) + A task can be expanded at runtime via dynamic task mapping, and each mapped instance + receives its own arguments. + +``asset-inlets-outlets`` (MAY) + A task's asset inlets and outlets are honored, so a successful task emits events for its + outlets. + +``task-callbacks`` (MAY) + Task-level callbacks (``on_failure_callback`` and the rest) run for a task of this SDK. + The two authoring modes ask different things of the SDK: in a Dag definition the SDK + declares the callback as a function in its own language, while for a ``@task.stub`` task + the callback is declared in the Python Dag and the SDK's execution path has to get it run. Compatibility matrix ~~~~~~~~~~~~~~~~~~~~ The dimensions above are prose; each SDK also declares them *machine-readably* in one -hand-authored ``/capabilities.yaml``, from which a prek hook generates its published tables: +hand-authored ``/capabilities.yaml``, from which prek hooks generate the published tables: .. code-block:: text java-sdk/capabilities.yaml <- the only file you edit | - | hook: update-java-sdk-readme-matrix + +-- hook: update-java-sdk-readme-matrix + | +--> java-sdk/README.md (user-facing landing page) + | +--> java-sdk/sdk/module.md (Dokka -> the published API reference) | - +--> java-sdk/README.md (user-facing landing page) - +--> java-sdk/sdk/module.md (Dokka -> the published API reference) + +-- hook: update-lang-sdk-central-matrix (reads every SDK in LANG_SDKS) + +--> airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst + (consolidated cross-SDK matrix) -The hook rewrites its targets and exits non-zero when either was stale, so a drifted table fails +Each hook rewrites its targets and exits non-zero when a target was stale, so a drifted table fails the build. Keep the manifest out of whatever the SDK publishes — it describes the SDK rather than being part of it; for Java that means sitting above every subproject in ``settings.gradle.kts``. -To add an SDK, register it in ``LANG_SDKS`` in ``scripts/ci/prek/lang_sdk_compat_matrix.py``, write +A supported dimension may also record ``sdk_version``, the SDK's own release that shipped it, +alongside ``since`` (the Airflow version). It is optional: name the release that ships, or will +ship, the dimension, and omit it when that is not settled yet. A shared capability holds one +entry per authoring mode: + +.. code-block:: yaml + + mapped-tasks: + dag-definition: + supported: false + stub-task-handler: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + +To add an SDK, register it in ``LANG_SDKS`` (its ``id``, ``display_name`` and file locations) in +``scripts/ci/prek/lang_sdk_compat_matrix.py``, write a ``capabilities.yaml`` in the same schema, and add the equivalent hook. Adding or renaming a dimension means editing ``STATE_DIMENSIONS`` / ``CAPABILITY_DIMENSIONS`` there **and** the prose above in the same PR — the renderer validates every manifest against that list. diff --git a/go-sdk/README.md b/go-sdk/README.md index 4091be2599b..9411654c232 100644 --- a/go-sdk/README.md +++ b/go-sdk/README.md @@ -117,46 +117,62 @@ prek hook regenerate it. -*Min. Airflow version: 3.4 · supervisor schema: 2026-10-30* +*Min. Airflow version: 3.4.0 · supervisor schema: 2026-10-30* -| Dimension | Tier | Supported | Since | Notes | -|---|---|---|---|---| -| **TaskInstance states** | | | | | -| state: `success` | MUST | ✓ | 3.3 | | -| state: `failed` | MUST | ✓ | 3.3 | | -| state: `up_for_retry` | MUST | ✓ | 3.3 | RetryTask | -| state: `skipped` | SHOULD | ✓ | 3.4 | airflow.TriggerDagRun with SkipWhenAlreadyExists only | -| state: `deferred` | MAY | ✓ | 3.4 | airflow.TriggerDagRun with Deferrable only; resumes in the Go runtime | -| state: `up_for_reschedule` | MAY | ✗ | – | runtime does not emit RescheduleTask yet | -| state: `awaiting_input` | MAY | ✗ | – | runtime does not emit AwaitInputTask yet | -| state: `removed` | MAY | ✓ | 3.3 | | -| **Runtime capabilities** | | | | | -| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.3 | @task.stub | -| capability: `taskflow-binding` | MUST | ✗ | – | bind @task.stub literal/XCom args to the native handler | -| capability: `task-logging` | MUST | ✓ | 3.3 | slog records streamed over the logs socket | -| capability: `xcom-read-write` | MUST | ✓ | 3.3 | PushXCom / GetXCom | -| capability: `connection-read` | MUST | ✓ | 3.3 | GetConnection | -| capability: `variable-read-write` | MUST | ✓ | 3.4 | | -| capability: `self-contained-bundle` | MUST | ✓ | 3.3 | AFBNDL01 native binary via airflow-go-pack | -| capability: `retry-policy` | MAY | ✗ | – | no task-facing retry-policy API yet | -| capability: `task-state-store` | MAY | ✓ | 3.4 | | -| capability: `asset-state-store` | MAY | ✗ | – | no task-facing state-store API yet | -| capability: `asset-event-emit` | MAY | ✗ | – | runtime does not emit asset events yet | -| capability: `asset-event-read` | MAY | ✗ | – | no task-facing asset-event API yet | -| **Native-Dag authoring** | | | | | -| capability: `native-dag-authoring` | SHOULD | ✗ | – | native Dag authoring not implemented yet | -| capability: `task-args` | MUST † | n/a | – | | -| capability: `dag-params` | MUST † | n/a | – | | -| capability: `taskflow-dependencies` | MUST † | n/a | – | | -| capability: `branching` | SHOULD † | n/a | – | | -| capability: `dag-test` | SHOULD † | n/a | – | | -| capability: `task-group` | MAY † | n/a | – | | -| capability: `dynamic-task-mapping` | MAY † | n/a | – | | -| capability: `asset-inlets-outlets` | MAY † | n/a | – | | -| capability: `asset-scheduling` | MAY † | n/a | – | | -| capability: `object-store` | MAY † | n/a | – | no object-storage API yet | +| Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes | +|---|---|---|---|---|---| +| **TaskInstance states** | | | | | | +| state: `success` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| state: `failed` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| state: `up_for_retry` | MUST | ✓ | 3.4.0 | 1.0.0 | RetryTask | +| state: `skipped` | SHOULD | ✓ | 3.4.0 | 1.0.0 | airflow.TriggerDagRun with SkipWhenAlreadyExists only | +| state: `deferred` | MAY | ✓ | 3.4.0 | 1.0.0 | airflow.TriggerDagRun with Deferrable only; resumes in the Go runtime | +| state: `up_for_reschedule` | MAY | ✗ | – | – | runtime does not emit RescheduleTask yet | +| state: `awaiting_input` | MAY | ✗ | – | – | runtime does not emit AwaitInputTask yet | +| state: `removed` | MAY | ✓ | 3.4.0 | 1.0.0 | | +| **Task runtime** | | | | | | +| capability: `task-logging` | MUST | ✓ | 3.4.0 | 1.0.0 | slog records streamed over the logs socket | +| capability: `xcom-read-write` | MUST | ✓ | 3.4.0 | 1.0.0 | PushXCom / GetXCom | +| capability: `connection-read` | MUST | ✓ | 3.4.0 | 1.0.0 | GetConnection | +| capability: `variable-read-write` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| capability: `self-contained-bundle` | MUST | ✓ | 3.4.0 | 1.0.0 | AFBNDL01 native binary via airflow-go-pack | +| capability: `retry-policy` | MAY | ✗ | – | – | no task-facing retry-policy API yet | +| capability: `task-state-store` | MAY | ✓ | 3.4.0 | 1.0.0 | | +| capability: `asset-state-store` | MAY | ✗ | – | – | no task-facing asset state-store API yet | +| capability: `asset-event-emit` | MAY | ✗ | – | – | runtime does not emit asset events yet | +| capability: `asset-event-read` | MAY | ✗ | – | – | no task-facing asset-event API yet | +| capability: `object-store` | MAY | ✗ | – | – | no object-storage API yet | +| capability: `listeners` | MAY | ✗ | – | – | no hook into the Python listener API, so a task emits no OpenLineage facets of its own | +| **Dag definition** | | | | | | +| capability: `dag-definition-authoring` | SHOULD | ✓ | 3.4.0 | 1.0.0 | airflow.Dag | +| capability: `dag-args` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-args` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `dag-params` | MUST † | ✗ | – | – | | +| capability: `dag-callbacks` | MAY † | ✗ | – | – | | +| capability: `explicit-dependencies` | MUST † | ✓ | 3.4.0 | 1.0.0 | Before / After | +| capability: `edge-labels` | MAY † | ✓ | 3.4.0 | 1.0.0 | airflow.Label(text).On(task) | +| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-group` | MAY † | ✓ | 3.4.0 | 1.0.0 | DagRef.TaskGroup | +| capability: `task-loops` | MAY † | ✗ | – | – | | +| capability: `if-else-branching` | SHOULD † | ✓ | 3.4.0 | 1.0.0 | DagRef.If | +| capability: `switch-case-branching` | SHOULD † | ✓ | 3.4.0 | 1.0.0 | DagRef.Switch | +| capability: `trigger-dag-run` | MAY † | ✓ | 3.4.0 | 1.0.0 | airflow.TriggerDagRun | +| capability: `dynamic-dag-generation` | MAY † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `human-in-the-loop` | MAY † | ✗ | – | – | | +| capability: `dag-test` | SHOULD † | ✗ | – | – | | +| capability: `asset-definition` | MAY † | ✗ | – | – | | +| capability: `asset-scheduling` | MAY † | ✗ | – | – | | +| capability: `mapped-tasks` | MAY † | ✗ | – | – | | +| capability: `asset-inlets-outlets` | MAY † | ✗ | – | – | | +| capability: `task-callbacks` | MAY † | ✗ | – | – | no SDK-defined task callbacks | +| **Python Dag with stub TaskHandler** | | | | | | +| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.4.0 | 1.0.0 | @task.stub | +| capability: `taskflow-binding` | MUST | ✓ | 3.4.0 | 1.0.0 | mapped stubs excluded | +| capability: `mapped-tasks` | MAY | ✗ | – | – | Airflow does not bind arguments to mapped stubs yet | +| capability: `asset-inlets-outlets` | MAY | ✗ | – | – | runtime does not report task outlets yet | +| capability: `task-callbacks` | MAY | ✗ | – | – | a Python-declared task callback does not run for a stub task | -*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.* +*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `dag-definition-authoring` is supported.* diff --git a/go-sdk/capabilities.yaml b/go-sdk/capabilities.yaml index 4527ffd1da1..95b39727d49 100644 --- a/go-sdk/capabilities.yaml +++ b/go-sdk/capabilities.yaml @@ -17,7 +17,7 @@ --- sdk: go -min_airflow_version: "3.4" +min_airflow_version: "3.4.0" # Keep in sync with SupervisorSchemaVersion in pkg/execution/messages.go, which is embedded in # bundle metadata. The render hook fails if the two disagree. @@ -29,21 +29,26 @@ supervisor_schema_version: "2026-10-30" states: success: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" failed: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" up_for_retry: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "RetryTask" skipped: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "airflow.TriggerDagRun with SkipWhenAlreadyExists only" deferred: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "airflow.TriggerDagRun with Deferrable only; resumes in the Go runtime" up_for_reschedule: supported: false @@ -53,73 +58,151 @@ states: note: "runtime does not emit AwaitInputTask yet" removed: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" -# Runtime capabilities reflect the task-facing CoordinatorClient surface; native-Dag authoring is -# not implemented yet, so every native capability is unsupported. +# Runtime capabilities apply to a task body in both authoring modes. capabilities: - mixed-lang-stub-target: - supported: true - since: "3.3" - note: "@task.stub" - taskflow-binding: - supported: false - note: "bind @task.stub literal/XCom args to the native handler" task-logging: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "slog records streamed over the logs socket" xcom-read-write: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "PushXCom / GetXCom" connection-read: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "GetConnection" variable-read-write: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" self-contained-bundle: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "AFBNDL01 native binary via airflow-go-pack" retry-policy: supported: false note: "no task-facing retry-policy API yet" task-state-store: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" asset-state-store: supported: false - note: "no task-facing state-store API yet" + note: "no task-facing asset state-store API yet" asset-event-emit: supported: false note: "runtime does not emit asset events yet" asset-event-read: supported: false note: "no task-facing asset-event API yet" - native-dag-authoring: - supported: false - note: "native Dag authoring not implemented yet" - task-args: - supported: false - dag-params: - supported: false - taskflow-dependencies: - supported: false - branching: - supported: false - dag-test: - supported: false - task-group: - supported: false - dynamic-task-mapping: - supported: false - asset-inlets-outlets: - supported: false - asset-scheduling: - supported: false object-store: supported: false note: "no object-storage API yet" + listeners: + supported: false + note: "no hook into the Python listener API, so a task emits no OpenLineage facets of its own" + # Dag definition: the whole Dag is declared in the SDK's language. + dag-definition-authoring: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "airflow.Dag" + dag-args: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + task-args: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + dag-params: + supported: false + dag-callbacks: + supported: false + explicit-dependencies: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "Before / After" + edge-labels: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "airflow.Label(text).On(task)" + taskflow-dependencies: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + task-group: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "DagRef.TaskGroup" + task-loops: + supported: false + if-else-branching: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "DagRef.If" + switch-case-branching: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "DagRef.Switch" + trigger-dag-run: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "airflow.TriggerDagRun" + dynamic-dag-generation: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + human-in-the-loop: + supported: false + dag-test: + supported: false + asset-definition: + supported: false + asset-scheduling: + supported: false + # Python Dag with stub TaskHandler. + mixed-lang-stub-target: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "@task.stub" + taskflow-binding: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "mapped stubs excluded" + # Shared by both authoring modes, so declared once per mode. + mapped-tasks: + dag-definition: + supported: false + stub-task-handler: + supported: false + note: "Airflow does not bind arguments to mapped stubs yet" + asset-inlets-outlets: + dag-definition: + supported: false + stub-task-handler: + supported: false + note: "runtime does not report task outlets yet" + task-callbacks: + dag-definition: + supported: false + note: "no SDK-defined task callbacks" + stub-task-handler: + supported: false + note: "a Python-declared task callback does not run for a stub task" diff --git a/java-sdk/README.md b/java-sdk/README.md index bcf73089cb0..e401378fb4e 100644 --- a/java-sdk/README.md +++ b/java-sdk/README.md @@ -149,46 +149,62 @@ prek hook regenerate it. -*Min. Airflow version: 3.3 · supervisor schema: 2026-10-30* +*Min. Airflow version: 3.4.0 · supervisor schema: 2026-10-30* -| Dimension | Tier | Supported | Since | Notes | -|---|---|---|---|---| -| **TaskInstance states** | | | | | -| state: `success` | MUST | ✓ | 3.3 | | -| state: `failed` | MUST | ✓ | 3.3 | | -| state: `up_for_retry` | MUST | ✓ | 3.3 | RetryTask | -| state: `skipped` | SHOULD | ✓ | 3.4 | a TriggerDagRun task that skips an existing run | -| state: `deferred` | MAY | ✓ | 3.4 | a TriggerDagRun task that waits with deferrable | -| state: `up_for_reschedule` | MAY | ✗ | – | runtime does not emit RescheduleTask yet | -| state: `awaiting_input` | MAY | ✗ | – | runtime does not emit AwaitInputTask yet | -| state: `removed` | MAY | ✓ | 3.3 | | -| **Runtime capabilities** | | | | | -| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.3 | @task.stub | -| capability: `taskflow-binding` | MUST | ✗ | – | bind @task.stub literal/XCom args to the native handler | -| capability: `task-logging` | MUST | ✓ | 3.3 | SLF4J + JPL bridged to the task log | -| capability: `xcom-read-write` | MUST | ✓ | 3.3 | | -| capability: `connection-read` | MUST | ✓ | 3.3 | | -| capability: `variable-read-write` | MUST | ✓ | 3.3 | | -| capability: `self-contained-bundle` | MUST | ✓ | 3.3 | Airflow metadata embedded in the jar artifact | -| capability: `retry-policy` | MAY | ✗ | – | no task-facing retry-policy API yet | -| capability: `task-state-store` | MAY | ✓ | 3.3 | Client.getTaskStateStore() get/set/delete/clear | -| capability: `asset-state-store` | MAY | ✗ | – | no task-facing state-store API yet | -| capability: `asset-event-emit` | MAY | ✗ | – | runtime does not emit asset events yet | -| capability: `asset-event-read` | MAY | ✗ | – | no task-facing asset-event API yet | -| **Native-Dag authoring** | | | | | -| capability: `native-dag-authoring` | SHOULD | ✓ | 3.4 | | -| capability: `task-args` | MUST † | ✓ | 3.4 | | -| capability: `dag-params` | MUST † | ✗ | – | | -| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4 | | -| capability: `branching` | SHOULD † | ✓ | 3.4 | DagDef.If and DagDef.Switch, or @Airflow.If and @Airflow.Switch | -| capability: `dag-test` | SHOULD † | ✗ | – | | -| capability: `task-group` | MAY † | ✓ | 3.4 | | -| capability: `dynamic-task-mapping` | MAY † | ✗ | – | | -| capability: `asset-inlets-outlets` | MAY † | ✗ | – | | -| capability: `asset-scheduling` | MAY † | ✗ | – | | -| capability: `object-store` | MAY † | ✗ | – | | +| Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes | +|---|---|---|---|---|---| +| **TaskInstance states** | | | | | | +| state: `success` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| state: `failed` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| state: `up_for_retry` | MUST | ✓ | 3.4.0 | 1.0.0 | RetryTask | +| state: `skipped` | SHOULD | ✓ | 3.4.0 | 1.0.0 | a TriggerDagRun task that skips an existing run | +| state: `deferred` | MAY | ✓ | 3.4.0 | 1.0.0 | a TriggerDagRun task that waits with deferrable | +| state: `up_for_reschedule` | MAY | ✗ | – | – | runtime does not emit RescheduleTask yet | +| state: `awaiting_input` | MAY | ✗ | – | – | runtime does not emit AwaitInputTask yet | +| state: `removed` | MAY | ✓ | 3.4.0 | 1.0.0 | | +| **Task runtime** | | | | | | +| capability: `task-logging` | MUST | ✓ | 3.4.0 | 1.0.0 | SLF4J + JPL bridged to the task log | +| capability: `xcom-read-write` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| capability: `connection-read` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| capability: `variable-read-write` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| capability: `self-contained-bundle` | MUST | ✓ | 3.4.0 | 1.0.0 | Airflow metadata embedded in the jar artifact | +| capability: `retry-policy` | MAY | ✗ | – | – | no task-facing retry-policy API yet | +| capability: `task-state-store` | MAY | ✓ | 3.4.0 | 1.0.0 | Client.getTaskStateStore() get/set/delete/clear | +| capability: `asset-state-store` | MAY | ✗ | – | – | no task-facing asset state-store API yet | +| capability: `asset-event-emit` | MAY | ✗ | – | – | runtime does not emit asset events yet | +| capability: `asset-event-read` | MAY | ✗ | – | – | no task-facing asset-event API yet | +| capability: `object-store` | MAY | ✗ | – | – | no object-storage API yet | +| capability: `listeners` | MAY | ✗ | – | – | no hook into the Python listener API, so a task emits no OpenLineage facets of its own | +| **Dag definition** | | | | | | +| capability: `dag-definition-authoring` | SHOULD | ✓ | 3.4.0 | 1.0.0 | | +| capability: `dag-args` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-args` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `dag-params` | MUST † | ✗ | – | – | | +| capability: `dag-callbacks` | MAY † | ✗ | – | – | | +| capability: `explicit-dependencies` | MUST † | ✓ | 3.4.0 | 1.0.0 | before / after, or @Airflow.Deps | +| capability: `edge-labels` | MAY † | ✓ | 3.4.0 | 1.0.0 | Label.of(text).on(task) | +| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-group` | MAY † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-loops` | MAY † | ✗ | – | – | | +| capability: `if-else-branching` | SHOULD † | ✓ | 3.4.0 | 1.0.0 | DagDef.If or @Airflow.If | +| capability: `switch-case-branching` | SHOULD † | ✓ | 3.4.0 | 1.0.0 | DagDef.Switch or @Airflow.Switch | +| capability: `trigger-dag-run` | MAY † | ✓ | 3.4.0 | 1.0.0 | TriggerDagRun | +| capability: `dynamic-dag-generation` | MAY † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `human-in-the-loop` | MAY † | ✗ | – | – | | +| capability: `dag-test` | SHOULD † | ✗ | – | – | | +| capability: `asset-definition` | MAY † | ✗ | – | – | | +| capability: `asset-scheduling` | MAY † | ✗ | – | – | | +| capability: `mapped-tasks` | MAY † | ✗ | – | – | | +| capability: `asset-inlets-outlets` | MAY † | ✗ | – | – | | +| capability: `task-callbacks` | MAY † | ✗ | – | – | no SDK-defined task callbacks | +| **Python Dag with stub TaskHandler** | | | | | | +| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.4.0 | 1.0.0 | @task.stub | +| capability: `taskflow-binding` | MUST | ✓ | 3.4.0 | 1.0.0 | mapped stubs excluded | +| capability: `mapped-tasks` | MAY | ✗ | – | – | Airflow does not bind arguments to mapped stubs yet | +| capability: `asset-inlets-outlets` | MAY | ✗ | – | – | runtime does not report task outlets yet | +| capability: `task-callbacks` | MAY | ✗ | – | – | a Python-declared task callback does not run for a stub task | -*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.* +*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `dag-definition-authoring` is supported.* diff --git a/java-sdk/capabilities.yaml b/java-sdk/capabilities.yaml index cf9ec1c5fa7..270aed68b7c 100644 --- a/java-sdk/capabilities.yaml +++ b/java-sdk/capabilities.yaml @@ -17,7 +17,7 @@ --- sdk: java -min_airflow_version: "3.3" +min_airflow_version: "3.4.0" # Keep in sync with airflowSupervisorSchemaVersion in gradle.properties, which is what stamps the # JAR manifest. The render hook fails if the two disagree. @@ -28,21 +28,26 @@ supervisor_schema_version: "2026-10-30" states: success: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" failed: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" up_for_retry: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "RetryTask" skipped: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "a TriggerDagRun task that skips an existing run" deferred: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "a TriggerDagRun task that waits with deferrable" up_for_reschedule: supported: false @@ -52,76 +57,148 @@ states: note: "runtime does not emit AwaitInputTask yet" removed: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" -# Runtime capabilities reflect the task-facing Client surface. Native-Dag capabilities describe -# what a Java-authored Dag can declare and deliver through Dag serialization. +# Runtime capabilities apply to a task body in both authoring modes. capabilities: - mixed-lang-stub-target: - supported: true - since: "3.3" - note: "@task.stub" - taskflow-binding: - supported: false - note: "bind @task.stub literal/XCom args to the native handler" task-logging: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "SLF4J + JPL bridged to the task log" xcom-read-write: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" connection-read: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" variable-read-write: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" self-contained-bundle: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "Airflow metadata embedded in the jar artifact" retry-policy: supported: false note: "no task-facing retry-policy API yet" task-state-store: supported: true - since: "3.3" + since: "3.4.0" + sdk_version: "1.0.0" note: "Client.getTaskStateStore() get/set/delete/clear" asset-state-store: supported: false - note: "no task-facing state-store API yet" + note: "no task-facing asset state-store API yet" asset-event-emit: supported: false note: "runtime does not emit asset events yet" asset-event-read: supported: false note: "no task-facing asset-event API yet" - native-dag-authoring: + object-store: + supported: false + note: "no object-storage API yet" + listeners: + supported: false + note: "no hook into the Python listener API, so a task emits no OpenLineage facets of its own" + # Dag definition: the whole Dag is declared in the SDK's language. + dag-definition-authoring: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" + dag-args: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" task-args: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" dag-params: supported: false + dag-callbacks: + supported: false + explicit-dependencies: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "before / after, or @Airflow.Deps" + edge-labels: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "Label.of(text).on(task)" taskflow-dependencies: supported: true - since: "3.4" - branching: - supported: true - since: "3.4" - note: "DagDef.If and DagDef.Switch, or @Airflow.If and @Airflow.Switch" - dag-test: - supported: false + since: "3.4.0" + sdk_version: "1.0.0" task-group: supported: true - since: "3.4" - dynamic-task-mapping: + since: "3.4.0" + sdk_version: "1.0.0" + task-loops: supported: false - asset-inlets-outlets: + if-else-branching: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "DagDef.If or @Airflow.If" + switch-case-branching: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "DagDef.Switch or @Airflow.Switch" + trigger-dag-run: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "TriggerDagRun" + dynamic-dag-generation: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + human-in-the-loop: + supported: false + dag-test: + supported: false + asset-definition: supported: false asset-scheduling: supported: false - object-store: - supported: false + # Python Dag with stub TaskHandler. + mixed-lang-stub-target: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "@task.stub" + taskflow-binding: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "mapped stubs excluded" + # Shared by both authoring modes, so declared once per mode. + mapped-tasks: + dag-definition: + supported: false + stub-task-handler: + supported: false + note: "Airflow does not bind arguments to mapped stubs yet" + asset-inlets-outlets: + dag-definition: + supported: false + stub-task-handler: + supported: false + note: "runtime does not report task outlets yet" + task-callbacks: + dag-definition: + supported: false + note: "no SDK-defined task callbacks" + stub-task-handler: + supported: false + note: "a Python-declared task callback does not run for a stub task" diff --git a/java-sdk/sdk/module.md b/java-sdk/sdk/module.md index 47fad6d1803..3971e49dd13 100644 --- a/java-sdk/sdk/module.md +++ b/java-sdk/sdk/module.md @@ -108,45 +108,61 @@ meaning of each dimension is defined in the -*Min. Airflow version: 3.3 · supervisor schema: 2026-10-30* +*Min. Airflow version: 3.4.0 · supervisor schema: 2026-10-30* -| Dimension | Tier | Supported | Since | Notes | -|---|---|---|---|---| -| **TaskInstance states** | | | | | -| state: `success` | MUST | ✓ | 3.3 | | -| state: `failed` | MUST | ✓ | 3.3 | | -| state: `up_for_retry` | MUST | ✓ | 3.3 | RetryTask | -| state: `skipped` | SHOULD | ✓ | 3.4 | a TriggerDagRun task that skips an existing run | -| state: `deferred` | MAY | ✓ | 3.4 | a TriggerDagRun task that waits with deferrable | -| state: `up_for_reschedule` | MAY | ✗ | – | runtime does not emit RescheduleTask yet | -| state: `awaiting_input` | MAY | ✗ | – | runtime does not emit AwaitInputTask yet | -| state: `removed` | MAY | ✓ | 3.3 | | -| **Runtime capabilities** | | | | | -| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.3 | @task.stub | -| capability: `taskflow-binding` | MUST | ✗ | – | bind @task.stub literal/XCom args to the native handler | -| capability: `task-logging` | MUST | ✓ | 3.3 | SLF4J + JPL bridged to the task log | -| capability: `xcom-read-write` | MUST | ✓ | 3.3 | | -| capability: `connection-read` | MUST | ✓ | 3.3 | | -| capability: `variable-read-write` | MUST | ✓ | 3.3 | | -| capability: `self-contained-bundle` | MUST | ✓ | 3.3 | Airflow metadata embedded in the jar artifact | -| capability: `retry-policy` | MAY | ✗ | – | no task-facing retry-policy API yet | -| capability: `task-state-store` | MAY | ✓ | 3.3 | Client.getTaskStateStore() get/set/delete/clear | -| capability: `asset-state-store` | MAY | ✗ | – | no task-facing state-store API yet | -| capability: `asset-event-emit` | MAY | ✗ | – | runtime does not emit asset events yet | -| capability: `asset-event-read` | MAY | ✗ | – | no task-facing asset-event API yet | -| **Native-Dag authoring** | | | | | -| capability: `native-dag-authoring` | SHOULD | ✓ | 3.4 | | -| capability: `task-args` | MUST † | ✓ | 3.4 | | -| capability: `dag-params` | MUST † | ✗ | – | | -| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4 | | -| capability: `branching` | SHOULD † | ✓ | 3.4 | DagDef.If and DagDef.Switch, or @Airflow.If and @Airflow.Switch | -| capability: `dag-test` | SHOULD † | ✗ | – | | -| capability: `task-group` | MAY † | ✓ | 3.4 | | -| capability: `dynamic-task-mapping` | MAY † | ✗ | – | | -| capability: `asset-inlets-outlets` | MAY † | ✗ | – | | -| capability: `asset-scheduling` | MAY † | ✗ | – | | -| capability: `object-store` | MAY † | ✗ | – | | +| Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes | +|---|---|---|---|---|---| +| **TaskInstance states** | | | | | | +| state: `success` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| state: `failed` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| state: `up_for_retry` | MUST | ✓ | 3.4.0 | 1.0.0 | RetryTask | +| state: `skipped` | SHOULD | ✓ | 3.4.0 | 1.0.0 | a TriggerDagRun task that skips an existing run | +| state: `deferred` | MAY | ✓ | 3.4.0 | 1.0.0 | a TriggerDagRun task that waits with deferrable | +| state: `up_for_reschedule` | MAY | ✗ | – | – | runtime does not emit RescheduleTask yet | +| state: `awaiting_input` | MAY | ✗ | – | – | runtime does not emit AwaitInputTask yet | +| state: `removed` | MAY | ✓ | 3.4.0 | 1.0.0 | | +| **Task runtime** | | | | | | +| capability: `task-logging` | MUST | ✓ | 3.4.0 | 1.0.0 | SLF4J + JPL bridged to the task log | +| capability: `xcom-read-write` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| capability: `connection-read` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| capability: `variable-read-write` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| capability: `self-contained-bundle` | MUST | ✓ | 3.4.0 | 1.0.0 | Airflow metadata embedded in the jar artifact | +| capability: `retry-policy` | MAY | ✗ | – | – | no task-facing retry-policy API yet | +| capability: `task-state-store` | MAY | ✓ | 3.4.0 | 1.0.0 | Client.getTaskStateStore() get/set/delete/clear | +| capability: `asset-state-store` | MAY | ✗ | – | – | no task-facing asset state-store API yet | +| capability: `asset-event-emit` | MAY | ✗ | – | – | runtime does not emit asset events yet | +| capability: `asset-event-read` | MAY | ✗ | – | – | no task-facing asset-event API yet | +| capability: `object-store` | MAY | ✗ | – | – | no object-storage API yet | +| capability: `listeners` | MAY | ✗ | – | – | no hook into the Python listener API, so a task emits no OpenLineage facets of its own | +| **Dag definition** | | | | | | +| capability: `dag-definition-authoring` | SHOULD | ✓ | 3.4.0 | 1.0.0 | | +| capability: `dag-args` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-args` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `dag-params` | MUST † | ✗ | – | – | | +| capability: `dag-callbacks` | MAY † | ✗ | – | – | | +| capability: `explicit-dependencies` | MUST † | ✓ | 3.4.0 | 1.0.0 | before / after, or @Airflow.Deps | +| capability: `edge-labels` | MAY † | ✓ | 3.4.0 | 1.0.0 | Label.of(text).on(task) | +| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-group` | MAY † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-loops` | MAY † | ✗ | – | – | | +| capability: `if-else-branching` | SHOULD † | ✓ | 3.4.0 | 1.0.0 | DagDef.If or @Airflow.If | +| capability: `switch-case-branching` | SHOULD † | ✓ | 3.4.0 | 1.0.0 | DagDef.Switch or @Airflow.Switch | +| capability: `trigger-dag-run` | MAY † | ✓ | 3.4.0 | 1.0.0 | TriggerDagRun | +| capability: `dynamic-dag-generation` | MAY † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `human-in-the-loop` | MAY † | ✗ | – | – | | +| capability: `dag-test` | SHOULD † | ✗ | – | – | | +| capability: `asset-definition` | MAY † | ✗ | – | – | | +| capability: `asset-scheduling` | MAY † | ✗ | – | – | | +| capability: `mapped-tasks` | MAY † | ✗ | – | – | | +| capability: `asset-inlets-outlets` | MAY † | ✗ | – | – | | +| capability: `task-callbacks` | MAY † | ✗ | – | – | no SDK-defined task callbacks | +| **Python Dag with stub TaskHandler** | | | | | | +| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.4.0 | 1.0.0 | @task.stub | +| capability: `taskflow-binding` | MUST | ✓ | 3.4.0 | 1.0.0 | mapped stubs excluded | +| capability: `mapped-tasks` | MAY | ✗ | – | – | Airflow does not bind arguments to mapped stubs yet | +| capability: `asset-inlets-outlets` | MAY | ✗ | – | – | runtime does not report task outlets yet | +| capability: `task-callbacks` | MAY | ✗ | – | – | a Python-declared task callback does not run for a stub task | -*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.* +*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `dag-definition-authoring` is supported.* diff --git a/scripts/ci/prek/lang_sdk_compat_matrix.py b/scripts/ci/prek/lang_sdk_compat_matrix.py index c9003354c56..8e7c743ec24 100644 --- a/scripts/ci/prek/lang_sdk_compat_matrix.py +++ b/scripts/ci/prek/lang_sdk_compat_matrix.py @@ -21,7 +21,7 @@ Every Language SDK (Go, Java, TypeScript) declares what it supports in a hand-au file, the *registry* of SDKs, and :func:`render_markdown_table`, which renders the per-SDK Markdown table that each SDK's own prek hook embeds in its docs. -Because the manifest is hand-authored, :func:`validate_capabilities` is the only thing standing +Because the manifest is hand-authored, :func:`parse_capabilities` is the only thing standing between a typo and a wrong published table, so it rejects unknown keys as well as missing ones. The normative meaning of each dimension lives in ``contributing-docs/30_new_language_sdk.rst`` @@ -31,7 +31,7 @@ The normative meaning of each dimension lives in ``contributing-docs/30_new_lang from __future__ import annotations from pathlib import Path -from typing import NamedTuple, TypedDict +from typing import Final, Literal, NamedTuple, TypedDict import yaml from common_prek_utils import AIRFLOW_ROOT_PATH @@ -43,17 +43,30 @@ README_MATRIX_FOOTER = "" SUPPORTED_MARK = "✓" UNSUPPORTED_MARK = "✗" -NA_MARK = "n/a" # a gated native-Dag capability while native-dag-authoring is unsupported +NA_MARK = "n/a" # a gated Dag-definition capability while dag-definition itself is unsupported NO_VERSION_MARK = "–" # "Since" placeholder for a dimension that is not supported -# The umbrella capability that gates the conditional native-Dag capabilities: when an SDK does -# not support it, every gated native-Dag capability is "not applicable" rather than unsupported. -NATIVE_DAG_GATE = "native-dag-authoring" +# An authoring mode: a Dag declared in the SDK's own language, or a stub TaskHandler implementing a +# task of a Python Dag. A shared capability applies to both and is declared once per mode, as +# ``: {dag-definition: , stub-task-handler: }``. +Mode = Literal["dag-definition", "stub-task-handler"] +RowKind = Literal["state", "capability"] +Tier = Literal["MUST", "SHOULD", "MAY"] +Group = Literal["runtime", "dag-definition", "stub-task-handler", "shared"] +EntryKey = tuple[RowKind, str, Mode | None] + +DAG_DEFINITION_MODE: Final[Mode] = "dag-definition" +STUB_TASK_HANDLER_MODE: Final[Mode] = "stub-task-handler" +AUTHORING_MODES: Final[tuple[Mode, ...]] = (DAG_DEFINITION_MODE, STUB_TASK_HANDLER_MODE) + +# The umbrella capability that gates the conditional Dag-definition capabilities: when an SDK does +# not support it, every gated capability is "not applicable" rather than unsupported. +DAG_DEFINITION_GATE: Final = "dag-definition-authoring" # TaskInstance states a subprocess can emit, in display order, with their conformance tier. # Scheduler-owned states (queued, scheduled, running, restarting, upstream_failed) are never # emitted by an SDK runtime and are deliberately excluded. -STATE_DIMENSIONS: list[tuple[str, str]] = [ +STATE_DIMENSIONS: list[tuple[str, Tier]] = [ ("success", "MUST"), ("failed", "MUST"), ("up_for_retry", "MUST"), @@ -67,17 +80,16 @@ STATE_DIMENSIONS: list[tuple[str, str]] = [ class Capability(NamedTuple): name: str - tier: str - group: str # "runtime" or "native" - gated: bool # renders n/a (not ✗) when NATIVE_DAG_GATE is unsupported + tier: Tier + group: Group + gated: bool # its Dag-definition entry renders n/a (not ✗) when DAG_DEFINITION_GATE is unsupported -# Capability flags, in display order. Runtime capabilities describe what a task body can do while -# it runs (in either a mixed-lang or native Dag); native-Dag capabilities describe authoring a -# whole Dag in the target language and are gated by NATIVE_DAG_GATE (except the gate itself). +# Capability flags, in display order. Runtime capabilities describe what a task body can do in +# either authoring mode; Dag-definition capabilities describe authoring a whole Dag in the target +# language; stub-task-handler capabilities describe running @task.stub tasks of a Python Dag; +# shared capabilities exist in both authoring modes and are supported per mode. CAPABILITY_DIMENSIONS: list[Capability] = [ - Capability("mixed-lang-stub-target", "MUST", "runtime", False), - Capability("taskflow-binding", "MUST", "runtime", False), Capability("task-logging", "MUST", "runtime", False), Capability("xcom-read-write", "MUST", "runtime", False), Capability("connection-read", "MUST", "runtime", False), @@ -88,54 +100,149 @@ CAPABILITY_DIMENSIONS: list[Capability] = [ Capability("asset-state-store", "MAY", "runtime", False), Capability("asset-event-emit", "MAY", "runtime", False), Capability("asset-event-read", "MAY", "runtime", False), - Capability(NATIVE_DAG_GATE, "SHOULD", "native", False), - Capability("task-args", "MUST", "native", True), - Capability("dag-params", "MUST", "native", True), - Capability("taskflow-dependencies", "MUST", "native", True), - Capability("branching", "SHOULD", "native", True), - Capability("dag-test", "SHOULD", "native", True), - Capability("task-group", "MAY", "native", True), - Capability("dynamic-task-mapping", "MAY", "native", True), - Capability("asset-inlets-outlets", "MAY", "native", True), - Capability("asset-scheduling", "MAY", "native", True), - Capability("object-store", "MAY", "native", True), + Capability("object-store", "MAY", "runtime", False), + Capability("listeners", "MAY", "runtime", False), + Capability(DAG_DEFINITION_GATE, "SHOULD", "dag-definition", False), + Capability("dag-args", "MUST", "dag-definition", True), + Capability("task-args", "MUST", "dag-definition", True), + Capability("dag-params", "MUST", "dag-definition", True), + Capability("dag-callbacks", "MAY", "dag-definition", True), + Capability("explicit-dependencies", "MUST", "dag-definition", True), + Capability("edge-labels", "MAY", "dag-definition", True), + Capability("taskflow-dependencies", "MUST", "dag-definition", True), + Capability("task-group", "MAY", "dag-definition", True), + Capability("task-loops", "MAY", "dag-definition", True), + Capability("if-else-branching", "SHOULD", "dag-definition", True), + Capability("switch-case-branching", "SHOULD", "dag-definition", True), + Capability("trigger-dag-run", "MAY", "dag-definition", True), + Capability("dynamic-dag-generation", "MAY", "dag-definition", True), + Capability("human-in-the-loop", "MAY", "dag-definition", True), + Capability("dag-test", "SHOULD", "dag-definition", True), + Capability("asset-definition", "MAY", "dag-definition", True), + Capability("asset-scheduling", "MAY", "dag-definition", True), + Capability("mixed-lang-stub-target", "MUST", "stub-task-handler", False), + Capability("taskflow-binding", "MUST", "stub-task-handler", False), + Capability("mapped-tasks", "MAY", "shared", True), + Capability("asset-inlets-outlets", "MAY", "shared", True), + Capability("task-callbacks", "MAY", "shared", True), ] CAPABILITY_NAMES = {cap.name for cap in CAPABILITY_DIMENSIONS} - -GROUP_LABELS = {"runtime": "Runtime capabilities", "native": "Native-Dag authoring"} -STATES_GROUP_LABEL = "TaskInstance states" +SHARED_CAPABILITY_NAMES = {cap.name for cap in CAPABILITY_DIMENSIONS if cap.group == "shared"} LEGEND = ( f"Marks: {SUPPORTED_MARK} supported · {UNSUPPORTED_MARK} not supported · " - f"{NA_MARK} not applicable. A tier marked † applies only when `{NATIVE_DAG_GATE}` is supported." + f"{NA_MARK} not applicable. A tier marked † applies only when `{DAG_DEFINITION_GATE}` is supported." ) +class MatrixRow(NamedTuple): + kind: RowKind + name: str + tier: Tier + gated: bool + mode: Mode | None # the authoring mode whose entry a shared capability shows + + @property + def entry_key(self) -> EntryKey: + return self.kind, self.name, self.mode + + +GATE_KEY: Final[EntryKey] = ("capability", DAG_DEFINITION_GATE, None) + + +class MatrixGroup(NamedTuple): + label: str + rows: list[MatrixRow] + + +class MatrixPart(NamedTuple): + title: str + groups: list[MatrixGroup] + + +class MatrixParts(NamedTuple): + common: MatrixPart + dag_definition: MatrixPart + stub_task_handler: MatrixPart + + +def _build_capability_rows(groups: tuple[Group, ...], mode: Mode | None) -> list[MatrixRow]: + return [ + MatrixRow( + "capability", + cap.name, + cap.tier, + cap.gated and mode != STUB_TASK_HANDLER_MODE, + mode if cap.group == "shared" else None, + ) + for group in groups + for cap in CAPABILITY_DIMENSIONS + if cap.group == group + ] + + +def build_matrix_parts() -> MatrixParts: + """Split the dimensions into the parts both renderers publish, in display order.""" + return MatrixParts( + MatrixPart( + "Common to both authoring modes", + [ + MatrixGroup( + "TaskInstance states", + [MatrixRow("state", state, tier, False, None) for state, tier in STATE_DIMENSIONS], + ), + MatrixGroup("Task runtime", _build_capability_rows(("runtime",), None)), + ], + ), + MatrixPart( + "Dag definition", + [ + MatrixGroup( + "Dag definition", + _build_capability_rows(("dag-definition", "shared"), DAG_DEFINITION_MODE), + ) + ], + ), + MatrixPart( + "Python Dag with stub TaskHandler", + [ + MatrixGroup( + "Python Dag with stub TaskHandler", + _build_capability_rows(("stub-task-handler", "shared"), STUB_TASK_HANDLER_MODE), + ) + ], + ), + ) + + class LangSdk(TypedDict): id: str + display_name: str capabilities_yaml: Path readme: Path -# The registry of Language SDKs and where each one's manifest and README live. The Go and Java SDKs -# declare manifests; the TypeScript entry records where its manifest goes when that runtime declares -# its capabilities. Because of that, `capabilities_yaml` is a declared location and not a promise the -# file exists — a consumer walking the whole registry must check `.exists()` before calling -# load_capabilities(). +# The registry of Language SDKs and where each one's manifest and README live. Because an SDK may be +# registered before it declares its capabilities, `capabilities_yaml` is a declared location and not +# a promise the file exists — a consumer walking the whole registry must check `.exists()` before +# calling load_capabilities(). LANG_SDKS: list[LangSdk] = [ { "id": "go", + "display_name": "Go", "capabilities_yaml": AIRFLOW_ROOT_PATH / "go-sdk" / "capabilities.yaml", "readme": AIRFLOW_ROOT_PATH / "go-sdk" / "README.md", }, { "id": "java", + "display_name": "Java", "capabilities_yaml": AIRFLOW_ROOT_PATH / "java-sdk" / "capabilities.yaml", "readme": AIRFLOW_ROOT_PATH / "java-sdk" / "README.md", }, { "id": "ts", + "display_name": "TypeScript", "capabilities_yaml": AIRFLOW_ROOT_PATH / "ts-sdk" / "capabilities.yaml", "readme": AIRFLOW_ROOT_PATH / "ts-sdk" / "README.md", }, @@ -147,34 +254,37 @@ VALID_SDK_IDS = {sdk["id"] for sdk in LANG_SDKS} class DimensionEntry(TypedDict, total=False): supported: bool since: str | None + sdk_version: str | None note: str -class CapabilitiesDoc(TypedDict): +class Capabilities(NamedTuple): + """A validated manifest, with every dimension resolved to the entry its matrix row shows.""" + sdk: str supervisor_schema_version: str min_airflow_version: str - states: dict[str, DimensionEntry] - capabilities: dict[str, DimensionEntry] + entries: dict[EntryKey, DimensionEntry] + + def get_entry(self, row: MatrixRow) -> DimensionEntry: + return self.entries[row.entry_key] class CapabilitiesError(ValueError): """Raised when a capabilities.yaml file does not match the expected schema.""" -def load_capabilities(path: Path, *, expected_sdk: str | None = None) -> CapabilitiesDoc: +def load_capabilities(path: Path, *, expected_sdk: str | None = None) -> Capabilities: """Load and validate a ``capabilities.yaml`` file. ``expected_sdk`` binds the file to the SDK it belongs to: passing it makes a manifest whose ``sdk`` field disagrees with the file's own SDK (e.g. ``go-sdk/capabilities.yaml`` declaring ``sdk: java``) a validation error instead of silently rendering in the wrong column. """ - doc = yaml.safe_load(path.read_text()) - validate_capabilities(doc, source=str(path), expected_sdk=expected_sdk) - return doc + return parse_capabilities(yaml.safe_load(path.read_text()), source=str(path), expected_sdk=expected_sdk) -def validate_capabilities(doc: object, *, source: str, expected_sdk: str | None = None) -> None: +def parse_capabilities(doc: object, *, source: str, expected_sdk: str | None = None) -> Capabilities: """Validate a decoded capabilities document, raising :class:`CapabilitiesError` on any issue. When ``expected_sdk`` is given, the document's ``sdk`` field must equal it. @@ -197,24 +307,45 @@ def validate_capabilities(doc: object, *, source: str, expected_sdk: str | None for field in ("supervisor_schema_version", "min_airflow_version"): if not isinstance(doc[field], str): raise CapabilitiesError(f"{source}: {field} must be a string") - _validate_entries( + _validate_keys( doc["states"], expected={state for state, _ in STATE_DIMENSIONS}, kind="states", source=source ) - _validate_entries(doc["capabilities"], expected=CAPABILITY_NAMES, kind="capabilities", source=source) - if not doc["capabilities"][NATIVE_DAG_GATE]["supported"]: + for name, entry in doc["states"].items(): + _validate_entry(entry, label=f"states.{name}", source=source) + capabilities = doc["capabilities"] + _validate_keys(capabilities, expected=CAPABILITY_NAMES, kind="capabilities", source=source) + entries: dict[EntryKey, DimensionEntry] = { + ("state", name, None): entry for name, entry in doc["states"].items() + } + for name, entry in capabilities.items(): + if name in SHARED_CAPABILITY_NAMES: + _validate_keys(entry, expected=set(AUTHORING_MODES), kind=f"capabilities.{name}", source=source) + for mode in AUTHORING_MODES: + _validate_entry(entry[mode], label=f"capabilities.{name}.{mode}", source=source) + entries["capability", name, mode] = entry[mode] + else: + _validate_entry(entry, label=f"capabilities.{name}", source=source) + entries["capability", name, None] = entry + if not entries[GATE_KEY]["supported"]: supported_gated = sorted( - cap.name - for cap in CAPABILITY_DIMENSIONS - if cap.gated and doc["capabilities"][cap.name]["supported"] + row.name + for row in _build_capability_rows(("dag-definition", "shared"), DAG_DEFINITION_MODE) + if row.gated and entries[row.entry_key]["supported"] ) if supported_gated: raise CapabilitiesError( - f"{source}: gated capabilities cannot be supported while {NATIVE_DAG_GATE!r} " + f"{source}: gated capabilities cannot be supported while {DAG_DEFINITION_GATE!r} " f"is not supported: {', '.join(supported_gated)}" ) + return Capabilities( + sdk=doc["sdk"], + supervisor_schema_version=doc["supervisor_schema_version"], + min_airflow_version=doc["min_airflow_version"], + entries=entries, + ) -def _validate_entries(entries: object, *, expected: set[str], kind: str, source: str) -> None: +def _validate_keys(entries: object, *, expected: set[str], kind: str, source: str) -> None: if not isinstance(entries, dict): raise CapabilitiesError(f"{source}: {kind!r} must be a mapping") actual = set(entries.keys()) @@ -227,79 +358,70 @@ def _validate_entries(entries: object, *, expected: set[str], kind: str, source: if unknown: problems.append(f"unknown {sorted(unknown)}") raise CapabilitiesError(f"{source}: {kind} keys mismatch: {'; '.join(problems)}") - for name, entry in entries.items(): - if not isinstance(entry, dict) or not isinstance(entry.get("supported"), bool): - raise CapabilitiesError(f"{source}: {kind}.{name} must be a mapping with a boolean 'supported'") - # A misspelled optional key would otherwise be dropped silently and render as a blank cell. - unknown_fields = entry.keys() - {"supported", "since", "note"} - if unknown_fields: - raise CapabilitiesError( - f"{source}: {kind}.{name} has unknown keys: {', '.join(sorted(unknown_fields))}" - ) - if not isinstance(entry.get("since", None), (str, type(None))): - raise CapabilitiesError(f"{source}: {kind}.{name}.since must be a string or null") - if not entry["supported"] and entry.get("since") is not None: - # "Since" means "supported since"; carrying one while unsupported is contradictory and + + +def _validate_entry(entry: object, *, label: str, source: str) -> None: + if not isinstance(entry, dict) or not isinstance(entry.get("supported"), bool): + raise CapabilitiesError(f"{source}: {label} must be a mapping with a boolean 'supported'") + # A misspelled optional key would otherwise be dropped silently and render as a blank cell. + unknown_fields = entry.keys() - {"supported", "since", "sdk_version", "note"} + if unknown_fields: + raise CapabilitiesError(f"{source}: {label} has unknown keys: {', '.join(sorted(unknown_fields))}") + for field in ("since", "sdk_version"): + if not isinstance(entry.get(field, None), (str, type(None))): + raise CapabilitiesError(f"{source}: {label}.{field} must be a string or null") + if not entry["supported"] and entry.get(field) is not None: + # A version means "supported since"; carrying one while unsupported is contradictory and # would silently render as the not-supported placeholder. Supported *without* a version # stays legal — an SDK may not know which release first shipped a dimension. raise CapabilitiesError( - f"{source}: {kind}.{name} is not supported but carries since=" - f"{entry['since']!r}; drop the version or mark it supported" + f"{source}: {label} is not supported but carries {field}=" + f"{entry[field]!r}; drop the version or mark it supported" ) - if not isinstance(entry.get("note", ""), str): - raise CapabilitiesError(f"{source}: {kind}.{name}.note must be a string") + if not isinstance(entry.get("note", ""), str): + raise CapabilitiesError(f"{source}: {label}.note must be a string") -def _state_mark(entry: DimensionEntry) -> str: - return SUPPORTED_MARK if entry.get("supported") else UNSUPPORTED_MARK - - -def _capability_mark(doc: CapabilitiesDoc, cap: Capability) -> str: - if cap.gated and not doc["capabilities"][NATIVE_DAG_GATE].get("supported"): +def get_mark(doc: Capabilities, row: MatrixRow) -> str: + if row.gated and not doc.entries[GATE_KEY]["supported"]: return NA_MARK - return SUPPORTED_MARK if doc["capabilities"][cap.name].get("supported") else UNSUPPORTED_MARK + return SUPPORTED_MARK if doc.get_entry(row)["supported"] else UNSUPPORTED_MARK -def _since(entry: DimensionEntry) -> str: +def format_tier(row: MatrixRow) -> str: + return f"{row.tier} †" if row.gated else row.tier + + +def _format_version(entry: DimensionEntry, field: Literal["since", "sdk_version"]) -> str: if not entry.get("supported"): return NO_VERSION_MARK - return entry.get("since") or NO_VERSION_MARK + return entry.get(field) or NO_VERSION_MARK -def _note(entry: DimensionEntry) -> str: +def _format_note(entry: DimensionEntry) -> str: return (entry.get("note") or "").replace("|", "\\|") -def _tier_label(cap: Capability) -> str: - return f"{cap.tier} †" if cap.gated else cap.tier - - -def render_markdown_table(doc: CapabilitiesDoc) -> list[str]: +def render_markdown_table(doc: Capabilities) -> list[str]: """Render the per-SDK Markdown compatibility table as a list of lines (trailing newlines).""" lines = [ "\n", - f"*Min. Airflow version: {doc['min_airflow_version']} · " - f"supervisor schema: {doc['supervisor_schema_version']}*\n", + f"*Min. Airflow version: {doc.min_airflow_version} · " + f"supervisor schema: {doc.supervisor_schema_version}*\n", "\n", - "| Dimension | Tier | Supported | Since | Notes |\n", - "|---|---|---|---|---|\n", - f"| **{STATES_GROUP_LABEL}** | | | | |\n", + "| Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes |\n", + "|---|---|---|---|---|---|\n", ] - for state, tier in STATE_DIMENSIONS: - entry = doc["states"][state] - lines.append( - f"| state: `{state}` | {tier} | {_state_mark(entry)} | {_since(entry)} | {_note(entry)} |\n" - ) - current_group = "" - for cap in CAPABILITY_DIMENSIONS: - if cap.group != current_group: - current_group = cap.group - lines.append(f"| **{GROUP_LABELS[cap.group]}** | | | | |\n") - entry = doc["capabilities"][cap.name] - lines.append( - f"| capability: `{cap.name}` | {_tier_label(cap)} | {_capability_mark(doc, cap)} | " - f"{_since(entry)} | {_note(entry)} |\n" - ) + for part in build_matrix_parts(): + for group in part.groups: + lines.append(f"| **{group.label}** | | | | | |\n") + for row in group.rows: + entry = doc.get_entry(row) + lines.append( + f"| {row.kind}: `{row.name}` | {format_tier(row)} | {get_mark(doc, row)} | " + f"{_format_version(entry, 'since')} | {_format_version(entry, 'sdk_version')} | " + f"{_format_note(entry)} |\n" + ) lines.append("\n") lines.append(f"*{LEGEND}*\n") lines.append("\n") diff --git a/scripts/ci/prek/update_go_sdk_readme_matrix.py b/scripts/ci/prek/update_go_sdk_readme_matrix.py index 25a33b7f60d..33c7b1fa653 100755 --- a/scripts/ci/prek/update_go_sdk_readme_matrix.py +++ b/scripts/ci/prek/update_go_sdk_readme_matrix.py @@ -40,7 +40,7 @@ from lang_sdk_compat_matrix import ( LANG_SDKS, README_MATRIX_FOOTER, README_MATRIX_HEADER, - CapabilitiesDoc, + Capabilities, load_capabilities, render_markdown_table, ) @@ -60,7 +60,7 @@ def read_go_schema_version() -> str | None: return None -def check_schema_version(doc: CapabilitiesDoc) -> bool: +def check_schema_version(doc: Capabilities) -> bool: """Whether the manifest agrees with the schema version embedded in Go bundle metadata.""" try: go_version = read_go_schema_version() @@ -75,7 +75,7 @@ def check_schema_version(doc: CapabilitiesDoc) -> bool: "Editing go-sdk/capabilities.yaml cannot fix this.[/]" ) return False - declared = doc["supervisor_schema_version"] + declared = doc.supervisor_schema_version if go_version == declared: return True console.print( diff --git a/scripts/ci/prek/update_java_sdk_readme_matrix.py b/scripts/ci/prek/update_java_sdk_readme_matrix.py index 05a531c169a..10e2adb91a0 100755 --- a/scripts/ci/prek/update_java_sdk_readme_matrix.py +++ b/scripts/ci/prek/update_java_sdk_readme_matrix.py @@ -40,7 +40,7 @@ from lang_sdk_compat_matrix import ( LANG_SDKS, README_MATRIX_FOOTER, README_MATRIX_HEADER, - CapabilitiesDoc, + Capabilities, load_capabilities, render_markdown_table, ) @@ -63,10 +63,10 @@ def read_gradle_schema_version() -> str | None: return None -def check_schema_version(doc: CapabilitiesDoc) -> bool: +def check_schema_version(doc: Capabilities) -> bool: """Whether the manifest agrees with the schema version the JAR manifest is stamped with.""" gradle_version = read_gradle_schema_version() - declared = doc["supervisor_schema_version"] + declared = doc.supervisor_schema_version if gradle_version is None or gradle_version == declared: return True console.print( diff --git a/scripts/ci/prek/update_lang_sdk_central_matrix.py b/scripts/ci/prek/update_lang_sdk_central_matrix.py new file mode 100755 index 00000000000..8db82e4913b --- /dev/null +++ b/scripts/ci/prek/update_lang_sdk_central_matrix.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# /// script +# requires-python = ">=3.11,<3.12" +# dependencies = ["PyYAML>=6.0", "rich>=13.6.0"] +# /// +"""Regenerate the consolidated Language SDK compatibility matrix in the user-facing docs. + +Reads every ``/capabilities.yaml`` listed in the ``LANG_SDKS`` registry (an SDK that has +not published one yet simply shows the "absent" mark) and rewrites the ``list-table`` blocks in +``airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst`` between the +AUTO-GENERATED markers. Exits non-zero when the file changed so the contributor re-stages it. +""" + +from __future__ import annotations + +import sys +from collections.abc import Sequence +from pathlib import Path +from typing import Final, NamedTuple + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import lang_sdk_compat_matrix as matrix +from common_prek_utils import console, insert_documentation + +CENTRAL_MATRIX_HEADER: Final = ".. BEGIN AUTO-GENERATED LANG-SDK COMPAT MATRIX" +CENTRAL_MATRIX_FOOTER: Final = ".. END AUTO-GENERATED LANG-SDK COMPAT MATRIX" +ABSENT_MARK: Final = "n/p" # the SDK has not published a manifest at all + +INDEX_RST: Final = ( + matrix.AIRFLOW_ROOT_PATH + / "airflow-core" + / "docs" + / "authoring-and-scheduling" + / "language-sdks" + / "index.rst" +) + + +class SdkCapabilities(NamedTuple): + display_name: str + doc: matrix.Capabilities | None + + +def load_sdk_capabilities(sdks: Sequence[matrix.LangSdk]) -> list[SdkCapabilities]: + """Load every published manifest in registry order.""" + return [ + SdkCapabilities( + display_name=sdk["display_name"], + doc=( + matrix.load_capabilities(sdk["capabilities_yaml"], expected_sdk=sdk["id"]) + if sdk["capabilities_yaml"].exists() + else None + ), + ) + for sdk in sdks + ] + + +def _render_row(label: str, values: Sequence[str]) -> list[str]: + return [f" * - {label}\n", *[f" - {value}".rstrip() + "\n" for value in values]] + + +def _render_table_header( + title: str, sdk_capabilities: Sequence[SdkCapabilities], first_column: str = "Dimension" +) -> list[str]: + widths = " ".join(["30"] + ["15"] * len(sdk_capabilities)) + return [ + "\n", + f".. list-table:: {title}\n", + " :header-rows: 1\n", + f" :widths: {widths}\n", + "\n", + *_render_row(first_column, [sdk.display_name for sdk in sdk_capabilities]), + ] + + +def _get_cell(sdk: SdkCapabilities, row: matrix.MatrixRow) -> str: + """Render a mark, with the SDK release that shipped a supported dimension, e.g. ``✓ 1.0.0``.""" + if sdk.doc is None: + return ABSENT_MARK + mark = matrix.get_mark(sdk.doc, row) + sdk_version = sdk.doc.get_entry(row).get("sdk_version") + return f"{mark} {sdk_version}" if mark == matrix.SUPPORTED_MARK and sdk_version else mark + + +def render_central_table(sdk_capabilities: Sequence[SdkCapabilities]) -> list[str]: + """Render the consolidated cross-SDK RST ``list-table`` for each part of the matrix.""" + lines = _render_table_header("Language SDK versions", sdk_capabilities, first_column="Version") + lines += _render_row( + "Min. Airflow version", + [sdk.doc.min_airflow_version if sdk.doc else ABSENT_MARK for sdk in sdk_capabilities], + ) + lines += _render_row( + "Current Supervisor Schema Version", + [sdk.doc.supervisor_schema_version if sdk.doc else ABSENT_MARK for sdk in sdk_capabilities], + ) + for part in matrix.build_matrix_parts(): + lines += _render_table_header(part.title, sdk_capabilities) + for group in part.groups: + if len(part.groups) > 1: + lines += _render_row(f"**{group.label}**", [""] * len(sdk_capabilities)) + for row in group.rows: + lines += _render_row( + f"``{row.name}`` ({matrix.format_tier(row)})", + [_get_cell(sdk, row) for sdk in sdk_capabilities], + ) + lines += [ + "\n", + f"*Marks:* ``{matrix.SUPPORTED_MARK}`` supported (with the SDK release that shipped it, when known) · " + f"``{matrix.UNSUPPORTED_MARK}`` not supported · ``{matrix.NA_MARK}`` not applicable · " + f"``{ABSENT_MARK}`` not published. A tier marked ``†`` applies only when " + f"``{matrix.DAG_DEFINITION_GATE}`` is supported.\n", + "\n", + ] + return lines + + +def main() -> int: + changed = insert_documentation( + INDEX_RST, + render_central_table(load_sdk_capabilities(matrix.LANG_SDKS)), + CENTRAL_MATRIX_HEADER, + CENTRAL_MATRIX_FOOTER, + extra_information="the Language SDK compatibility matrix", + ) + if changed: + console.print( + "[yellow]Regenerated the Language SDK compatibility matrix in index.rst; re-stage the file.[/]" + ) + return 1 + return 0 + + +if __name__ in ("__main__", "__mp_main__"): + raise SystemExit(main()) diff --git a/scripts/ci/prek/update_ts_sdk_readme_matrix.py b/scripts/ci/prek/update_ts_sdk_readme_matrix.py index d4c59645985..57c6c8859bd 100644 --- a/scripts/ci/prek/update_ts_sdk_readme_matrix.py +++ b/scripts/ci/prek/update_ts_sdk_readme_matrix.py @@ -40,7 +40,7 @@ from lang_sdk_compat_matrix import ( LANG_SDKS, README_MATRIX_FOOTER, README_MATRIX_HEADER, - CapabilitiesDoc, + Capabilities, load_capabilities, render_markdown_table, ) @@ -60,7 +60,7 @@ def read_ts_schema_version() -> str | None: return None -def check_schema_version(doc: CapabilitiesDoc) -> bool: +def check_schema_version(doc: Capabilities) -> bool: """Whether the manifest agrees with the generated TypeScript supervisor schema version. Unlike the Java sibling hook (which treats an unreadable gradle property as an implicit pass), @@ -69,7 +69,7 @@ def check_schema_version(doc: CapabilitiesDoc) -> bool: genuine mismatch because editing capabilities.yaml cannot fix a missing constant. """ ts_version = read_ts_schema_version() - declared = doc["supervisor_schema_version"] + declared = doc.supervisor_schema_version if ts_version is None: console.print( "[red]Could not read SUPERVISOR_API_VERSION from ts-sdk/src/generated/supervisor.ts: " diff --git a/scripts/tests/ci/prek/conftest.py b/scripts/tests/ci/prek/conftest.py index 6ef372f3481..b09b89f7801 100644 --- a/scripts/tests/ci/prek/conftest.py +++ b/scripts/tests/ci/prek/conftest.py @@ -21,6 +21,7 @@ from pathlib import Path import pytest import yaml +from ci.prek import lang_sdk_compat_matrix as matrix @pytest.fixture @@ -74,3 +75,57 @@ def create_provider_tree(tmp_path): return test_file return _create + + +@pytest.fixture +def capabilities_entry(): + """Factory fixture: one dimension entry of a ``capabilities.yaml`` manifest.""" + + def _entry( + supported: bool, since: str | None = None, sdk_version: str | None = None, note: str = "" + ) -> dict: + return {"supported": supported, "since": since, "sdk_version": sdk_version, "note": note} + + return _entry + + +@pytest.fixture +def capabilities_doc(capabilities_entry): + """Factory fixture: a fully-populated, valid capabilities document for an SDK.""" + + def _doc(sdk_id: str = "go", **overrides) -> dict: + doc = { + "sdk": sdk_id, + "supervisor_schema_version": "2026-06-16", + "min_airflow_version": "3.3", + "states": {state: capabilities_entry(True, since="3.3") for state, _ in matrix.STATE_DIMENSIONS}, + "capabilities": { + cap.name: {mode: capabilities_entry(True, since="3.3") for mode in matrix.AUTHORING_MODES} + if cap.name in matrix.SHARED_CAPABILITY_NAMES + else capabilities_entry(True, since="3.3") + for cap in matrix.CAPABILITY_DIMENSIONS + }, + } + doc.update(overrides) + return doc + + return _doc + + +@pytest.fixture +def capabilities_doc_without_dag_definition(capabilities_doc, capabilities_entry): + """Factory fixture: a valid document for an SDK that does not define Dags in its own language.""" + + def _doc(sdk_id: str = "go") -> dict: + doc = capabilities_doc(sdk_id) + doc["capabilities"][matrix.DAG_DEFINITION_GATE] = capabilities_entry(False) + for row in matrix.build_matrix_parts().dag_definition.groups[0].rows: + if not row.gated: + continue + if row.name in matrix.SHARED_CAPABILITY_NAMES: + doc["capabilities"][row.name][matrix.DAG_DEFINITION_MODE] = capabilities_entry(False) + else: + doc["capabilities"][row.name] = capabilities_entry(False) + return doc + + return _doc diff --git a/scripts/tests/ci/prek/test_lang_sdk_compat_matrix.py b/scripts/tests/ci/prek/test_lang_sdk_compat_matrix.py index 38a0a29f8f2..05690fd6d1a 100644 --- a/scripts/tests/ci/prek/test_lang_sdk_compat_matrix.py +++ b/scripts/tests/ci/prek/test_lang_sdk_compat_matrix.py @@ -21,126 +21,169 @@ import yaml from ci.prek import lang_sdk_compat_matrix as matrix -def _entry(supported: bool, since: str | None = None, note: str = "") -> dict: - return {"supported": supported, "since": since, "note": note} - - -def _doc(sdk_id: str = "go", **overrides) -> dict: - """A fully-populated, valid capabilities document for ``sdk_id``.""" - doc = { - "sdk": sdk_id, - "supervisor_schema_version": "2026-06-16", - "min_airflow_version": "3.3", - "states": {state: _entry(True, since="3.3") for state, _ in matrix.STATE_DIMENSIONS}, - "capabilities": {cap.name: _entry(True, since="3.3") for cap in matrix.CAPABILITY_DIMENSIONS}, - } - doc.update(overrides) - return doc - - -class TestValidateCapabilities: - def test_valid_doc_passes(self): - matrix.validate_capabilities(_doc(), source="test") +class TestParseCapabilities: + def test_valid_doc_passes(self, capabilities_doc): + matrix.parse_capabilities(capabilities_doc(), source="test") def test_non_dict_raises(self): with pytest.raises(matrix.CapabilitiesError, match="must be a mapping"): - matrix.validate_capabilities([], source="test") + matrix.parse_capabilities([], source="test") @pytest.mark.parametrize( "key", ["sdk", "supervisor_schema_version", "min_airflow_version", "states", "capabilities"] ) - def test_missing_required_key_raises(self, key): - doc = _doc() + def test_missing_required_key_raises(self, key, capabilities_doc): + doc = capabilities_doc() del doc[key] with pytest.raises(matrix.CapabilitiesError, match="missing required keys"): - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") - def test_unknown_top_level_key_raises(self): + def test_unknown_top_level_key_raises(self, capabilities_doc): with pytest.raises(matrix.CapabilitiesError, match="unknown top-level keys: capability"): - matrix.validate_capabilities(_doc(capability={}), source="test") + matrix.parse_capabilities(capabilities_doc(capability={}), source="test") - def test_unknown_entry_key_raises(self): - doc = _doc() - doc["capabilities"]["branching"] = {"supported": True, "supported_since": "3.3"} - with pytest.raises(matrix.CapabilitiesError, match="branching has unknown keys: supported_since"): - matrix.validate_capabilities(doc, source="test") + def test_unknown_entry_key_raises(self, capabilities_doc): + doc = capabilities_doc() + doc["capabilities"]["if-else-branching"] = {"supported": True, "supported_since": "3.3"} + with pytest.raises( + matrix.CapabilitiesError, match="if-else-branching has unknown keys: supported_since" + ): + matrix.parse_capabilities(doc, source="test") - def test_unknown_sdk_raises(self): + def test_unknown_sdk_raises(self, capabilities_doc): with pytest.raises(matrix.CapabilitiesError, match="unknown sdk"): - matrix.validate_capabilities(_doc(sdk_id="rust"), source="test") + matrix.parse_capabilities(capabilities_doc(sdk_id="rust"), source="test") - def test_missing_state_raises(self): - doc = _doc() + def test_missing_state_raises(self, capabilities_doc): + doc = capabilities_doc() del doc["states"]["deferred"] with pytest.raises(matrix.CapabilitiesError, match="states keys mismatch"): - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") - def test_unknown_capability_raises(self): - doc = _doc() - doc["capabilities"]["telepathy"] = _entry(True) + def test_unknown_capability_raises(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["telepathy"] = capabilities_entry(True) with pytest.raises(matrix.CapabilitiesError, match="capabilities keys mismatch"): - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") - def test_non_boolean_supported_raises(self): - doc = _doc() + def test_non_boolean_supported_raises(self, capabilities_doc): + doc = capabilities_doc() doc["states"]["success"] = {"supported": "true"} with pytest.raises(matrix.CapabilitiesError, match="boolean 'supported'"): - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") - def test_sdk_mismatch_against_expected_raises(self): + def test_sdk_mismatch_against_expected_raises(self, capabilities_doc): with pytest.raises(matrix.CapabilitiesError, match="belongs to 'go'"): - matrix.validate_capabilities(_doc(sdk_id="java"), source="test", expected_sdk="go") + matrix.parse_capabilities(capabilities_doc(sdk_id="java"), source="test", expected_sdk="go") - def test_expected_sdk_match_passes(self): - matrix.validate_capabilities(_doc(sdk_id="go"), source="test", expected_sdk="go") + def test_expected_sdk_match_passes(self, capabilities_doc): + matrix.parse_capabilities(capabilities_doc(sdk_id="go"), source="test", expected_sdk="go") - def test_non_string_note_raises(self): - doc = _doc() + def test_non_string_note_raises(self, capabilities_doc): + doc = capabilities_doc() doc["states"]["success"] = {"supported": True, "since": "3.3", "note": 123} with pytest.raises(matrix.CapabilitiesError, match="note must be a string"): - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") - def test_non_string_since_raises(self): - doc = _doc() + def test_non_string_since_raises(self, capabilities_doc): + doc = capabilities_doc() doc["capabilities"]["xcom-read-write"] = {"supported": True, "since": 3, "note": ""} with pytest.raises(matrix.CapabilitiesError, match="since must be a string or null"): - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") - def test_non_string_schema_version_raises(self): + def test_non_string_schema_version_raises(self, capabilities_doc): with pytest.raises(matrix.CapabilitiesError, match="supervisor_schema_version must be a string"): - matrix.validate_capabilities(_doc(supervisor_schema_version=123), source="test") + matrix.parse_capabilities(capabilities_doc(supervisor_schema_version=123), source="test") - def test_unsupported_entry_carrying_since_raises(self): - doc = _doc() - doc["capabilities"]["branching"] = _entry(False, since="3.3") + def test_unsupported_entry_carrying_since_raises(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["if-else-branching"] = capabilities_entry(False, since="3.3") with pytest.raises(matrix.CapabilitiesError, match="not supported but carries since"): - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") - def test_supported_gated_capability_without_native_dag_authoring_raises(self): - doc = _doc() - doc["capabilities"][matrix.NATIVE_DAG_GATE] = _entry(False) + def test_unsupported_entry_carrying_sdk_version_raises(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["if-else-branching"] = capabilities_entry(False, sdk_version="1.0.0") + with pytest.raises(matrix.CapabilitiesError, match="not supported but carries sdk_version"): + matrix.parse_capabilities(doc, source="test") + + def test_non_string_sdk_version_raises(self, capabilities_doc): + doc = capabilities_doc() + doc["capabilities"]["xcom-read-write"] = {"supported": True, "since": "3.3", "sdk_version": 3} + with pytest.raises(matrix.CapabilitiesError, match="sdk_version must be a string or null"): + matrix.parse_capabilities(doc, source="test") + + def test_supported_gated_capability_without_dag_definition_raises( + self, capabilities_doc, capabilities_entry + ): + doc = capabilities_doc() + doc["capabilities"][matrix.DAG_DEFINITION_GATE] = capabilities_entry(False) with pytest.raises( matrix.CapabilitiesError, - match="gated capabilities cannot be supported.*branching", + match="gated capabilities cannot be supported.*if-else-branching", ): - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") - def test_supported_entry_without_since_passes(self): - doc = _doc() - doc["states"]["success"] = _entry(True, since=None) - matrix.validate_capabilities(doc, source="test") + def test_shared_capability_missing_a_mode_raises(self, capabilities_doc): + doc = capabilities_doc() + del doc["capabilities"]["mapped-tasks"][matrix.STUB_TASK_HANDLER_MODE] + with pytest.raises(matrix.CapabilitiesError, match="capabilities.mapped-tasks keys mismatch"): + matrix.parse_capabilities(doc, source="test") - def test_entry_omitting_optional_keys_passes(self): - doc = _doc() + def test_shared_capability_declared_as_a_single_entry_raises(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["mapped-tasks"] = capabilities_entry(True, since="3.3") + with pytest.raises(matrix.CapabilitiesError, match="capabilities.mapped-tasks keys mismatch"): + matrix.parse_capabilities(doc, source="test") + + def test_invalid_shared_mode_entry_raises(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["asset-inlets-outlets"][matrix.STUB_TASK_HANDLER_MODE] = capabilities_entry( + False, since="3.3" + ) + with pytest.raises( + matrix.CapabilitiesError, + match="capabilities.asset-inlets-outlets.stub-task-handler is not supported", + ): + matrix.parse_capabilities(doc, source="test") + + def test_shared_capability_dag_definition_mode_is_gated( + self, capabilities_doc_without_dag_definition, capabilities_entry + ): + # Everything gated is off except the Dag-definition side of mapped-tasks, so the + # error names that one shared capability and nothing else. + doc = capabilities_doc_without_dag_definition() + doc["capabilities"]["mapped-tasks"][matrix.DAG_DEFINITION_MODE] = capabilities_entry( + True, since="3.3" + ) + with pytest.raises(matrix.CapabilitiesError, match="cannot be supported.*: mapped-tasks$"): + matrix.parse_capabilities(doc, source="test") + + def test_shared_capability_stub_task_handler_mode_is_not_gated( + self, capabilities_doc_without_dag_definition + ): + # Every gated dimension is off, but the stub-task-handler side of a shared capability + # stays supported: the gate covers the Dag-definition side only. + matrix.parse_capabilities(capabilities_doc_without_dag_definition(), source="test") + + def test_supported_entry_without_since_passes(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["states"]["success"] = capabilities_entry(True, since=None) + matrix.parse_capabilities(doc, source="test") + + def test_entry_omitting_optional_keys_passes(self, capabilities_doc): + doc = capabilities_doc() doc["states"]["success"] = {"supported": True} - matrix.validate_capabilities(doc, source="test") + matrix.parse_capabilities(doc, source="test") class TestLoadCapabilities: - def test_reads_yaml(self, tmp_path): + def test_reads_yaml(self, tmp_path, capabilities_doc): path = tmp_path / "capabilities.yaml" - path.write_text(yaml.safe_dump(_doc())) - assert matrix.load_capabilities(path, expected_sdk="go") == _doc() + path.write_text(yaml.safe_dump(capabilities_doc())) + assert matrix.load_capabilities(path, expected_sdk="go") == matrix.parse_capabilities( + capabilities_doc(), source="test" + ) def test_committed_manifests_are_valid(self): """Every manifest an SDK has actually committed passes validation.""" @@ -150,54 +193,129 @@ class TestLoadCapabilities: matrix.load_capabilities(sdk["capabilities_yaml"], expected_sdk=sdk["id"]) +class TestEntryLookup: + def test_shared_capability_resolves_per_authoring_mode(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["mapped-tasks"][matrix.STUB_TASK_HANDLER_MODE] = capabilities_entry(False) + parsed = matrix.parse_capabilities(doc, source="test") + parts = matrix.build_matrix_parts() + dag_definition_row, stub_row = ( + next(row for row in part.groups[0].rows if row.name == "mapped-tasks") + for part in (parts.dag_definition, parts.stub_task_handler) + ) + + assert parsed.get_entry(dag_definition_row)["supported"] is True + assert parsed.get_entry(stub_row)["supported"] is False + + def test_unshared_capability_carries_no_mode(self, capabilities_doc): + parsed = matrix.parse_capabilities(capabilities_doc(), source="test") + row = next( + row + for row in matrix.build_matrix_parts().dag_definition.groups[0].rows + if row.name == "task-group" + ) + + assert row.mode is None + assert row.entry_key == ("capability", "task-group", None) + assert parsed.get_entry(row) is parsed.entries["capability", "task-group", None] + + def test_state_and_capability_of_the_same_name_do_not_collide(self): + state_row = matrix.MatrixRow("state", "removed", "MAY", False, None) + capability_row = matrix.MatrixRow("capability", "removed", "MAY", False, None) + + assert state_row.entry_key != capability_row.entry_key + + class TestRenderMarkdownTable: def test_retry_policy_is_an_optional_runtime_capability(self): retry_policy = next(cap for cap in matrix.CAPABILITY_DIMENSIONS if cap.name == "retry-policy") assert retry_policy == matrix.Capability("retry-policy", "MAY", "runtime", False) - def test_rows_cover_every_dimension(self): - rendered = "".join(matrix.render_markdown_table(_doc())) + def test_rows_cover_every_dimension(self, capabilities_doc): + rendered = "".join( + matrix.render_markdown_table(matrix.parse_capabilities(capabilities_doc(), source="test")) + ) for state, tier in matrix.STATE_DIMENSIONS: assert f"| state: `{state}` | {tier} |" in rendered for cap in matrix.CAPABILITY_DIMENSIONS: - assert f"| capability: `{cap.name}` | {matrix._tier_label(cap)} |" in rendered + assert f"| capability: `{cap.name}` |" in rendered assert "supervisor schema: 2026-06-16" in rendered - def test_supported_and_unsupported_marks(self): - doc = _doc() - doc["states"]["deferred"] = _entry(False, note="no triggerer bridge") - rendered = "".join(matrix.render_markdown_table(doc)) + def test_parts_split_dimensions_by_authoring_mode(self): + common, dag_definition, stub_task_handler = matrix.build_matrix_parts() + dag_definition_names = [row.name for row in dag_definition.groups[0].rows] + stub_names = [row.name for row in stub_task_handler.groups[0].rows] + + assert [group.label for group in common.groups] == ["TaskInstance states", "Task runtime"] + assert "object-store" in [row.name for row in common.groups[1].rows] + assert dag_definition_names[0] == matrix.DAG_DEFINITION_GATE + assert stub_names == [ + "mixed-lang-stub-target", + "taskflow-binding", + "mapped-tasks", + "asset-inlets-outlets", + "task-callbacks", + ] + assert set(dag_definition_names) >= matrix.SHARED_CAPABILITY_NAMES + assert all(not row.gated for row in stub_task_handler.groups[0].rows) + + def test_shared_capability_renders_each_mode(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["mapped-tasks"][matrix.DAG_DEFINITION_MODE] = capabilities_entry( + False, note="no expand API" + ) + rendered = "".join(matrix.render_markdown_table(matrix.parse_capabilities(doc, source="test"))) + assert f"| capability: `mapped-tasks` | MAY † | {matrix.UNSUPPORTED_MARK} |" in rendered + assert f"| capability: `mapped-tasks` | MAY | {matrix.SUPPORTED_MARK} | 3.3 |" in rendered + + def test_supported_and_unsupported_marks(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["states"]["deferred"] = capabilities_entry(False, note="no triggerer bridge") + rendered = "".join(matrix.render_markdown_table(matrix.parse_capabilities(doc, source="test"))) assert f"| state: `success` | MUST | {matrix.SUPPORTED_MARK} | 3.3 |" in rendered assert ( f"| state: `deferred` | MAY | {matrix.UNSUPPORTED_MARK} | {matrix.NO_VERSION_MARK} |" in rendered ) - def test_gated_capabilities_are_na_without_the_native_dag_gate(self): - doc = _doc() - doc["capabilities"][matrix.NATIVE_DAG_GATE] = _entry(False) - rendered = "".join(matrix.render_markdown_table(doc)) - assert f"| capability: `{matrix.NATIVE_DAG_GATE}` | SHOULD | {matrix.UNSUPPORTED_MARK} |" in rendered - for cap in matrix.CAPABILITY_DIMENSIONS: - if cap.gated: - assert f"| capability: `{cap.name}` | {cap.tier} † | {matrix.NA_MARK} |" in rendered + def test_gated_capabilities_are_na_without_the_dag_definition_gate( + self, capabilities_doc_without_dag_definition + ): + doc = capabilities_doc_without_dag_definition() + rendered = "".join(matrix.render_markdown_table(matrix.parse_capabilities(doc, source="test"))) + assert ( + f"| capability: `{matrix.DAG_DEFINITION_GATE}` | SHOULD | {matrix.UNSUPPORTED_MARK} |" in rendered + ) + for row in matrix.build_matrix_parts().dag_definition.groups[0].rows: + if row.gated: + assert f"| capability: `{row.name}` | {row.tier} † | {matrix.NA_MARK} |" in rendered - def test_gated_capabilities_resolve_to_their_own_mark_once_the_gate_is_supported(self): - doc = _doc() - doc["capabilities"]["branching"] = _entry(False, note="no branch construct") - rendered = "".join(matrix.render_markdown_table(doc)) - gated = [cap for cap in matrix.CAPABILITY_DIMENSIONS if cap.gated] + def test_gated_capabilities_resolve_to_their_own_mark_once_the_gate_is_supported( + self, capabilities_doc, capabilities_entry + ): + doc = capabilities_doc() + doc["capabilities"]["if-else-branching"] = capabilities_entry(False, note="no branch construct") + rendered = "".join(matrix.render_markdown_table(matrix.parse_capabilities(doc, source="test"))) + gated = [row for row in matrix.build_matrix_parts().dag_definition.groups[0].rows if row.gated] assert gated, "expected at least one gated capability" # With the gate supported, a gated capability reports its real support, never n/a. - for cap in gated: - if cap.name != "branching": + for row in gated: + if row.name != "if-else-branching": assert ( - f"| capability: `{cap.name}` | {cap.tier} † | {matrix.SUPPORTED_MARK} | 3.3 |" in rendered + f"| capability: `{row.name}` | {row.tier} † | {matrix.SUPPORTED_MARK} | 3.3 |" in rendered ) - assert f"| capability: `branching` | SHOULD † | {matrix.UNSUPPORTED_MARK} |" in rendered + assert f"| capability: `if-else-branching` | SHOULD † | {matrix.UNSUPPORTED_MARK} |" in rendered assert matrix.NA_MARK not in rendered.replace(matrix.LEGEND, "") - def test_pipe_in_note_is_escaped(self): - doc = _doc() - doc["capabilities"]["xcom-read-write"] = _entry(True, since="3.3", note="read | write") - rendered = "".join(matrix.render_markdown_table(doc)) + def test_sdk_version_renders_in_its_own_column(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["xcom-read-write"] = capabilities_entry(True, since="3.4.0", sdk_version="1.0.0") + rendered = "".join(matrix.render_markdown_table(matrix.parse_capabilities(doc, source="test"))) + assert ( + f"| capability: `xcom-read-write` | MUST | {matrix.SUPPORTED_MARK} | 3.4.0 | 1.0.0 |" in rendered + ) + + def test_pipe_in_note_is_escaped(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["xcom-read-write"] = capabilities_entry(True, since="3.3", note="read | write") + rendered = "".join(matrix.render_markdown_table(matrix.parse_capabilities(doc, source="test"))) assert "read \\| write" in rendered diff --git a/scripts/tests/ci/prek/test_update_go_sdk_readme_matrix.py b/scripts/tests/ci/prek/test_update_go_sdk_readme_matrix.py index 4bfdde594c8..dc0cf1ed944 100644 --- a/scripts/tests/ci/prek/test_update_go_sdk_readme_matrix.py +++ b/scripts/tests/ci/prek/test_update_go_sdk_readme_matrix.py @@ -23,28 +23,17 @@ from ci.prek import lang_sdk_compat_matrix as matrix, update_go_sdk_readme_matri SCHEMA_VERSION = "2026-06-16" -def _doc() -> dict: - def entry(supported: bool) -> dict: - return {"supported": supported, "since": "3.3" if supported else None, "note": ""} - - return { - "sdk": "go", - "supervisor_schema_version": SCHEMA_VERSION, - "min_airflow_version": "3.3", - "states": {state: entry(True) for state, _ in matrix.STATE_DIMENSIONS}, - "capabilities": {cap.name: entry(True) for cap in matrix.CAPABILITY_DIMENSIONS}, - } - - class TestMain: @pytest.fixture - def wired(self, tmp_path, monkeypatch): + def wired(self, tmp_path, monkeypatch, capabilities_doc): + doc = capabilities_doc("go") capabilities_yaml = tmp_path / "capabilities.yaml" - capabilities_yaml.write_text(yaml.safe_dump(_doc())) + capabilities_yaml.write_text(yaml.safe_dump(doc)) readme = tmp_path / "README.md" readme.write_text(f"intro\n\n{matrix.README_MATRIX_HEADER}\n{matrix.README_MATRIX_FOOTER}\n\noutro\n") go_messages = tmp_path / "messages.go" - go_messages.write_text(f'package execution\n\nconst SupervisorSchemaVersion = "{SCHEMA_VERSION}"\n') + schema_version = doc["supervisor_schema_version"] + go_messages.write_text(f'package execution\n\nconst SupervisorSchemaVersion = "{schema_version}"\n') monkeypatch.setattr( hook, "LANG_SDKS", @@ -56,7 +45,7 @@ class TestMain: def test_generates_target_then_is_idempotent(self, wired): assert hook.main() == 1 content = wired.read_text() - assert "| Dimension | Tier | Supported | Since | Notes |" in content + assert "| Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes |" in content assert matrix.SUPPORTED_MARK in content assert content.startswith("intro\n") and content.endswith("outro\n") @@ -77,7 +66,7 @@ class TestMain: def test_schema_version_matches_go_source(self): sdk = next(entry for entry in matrix.LANG_SDKS if entry["id"] == hook.SDK_ID) doc = matrix.load_capabilities(sdk["capabilities_yaml"], expected_sdk=hook.SDK_ID) - assert doc["supervisor_schema_version"] == hook.read_go_schema_version() + assert doc.supervisor_schema_version == hook.read_go_schema_version() @pytest.mark.parametrize( "declaration", diff --git a/scripts/tests/ci/prek/test_update_java_sdk_readme_matrix.py b/scripts/tests/ci/prek/test_update_java_sdk_readme_matrix.py index 730f030c64a..45a8a44f3a9 100644 --- a/scripts/tests/ci/prek/test_update_java_sdk_readme_matrix.py +++ b/scripts/tests/ci/prek/test_update_java_sdk_readme_matrix.py @@ -20,31 +20,18 @@ import pytest import yaml from ci.prek import lang_sdk_compat_matrix as matrix, update_java_sdk_readme_matrix as hook -SCHEMA_VERSION = "2026-06-16" - - -def _doc() -> dict: - def entry(supported: bool) -> dict: - return {"supported": supported, "since": "3.3" if supported else None, "note": ""} - - return { - "sdk": "java", - "supervisor_schema_version": SCHEMA_VERSION, - "min_airflow_version": "3.3", - "states": {state: entry(True) for state, _ in matrix.STATE_DIMENSIONS}, - "capabilities": {cap.name: entry(True) for cap in matrix.CAPABILITY_DIMENSIONS}, - } - class TestMain: @pytest.fixture - def wired(self, tmp_path, monkeypatch): + def wired(self, tmp_path, monkeypatch, capabilities_doc): """Point the hook at a temp capabilities.yaml plus temp README and Dokka module doc.""" + doc = capabilities_doc("java") capabilities_yaml = tmp_path / "capabilities.yaml" - capabilities_yaml.write_text(yaml.safe_dump(_doc())) + capabilities_yaml.write_text(yaml.safe_dump(doc)) gradle_properties = tmp_path / "gradle.properties" + schema_version = doc["supervisor_schema_version"] gradle_properties.write_text( - f"projectVersion=1.0.0-SNAPSHOT\n{hook.SCHEMA_VERSION_PROPERTY}={SCHEMA_VERSION}\n" + f"projectVersion=1.0.0-SNAPSHOT\n{hook.SCHEMA_VERSION_PROPERTY}={schema_version}\n" ) targets = [] for name in ("README.md", "module.md"): @@ -67,7 +54,7 @@ class TestMain: assert hook.main() == 1 for target in wired: content = target.read_text() - assert "| Dimension | Tier | Supported | Since | Notes |" in content + assert "| Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes |" in content assert matrix.SUPPORTED_MARK in content assert content.startswith("intro\n") and content.endswith("outro\n") @@ -83,4 +70,4 @@ class TestMain: """The committed manifest agrees with the version that stamps the JAR.""" sdk = next(entry for entry in matrix.LANG_SDKS if entry["id"] == hook.SDK_ID) doc = matrix.load_capabilities(sdk["capabilities_yaml"], expected_sdk=hook.SDK_ID) - assert doc["supervisor_schema_version"] == hook.read_gradle_schema_version() + assert doc.supervisor_schema_version == hook.read_gradle_schema_version() diff --git a/scripts/tests/ci/prek/test_update_lang_sdk_central_matrix.py b/scripts/tests/ci/prek/test_update_lang_sdk_central_matrix.py new file mode 100644 index 00000000000..0d831cac415 --- /dev/null +++ b/scripts/tests/ci/prek/test_update_lang_sdk_central_matrix.py @@ -0,0 +1,148 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import pytest +import yaml +from ci.prek import lang_sdk_compat_matrix as matrix, update_lang_sdk_central_matrix as hook + + +class TestLoadSdkCapabilities: + def test_loads_published_manifest_and_preserves_unpublished_sdk(self, tmp_path, capabilities_doc): + go_yaml = tmp_path / "go.yaml" + go_yaml.write_text(yaml.safe_dump(capabilities_doc())) + registry = [ + { + "id": "go", + "display_name": "Go", + "capabilities_yaml": go_yaml, + "readme": tmp_path / "go.md", + }, + { + "id": "ts", + "display_name": "TypeScript", + "capabilities_yaml": tmp_path / "ts.yaml", + "readme": tmp_path / "ts.md", + }, + ] + + loaded = hook.load_sdk_capabilities(registry) + + assert loaded == [ + hook.SdkCapabilities("Go", matrix.parse_capabilities(capabilities_doc(), source="test")), + hook.SdkCapabilities("TypeScript", None), + ] + + +class TestRenderCentralTable: + def test_absent_mark_is_distinct_from_the_other_marks(self): + """ "Not published" must not read as "not applicable" or as a missing version.""" + assert hook.ABSENT_MARK == "n/p" + assert hook.ABSENT_MARK not in ( + matrix.NA_MARK, + matrix.NO_VERSION_MARK, + matrix.UNSUPPORTED_MARK, + ) + + def test_lists_registered_sdks_and_absent_marks(self): + sdk_capabilities = [ + hook.SdkCapabilities("Go", None), + hook.SdkCapabilities("Java", None), + hook.SdkCapabilities("TypeScript", None), + ] + + rendered = "".join(hook.render_central_table(sdk_capabilities)) + + for title in ( + "Language SDK versions", + "Common to both authoring modes", + "Dag definition", + "Python Dag with stub TaskHandler", + ): + assert f".. list-table:: {title}\n" in rendered + for display_name in ("Go", "Java", "TypeScript"): + assert f" - {display_name}\n" in rendered + assert f" - {matrix.SUPPORTED_MARK}\n" not in rendered + assert f" * - Min. Airflow version\n - {hook.ABSENT_MARK}\n" in rendered + + def test_renders_manifest_marks_and_metadata(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"][matrix.DAG_DEFINITION_GATE] = capabilities_entry(False) + for capability in matrix.CAPABILITY_DIMENSIONS: + if capability.gated and capability.name in matrix.SHARED_CAPABILITY_NAMES: + doc["capabilities"][capability.name][matrix.DAG_DEFINITION_MODE] = capabilities_entry(False) + elif capability.gated: + doc["capabilities"][capability.name] = capabilities_entry(False) + rendered = "".join( + hook.render_central_table( + [hook.SdkCapabilities("Go", matrix.parse_capabilities(doc, source="test"))] + ) + ) + + assert f" * - ``success`` (MUST)\n - {matrix.SUPPORTED_MARK}\n" in rendered + assert f" * - ``object-store`` (MAY)\n - {matrix.SUPPORTED_MARK}\n" in rendered + assert f" * - ``dag-definition-authoring`` (SHOULD)\n - {matrix.UNSUPPORTED_MARK}\n" in rendered + assert f" * - ``task-args`` (MUST †)\n - {matrix.NA_MARK}\n" in rendered + assert f" * - ``mapped-tasks`` (MAY †)\n - {matrix.NA_MARK}\n" in rendered + assert f" * - ``mapped-tasks`` (MAY)\n - {matrix.SUPPORTED_MARK}\n" in rendered + assert " * - Min. Airflow version\n - 3.3\n" in rendered + assert "2026-06-16" in rendered + + def test_supported_mark_shows_sdk_version_when_present(self, capabilities_doc, capabilities_entry): + doc = capabilities_doc() + doc["capabilities"]["xcom-read-write"] = capabilities_entry(True, since="3.4.0", sdk_version="1.0.0") + + rendered = "".join( + hook.render_central_table( + [hook.SdkCapabilities("Java", matrix.parse_capabilities(doc, source="test"))] + ) + ) + + assert f" * - ``xcom-read-write`` (MUST)\n - {matrix.SUPPORTED_MARK} 1.0.0\n" in rendered + + +class TestMain: + @pytest.fixture + def wired(self, tmp_path, monkeypatch, capabilities_doc): + """Point the hook at a temporary index and registry.""" + index = tmp_path / "index.rst" + index.write_text(f"intro\n\n{hook.CENTRAL_MATRIX_HEADER}\n{hook.CENTRAL_MATRIX_FOOTER}\n\noutro\n") + capabilities_yaml = tmp_path / "capabilities.yaml" + capabilities_yaml.write_text(yaml.safe_dump(capabilities_doc())) + monkeypatch.setattr(hook, "INDEX_RST", index) + monkeypatch.setattr( + matrix, + "LANG_SDKS", + [ + { + "id": "go", + "display_name": "Go", + "capabilities_yaml": capabilities_yaml, + "readme": tmp_path / "README.md", + } + ], + ) + return index + + def test_generates_then_is_idempotent(self, wired): + assert hook.main() == 1 + content = wired.read_text() + assert ".. list-table:: Dag definition" in content + assert content.startswith("intro\n") and content.endswith("outro\n") + assert matrix.SUPPORTED_MARK in content + + assert hook.main() == 0 diff --git a/scripts/tests/ci/prek/test_update_ts_sdk_readme_matrix.py b/scripts/tests/ci/prek/test_update_ts_sdk_readme_matrix.py index 57f84edb94f..3ce510afe08 100644 --- a/scripts/tests/ci/prek/test_update_ts_sdk_readme_matrix.py +++ b/scripts/tests/ci/prek/test_update_ts_sdk_readme_matrix.py @@ -20,31 +20,18 @@ import pytest import yaml from ci.prek import lang_sdk_compat_matrix as matrix, update_ts_sdk_readme_matrix as hook -SCHEMA_VERSION = "2026-06-16" - - -def _doc() -> dict: - def entry(supported: bool) -> dict: - return {"supported": supported, "since": "3.3" if supported else None, "note": ""} - - return { - "sdk": "ts", - "supervisor_schema_version": SCHEMA_VERSION, - "min_airflow_version": "3.3", - "states": {state: entry(True) for state, _ in matrix.STATE_DIMENSIONS}, - "capabilities": {cap.name: entry(True) for cap in matrix.CAPABILITY_DIMENSIONS}, - } - class TestMain: @pytest.fixture - def wired(self, tmp_path, monkeypatch): + def wired(self, tmp_path, monkeypatch, capabilities_doc): + doc = capabilities_doc("ts") capabilities_yaml = tmp_path / "capabilities.yaml" - capabilities_yaml.write_text(yaml.safe_dump(_doc())) + capabilities_yaml.write_text(yaml.safe_dump(doc)) readme = tmp_path / "README.md" readme.write_text(f"intro\n\n{matrix.README_MATRIX_HEADER}\n{matrix.README_MATRIX_FOOTER}\n\noutro\n") ts_supervisor = tmp_path / "supervisor.ts" - ts_supervisor.write_text(f'export const SUPERVISOR_API_VERSION = "{SCHEMA_VERSION}" as const;\n') + schema_version = doc["supervisor_schema_version"] + ts_supervisor.write_text(f'export const SUPERVISOR_API_VERSION = "{schema_version}" as const;\n') monkeypatch.setattr( hook, "LANG_SDKS", @@ -56,7 +43,7 @@ class TestMain: def test_generates_target_then_is_idempotent(self, wired): assert hook.main() == 1 content = wired.read_text() - assert "| Dimension | Tier | Supported | Since | Notes |" in content + assert "| Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes |" in content assert matrix.SUPPORTED_MARK in content assert content.startswith("intro\n") and content.endswith("outro\n") @@ -70,4 +57,4 @@ class TestMain: def test_schema_version_matches_ts_source(self): sdk = next(entry for entry in matrix.LANG_SDKS if entry["id"] == hook.SDK_ID) doc = matrix.load_capabilities(sdk["capabilities_yaml"], expected_sdk=hook.SDK_ID) - assert doc["supervisor_schema_version"] == hook.read_ts_schema_version() + assert doc.supervisor_schema_version == hook.read_ts_schema_version() diff --git a/ts-sdk/README.md b/ts-sdk/README.md index 33f8178ea22..490c031bfe6 100644 --- a/ts-sdk/README.md +++ b/ts-sdk/README.md @@ -100,46 +100,62 @@ Do not edit the table by hand. Update the manifest and run the `update-ts-sdk-re -*Min. Airflow version: 3.4 · supervisor schema: 2026-10-30* +*Min. Airflow version: 3.4.0 · supervisor schema: 2026-10-30* -| Dimension | Tier | Supported | Since | Notes | -|---|---|---|---|---| -| **TaskInstance states** | | | | | -| state: `success` | MUST | ✓ | 3.4 | | -| state: `failed` | MUST | ✓ | 3.4 | | -| state: `up_for_retry` | MUST | ✓ | 3.4 | RetryTask | -| state: `skipped` | SHOULD | ✗ | – | runtime does not emit TaskState skipped yet | -| state: `deferred` | MAY | ✗ | – | runtime does not emit DeferTask yet | -| state: `up_for_reschedule` | MAY | ✗ | – | runtime does not emit RescheduleTask yet | -| state: `awaiting_input` | MAY | ✓ | 3.4 | approval() / hitl() in a native Dag | -| state: `removed` | MAY | ✓ | 3.4 | | -| **Runtime capabilities** | | | | | -| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.4 | @task.stub | -| capability: `taskflow-binding` | MUST | ✗ | – | bind @task.stub literal/XCom args to the native handler | -| capability: `task-logging` | MUST | ✓ | 3.4 | structured records over the log socket | -| capability: `xcom-read-write` | MUST | ✓ | 3.4 | getXCom / setXCom | -| capability: `connection-read` | MUST | ✓ | 3.4 | getConnection | -| capability: `variable-read-write` | MUST | ✓ | 3.4 | getVariable / setVariable / deleteVariable | -| capability: `self-contained-bundle` | MUST | ✓ | 3.4 | Airflow metadata embedded in the bundle | -| capability: `retry-policy` | MAY | ✗ | – | no task-facing retry-policy API yet | -| capability: `task-state-store` | MAY | ✓ | 3.4 | taskStateStore.get / set / delete / clear | -| capability: `asset-state-store` | MAY | ✗ | – | no task-facing state-store API yet | -| capability: `asset-event-emit` | MAY | ✗ | – | runtime does not emit asset events yet | -| capability: `asset-event-read` | MAY | ✗ | – | no task-facing asset-event API yet | -| **Native-Dag authoring** | | | | | -| capability: `native-dag-authoring` | SHOULD | ✗ | – | native Dag authoring not implemented yet | -| capability: `task-args` | MUST † | n/a | – | | -| capability: `dag-params` | MUST † | n/a | – | | -| capability: `taskflow-dependencies` | MUST † | n/a | – | | -| capability: `branching` | SHOULD † | n/a | – | | -| capability: `dag-test` | SHOULD † | n/a | – | | -| capability: `task-group` | MAY † | n/a | – | | -| capability: `dynamic-task-mapping` | MAY † | n/a | – | | -| capability: `asset-inlets-outlets` | MAY † | n/a | – | | -| capability: `asset-scheduling` | MAY † | n/a | – | | -| capability: `object-store` | MAY † | n/a | – | no object-storage API yet | +| Dimension | Tier | Supported | Since (Airflow) | Since (SDK) | Notes | +|---|---|---|---|---|---| +| **TaskInstance states** | | | | | | +| state: `success` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| state: `failed` | MUST | ✓ | 3.4.0 | 1.0.0 | | +| state: `up_for_retry` | MUST | ✓ | 3.4.0 | 1.0.0 | RetryTask | +| state: `skipped` | SHOULD | ✓ | 3.4.0 | 1.0.0 | triggerDagRun with skipWhenAlreadyExists only | +| state: `deferred` | MAY | ✓ | 3.4.0 | 1.0.0 | triggerDagRun with deferrable only | +| state: `up_for_reschedule` | MAY | ✗ | – | – | runtime does not emit RescheduleTask yet | +| state: `awaiting_input` | MAY | ✓ | 3.4.0 | 1.0.0 | approval() / hitl() in a Dag definition | +| state: `removed` | MAY | ✓ | 3.4.0 | 1.0.0 | | +| **Task runtime** | | | | | | +| capability: `task-logging` | MUST | ✓ | 3.4.0 | 1.0.0 | structured records over the log socket | +| capability: `xcom-read-write` | MUST | ✓ | 3.4.0 | 1.0.0 | getXCom / setXCom | +| capability: `connection-read` | MUST | ✓ | 3.4.0 | 1.0.0 | getConnection | +| capability: `variable-read-write` | MUST | ✓ | 3.4.0 | 1.0.0 | getVariable / setVariable / deleteVariable | +| capability: `self-contained-bundle` | MUST | ✓ | 3.4.0 | 1.0.0 | Airflow metadata embedded in the bundle | +| capability: `retry-policy` | MAY | ✗ | – | – | no task-facing retry-policy API yet | +| capability: `task-state-store` | MAY | ✓ | 3.4.0 | 1.0.0 | taskStateStore.get / set / delete / clear | +| capability: `asset-state-store` | MAY | ✗ | – | – | no task-facing asset state-store API yet | +| capability: `asset-event-emit` | MAY | ✗ | – | – | runtime does not emit asset events yet | +| capability: `asset-event-read` | MAY | ✗ | – | – | no task-facing asset-event API yet | +| capability: `object-store` | MAY | ✗ | – | – | no object-storage API yet | +| capability: `listeners` | MAY | ✗ | – | – | no hook into the Python listener API, so a task emits no OpenLineage facets of its own | +| **Dag definition** | | | | | | +| capability: `dag-definition-authoring` | SHOULD | ✓ | 3.4.0 | 1.0.0 | new Dag | +| capability: `dag-args` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-args` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `dag-params` | MUST † | ✗ | – | – | | +| capability: `dag-callbacks` | MAY † | ✗ | – | – | | +| capability: `explicit-dependencies` | MUST † | ✓ | 3.4.0 | 1.0.0 | before / after | +| capability: `edge-labels` | MAY † | ✓ | 3.4.0 | 1.0.0 | label(task, text) | +| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `task-group` | MAY † | ✓ | 3.4.0 | 1.0.0 | dag.taskGroup | +| capability: `task-loops` | MAY † | ✗ | – | – | | +| capability: `if-else-branching` | SHOULD † | ✓ | 3.4.0 | 1.0.0 | dag.if | +| capability: `switch-case-branching` | SHOULD † | ✓ | 3.4.0 | 1.0.0 | dag.switch | +| capability: `trigger-dag-run` | MAY † | ✓ | 3.4.0 | 1.0.0 | triggerDagRun | +| capability: `dynamic-dag-generation` | MAY † | ✓ | 3.4.0 | 1.0.0 | | +| capability: `human-in-the-loop` | MAY † | ✓ | 3.4.0 | 1.0.0 | approval() / hitl() | +| capability: `dag-test` | SHOULD † | ✗ | – | – | | +| capability: `asset-definition` | MAY † | ✗ | – | – | | +| capability: `asset-scheduling` | MAY † | ✗ | – | – | | +| capability: `mapped-tasks` | MAY † | ✗ | – | – | | +| capability: `asset-inlets-outlets` | MAY † | ✗ | – | – | | +| capability: `task-callbacks` | MAY † | ✗ | – | – | no SDK-defined task callbacks | +| **Python Dag with stub TaskHandler** | | | | | | +| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.4.0 | 1.0.0 | @task.stub | +| capability: `taskflow-binding` | MUST | ✓ | 3.4.0 | 1.0.0 | mapped stubs excluded | +| capability: `mapped-tasks` | MAY | ✗ | – | – | Airflow does not bind arguments to mapped stubs yet | +| capability: `asset-inlets-outlets` | MAY | ✗ | – | – | runtime does not report task outlets yet | +| capability: `task-callbacks` | MAY | ✗ | – | – | a Python-declared task callback does not run for a stub task | -*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.* +*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `dag-definition-authoring` is supported.* diff --git a/ts-sdk/capabilities.yaml b/ts-sdk/capabilities.yaml index fc5b74dc6d1..6fef910ad01 100644 --- a/ts-sdk/capabilities.yaml +++ b/ts-sdk/capabilities.yaml @@ -17,109 +17,199 @@ --- sdk: ts -min_airflow_version: "3.4" +min_airflow_version: "3.4.0" # Keep in sync with SUPERVISOR_API_VERSION in src/generated/supervisor.ts. The render hook fails # if the two disagree. supervisor_schema_version: "2026-10-30" # The runtime terminates a task with SucceedTask, RetryTask, TaskState (failed/removed), or, for a -# human-in-the-loop task, AwaitInputTask; it does not yet emit skipped, DeferTask, or RescheduleTask. +# human-in-the-loop task, AwaitInputTask; only a triggerDagRun task emits TaskState skipped or +# DeferTask. It does not emit RescheduleTask. states: success: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" failed: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" up_for_retry: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "RetryTask" skipped: - supported: false - note: "runtime does not emit TaskState skipped yet" + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "triggerDagRun with skipWhenAlreadyExists only" deferred: - supported: false - note: "runtime does not emit DeferTask yet" + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "triggerDagRun with deferrable only" up_for_reschedule: supported: false note: "runtime does not emit RescheduleTask yet" awaiting_input: supported: true - since: "3.4" - note: "approval() / hitl() in a native Dag" + since: "3.4.0" + sdk_version: "1.0.0" + note: "approval() / hitl() in a Dag definition" removed: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" -# Runtime capabilities reflect the task-facing coordinator client surface; native-Dag authoring is -# not implemented yet, so every native capability is unsupported. +# Runtime capabilities apply to a task body in both authoring modes. capabilities: - mixed-lang-stub-target: - supported: true - since: "3.4" - note: "@task.stub" - taskflow-binding: - supported: false - note: "bind @task.stub literal/XCom args to the native handler" task-logging: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "structured records over the log socket" xcom-read-write: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "getXCom / setXCom" connection-read: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "getConnection" variable-read-write: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "getVariable / setVariable / deleteVariable" self-contained-bundle: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "Airflow metadata embedded in the bundle" retry-policy: supported: false note: "no task-facing retry-policy API yet" task-state-store: supported: true - since: "3.4" + since: "3.4.0" + sdk_version: "1.0.0" note: "taskStateStore.get / set / delete / clear" asset-state-store: supported: false - note: "no task-facing state-store API yet" + note: "no task-facing asset state-store API yet" asset-event-emit: supported: false note: "runtime does not emit asset events yet" asset-event-read: supported: false note: "no task-facing asset-event API yet" - native-dag-authoring: - supported: false - note: "native Dag authoring not implemented yet" - task-args: - supported: false - dag-params: - supported: false - taskflow-dependencies: - supported: false - branching: - supported: false - dag-test: - supported: false - task-group: - supported: false - dynamic-task-mapping: - supported: false - asset-inlets-outlets: - supported: false - asset-scheduling: - supported: false object-store: supported: false note: "no object-storage API yet" + listeners: + supported: false + note: "no hook into the Python listener API, so a task emits no OpenLineage facets of its own" + # Dag definition: the whole Dag is declared in the SDK's language. + dag-definition-authoring: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "new Dag" + dag-args: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + task-args: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + dag-params: + supported: false + dag-callbacks: + supported: false + explicit-dependencies: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "before / after" + edge-labels: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "label(task, text)" + taskflow-dependencies: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + task-group: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "dag.taskGroup" + task-loops: + supported: false + if-else-branching: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "dag.if" + switch-case-branching: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "dag.switch" + trigger-dag-run: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "triggerDagRun" + dynamic-dag-generation: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + human-in-the-loop: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "approval() / hitl()" + dag-test: + supported: false + asset-definition: + supported: false + asset-scheduling: + supported: false + # Python Dag with stub TaskHandler. + mixed-lang-stub-target: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "@task.stub" + taskflow-binding: + supported: true + since: "3.4.0" + sdk_version: "1.0.0" + note: "mapped stubs excluded" + # Shared by both authoring modes, so declared once per mode. + mapped-tasks: + dag-definition: + supported: false + stub-task-handler: + supported: false + note: "Airflow does not bind arguments to mapped stubs yet" + asset-inlets-outlets: + dag-definition: + supported: false + stub-task-handler: + supported: false + note: "runtime does not report task outlets yet" + task-callbacks: + dag-definition: + supported: false + note: "no SDK-defined task callbacks" + stub-task-handler: + supported: false + note: "a Python-declared task callback does not run for a stub task"