pierrejeambrun commented on code in PR #72047: URL: https://github.com/apache/airflow/pull/72047#discussion_r3933245933
########## ts-sdk/adr/0002-native-dag-interface.md: ########## @@ -0,0 +1,143 @@ +<!-- + 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. + --> + +# ADR-0002: Native TypeScript Dag — Interface Design + +## Status + +Proposed + +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case is the separate `MixedLangDag` class, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). This ADR shares the injectable `ctx`/`client` question raised there, and shares its protocol substrate (the argument-binding spec) with [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Decision + +`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow itself uses for `load(transformed=transform(...))`: + +```ts +const dag = new Dag("ts_etl"); + +const extract = dag.task("extract", async ({ client }): Promise<number> => { + const rows = 42; + await client.setXCom({ key: "row_count", value: rows }); + return rows; +}); + +interface TransformArgs { + extracted: number; +} + +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); + +interface LoadArgs { + transformed: number; +} + +const load = dag.task("load", async ({ ctx, client, transformed }: LoadArgs & TaskHandlerArgs) => { + if (transformed <= 0) { + throw new Error(`task ${ctx.taskId} received a non-positive value: ${transformed}`); + } + await client.setXCom({ key: "loaded", value: transformed }); +}); + +load({ transformed: transform({ extracted: extract() }) }); +``` + +The call graph is the task graph. `tsc` checks every wired key against the handler's own parameter type, and a `TaskRef` only exists once its producing call has returned, so a cycle is unrepresentable rather than merely rejected by a validator. Every task must be called exactly once; an uncalled task fails when the Dag is read, so a task can't be silently left out of the graph. + +## If `ctx`/`client` were real getter methods instead of injected arguments + +The intersection type above, `TransformArgs & TaskHandlerArgs`, exists for one reason: today's handler signature carries `ctx`/`client` as arguments, so a handler that wants type safety on its own data has to say so explicitly. If `ctx`/`client` came from getter functions instead, a handler's parameter type would be exactly its own data: + +```ts +// Before: ctx/client share the handler's one argument object, so +// TransformArgs must be intersected with TaskHandlerArgs. +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); +``` + +```ts +// After: ctx/client come from getters, so the handler's parameter type is +// exactly its own data, with no TaskHandlerArgs intersection needed. +import { getClient } from "@apache-airflow/ts-sdk"; + +const transform = dag.task("transform", async ({ extracted }: TransformArgs) => { + const client = getClient(); + await client.setXCom({ key: "doubled", value: extracted * 2 }); + return extracted * 2; +}); +``` Review Comment: This is nice too. The only point for the injected arguments is that it mimics the python interface. But does it have to really? Not sure, I probably like this one better. Makes thinks (types) and signatures more readable I believe. ########## ts-sdk/adr/0002-native-dag-interface.md: ########## @@ -0,0 +1,143 @@ +<!-- + 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. + --> + +# ADR-0002: Native TypeScript Dag — Interface Design + +## Status + +Proposed + +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case is the separate `MixedLangDag` class, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). This ADR shares the injectable `ctx`/`client` question raised there, and shares its protocol substrate (the argument-binding spec) with [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Decision + +`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow itself uses for `load(transformed=transform(...))`: + +```ts +const dag = new Dag("ts_etl"); + +const extract = dag.task("extract", async ({ client }): Promise<number> => { + const rows = 42; + await client.setXCom({ key: "row_count", value: rows }); + return rows; +}); + +interface TransformArgs { + extracted: number; +} + +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); + +interface LoadArgs { + transformed: number; +} + +const load = dag.task("load", async ({ ctx, client, transformed }: LoadArgs & TaskHandlerArgs) => { + if (transformed <= 0) { + throw new Error(`task ${ctx.taskId} received a non-positive value: ${transformed}`); + } + await client.setXCom({ key: "loaded", value: transformed }); +}); + +load({ transformed: transform({ extracted: extract() }) }); Review Comment: THis is hard to read, and if there is 25 tasks like this, it basically becomes unreadable no ? How do you handle `[task1, task2] >> task3` ? Via the params `transform({ extracted: extract(), extrtacted2: extract2() })`? ########## ts-sdk/adr/0001-mixed-lang-dag-interface.md: ########## @@ -0,0 +1,89 @@ +<!-- + 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. + --> + +# ADR-0001: Mixed-Lang Dag — TypeScript Task Interface + +## Status + +Proposed + +## Context + +The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume those bindings directly as named arguments, instead of re-fetching each value with `client.getXCom(...)`. + +This ADR covers only the TypeScript call-site interface. The argument-binding spec itself (its shape, how it's materialized, how it travels over the wire) is a separate, protocol-level decision recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). Given that spec, this ADR only answers what TypeScript code a user writes. + +A mixed-language Dag declares its structure in Python and supplies task bodies from TypeScript, using `MixedLangDag`, a class dedicated to this mode. The native case, where TypeScript owns the graph too, is a separate `Dag` class, covered in [ADR-0002](0002-native-dag-interface.md). + +## Decision + +TypeScript uses one syntax for the TaskFlow binding: named arguments merged onto the handler's single parameter object, alongside the SDK's own `ctx`/`client`. + +```ts +const dag = new MixedLangDag("etl"); + +interface TransformArgs { + region_code: string; + threshold: number; +} + +async function transform({ ctx, client, region_code, threshold }: TransformArgs & TaskHandlerArgs) { + const rows = await client.getXCom<number>({ key: "return_value", taskId: "extract" }); + if (rows === null) { + throw new Error(`task ${ctx.taskId} has no upstream row count to transform`); + } + const passed = rows >= threshold; + await client.setXCom({ key: "region", value: region_code }); + return { region_code, passed }; +} + +dag.task("transform", transform); +``` + +Renaming a Python name that isn't idiomatic TypeScript is ordinary destructuring, not a separate mechanism: + +```ts +async function report({ run_label: runLabel }: { run_label: string } & TaskHandlerArgs) { + if (runLabel !== "nightly") { + throw new Error(`expected run label "nightly" but got "${runLabel}"`); + } +} + Review Comment: I don't know if it's possible, but ideally, all those python names are transformed into ts names in the deeper layer. (The ts sdk I suppose which should have a transformer to automatically modify all those names, similarly to what we do in the public API client.) To avoid having to do that manually everywhere because it's not pretty and annoying to having to rename everything. ########## ts-sdk/adr/0001-mixed-lang-dag-interface.md: ########## @@ -0,0 +1,89 @@ +<!-- + 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. + --> + +# ADR-0001: Mixed-Lang Dag — TypeScript Task Interface + +## Status + +Proposed + +## Context + +The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume those bindings directly as named arguments, instead of re-fetching each value with `client.getXCom(...)`. + +This ADR covers only the TypeScript call-site interface. The argument-binding spec itself (its shape, how it's materialized, how it travels over the wire) is a separate, protocol-level decision recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). Given that spec, this ADR only answers what TypeScript code a user writes. + +A mixed-language Dag declares its structure in Python and supplies task bodies from TypeScript, using `MixedLangDag`, a class dedicated to this mode. The native case, where TypeScript owns the graph too, is a separate `Dag` class, covered in [ADR-0002](0002-native-dag-interface.md). + +## Decision + +TypeScript uses one syntax for the TaskFlow binding: named arguments merged onto the handler's single parameter object, alongside the SDK's own `ctx`/`client`. + +```ts +const dag = new MixedLangDag("etl"); + +interface TransformArgs { + region_code: string; + threshold: number; +} + +async function transform({ ctx, client, region_code, threshold }: TransformArgs & TaskHandlerArgs) { Review Comment: I would move `&` to be done directly above (basically extends the `TaskHandlerArgs` when defining the TasnformArgs. This way `transform` type is straightforward and easier to read. ########## ts-sdk/adr/0001-mixed-lang-dag-interface.md: ########## @@ -0,0 +1,89 @@ +<!-- + 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. + --> + +# ADR-0001: Mixed-Lang Dag — TypeScript Task Interface + +## Status + +Proposed + +## Context + +The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume those bindings directly as named arguments, instead of re-fetching each value with `client.getXCom(...)`. + +This ADR covers only the TypeScript call-site interface. The argument-binding spec itself (its shape, how it's materialized, how it travels over the wire) is a separate, protocol-level decision recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). Given that spec, this ADR only answers what TypeScript code a user writes. + +A mixed-language Dag declares its structure in Python and supplies task bodies from TypeScript, using `MixedLangDag`, a class dedicated to this mode. The native case, where TypeScript owns the graph too, is a separate `Dag` class, covered in [ADR-0002](0002-native-dag-interface.md). + +## Decision + +TypeScript uses one syntax for the TaskFlow binding: named arguments merged onto the handler's single parameter object, alongside the SDK's own `ctx`/`client`. + +```ts +const dag = new MixedLangDag("etl"); + +interface TransformArgs { + region_code: string; + threshold: number; +} + +async function transform({ ctx, client, region_code, threshold }: TransformArgs & TaskHandlerArgs) { + const rows = await client.getXCom<number>({ key: "return_value", taskId: "extract" }); + if (rows === null) { + throw new Error(`task ${ctx.taskId} has no upstream row count to transform`); + } + const passed = rows >= threshold; + await client.setXCom({ key: "region", value: region_code }); + return { region_code, passed }; +} + +dag.task("transform", transform); +``` + +Renaming a Python name that isn't idiomatic TypeScript is ordinary destructuring, not a separate mechanism: + +```ts +async function report({ run_label: runLabel }: { run_label: string } & TaskHandlerArgs) { + if (runLabel !== "nightly") { + throw new Error(`expected run label "nightly" but got "${runLabel}"`); + } +} + +dag.task("report", report); +``` + +### How + +- `MixedLangDag.task()` returns a plain `TaskRef`, not a callable factory. There is nothing to wire, since the Python file already owns task order. Calling it the way a native `Dag`'s factory is called (`transform()`) is a compile error, since `TaskRef` has no call signature, not a runtime throw. +- Wire names match the Python parameter names character for character, with no case- or separator-insensitive fallback. Renaming happens once, at the destructuring site. +- `ctx` and `client` are reserved, permanently: bound arguments are merged flat into the same object alongside them, and a bound name that collides with either fails the task at dispatch. +- An upstream's return value is not delivered as a bound argument. Read it explicitly via `client.getXCom({ key: "return_value", taskId: "..." })`. +- `tsc` cannot check a handler's destructuring pattern against the Python call site. A typo binds `undefined` silently; the runtime logs the bound names at dispatch and includes them in a handler-failure message, so the mismatch is diagnosable from the task log. + +## Open Questions + +- Should the decoded bindings also be exposed as a public, positional/raw accessor (name/value pairs, no interface required), or should that stay an internal runtime detail? +- Should `ctx` and `client` become explicit getter functions (`getClient()`, `getContext()`) instead of arguments merged into the handler's object? (See [ADR-0002](0002-native-dag-interface.md) for where this same question resurfaces on the native-Dag side.) Review Comment: We could check how the gosdk does this. I would love to get rid of the `TaskHandlerArgs` somehow. Which I do not find explicit. (client and ctx are) ########## ts-sdk/adr/0002-native-dag-interface.md: ########## @@ -0,0 +1,143 @@ +<!-- + 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. + --> + +# ADR-0002: Native TypeScript Dag — Interface Design + +## Status + +Proposed + +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case is the separate `MixedLangDag` class, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). This ADR shares the injectable `ctx`/`client` question raised there, and shares its protocol substrate (the argument-binding spec) with [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Decision + +`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow itself uses for `load(transformed=transform(...))`: + +```ts +const dag = new Dag("ts_etl"); + +const extract = dag.task("extract", async ({ client }): Promise<number> => { + const rows = 42; + await client.setXCom({ key: "row_count", value: rows }); + return rows; +}); + +interface TransformArgs { + extracted: number; +} + +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); + +interface LoadArgs { + transformed: number; +} + +const load = dag.task("load", async ({ ctx, client, transformed }: LoadArgs & TaskHandlerArgs) => { + if (transformed <= 0) { + throw new Error(`task ${ctx.taskId} received a non-positive value: ${transformed}`); + } + await client.setXCom({ key: "loaded", value: transformed }); +}); + +load({ transformed: transform({ extracted: extract() }) }); +``` + +The call graph is the task graph. `tsc` checks every wired key against the handler's own parameter type, and a `TaskRef` only exists once its producing call has returned, so a cycle is unrepresentable rather than merely rejected by a validator. Every task must be called exactly once; an uncalled task fails when the Dag is read, so a task can't be silently left out of the graph. + +## If `ctx`/`client` were real getter methods instead of injected arguments + +The intersection type above, `TransformArgs & TaskHandlerArgs`, exists for one reason: today's handler signature carries `ctx`/`client` as arguments, so a handler that wants type safety on its own data has to say so explicitly. If `ctx`/`client` came from getter functions instead, a handler's parameter type would be exactly its own data: + +```ts +// Before: ctx/client share the handler's one argument object, so +// TransformArgs must be intersected with TaskHandlerArgs. +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); +``` + +```ts +// After: ctx/client come from getters, so the handler's parameter type is +// exactly its own data, with no TaskHandlerArgs intersection needed. +import { getClient } from "@apache-airflow/ts-sdk"; + +const transform = dag.task("transform", async ({ extracted }: TransformArgs) => { + const client = getClient(); + await client.setXCom({ key: "doubled", value: extracted * 2 }); + return extracted * 2; +}); +``` + +Open question this ADR does not resolve: + +- What backs `getClient()`/`getContext()` at run time. A Node `AsyncLocalStorage` scoped to the handler's execution is the likely mechanism, but confirming it survives every `await` inside a handler, and any user-spawned concurrency, is a runtime-dispatch question, not a call-site question. This ADR only proposes the shape. + +If that question is answered, the implications are: + +- The top-level argument namespace closes automatically: `ctx`/`client` can never collide with a Dag author's own parameter name, because they no longer share an object with one. +- A handler becomes directly unit-testable with a plain data argument. No `TaskHandlerArgs` fixture is needed, and there's no risk of forgetting to intersect it in. +- The wiring surface (`TaskInputs<TArgs>`, the mapped type checked at the call site) is unaffected either way, since it already derives from the handler's own declared parameter type, whichever shape that type takes. + +### This also reopens positional wiring + +The "after" handler above still takes one object (`{ extracted }: TransformArgs`), out of habit. But the *reason* wiring is named rather than positional today is that a handler takes one destructured object, so `Parameters<typeof handler>` is always a one-element tuple with no per-field position to check against. That reason goes away once data no longer has to share an object with `ctx`/`client`: a handler can drop the object entirely and take its arguments positionally, the way an ordinary TypeScript function does. + +```ts +// Before (named wiring, current): the wiring object's keys match each +// handler's destructured object. +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); +const load = dag.task("load", async ({ transformed }: LoadArgs & TaskHandlerArgs) => { + /* ... */ +}); +load({ transformed: transform({ extracted: extract() }) }); +``` + +```ts +// After (positional wiring, hypothetical): each handler takes its data +// positionally, so `Parameters<typeof handler>` is a real tuple (`[number]` +// for both transform and load), and a positional TaskFactory can type-check +// a call against it, the same way today's `Wiring<TParams>` type-checks a +// named one. +const extract = dag.task("extract", async () => 42); +const transform = dag.task("transform", async (extracted: number) => extracted * 2); +const load = dag.task("load", async (transformed: number) => { + const client = getClient(); + await client.setXCom({ key: "loaded", value: transformed }); +}); + +load(transform(extract())); +``` + Review Comment: One more point for getters. -- 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]
