Go-SDK: Support TaskFlow#68311
Conversation
jason810496
left a comment
jason810496
left a comment
There was a problem hiding this comment.
This is a final user interface changes before the Go-SDK beta release.
Would appreciate review and welcome any suggestion, thanks.
| // ExtractInput declares extract's XCom inputs. The `xcom:"python_task_1"` tag | ||
| // binds this field to the return_value of the upstream Python task | ||
| // `python_task_1`, so the Go task receives a Python task's output without | ||
| // calling client.GetXCom itself. The runtime pulls and decodes it before | ||
| // extract runs. | ||
| type ExtractInput struct { | ||
| FromPython string `xcom:"python_task_1"` | ||
| } |
There was a problem hiding this comment.
No matter how many upstream tasks there are, we will only inject all the upstream XCom result as one struct.
User should define all the upstream task_id in the xcom:<task_id> (or xcom:<task_id>,<key> if the they are not using default return_value key) tag.
Other directions I had consider:
- Define the injection at
AddTasklevel -> This will be ambiguous for user IMO as we still need to define the Dag structure at Python side at this stage. If we make the interface likeAddTask(<task_id>, <other task function handle> ... <or task id>). User will be confuse about: Are we defining the edge in Go slide instead of Python side.
| // TransformResult is transform's return value. | ||
| type TransformResult struct { | ||
| Variable string `json:"variable"` | ||
| Extracted ExtractResult `json:"extracted"` | ||
| } |
There was a problem hiding this comment.
For the return type, user should define the all the fields with json tag.
| "github.com/apache/airflow/go-sdk/pkg/sdkcontext" | ||
| "github.com/apache/airflow/go-sdk/sdk" | ||
| ) | ||
|
|
There was a problem hiding this comment.
Extract most of the reflect and parameter binding into a new binding module to not overwhelm the task module.
| // xcom_pull yields None when the key is missing), so surface it as | ||
| // not-found, matching how the HTTP client maps a 404 and how | ||
| // GetVariable handles the same condition above. | ||
| return nil, fmt.Errorf("%w: %q", sdk.XComNotFound, key) |
There was a problem hiding this comment.
Intentionally raise as error or it will show as decoding error even if the XCom not found.
|
This is probably not tied to 3.3.0 since there’s not change needed to Airflow? Go SDK is released separately. |
We use 3.3.0 milestone to keep track of the AIP-108 progress (we don't have a specific milestone for Go and Java SDK). I add this to milestone as it would be better to support TaskFlow in the initial release that align with Java SDK. |
2 participants
Why
A Go task could only read upstream XComs by calling the
clientexplicitly without this PR's changes, matching the Java SDK's TaskFlow shape (#66332).How
xcom:"<task_id>[,key=<key>]"tag is treated as an XCom-input struct: at execution the runtime pulls each field from the named upstream in the current dag run and decodes it into the field's type. Binding lives on the function, not the registry, soAddTaskis unchanged.map[string]any/anyfield decodes loosely; a null value is only accepted by a nilable field. Signature/tag mistakes fail at registration.pkg/binding(shared by the coordinator and Edge-worker paths), keepingbundlev1/task.gothin.What
pkg/binding: newPlan/Analyze/ResolveAPI for parameter analysis, XCom pull, and strict decodepkg/executioncoordinator fixes uncovered while wiring the example end-to-end:GetXCommaps a null supervisor response tosdk.XComNotFound(mirrors the HTTP 404 mapping andxcom_pullreturningNone).jsonstruct tags so a typed XCom round-tripstimestamp/event/levelfields last so a user attribute can't corrupt the frame the supervisor parses. (Caught bymain.goexample when usingtimestampas XCom field name)Was generative AI tooling used to co-author this PR?