Support TaskFlow Dag definitions natively in the Go-SDK#70158
Draft
jason810496 wants to merge 9 commits into
Draft
Support TaskFlow Dag definitions natively in the Go-SDK#70158jason810496 wants to merge 9 commits into
jason810496 wants to merge 9 commits into
Conversation
- Add `messages_test.go` to test message decoding and encoding functionalities. - Introduce `serde.go` for serialization of various data types to Airflow's format. - Create `serde_test.go` to validate serialization logic and ensure correctness. - Implement `server.go` to handle communication with the supervisor and manage task execution. - Add `task_runner.go` to execute tasks based on received startup details and handle success/failure.
…proved configuration flexibility
Re-add the DagFileParseRequest message, its decoder wiring in decodeIncomingBody, and the server dispatch case so the supervisor can request DAG parsing and receive a DagFileParsingResult. Also fix two AddTaskWithName test calls broken by the registration signature change.
Align the Go bundle serializer with Python's serialize_dag so a Go-authored Dag produces the same serialized shape Python would for the fields the SDK models, verified field-by-field against a reference dump: - Always emit the DAG fields that have no JSON-schema default and that Python therefore never omits: catchup, disable_bundle_versioning, max_active_tasks, max_active_runs (default 16 from [core]) and max_consecutive_failed_dag_runs. - Sort tags, matching Python's set-backed serialization (stable dag_hash). - Always emit template_fields on each task. Cron schedules still serialize to CronTriggerTimetable, which matches only the default [scheduler] create_cron_data_intervals=False; honoring the non-default value needs the supervisor to pass scheduler config over the coordinator protocol, tracked in apache#67938.
Replace the hand-written TaskSpec struct and its hand-mapped serializer omit-if-default rules with spec.gen.go, generated from the "operator" definition in airflow-core/src/airflow/serialization/schema.json by a local gen tool (same pattern as pkg/execution/genmodels). Field types and schema defaults can no longer drift from what the scheduler deserializes; regenerate with `just generate-specs`. DagSpec and the registration Info structs move to the hand-written spec.go next to it: Schedule is an SDK-level concept the schema has no scalar for, and several dag keys are always-emitted with [core]-config fallbacks the schema cannot express.
jason810496
force-pushed
the
feature/go-sdk/taskflow-syntax
branch
from
July 21, 2026 03:39
deaf1ff to
b8d362b
Compare
The generator read field types and defaults from schema.json, but the field list itself was a hand-written allowlist, so a new author-settable operator key would go silently missing from TaskSpec. Selecting every eligible scalar key from the schema - with serializer-owned and Python-only keys excluded by documented rules - inverts the drift direction: new schema keys surface in the regenerated spec.gen.go diff, and stale exclusions fail generation. This also surfaces ignore_first_depends_on_past and wait_for_past_depends_before_skipping, which the hand list had missed.
jason810496
force-pushed
the
feature/go-sdk/taskflow-syntax
branch
from
July 21, 2026 04:49
b8d362b to
04c6c1b
Compare
Dag.Task(fn, opts...) replaces AddTask/AddTaskWithName and returns a
*TaskRef handle. Passing handles via v1.Inputs(...) wires the dependency
edge and feeds each upstream's return value into the matching data
parameter of the task function at run time: the runtime pulls the
upstream's return-value XCom and strictly decodes it into the declared
parameter type, so a Go dag reads like plain function composition:
d := reg.AddDag(v1.DagSpec{DagId: "etl", Schedule: "@daily"})
extracted := d.Task(extract)
d.Task(transform, v1.Inputs(extracted))
v1.After(...) declares ordering-only edges. Task ids move into the
specs: TaskSpec.TaskId (generated from schema.json as an identity
field) and DagSpec.DagId replace positional ids and AddTaskWithName.
Mismatched input counts or types, refs from another dag, and
non-decodable parameters all panic at registration, i.e. dag-parse
time. Binding is shared by the coordinator and Edge worker paths.
jason810496
force-pushed
the
feature/go-sdk/taskflow-syntax
branch
from
July 21, 2026 07:15
04c6c1b to
0ea0905
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
refactor/go-sdk/dag-authoring-apiPR (this branch is stacked on it; only the tip commit is new here).Why
The Go SDK's dag-authoring API wired dependencies with string task ids (
AddTask(fn, spec, []string{"extract"})), which is typo-prone, forces upstream-first registration, and carries no data — downstream tasks had to pull upstream results from XCom by hand. Python's TaskFlow shows the better model: dependencies fall out of data flow. This PR brings that model to Go.Was generative AI tooling used to co-author this PR?