jason810496 commented on code in PR #72019: URL: https://github.com/apache/airflow/pull/72019#discussion_r4005569889
########## java-sdk/adr/0002-native-dag-interface.md: ########## @@ -0,0 +1,322 @@ +<!-- + 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 Java 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 +Java itself must express the graph, the Dag/task configuration, and the task bodies. This ADR is +scoped to what that Java call site looks like for a user. It also settles the injectable +`client`/`context` question shared with [ADR-0001](0001-mixed-lang-dag-interface.md): they are +**injected as method arguments**, not exposed through getters. It shares the 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 + +### Annotation based + +`Client` and `Context` are injected as **method arguments**, exactly as on the mixed-language +surface ([ADR-0001](0001-mixed-lang-dag-interface.md)) — there are no getters and no SDK base class +to extend. A task method's signature is its injected arguments, if any, followed by its data: + +```java [email protected](id = "java_etl", schedule = "@daily") +public class EtlPipeline { // extends nothing of ours; your own base class stays free + + @Builder.Task(id = "extract", retries = 2) + public long extract(Client client) { + return ((Number) client.getVariable("row_count")).longValue(); + } + + @Builder.Task(id = "transform") + public long transform(Client client, Context context, long extracted, double threshold) { + client.setXCom("scaled_in", context.runId); + return (long) (extracted * threshold); + } + + @Builder.Task(id = "load") + public void load(Context context, long transformed) { + log.log(INFO, "Loaded {0} for run {1}", transformed, context.runId); + } + + @Builder.Task(id = "audit") + public void audit(Client client) { /* side effect only, no data in or out */ } + + @Builder.Task(id = "notify") + public void notify(Client client) { /* side effect only, no data in or out */ } + + @Builder.Deps + static class Wiring implements EtlPipelineDeps { + void depends() { + // TaskFlow (data) edges: implied by passing a TaskRef; a constant is wrapped with lit(...) + var rows = extract(); + var scaled = transform(rows, lit(0.9)); + var loaded = load(scaled); + + // non-TaskFlow (ordering-only) edges: sequence with no data flowing + rows.then(audit()); // extract >> audit Review Comment: We standardize the term using `.before` and `.after` in https://github.com/apache/airflow/pull/72765. ```suggestion rows.before(audit()); // extract >> audit ``` ########## java-sdk/adr/0002-native-dag-interface.md: ########## @@ -94,18 +138,57 @@ public final class EtlPipeline_Dag { public void execute(Context context, Client client) throws Exception { TaskArgs args = TaskArgs.of(context); long extracted = args.require(0, Long.class); - EtlPipeline tasks = new EtlPipeline(); - tasks.bind(context, client); - client.setXCom(tasks.transform(extracted)); + double threshold = 0.9; // baked from lit(0.9) at Dag-build time + client.setXCom(new EtlPipeline().transform(client, context, extracted, threshold)); } } - // Extract and Load follow the same shape. + // Extract, Load, Audit, and Notify follow the same shape. } ``` -`bind(context, client)` is what `getClient()` and `getContext()` return for that invocation, and a -getter called with nothing bound throws. +The injected arguments are passed straight into the user method, and the data arguments bind by +position through the same internal `TaskArgs` the mixed-language surface uses +([ADR-0001](0001-mixed-lang-dag-interface.md)) — no getters, no `bind()`, nothing ambient. + +**Literals.** A data argument is an `Arg<T>`, which a `TaskRef` satisfies; a constant is wrapped +with `lit(...)` — `transform(extract(), lit(0.9))` — and recorded as a baked value with no edge. Review Comment: Cross Lang SDK alignment question: Do we need to support the Literal for the Lang SDK case now? I agreed it's a valid feature (it was raised up several time during my discussion with Agent as well), but both Go and TS haven't design this feature. We can make Java be the first one to support it, as long as we have track this down in the compatible matrix. ########## java-sdk/adr/0002-native-dag-interface.md: ########## @@ -94,18 +138,57 @@ public final class EtlPipeline_Dag { public void execute(Context context, Client client) throws Exception { TaskArgs args = TaskArgs.of(context); long extracted = args.require(0, Long.class); - EtlPipeline tasks = new EtlPipeline(); - tasks.bind(context, client); - client.setXCom(tasks.transform(extracted)); + double threshold = 0.9; // baked from lit(0.9) at Dag-build time + client.setXCom(new EtlPipeline().transform(client, context, extracted, threshold)); } } - // Extract and Load follow the same shape. + // Extract, Load, Audit, and Notify follow the same shape. } ``` -`bind(context, client)` is what `getClient()` and `getContext()` return for that invocation, and a -getter called with nothing bound throws. +The injected arguments are passed straight into the user method, and the data arguments bind by +position through the same internal `TaskArgs` the mixed-language surface uses +([ADR-0001](0001-mixed-lang-dag-interface.md)) — no getters, no `bind()`, nothing ambient. + +**Literals.** A data argument is an `Arg<T>`, which a `TaskRef` satisfies; a constant is wrapped +with `lit(...)` — `transform(extract(), lit(0.9))` — and recorded as a baked value with no edge. +Wrapping is required because a bare `Integer` cannot implement `Arg`; boxed types only, no +primitives. + +### Non-TaskFlow dependencies + +A TaskFlow (data) edge comes for free from passing a `TaskRef` into another wiring method. A +dependency where **no data flows** — Python's `a >> b`, and the list forms `a >> [b, c]` and +`[x, y] >> z` — is expressed instead with two variadic verbs that every `TaskRef` carries, `then` +and `after`. Both live on a small `Chain` interface, and a `TaskRef` is a `Chain` of one +(`interface TaskRef<T> extends Arg<T>, Chain`): + +```java +a.then(b, c); // a >> [b, c] +z.after(x, y); // z << [x, y] +``` + +`then` and `after` are mirror images, and each returns the **new frontier** (the set it just pointed +at), the way `>>`/`<<` evaluate to their right operand, so a chain walks through a fan: + +```java +a.then(b, c).then(d); // a >> [b, c] >> d (a->b, a->c, then b->d, c->d) +``` + +The one thing the verbs cannot do is start from a *set*: Java can't overload `>>` the way Python +does, and there is no list literal to call `.then` on, so `Flow.of` opens a chain from one: + +```java +Flow.of(a, b).then(c); // [a, b] >> c Review Comment: Would `Order.of` be more straightforward? No strong opinion on the naming though. -- 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]
