jason810496 commented on code in PR #71757:
URL: https://github.com/apache/airflow/pull/71757#discussion_r4347607945
##########
ts-sdk/capabilities.yaml:
##########
@@ -17,109 +17,187 @@
---
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"
+ since: "3.4.0"
+ sdk_version: "1.0.0"
note: "approval() / hitl() in a native Dag"
Review Comment:
Applied in 4b7c04f.
##########
scripts/ci/prek/lang_sdk_compat_matrix.py:
##########
@@ -43,17 +43,30 @@
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
+# ``<capability>: {dag-definition: <entry>, stub-task-handler: <entry>}``.
+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[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"
Review Comment:
Agreed. Renamed to `dag-definition-authoring` in 4b7c04f, so the capability
and the authoring mode no longer share a spelling.
##########
java-sdk/capabilities.yaml:
##########
@@ -52,76 +57,136 @@ 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"
+ # Dag definition: the whole Dag is declared in the SDK's language.
+ dag-definition:
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
- taskflow-dependencies:
+ explicit-dependencies:
supported: true
- since: "3.4"
- branching:
+ since: "3.4.0"
+ sdk_version: "1.0.0"
+ note: "before / after, or @Airflow.Deps"
+ edge-labels:
Review Comment:
The v3-4 backport will wait for #75365 just merged.
##########
scripts/tests/ci/prek/test_lang_sdk_compat_matrix.py:
##########
@@ -32,19 +32,38 @@ def _doc(sdk_id: str = "go", **overrides) -> dict:
"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},
+ "capabilities": {
+ cap.name: {mode: _entry(True, since="3.3") for mode in
matrix.AUTHORING_MODES}
+ if cap.name in matrix.SHARED_CAPABILITY_NAMES
+ else _entry(True, since="3.3")
+ for cap in matrix.CAPABILITY_DIMENSIONS
+ },
}
doc.update(overrides)
return doc
-class TestValidateCapabilities:
+def _doc_without_dag_definition(sdk_id: str = "go") -> dict:
+ """A valid document for an SDK that does not define Dags in its own
language."""
+ doc = _doc(sdk_id)
+ doc["capabilities"][matrix.DAG_DEFINITION_GATE] = _entry(False)
+ for row in matrix.build_matrix_parts()[1].groups[0].rows:
Review Comment:
Done in 4b7c04f: `build_matrix_parts()` returns a `MatrixParts` NamedTuple
(`common`, `dag_definition`, `stub_task_handler`), so a test names the part
instead of indexing it. Both renderers still iterate it unchanged.
##########
scripts/ci/prek/lang_sdk_compat_matrix.py:
##########
@@ -43,17 +43,30 @@
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
+# ``<capability>: {dag-definition: <entry>, stub-task-handler: <entry>}``.
+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[str, Mode | None]
Review Comment:
Done in 4b7c04f: `EntryKey` is `tuple[RowKind, str, Mode | None]`, states
are keyed `("state", name, None)` and capabilities `("capability", name,
mode_or_None)`.
##########
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 <language-sdks/dag-definition>`, and the features of a
+:ref:`Python Dag with stub TaskHandler <language-sdks/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 †)
+ - ✗
+ - ✗
+ - ✗
+ * - ``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
+ * - ``provider-operators`` (MAY †)
+ - ✗
+ - ✗
+ - ✗
+ * - ``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 †)
+ - ✗
+ - ✗
+ - ✗
+ * - ``dynamic-task-mapping`` (MAY †)
Review Comment:
Cool, I didn't know that! Addressed in
https://github.com/apache/airflow/pull/71757/commits/0915399b0c2162802ae8f05e7d2c270213a70574
, thanks.
##########
scripts/ci/prek/lang_sdk_compat_matrix.py:
##########
@@ -88,54 +100,136 @@ class Capability(NamedTuple):
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(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("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("dynamic-task-mapping", "MAY", "shared", True),
+ Capability("asset-inlets-outlets", "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:
+ # Only a shared capability carries one entry per authoring mode; every
other dimension is
+ # declared once, so its mode is dropped from the key.
+ return self.name, (self.mode if self.name in SHARED_CAPABILITY_NAMES
else None)
+
+
+GATE_KEY: Final[EntryKey] = (DAG_DEFINITION_GATE, None)
+
+
+class MatrixGroup(NamedTuple):
+ label: str
+ rows: list[MatrixRow]
+
+
+class MatrixPart(NamedTuple):
+ title: str
+ groups: list[MatrixGroup]
+
+
+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)
Review Comment:
Applied in 4b7c04f, together with the `RowKind` key above, so `entry_key` is
`return self.kind, self.name, self.mode` with no `SHARED_CAPABILITY_NAMES`
check left.
##########
scripts/tests/ci/prek/test_update_lang_sdk_central_matrix.py:
##########
@@ -0,0 +1,169 @@
+# 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
+
+
+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}
+
+
+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: _entry(True, since="3.3") for state, _ in
matrix.STATE_DIMENSIONS},
+ "capabilities": {
+ cap.name: {mode: _entry(True, since="3.3") for mode in
matrix.AUTHORING_MODES}
+ if cap.name in matrix.SHARED_CAPABILITY_NAMES
+ else _entry(True, since="3.3")
+ for cap in matrix.CAPABILITY_DIMENSIONS
+ },
+ }
+ doc.update(overrides)
+ return doc
Review Comment:
Done in 4b7c04f: `capabilities_entry`, `capabilities_doc` and
`capabilities_doc_without_dag_definition` are now factory fixtures in
`scripts/tests/ci/prek/conftest.py`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
