You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@airflow.apache.org by "jason810496 (via GitHub)" <gi...@apache.org> on 2026/07/22 03:21:51 UTC

[PR] Bind TaskFlow stub-task call arguments in the Go SDK runtime [airflow]

jason810496 opened a new pull request, #70209:
URL: https://github.com/apache/airflow/pull/70209

   - **Depends on https://github.com/apache/airflow/pull/69757 get merged first** (this branch is stacked on `feature/lang-sdk/taskflow-stub-dag`; only the tip commit is new here).
   - **Diff for early review**: https://github.com/jason810496/airflow/compare/feature/lang-sdk/taskflow-stub-dag...feature/go-sdk/taskflow-arg-binding
   - related: split out of #69757 to keep that PR Core + Task SDK only.
   
   ## Why
   
   #69757 ships the Python-side contract: a `@task.stub` TaskFlow call is serialized as an ordered arg-binding spec and returned by `ti_run` as `TIRunContext.arg_bindings`. This PR makes the Go SDK actually consume it — Go task functions receive the Dag file's literals and upstream XComs as typed parameters instead of hand-writing `GetXCom` calls with hard-coded upstream task ids:
   
   ```python
   @task.stub(queue="golang")
   def transform(country: str, extracted: dict): ...
   
   with DAG(...):
       transform("uk", extract())   # extract() is a normal Python @task
   ```
   
   ```go
   // The runtime binds "uk" onto country and pulls extract's XCom into extracted.
   func transform(ctx sdk.TIRunContext, log *slog.Logger, country string, extracted map[string]any) error
   ```
   
   ## Supported TaskFlow syntax
   
   The example bundle's `taskflow_binding_dag` (`go-sdk/dags/go_examples.py` + `go-sdk/example/bundle/taskflowbinding/`) exercises the full surface end to end:
   
   ```python
   @task.stub(queue="golang")
   def via_flat_args(
       name: str, count: int, ratio: float, enabled: bool,
       tags: list, config: dict, numbers: list, note: str | None = None,
   ): ...
   
   @dag(dag_id="taskflow_binding_dag")
   def taskflow_binding_dag():
       via_flat_args(
           "summary", 3, 2.5, True,       # positional scalar literals (str/int/float/bool)
           ["metrics", "hourly"],         # array literal
           config=make_config(),          # keyword arg: XCom from another @task.stub
           numbers=make_numbers(),        # XCom binding onto a typed array parameter
       )                                  # `note` unpassed: its None default is captured as from_default
       region = make_region()
       via_struct_no_tags(RegionCode=region, Threshold=0.75)   # one XCom fanned into several calls
       via_struct_arg_tag(region_code=region, threshold=0.75)  # literal + XCom mixed as kwargs
       via_struct_unmatched_arg(region_code=region)            # defaulted param left unpassed
   ```
   
   On the Go side a task declares **either** flat positional data parameters **or** one keyword-style `sdk.TaskInput` struct (mixing the two shapes is rejected at registration as too ambiguous):
   
   ```go
   // Flat parameters: bound in declaration order after the injectables
   // (sdk.TIRunContext, *slog.Logger, context.Context, client interfaces).
   // Arity or declared-type mismatches fail the task before its body runs.
   func ViaFlatArgs(ctx sdk.TIRunContext, log *slog.Logger,
       name string, count int, ratio float64, enabled bool,
       tags []string, config Config, numbers []int, note *string) (any, error)
   
   // TaskInput struct: exported fields bind per call-argument name.
   type ViaStructArgTagInput struct {
       sdk.TaskInput
       Region    string  `arg:"region_code"` // explicit argument name via tag
       Threshold float64 `arg:"threshold"`   // untagged fields bind their verbatim Go field name
   }
   func ViaStructArgTag(ctx sdk.TIRunContext, log *slog.Logger, input ViaStructArgTagInput) (any, error)
   ```
   
   Binding semantics, mirroring positional vs keyword calls:
   
   - **Flat parameters** are positional: every data parameter must be filled, literals decode directly, XCom entries pull on demand, and `None` onto a pointer arrives as `nil`.
   - **`sdk.TaskInput` fields** are keyword-like: a field whose name/tag matches no call argument stays at its Go zero value; an argument the Dag author *explicitly passed* that no field claims fails the task; `from_default` entries (captured stub defaults) may go unclaimed.
   - A malformed or unknown-kind spec, and a missing spec for a struct with bindable fields, fail the task loudly before its body runs — replacing the old silent `reflect.Zero` fill.
   
   ## How
   
   - New `pkg/binding` package: `Analyze` classifies parameters once at registration (injectables vs JSON-decodable *data parameters*, or the single `TaskInput` struct); `Resolve` binds the wire spec onto them. The wire union surfaces as a sealed sum type (`binding.XComArg`/`binding.LiteralArg`) whose variants and `DataType` vocabulary are defined in terms of the generated `genmodels` schema types, so the runtime types cannot drift from the wire model.
   - `TaskWithArgs` dispatch in the task runner with strict wire-spec validation (missing/empty `name` or xcom `task_id` fails the task before its body runs); arity and malformed-spec errors share retry semantics.
   - Ad hoc `xcom:`/`xcom-key:` struct tags were considered and dropped — they would hard-code upstream task ids in the Go binary, hiding wiring the Dag file owns; a task that needs an extra XCom still asks the injected client explicitly.
   - msgpack encoder now honours `json` struct tags so typed XComs round-trip.
   - Regenerated `genmodels` from the supervisor schema `2026-07-30` + `SupervisorSchemaVersion` bump.
   
   ## What
   
   - `go-sdk`: `pkg/binding` (flat positional binding + `sdk.TaskInput` struct-field injection); task-runner dispatch; regenerated genmodels; msgpack `json`-tag support; example bundle (`taskflow_binding_dag` covering the full binding surface, with every `via_struct_*` task binding an XCom-sourced field from `make_region` alongside a literal); README + ADR 0003 updated.
   - `airflow-e2e-tests`: `test_go_sdk_taskflow_binding.py` drives the example Dag end to end and asserts every task bound its arguments.
   - Go tests: binding/arity/type/TaskInput suites and task-runner integration tests (including malformed- and incomplete-spec fail-fast).
   
   ---
   
   ##### Was generative AI tooling used to co-author this PR?
   
   - [x] Yes, with help of Claude Code (Fable 5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
   


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org