jason810496 commented on code in PR #74319:
URL: https://github.com/apache/airflow/pull/74319#discussion_r4191906019
##########
go-sdk/airflow/dag.go:
##########
@@ -82,11 +91,52 @@ func Dag(dagID string, spec ...DagSpec) *DagRef {
}
d := &DagRef{dagID: dagID}
if len(spec) == 1 {
+ if err := checkDagSpec(spec[0]); err != nil {
+ panic(fmt.Sprintf("airflow.Dag: Dag %q: %v", dagID,
err))
+ }
d.spec = copySpec(spec[0])
}
return d
}
+// tagMaxLength is the longest tag that Python's DAG accepts, counted in
characters. Airflow stores
+// a tag in a column of that length.
+const tagMaxLength = 100
+
+// checkDagSpec rejects a DagSpec with a value that Python rejects when it
builds or validates a
+// Dag. Depending on the value, Airflow would otherwise fail to load the
serialized Dag, fail to
+// store the Dag, or never schedule the Dag.
+func checkDagSpec(spec DagSpec) error {
Review Comment:
Let's add a TODO here saying that the validation should only be invoked at
the build-time eventually.
##########
go-sdk/airflow/recorded_payload_test.go:
##########
@@ -0,0 +1,155 @@
+// 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.
+
+package airflow
+
+import (
+ "encoding/json"
+ "flag"
+ "os"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// recordedPayloadPath is the path of a fixture of the Airflow core tests. For
each serialized Dag
+// in the file, those tests run DagSerialization.validate_serialized_dag and
put the Dag in a Dag
+// bag. If a change to Airflow stops Airflow from loading what this SDK
writes, those tests fail.
+const recordedPayloadPath = "../../airflow-core/tests/unit/dag_processing/" +
+ "lang_sdk_fixtures/go_native.json"
+
+var updateRecordedPayload = flag.Bool(
+ "update-recorded-payload", false, "rewrite "+recordedPayloadPath,
+)
+
+// recordedDags builds the Dags of the fixture. They also use constructs that
the Dags of
+// scripts/ci/lang_sdk_serialization/test_dags.yaml leave out, such as dag.If,
TriggerDagRun and
+// Label.
+func recordedDags() []*DagRef {
+ dag := Dag("go_native", DagSpec{
Review Comment:
I agreed this is a nice robust test case.
Instead of storing the excepted JSON output in the go-sdk test dir, would it
be better to make it part of the YAML test case that we define for all the
languages so that the TS and Java serde can also be verified with this
sophisticated case.
##########
go-sdk/airflow/conformance_test.go:
##########
@@ -0,0 +1,214 @@
+// 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.
+
+package airflow
+
+import (
+ "encoding/json"
+ "flag"
+ "fmt"
+ "os"
+ "reflect"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+ "gopkg.in/yaml.v3"
+)
+
+// TestSerializeConformanceDags builds the Dags of
scripts/ci/lang_sdk_serialization/test_dags.yaml
+// with this SDK, serializes them, and writes them to a JSON file keyed by
dag_id.
+// serialize_python.py in the same directory does the same with Airflow's own
serializer. compare.py
+// there runs this test as the serializer of the Go SDK for the
+// check-go-sdk-serialization-conformance prek hook, and passes the two paths
after -args:
+//
+// go -C go-sdk test ./airflow -count=1 -run
'^TestSerializeConformanceDags$' -args <test_dags.yaml> <output.json>
+//
+// A plain go test run passes no paths, so the test skips.
+func TestSerializeConformanceDags(t *testing.T) {
+ args := flag.Args()
+ if len(args) != 2 {
+ t.Skip(
+ "compare.py runs this with the paths of test_dags.yaml
and of the output after -args",
+ )
+ }
+ raw, err := os.ReadFile(args[0])
+ require.NoError(t, err)
+ var doc yaml.Node
+ require.NoError(t, yaml.Unmarshal(raw, &doc))
+ var file struct {
+ Dags []conformanceDag `yaml:"dags"`
+ }
+ require.NoError(t, doc.Decode(&file))
+
+ bundle := Bundle()
+ dags := make([]*DagRef, len(file.Dags))
+ for i, dagCase := range file.Dags {
+ dags[i] = buildConformanceDag(t, dagCase)
+ bundle.Register(dags[i])
+ }
+ serialized := make(map[string]any, len(dags))
+ for _, dag := range dags {
+ // compare.py does not compare fileloc, which names the file
that declares a Dag, so any path
+ // works here. Airflow still needs a fileloc to load the Dag.
+ serialized[dag.dagID] = dag.serialize("/bundles/app/etl", "etl")
+ }
+ out, err := json.MarshalIndent(serialized, "", " ")
+ require.NoError(t, err)
+ require.NoError(t, os.WriteFile(args[1], append(out, '\n'), 0o644))
+}
+
+// conformanceDag is one Dag of test_dags.yaml. A spec maps the snake_case
keys of the serialization
+// schema to values. The tag !datetime marks a value that is a moment, and
!timedelta marks a number
+// of seconds.
+type conformanceDag struct {
+ DagID string `yaml:"dag_id"`
+ Spec yaml.Node `yaml:"spec"`
+ Groups []string `yaml:"groups"`
+ Tasks []conformanceTask `yaml:"tasks"`
+ OrderEdges [][2]string `yaml:"order_edges"`
+}
+
+type conformanceTask struct {
+ TaskID string `yaml:"task_id"`
+ Group string `yaml:"group"`
+ Upstream []string `yaml:"upstream"`
+ Spec yaml.Node `yaml:"spec"`
+}
+
+func buildConformanceDag(t *testing.T, dagCase conformanceDag) *DagRef {
+ t.Helper()
+ var spec DagSpec
+ setConformanceSpec(t, &spec, dagSpecRules, dagCase.Spec, dagCase.DagID)
+ dag := Dag(dagCase.DagID, spec)
+
+ // test_dags.yaml gives the full group_id, so the parent of a group is
what comes before the last
+ // dot. A parent comes before the groups it holds.
+ groups := make(map[string]*TaskGroupRef)
+ for _, groupID := range dagCase.Groups {
+ cut := strings.LastIndex(groupID, ".")
+ if cut < 0 {
+ groups[groupID] = dag.TaskGroup(groupID)
+ continue
+ }
+ parent, ok := groups[groupID[:cut]]
+ require.True(t, ok, "%s: group %q comes before its parent",
dagCase.DagID, groupID)
+ groups[groupID] = parent.TaskGroup(groupID[cut+1:])
+ }
+
+ tasks := make(map[string]*TaskRef)
+ for _, task := range dagCase.Tasks {
+ var taskSpec TaskSpec
+ label := dagCase.DagID + "." + task.TaskID
+ setConformanceSpec(t, &taskSpec, taskSpecRules, task.Spec,
label)
+ taskSpec.TaskID = task.TaskID
+ // Each upstream task passes its result to a parameter of the
task, which gives the task the edge
+ // that test_dags.yaml asks for.
+ upstreams := make([]*TaskRef, len(task.Upstream))
+ for i, upstreamID := range task.Upstream {
+ upstream, ok := tasks[upstreamID]
+ require.True(t, ok, "%s: upstream %q comes after the
task", label, upstreamID)
+ upstreams[i] = upstream
+ }
+ fn := conformanceTaskFunction(len(upstreams))
+ opts := []TaskOption{taskSpec, Inputs(upstreams...)}
+ var added *TaskRef
+ if task.Group == "" {
+ added = dag.Task(fn, opts...)
+ } else {
+ group, ok := groups[task.Group]
+ require.True(t, ok, "%s: no group %q", label,
task.Group)
+ added = group.Task(fn, opts...)
+ }
+ tasks[added.taskID] = added
+ }
+
+ node := func(id string) Node {
+ if group, ok := groups[id]; ok {
+ return group
+ }
+ task, ok := tasks[id]
+ require.True(t, ok, "%s: no task or group %q", dagCase.DagID,
id)
+ return task
+ }
+ for _, edge := range dagCase.OrderEdges {
+ node(edge[0]).Before(node(edge[1]))
Review Comment:
Would it be possible to just invoke the public API to construct the whole
Dag?
##########
go-sdk/airflow/dag.go:
##########
@@ -82,11 +91,52 @@ func Dag(dagID string, spec ...DagSpec) *DagRef {
}
d := &DagRef{dagID: dagID}
if len(spec) == 1 {
+ if err := checkDagSpec(spec[0]); err != nil {
+ panic(fmt.Sprintf("airflow.Dag: Dag %q: %v", dagID,
err))
+ }
d.spec = copySpec(spec[0])
}
return d
}
+// tagMaxLength is the longest tag that Python's DAG accepts, counted in
characters. Airflow stores
+// a tag in a column of that length.
+const tagMaxLength = 100
+
+// checkDagSpec rejects a DagSpec with a value that Python rejects when it
builds or validates a
+// Dag. Depending on the value, Airflow would otherwise fail to load the
serialized Dag, fail to
+// store the Dag, or never schedule the Dag.
+func checkDagSpec(spec DagSpec) error {
+ if err := checkSchedule(spec.Schedule); err != nil {
+ return err
+ }
+ // An unset MaxActiveRuns takes [core] max_active_runs_per_dag, which
is 16 by default.
+ if spec.Schedule == "@continuous" && (spec.MaxActiveRuns == 0 ||
spec.MaxActiveRuns > 1) {
Review Comment:
Negative `MaxActiveRuns` isn't rejected: `DagSpec{Schedule: "@continuous",
MaxActiveRuns: -5}` builds and serializes `max_active_runs: -5`.
```suggestion
if spec.Schedule == "@continuous" && (spec.MaxActiveRuns <= 0 ||
spec.MaxActiveRuns > 1) {
```
##########
go-sdk/airflow/serialize.go:
##########
@@ -0,0 +1,779 @@
+// 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.
+
+package airflow
+
+import (
+ "cmp"
+ "fmt"
+ "reflect"
+ "regexp"
+ "slices"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// A serialized Dag is the Dag JSON that Airflow stores for a Dag and that its
scheduler reads.
+// Airflow core owns the format.
airflow-core/src/airflow/serialization/schema.json describes it,
+// and Python's DagSerialization writes it for a Dag authored in Python. The
serializer in this file
+// does not write the same bytes as Python. Python picks the fields to leave
out with a
+// client_defaults table that a bundle never gets, so Python writes some
fields at their schema
+// default, such as retry_delay, that this serializer leaves out. Airflow
reads a missing field as
+// its schema default, so it loads the same Dag from either serialization.
+// scripts/ci/lang_sdk_serialization/compare.py checks that for the Dags in
test_dags.yaml in the
+// same directory.
+
+// dagSerializationVersion is the __version of a serialized Dag. It is the
SERIALIZER_VERSION that
+// Airflow core's DagSerialization checks when it loads a serialized Dag. The
supervisor schema has
+// versions named by date, which Airflow negotiates with each runtime, but the
Dag JSON that a
+// bundle returns has only this number.
+const dagSerializationVersion = 3
+
+// A task that runs a Go function gets this task_type and _task_module, where
a task of a Python Dag
+// names the class and the module of its operator. Airflow never imports
_task_module, so the two
+// can name the coordinator that runs the task instead, as the TypeScript SDK
does for its tasks.
+// Nothing in Airflow reads language. It tells a reader of the serialized Dag
which SDK wrote the
+// task.
+const (
+ goTaskType = "GoOperator"
+ goTaskModule = "airflow.sdk.coordinators.executable"
+ goLanguage = "go"
+)
+
+// Python's serializer writes these values for a TriggerDagRunOperator. The
operator class is in
+//
providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py.
+const (
+ triggerDagRunTaskType = "TriggerDagRunOperator"
+ triggerDagRunTaskModule =
"airflow.providers.standard.operators.trigger_dagrun"
+ triggerDagRunUIColor = "#ffefeb"
+)
+
+var triggerDagRunTemplateFields = []string{
+ "trigger_dag_id",
+ "trigger_run_id",
+ "logical_date",
+ "conf",
+ "wait_for_completion",
+ "skip_when_already_exists",
+}
+
+// The ui_color and ui_fgcolor that Python's TaskGroup uses when the Dag
author sets neither.
+const (
+ defaultGroupUIColor = "CornflowerBlue"
+ defaultGroupUIFgColor = "#000"
+)
+
+// A Schedule maps to one of these timetables. Python's DAG picks the same
ones for a schedule
+// string while [scheduler] create_cron_data_intervals is false, which is its
default. When the
+// option is true, Python picks CronDataIntervalTimetable for a cron
expression instead. A bundle
+// cannot read the option, so a Go Dag always gets CronTriggerTimetable
+// (https://github.com/apache/airflow/issues/67938).
+const (
+ nullTimetable = "airflow.timetables.simple.NullTimetable"
+ onceTimetable = "airflow.timetables.simple.OnceTimetable"
+ continuousTimetable = "airflow.timetables.simple.ContinuousTimetable"
+ cronTriggerTimetable = "airflow.timetables.trigger.CronTriggerTimetable"
+)
+
+// cronPresets maps each cron preset to the expression that Python records for
it. It holds the same
+// entries as cron_presets in airflow-core/src/airflow/utils/dates.py.
+var cronPresets = map[string]string{
+ "@hourly": "0 * * * *",
+ "@daily": "0 0 * * *",
+ "@weekly": "0 0 * * 0",
+ "@monthly": "0 0 1 * *",
+ "@quarterly": "0 0 1 */3 *",
+ "@yearly": "0 0 1 1 *",
+}
+
+// croniterAliases are the presets that croniter, which reads a cron
expression for Airflow, accepts
+// in upper or lower case, such as "@Daily". Python passes such a preset to
croniter unchanged
+// unless the preset is a key of cronPresets, and a Go Dag does the same.
+var croniterAliases = map[string]bool{
+ "@midnight": true,
+ "@hourly": true,
+ "@daily": true,
+ "@weekly": true,
+ "@monthly": true,
+ "@yearly": true,
+ "@annually": true,
+}
+
+// cronItem matches one comma-separated item of a field of a cron expression.
An item is one value,
+// or values joined by -, / or #. A value is one of these:
+// - a number, with or without an L or a W before or after it, such as 15W
or L5
+// - one of *, ?, L, W and R
+// - R with a range, such as R(0-30)
+// - three letters, as in a month or weekday name such as JAN or MON
+//
+// cronItem checks only the shape of an item. It takes some items that
croniter rejects, such as a
+// minute of 61 or a weekday of 5L. croniter rejects those only when Airflow
tries to schedule the
+// Dag.
+var cronItem = regexp.MustCompile(`^(?i:` + cronValue + `(?:[-/#]` + cronValue
+ `)*)$`)
+
+const cronValue =
`(?:[0-9]+[LW]?|[LW][0-9]+|[*?LWR]|R\([0-9]+-[0-9]+\)|[A-Z]{3})`
+
+// checkSchedule returns an error unless the Schedule is empty, a preset, or
shaped like a cron
+// expression of five to seven fields. Airflow does not check a cron
expression when it loads a
+// serialized Dag. Without this check, a Schedule such as "every day" would
reach Airflow, and
+// Airflow would never schedule the Dag, with only a line in its log to say
why.
+func checkSchedule(schedule string) error {
+ switch schedule {
+ case "", "@once", "@continuous":
+ return nil
+ }
+ if _, ok := cronPresets[schedule]; ok ||
croniterAliases[strings.ToLower(schedule)] {
+ return nil
+ }
+ fields := strings.Fields(schedule)
+ shaped := len(fields) >= 5 && len(fields) <= 7
+ for _, field := range fields {
+ for item := range strings.SplitSeq(field, ",") {
+ shaped = shaped && cronItem.MatchString(item)
+ }
+ }
+ if !shaped {
+ return fmt.Errorf(
+ "airflow.DagSpec.Schedule is %q, which is not a cron
expression or a preset; set a "+
+ "cron expression such as \"0 3 * * *\", a
preset such as \"@daily\", \"@once\" or "+
+ "\"@continuous\", or no Schedule for a Dag that
runs only when something triggers it",
+ schedule,
+ )
+ }
+ return nil
+}
+
+// serializeTimetable returns the timetable that Python's DAG builds from the
same schedule string.
+// Dag has already checked the Schedule with checkSchedule.
+func serializeTimetable(schedule string) map[string]any {
+ switch schedule {
+ case "":
+ return map[string]any{"__type": nullTimetable, "__var":
map[string]any{}}
+ case "@once":
+ return map[string]any{"__type": onceTimetable, "__var":
map[string]any{}}
+ case "@continuous":
+ return map[string]any{"__type": continuousTimetable, "__var":
map[string]any{}}
+ }
+ expression := schedule
+ if preset, ok := cronPresets[schedule]; ok {
+ expression = preset
+ }
+ return map[string]any{
+ "__type": cronTriggerTimetable,
+ "__var": map[string]any{
+ "expression": expression,
+ "timezone": "UTC",
+ "interval": 0.0,
+ "run_immediately": false,
+ },
+ }
+}
+
+// fieldRule says how the serializer writes one field of DagSpec, TaskSpec or
TaskGroupSpec. Each of
+// the three structs has a table with an entry for every field. When the
generator adds a field to a
+// struct, TestSpecRulesCoverEveryField fails until the table gets an entry
for it.
+type fieldRule struct {
+ // key is the property of the serialization schema that the field sets.
A field with no key, such
+ // as DagSpec.Schedule, has code of its own that writes it, which the
comment on its entry names.
+ key string
+ // schemaDefault is the default that the serialization schema gives the
property, or nil when the
+ // schema has no default for it. writeSpecFields leaves out a value
equal to the default, because
+ // Airflow reads a missing property as its schema default.
+ schemaDefault any
+ // set marks a list that Python keeps in a set, so Python writes the
list sorted and without
+ // repeats. writeSpecFields writes it the same way.
+ set bool
+ // skip marks a field that has a key but that writeSpecFields never
writes. The comment on its
+ // entry says why.
+ skip bool
+}
+
+var dagSpecRules = map[string]fieldRule{
+ "Catchup": {key: "catchup"},
+ "DagDisplayName": {key: "dag_display_name"},
+ "DagrunTimeout": {key: "dagrun_timeout"},
+ "Description": {key: "description"},
+ "DisableBundleVersioning": {key: "disable_bundle_versioning"},
+ "DocMD": {key: "doc_md"},
+ "EndDate": {key: "end_date"},
+ "FailFast": {key: "fail_fast", schemaDefault: false},
+ "IsPausedUponCreation": {key: "is_paused_upon_creation"},
+ "MaxActiveRuns": {key: "max_active_runs"},
+ "MaxActiveTasks": {key: "max_active_tasks"},
+ "MaxConsecutiveFailedDagRuns": {key: "max_consecutive_failed_dag_runs"},
+ // serializeTaskLocked writes Queue as the queue of each Go task whose
TaskSpec sets no Queue.
+ "Queue": {},
+ "RenderTemplateAsNativeObj": {key: "render_template_as_native_obj",
schemaDefault: false},
+ // serializeTimetable writes Schedule as the timetable.
+ "Schedule": {},
+ "StartDate": {key: "start_date"},
+ "Tags": {key: "tags", set: true},
+}
+
+var taskSpecRules = map[string]fieldRule{
+ "TaskDisplayName": {key: "_task_display_name"},
+ "DependsOnPast": {key: "depends_on_past", schemaDefault: false},
+ "DoXComPush": {key: "do_xcom_push", schemaDefault: true},
+ "DocMD": {key: "doc_md"},
+ // Python writes email_on_failure and email_on_retry only for an
operator that has an email
Review Comment:
We shouldn't hand write the schema default in the SDK side serde.
It would be better to define the schema default at the Spec generator stage
to prevent the drift like the TS SDK side.
##########
go-sdk/airflow/serialize.go:
##########
@@ -0,0 +1,779 @@
+// 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.
+
+package airflow
+
+import (
+ "cmp"
+ "fmt"
+ "reflect"
+ "regexp"
+ "slices"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// A serialized Dag is the Dag JSON that Airflow stores for a Dag and that its
scheduler reads.
+// Airflow core owns the format.
airflow-core/src/airflow/serialization/schema.json describes it,
+// and Python's DagSerialization writes it for a Dag authored in Python. The
serializer in this file
+// does not write the same bytes as Python. Python picks the fields to leave
out with a
+// client_defaults table that a bundle never gets, so Python writes some
fields at their schema
+// default, such as retry_delay, that this serializer leaves out. Airflow
reads a missing field as
+// its schema default, so it loads the same Dag from either serialization.
+// scripts/ci/lang_sdk_serialization/compare.py checks that for the Dags in
test_dags.yaml in the
+// same directory.
+
+// dagSerializationVersion is the __version of a serialized Dag. It is the
SERIALIZER_VERSION that
+// Airflow core's DagSerialization checks when it loads a serialized Dag. The
supervisor schema has
+// versions named by date, which Airflow negotiates with each runtime, but the
Dag JSON that a
+// bundle returns has only this number.
+const dagSerializationVersion = 3
+
+// A task that runs a Go function gets this task_type and _task_module, where
a task of a Python Dag
+// names the class and the module of its operator. Airflow never imports
_task_module, so the two
+// can name the coordinator that runs the task instead, as the TypeScript SDK
does for its tasks.
+// Nothing in Airflow reads language. It tells a reader of the serialized Dag
which SDK wrote the
+// task.
+const (
+ goTaskType = "GoOperator"
+ goTaskModule = "airflow.sdk.coordinators.executable"
+ goLanguage = "go"
+)
+
+// Python's serializer writes these values for a TriggerDagRunOperator. The
operator class is in
+//
providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py.
+const (
+ triggerDagRunTaskType = "TriggerDagRunOperator"
+ triggerDagRunTaskModule =
"airflow.providers.standard.operators.trigger_dagrun"
+ triggerDagRunUIColor = "#ffefeb"
+)
+
+var triggerDagRunTemplateFields = []string{
+ "trigger_dag_id",
+ "trigger_run_id",
+ "logical_date",
+ "conf",
+ "wait_for_completion",
+ "skip_when_already_exists",
+}
+
+// The ui_color and ui_fgcolor that Python's TaskGroup uses when the Dag
author sets neither.
+const (
+ defaultGroupUIColor = "CornflowerBlue"
+ defaultGroupUIFgColor = "#000"
+)
+
+// A Schedule maps to one of these timetables. Python's DAG picks the same
ones for a schedule
+// string while [scheduler] create_cron_data_intervals is false, which is its
default. When the
+// option is true, Python picks CronDataIntervalTimetable for a cron
expression instead. A bundle
+// cannot read the option, so a Go Dag always gets CronTriggerTimetable
+// (https://github.com/apache/airflow/issues/67938).
+const (
+ nullTimetable = "airflow.timetables.simple.NullTimetable"
+ onceTimetable = "airflow.timetables.simple.OnceTimetable"
+ continuousTimetable = "airflow.timetables.simple.ContinuousTimetable"
+ cronTriggerTimetable = "airflow.timetables.trigger.CronTriggerTimetable"
+)
+
+// cronPresets maps each cron preset to the expression that Python records for
it. It holds the same
+// entries as cron_presets in airflow-core/src/airflow/utils/dates.py.
+var cronPresets = map[string]string{
+ "@hourly": "0 * * * *",
+ "@daily": "0 0 * * *",
+ "@weekly": "0 0 * * 0",
+ "@monthly": "0 0 1 * *",
+ "@quarterly": "0 0 1 */3 *",
+ "@yearly": "0 0 1 1 *",
+}
+
+// croniterAliases are the presets that croniter, which reads a cron
expression for Airflow, accepts
+// in upper or lower case, such as "@Daily". Python passes such a preset to
croniter unchanged
+// unless the preset is a key of cronPresets, and a Go Dag does the same.
+var croniterAliases = map[string]bool{
+ "@midnight": true,
+ "@hourly": true,
+ "@daily": true,
+ "@weekly": true,
+ "@monthly": true,
+ "@yearly": true,
+ "@annually": true,
+}
+
+// cronItem matches one comma-separated item of a field of a cron expression.
An item is one value,
+// or values joined by -, / or #. A value is one of these:
+// - a number, with or without an L or a W before or after it, such as 15W
or L5
+// - one of *, ?, L, W and R
+// - R with a range, such as R(0-30)
+// - three letters, as in a month or weekday name such as JAN or MON
+//
+// cronItem checks only the shape of an item. It takes some items that
croniter rejects, such as a
+// minute of 61 or a weekday of 5L. croniter rejects those only when Airflow
tries to schedule the
+// Dag.
+var cronItem = regexp.MustCompile(`^(?i:` + cronValue + `(?:[-/#]` + cronValue
+ `)*)$`)
+
+const cronValue =
`(?:[0-9]+[LW]?|[LW][0-9]+|[*?LWR]|R\([0-9]+-[0-9]+\)|[A-Z]{3})`
+
+// checkSchedule returns an error unless the Schedule is empty, a preset, or
shaped like a cron
+// expression of five to seven fields. Airflow does not check a cron
expression when it loads a
+// serialized Dag. Without this check, a Schedule such as "every day" would
reach Airflow, and
+// Airflow would never schedule the Dag, with only a line in its log to say
why.
+func checkSchedule(schedule string) error {
+ switch schedule {
+ case "", "@once", "@continuous":
+ return nil
+ }
+ if _, ok := cronPresets[schedule]; ok ||
croniterAliases[strings.ToLower(schedule)] {
+ return nil
+ }
+ fields := strings.Fields(schedule)
+ shaped := len(fields) >= 5 && len(fields) <= 7
+ for _, field := range fields {
+ for item := range strings.SplitSeq(field, ",") {
+ shaped = shaped && cronItem.MatchString(item)
+ }
+ }
+ if !shaped {
+ return fmt.Errorf(
+ "airflow.DagSpec.Schedule is %q, which is not a cron
expression or a preset; set a "+
+ "cron expression such as \"0 3 * * *\", a
preset such as \"@daily\", \"@once\" or "+
+ "\"@continuous\", or no Schedule for a Dag that
runs only when something triggers it",
+ schedule,
+ )
+ }
+ return nil
+}
+
+// serializeTimetable returns the timetable that Python's DAG builds from the
same schedule string.
+// Dag has already checked the Schedule with checkSchedule.
+func serializeTimetable(schedule string) map[string]any {
+ switch schedule {
+ case "":
+ return map[string]any{"__type": nullTimetable, "__var":
map[string]any{}}
+ case "@once":
+ return map[string]any{"__type": onceTimetable, "__var":
map[string]any{}}
+ case "@continuous":
+ return map[string]any{"__type": continuousTimetable, "__var":
map[string]any{}}
+ }
+ expression := schedule
+ if preset, ok := cronPresets[schedule]; ok {
+ expression = preset
+ }
+ return map[string]any{
+ "__type": cronTriggerTimetable,
+ "__var": map[string]any{
+ "expression": expression,
+ "timezone": "UTC",
+ "interval": 0.0,
+ "run_immediately": false,
+ },
+ }
+}
+
+// fieldRule says how the serializer writes one field of DagSpec, TaskSpec or
TaskGroupSpec. Each of
+// the three structs has a table with an entry for every field. When the
generator adds a field to a
+// struct, TestSpecRulesCoverEveryField fails until the table gets an entry
for it.
+type fieldRule struct {
+ // key is the property of the serialization schema that the field sets.
A field with no key, such
+ // as DagSpec.Schedule, has code of its own that writes it, which the
comment on its entry names.
+ key string
+ // schemaDefault is the default that the serialization schema gives the
property, or nil when the
+ // schema has no default for it. writeSpecFields leaves out a value
equal to the default, because
+ // Airflow reads a missing property as its schema default.
+ schemaDefault any
+ // set marks a list that Python keeps in a set, so Python writes the
list sorted and without
+ // repeats. writeSpecFields writes it the same way.
+ set bool
+ // skip marks a field that has a key but that writeSpecFields never
writes. The comment on its
+ // entry says why.
+ skip bool
+}
+
+var dagSpecRules = map[string]fieldRule{
+ "Catchup": {key: "catchup"},
+ "DagDisplayName": {key: "dag_display_name"},
+ "DagrunTimeout": {key: "dagrun_timeout"},
+ "Description": {key: "description"},
+ "DisableBundleVersioning": {key: "disable_bundle_versioning"},
+ "DocMD": {key: "doc_md"},
+ "EndDate": {key: "end_date"},
+ "FailFast": {key: "fail_fast", schemaDefault: false},
+ "IsPausedUponCreation": {key: "is_paused_upon_creation"},
+ "MaxActiveRuns": {key: "max_active_runs"},
+ "MaxActiveTasks": {key: "max_active_tasks"},
+ "MaxConsecutiveFailedDagRuns": {key: "max_consecutive_failed_dag_runs"},
+ // serializeTaskLocked writes Queue as the queue of each Go task whose
TaskSpec sets no Queue.
+ "Queue": {},
+ "RenderTemplateAsNativeObj": {key: "render_template_as_native_obj",
schemaDefault: false},
+ // serializeTimetable writes Schedule as the timetable.
+ "Schedule": {},
+ "StartDate": {key: "start_date"},
+ "Tags": {key: "tags", set: true},
+}
+
+var taskSpecRules = map[string]fieldRule{
+ "TaskDisplayName": {key: "_task_display_name"},
+ "DependsOnPast": {key: "depends_on_past", schemaDefault: false},
+ "DoXComPush": {key: "do_xcom_push", schemaDefault: true},
+ "DocMD": {key: "doc_md"},
+ // Python writes email_on_failure and email_on_retry only for an
operator that has an email
+ // recipient, and a TaskSpec has no field for a recipient.
+ "EmailOnFailure": {key: "email_on_failure", schemaDefault:
true, skip: true},
+ "EmailOnRetry": {key: "email_on_retry", schemaDefault:
true, skip: true},
+ "EndDate": {key: "end_date"},
+ "ExecutionTimeout": {key: "execution_timeout"},
+ "Executor": {key: "executor"},
+ "IgnoreFirstDependsOnPast": {key: "ignore_first_depends_on_past",
schemaDefault: false},
+ "MapIndexTemplate": {key: "map_index_template"},
+ "MaxActiveTisPerDag": {key: "max_active_tis_per_dag"},
+ "MaxActiveTisPerDagrun": {key: "max_active_tis_per_dagrun"},
+ "MaxRetryDelay": {key: "max_retry_delay"},
+ "Owner": {key: "owner", schemaDefault: "airflow"},
+ "Pool": {key: "pool", schemaDefault:
"default_pool"},
+ "PoolSlots": {key: "pool_slots", schemaDefault: 1},
+ "PriorityWeight": {key: "priority_weight", schemaDefault: 1},
+ "Queue": {key: "queue", schemaDefault: "default"},
+ "Retries": {key: "retries", schemaDefault: 0},
+ "RetryDelay": {key: "retry_delay", schemaDefault: 300.0},
+ "RetryExponentialBackoff": {key: "retry_exponential_backoff",
schemaDefault: 0},
+ "StartDate": {key: "start_date"},
+ // serializeTaskLocked writes the task_id of the TaskRef, where the
group_ids of the groups that
+ // hold the task prefix it.
+ "TaskID": {key: "task_id", skip: true},
+ "TriggerRule": {key: "trigger_rule", schemaDefault:
"all_success"},
+ "WaitForDownstream": {key: "wait_for_downstream", schemaDefault: false},
+ "WaitForPastDependsBeforeSkipping": {
+ key: "wait_for_past_depends_before_skipping",
+ schemaDefault: false,
+ },
+ "WeightRule": {key: "weight_rule", schemaDefault: "downstream"},
+}
+
+// The schema gives no property of a task group a default. serializeTaskGroup
starts each group
+// from the defaults of Python's TaskGroup instead.
+var taskGroupSpecRules = map[string]fieldRule{
+ "DocMD": {key: "doc_md"},
+ "GroupDisplayName": {key: "group_display_name"},
+ "PrefixGroupID": {key: "prefix_group_id"},
+ "Tooltip": {key: "tooltip"},
+ "UIColor": {key: "ui_color"},
+ "UIFgColor": {key: "ui_fgcolor"},
+}
+
+// writeSpecFields writes into data each field of spec that has a key in
rules. It leaves out a
+// field that is unset or that holds its schema default. A field is unset when
it holds its zero
+// value. A pointer field is unset only when it is nil, so a pointer field can
set the zero value of
+// the type it points to, such as false or 0.
+func writeSpecFields(data map[string]any, spec any, rules
map[string]fieldRule) {
+ value := reflect.ValueOf(spec)
+ for i := range value.NumField() {
+ name := value.Type().Field(i).Name
+ field, ok := rules[name]
+ if !ok {
+ panic(fmt.Sprintf(
+ "airflow: the serializer has no rule for
%s.%s", value.Type().Name(), name,
+ ))
+ }
+ if field.key == "" || field.skip {
+ continue
+ }
+ encoded, set := encodeSpecValue(value.Field(i), field.set)
+ if !set || (field.schemaDefault != nil && isSameJSON(encoded,
field.schemaDefault)) {
+ continue
+ }
+ data[field.key] = encoded
+ }
+}
+
+// encodeSpecValue returns the value of a spec field in the form that Python's
serializer writes for
+// the property, or false when the field is unset. set says that Python keeps
the list in a set.
+func encodeSpecValue(value reflect.Value, set bool) (any, bool) {
+ if value.Kind() == reflect.Pointer {
+ if value.IsNil() {
+ return nil, false
+ }
+ value = value.Elem()
+ } else if value.IsZero() {
+ return nil, false
+ }
+ switch v := value.Interface().(type) {
+ case time.Time:
+ // The zero Time moved to a location with In is not the zero
value of time.Time, so value.IsZero
+ // above misses it.
+ if v.IsZero() {
+ return nil, false
+ }
+ return encodeTime(v), true
+ case time.Duration:
+ return encodeDuration(v), true
+ }
+ switch value.Kind() {
+ case reflect.String:
+ return value.String(), true
+ case reflect.Bool:
+ return value.Bool(), true
+ case reflect.Int:
+ return int(value.Int()), true
+ case reflect.Float64:
+ return value.Float(), true
+ case reflect.Slice:
+ if value.Type().Elem().Kind() == reflect.String {
+ items := make([]string, value.Len())
+ for i := range items {
+ items[i] = value.Index(i).String()
+ }
+ if set {
+ slices.Sort(items)
+ items = slices.Compact(items)
+ }
+ return items, true
+ }
+ }
+ panic(fmt.Sprintf("airflow: the serializer cannot write a spec field of
type %s", value.Type()))
+}
+
+// isSameJSON reports whether two encoded values are the same JSON value. Like
Python, it takes 1
+// and 1.0 for the same number.
+func isSameJSON(a, b any) bool {
+ if x, ok := jsonNumber(a); ok {
+ y, ok := jsonNumber(b)
+ return ok && x == y
+ }
+ return a == b
+}
+
+func jsonNumber(value any) (float64, bool) {
+ switch v := value.(type) {
+ case int:
+ return float64(v), true
+ case float64:
+ return v, true
+ }
+ return 0, false
+}
+
+// encodeTime returns t as Python's serializer writes a datetime: the seconds
since the Unix epoch,
+// to the microsecond. A Python datetime holds nothing finer than a
microsecond.
+func encodeTime(t time.Time) float64 { return float64(t.UnixMicro()) / 1e6 }
+
+// checkTime returns an error for a time that a Python datetime cannot hold,
which is a time whose
+// year in UTC is not from 1 to 9999. Airflow would reject a serialized Dag
with such a time when it
+// loads the Dag. field names the field that holds t. The zero Time passes,
because it means that
+// the field is unset.
+func checkTime(field string, t time.Time) error {
+ if year := t.UTC().Year(); !t.IsZero() && (year < 1 || year > 9999) {
+ return fmt.Errorf(
+ "%s is %s in UTC; Airflow takes a time only from year 1
to year 9999",
+ field, t.UTC().Format(time.RFC3339Nano),
+ )
+ }
+ return nil
+}
+
+// encodeDuration returns d as Python's serializer writes a timedelta: a
number of seconds, to the
+// microsecond. A Python timedelta holds nothing finer than a microsecond.
+func encodeDuration(d time.Duration) float64 { return
float64(d.Microseconds()) / 1e6 }
+
+// serialize returns the serialized Dag that Airflow stores for d. A
DagFileParsingResult carries it
+// as the data of one entry of serialized_dags. Python's
DagSerialization.to_dict returns the same
+// shape for a Dag authored in Python. fileloc is the path of the file that
declares the Dag, and
+// relativeFileloc is that path relative to the root of its Dag bundle.
+//
+// serialize panics unless d is registered. Registration expands each edge to
or from a task group
+// into edges between tasks, which the downstream_task_ids of a serialized Dag
hold.
+func (d *DagRef) serialize(fileloc, relativeFileloc string) map[string]any {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+
+ if !d.registered {
+ panic(fmt.Sprintf(
+ "airflow: Dag %q is not registered, so its group edges
are not expanded yet", d.dagID,
+ ))
+ }
+ dag := map[string]any{
+ "dag_id": d.dagID,
+ "fileloc": fileloc,
+ "relative_fileloc": relativeFileloc,
+ // A Go Dag has no timezone of its own. Airflow reads its cron
expression in UTC, the timezone
+ // that serializeTimetable writes.
+ "timezone": "UTC",
+ "timetable": serializeTimetable(d.spec.Schedule),
+ "tasks": d.serializeTasksLocked(),
+ "dag_dependencies": d.serializeDagDependenciesLocked(),
+ "task_group": d.serializeTaskGroupsLocked(),
+ "edge_info": d.serializeEdgeInfoLocked(),
+ // Python's serializer writes params, deadline and
allowed_run_types for every Dag. A Go Dag
+ // cannot set any of them yet.
+ "params": []any{},
+ "deadline": nil,
+ "allowed_run_types": nil,
+ }
+ // writeSpecFields leaves out each field that the Dag does not set.
That includes max_active_tasks,
+ // max_active_runs, max_consecutive_failed_dag_runs, catchup and
disable_bundle_versioning, which a
+ // Python Dag takes from the Airflow config when it does not set them.
A bundle cannot read the
+ // config, so Airflow fills those five in from its own config when it
receives the Dag.
+ writeSpecFields(dag, d.spec, dagSpecRules)
+ return map[string]any{"__version": dagSerializationVersion, "dag": dag}
+}
+
+// serializeTasksLocked writes the tasks of d in the order they were added, as
Python writes the
+// tasks of a Dag. The caller holds d.mu.
+func (d *DagRef) serializeTasksLocked() []any {
+ tasks := make([]any, len(d.tasks))
+ for i, task := range d.tasks {
+ tasks[i] = d.serializeTaskLocked(task)
+ }
+ return tasks
+}
+
+// serializeTaskLocked writes one task of d. The caller holds d.mu.
+func (d *DagRef) serializeTaskLocked(task *TaskRef) map[string]any {
+ data := map[string]any{"task_id": task.taskID}
+ spec := task.spec
+ if task.triggerDagRun != nil {
+ writeTriggerDagRun(data, *task.triggerDagRun)
+ } else {
+ data["task_type"] = goTaskType
+ data["_task_module"] = goTaskModule
+ data["language"] = goLanguage
+ // Python writes template_fields for every operator. A Go task
has no template fields, so the list
+ // is empty.
+ data["template_fields"] = []string{}
+ // is_stub makes the API server send the task its arguments
from _arg_bindings when the
+ // task runs, as it does for a @task.stub task of a Python Dag.
+ data["is_stub"] = true
+ if len(task.inputs) > 0 {
+ data["_arg_bindings"] =
serializeArgBindings(task.inputs)
+ }
+ if task.ifRef != nil {
+ // Python writes _can_skip_downstream for a branch
operator. Airflow reads the skipmixin_key XCom
+ // of the condition only when the flag is set, before
it runs a task that comes after the
+ // condition.
+ data["_can_skip_downstream"] = true
+ }
+ if spec.Queue == "" {
+ spec.Queue = d.spec.Queue
+ }
+ }
+ writeSpecFields(data, spec, taskSpecRules)
+ if len(task.downstreams) > 0 {
+ downstreams := taskIDs(task.downstreams)
+ slices.Sort(downstreams)
+ data["downstream_task_ids"] = downstreams
+ }
+ return map[string]any{"__type": "operator", "__var": data}
+}
+
+// serializeArgBindings returns the arg bindings for the tasks that Inputs
passed. When the task
+// runs, the API server sends it these bindings, one for each parameter after
the Context, in order.
+// Package reflect cannot read the names of the parameters of a Go function,
and a Go task takes its
+// arguments by position. So each binding is named after the position of its
parameter: arg0 for the
+// first parameter after the Context, arg1 for the next, and so on.
+func serializeArgBindings(inputs []*TaskRef) []any {
Review Comment:
Though this feature gap is not caused by this PR, but it's a highly related
feature gap for Go SDK side.
Let's also support the `Literal(value)` to support injected the JSON
serializable literal at the `Input` in this PR.
TS SDK already supported it, and the Java SDK will support it soon as well
in https://github.com/apache/airflow/pull/71190.
##########
go-sdk/airflow/serialize.go:
##########
@@ -0,0 +1,779 @@
+// 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.
+
+package airflow
+
+import (
+ "cmp"
+ "fmt"
+ "reflect"
+ "regexp"
+ "slices"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// A serialized Dag is the Dag JSON that Airflow stores for a Dag and that its
scheduler reads.
+// Airflow core owns the format.
airflow-core/src/airflow/serialization/schema.json describes it,
+// and Python's DagSerialization writes it for a Dag authored in Python. The
serializer in this file
+// does not write the same bytes as Python. Python picks the fields to leave
out with a
+// client_defaults table that a bundle never gets, so Python writes some
fields at their schema
+// default, such as retry_delay, that this serializer leaves out. Airflow
reads a missing field as
+// its schema default, so it loads the same Dag from either serialization.
+// scripts/ci/lang_sdk_serialization/compare.py checks that for the Dags in
test_dags.yaml in the
+// same directory.
+
+// dagSerializationVersion is the __version of a serialized Dag. It is the
SERIALIZER_VERSION that
+// Airflow core's DagSerialization checks when it loads a serialized Dag. The
supervisor schema has
+// versions named by date, which Airflow negotiates with each runtime, but the
Dag JSON that a
+// bundle returns has only this number.
+const dagSerializationVersion = 3
+
+// A task that runs a Go function gets this task_type and _task_module, where
a task of a Python Dag
+// names the class and the module of its operator. Airflow never imports
_task_module, so the two
+// can name the coordinator that runs the task instead, as the TypeScript SDK
does for its tasks.
+// Nothing in Airflow reads language. It tells a reader of the serialized Dag
which SDK wrote the
+// task.
+const (
+ goTaskType = "GoOperator"
+ goTaskModule = "airflow.sdk.coordinators.executable"
+ goLanguage = "go"
+)
+
+// Python's serializer writes these values for a TriggerDagRunOperator. The
operator class is in
+//
providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py.
+const (
+ triggerDagRunTaskType = "TriggerDagRunOperator"
+ triggerDagRunTaskModule =
"airflow.providers.standard.operators.trigger_dagrun"
+ triggerDagRunUIColor = "#ffefeb"
+)
+
+var triggerDagRunTemplateFields = []string{
+ "trigger_dag_id",
+ "trigger_run_id",
+ "logical_date",
+ "conf",
+ "wait_for_completion",
+ "skip_when_already_exists",
+}
+
+// The ui_color and ui_fgcolor that Python's TaskGroup uses when the Dag
author sets neither.
+const (
+ defaultGroupUIColor = "CornflowerBlue"
+ defaultGroupUIFgColor = "#000"
+)
+
+// A Schedule maps to one of these timetables. Python's DAG picks the same
ones for a schedule
+// string while [scheduler] create_cron_data_intervals is false, which is its
default. When the
+// option is true, Python picks CronDataIntervalTimetable for a cron
expression instead. A bundle
+// cannot read the option, so a Go Dag always gets CronTriggerTimetable
+// (https://github.com/apache/airflow/issues/67938).
+const (
+ nullTimetable = "airflow.timetables.simple.NullTimetable"
+ onceTimetable = "airflow.timetables.simple.OnceTimetable"
+ continuousTimetable = "airflow.timetables.simple.ContinuousTimetable"
+ cronTriggerTimetable = "airflow.timetables.trigger.CronTriggerTimetable"
+)
+
+// cronPresets maps each cron preset to the expression that Python records for
it. It holds the same
+// entries as cron_presets in airflow-core/src/airflow/utils/dates.py.
+var cronPresets = map[string]string{
+ "@hourly": "0 * * * *",
+ "@daily": "0 0 * * *",
+ "@weekly": "0 0 * * 0",
+ "@monthly": "0 0 1 * *",
+ "@quarterly": "0 0 1 */3 *",
+ "@yearly": "0 0 1 1 *",
+}
+
+// croniterAliases are the presets that croniter, which reads a cron
expression for Airflow, accepts
+// in upper or lower case, such as "@Daily". Python passes such a preset to
croniter unchanged
+// unless the preset is a key of cronPresets, and a Go Dag does the same.
+var croniterAliases = map[string]bool{
+ "@midnight": true,
+ "@hourly": true,
+ "@daily": true,
+ "@weekly": true,
+ "@monthly": true,
+ "@yearly": true,
+ "@annually": true,
+}
+
+// cronItem matches one comma-separated item of a field of a cron expression.
An item is one value,
+// or values joined by -, / or #. A value is one of these:
+// - a number, with or without an L or a W before or after it, such as 15W
or L5
+// - one of *, ?, L, W and R
+// - R with a range, such as R(0-30)
+// - three letters, as in a month or weekday name such as JAN or MON
+//
+// cronItem checks only the shape of an item. It takes some items that
croniter rejects, such as a
+// minute of 61 or a weekday of 5L. croniter rejects those only when Airflow
tries to schedule the
+// Dag.
+var cronItem = regexp.MustCompile(`^(?i:` + cronValue + `(?:[-/#]` + cronValue
+ `)*)$`)
+
+const cronValue =
`(?:[0-9]+[LW]?|[LW][0-9]+|[*?LWR]|R\([0-9]+-[0-9]+\)|[A-Z]{3})`
+
+// checkSchedule returns an error unless the Schedule is empty, a preset, or
shaped like a cron
+// expression of five to seven fields. Airflow does not check a cron
expression when it loads a
+// serialized Dag. Without this check, a Schedule such as "every day" would
reach Airflow, and
+// Airflow would never schedule the Dag, with only a line in its log to say
why.
+func checkSchedule(schedule string) error {
+ switch schedule {
+ case "", "@once", "@continuous":
+ return nil
+ }
+ if _, ok := cronPresets[schedule]; ok ||
croniterAliases[strings.ToLower(schedule)] {
+ return nil
+ }
+ fields := strings.Fields(schedule)
+ shaped := len(fields) >= 5 && len(fields) <= 7
+ for _, field := range fields {
+ for item := range strings.SplitSeq(field, ",") {
+ shaped = shaped && cronItem.MatchString(item)
+ }
+ }
+ if !shaped {
+ return fmt.Errorf(
+ "airflow.DagSpec.Schedule is %q, which is not a cron
expression or a preset; set a "+
+ "cron expression such as \"0 3 * * *\", a
preset such as \"@daily\", \"@once\" or "+
+ "\"@continuous\", or no Schedule for a Dag that
runs only when something triggers it",
+ schedule,
+ )
+ }
+ return nil
+}
+
+// serializeTimetable returns the timetable that Python's DAG builds from the
same schedule string.
+// Dag has already checked the Schedule with checkSchedule.
+func serializeTimetable(schedule string) map[string]any {
+ switch schedule {
+ case "":
+ return map[string]any{"__type": nullTimetable, "__var":
map[string]any{}}
+ case "@once":
+ return map[string]any{"__type": onceTimetable, "__var":
map[string]any{}}
+ case "@continuous":
+ return map[string]any{"__type": continuousTimetable, "__var":
map[string]any{}}
+ }
+ expression := schedule
+ if preset, ok := cronPresets[schedule]; ok {
+ expression = preset
+ }
+ return map[string]any{
+ "__type": cronTriggerTimetable,
+ "__var": map[string]any{
+ "expression": expression,
+ "timezone": "UTC",
+ "interval": 0.0,
+ "run_immediately": false,
+ },
+ }
+}
+
+// fieldRule says how the serializer writes one field of DagSpec, TaskSpec or
TaskGroupSpec. Each of
+// the three structs has a table with an entry for every field. When the
generator adds a field to a
+// struct, TestSpecRulesCoverEveryField fails until the table gets an entry
for it.
+type fieldRule struct {
+ // key is the property of the serialization schema that the field sets.
A field with no key, such
+ // as DagSpec.Schedule, has code of its own that writes it, which the
comment on its entry names.
+ key string
+ // schemaDefault is the default that the serialization schema gives the
property, or nil when the
+ // schema has no default for it. writeSpecFields leaves out a value
equal to the default, because
+ // Airflow reads a missing property as its schema default.
+ schemaDefault any
+ // set marks a list that Python keeps in a set, so Python writes the
list sorted and without
+ // repeats. writeSpecFields writes it the same way.
+ set bool
+ // skip marks a field that has a key but that writeSpecFields never
writes. The comment on its
+ // entry says why.
+ skip bool
+}
+
+var dagSpecRules = map[string]fieldRule{
Review Comment:
These ~20 static JSON-schema defaults are hand-typed as Go literals instead
of generated from `go-sdk/schema/dag-schema.json`, unlike ts-sdk's generated
`dag-schema-fields.ts`. `TestSpecRulesMatchTheSchema` catches drift before
merge, but a generated table can't drift by construction. Worth considering for
consistency with ts-sdk.
##########
go-sdk/airflow/serialize.go:
##########
@@ -0,0 +1,779 @@
+// 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.
+
+package airflow
+
+import (
+ "cmp"
+ "fmt"
+ "reflect"
+ "regexp"
+ "slices"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// A serialized Dag is the Dag JSON that Airflow stores for a Dag and that its
scheduler reads.
+// Airflow core owns the format.
airflow-core/src/airflow/serialization/schema.json describes it,
+// and Python's DagSerialization writes it for a Dag authored in Python. The
serializer in this file
+// does not write the same bytes as Python. Python picks the fields to leave
out with a
+// client_defaults table that a bundle never gets, so Python writes some
fields at their schema
+// default, such as retry_delay, that this serializer leaves out. Airflow
reads a missing field as
+// its schema default, so it loads the same Dag from either serialization.
+// scripts/ci/lang_sdk_serialization/compare.py checks that for the Dags in
test_dags.yaml in the
+// same directory.
+
+// dagSerializationVersion is the __version of a serialized Dag. It is the
SERIALIZER_VERSION that
+// Airflow core's DagSerialization checks when it loads a serialized Dag. The
supervisor schema has
+// versions named by date, which Airflow negotiates with each runtime, but the
Dag JSON that a
+// bundle returns has only this number.
+const dagSerializationVersion = 3
+
+// A task that runs a Go function gets this task_type and _task_module, where
a task of a Python Dag
+// names the class and the module of its operator. Airflow never imports
_task_module, so the two
+// can name the coordinator that runs the task instead, as the TypeScript SDK
does for its tasks.
+// Nothing in Airflow reads language. It tells a reader of the serialized Dag
which SDK wrote the
+// task.
+const (
+ goTaskType = "GoOperator"
+ goTaskModule = "airflow.sdk.coordinators.executable"
+ goLanguage = "go"
+)
+
+// Python's serializer writes these values for a TriggerDagRunOperator. The
operator class is in
+//
providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py.
+const (
+ triggerDagRunTaskType = "TriggerDagRunOperator"
+ triggerDagRunTaskModule =
"airflow.providers.standard.operators.trigger_dagrun"
+ triggerDagRunUIColor = "#ffefeb"
+)
+
+var triggerDagRunTemplateFields = []string{
+ "trigger_dag_id",
+ "trigger_run_id",
+ "logical_date",
+ "conf",
+ "wait_for_completion",
+ "skip_when_already_exists",
+}
+
+// The ui_color and ui_fgcolor that Python's TaskGroup uses when the Dag
author sets neither.
+const (
+ defaultGroupUIColor = "CornflowerBlue"
+ defaultGroupUIFgColor = "#000"
+)
+
+// A Schedule maps to one of these timetables. Python's DAG picks the same
ones for a schedule
+// string while [scheduler] create_cron_data_intervals is false, which is its
default. When the
+// option is true, Python picks CronDataIntervalTimetable for a cron
expression instead. A bundle
+// cannot read the option, so a Go Dag always gets CronTriggerTimetable
+// (https://github.com/apache/airflow/issues/67938).
+const (
+ nullTimetable = "airflow.timetables.simple.NullTimetable"
+ onceTimetable = "airflow.timetables.simple.OnceTimetable"
+ continuousTimetable = "airflow.timetables.simple.ContinuousTimetable"
+ cronTriggerTimetable = "airflow.timetables.trigger.CronTriggerTimetable"
+)
+
+// cronPresets maps each cron preset to the expression that Python records for
it. It holds the same
+// entries as cron_presets in airflow-core/src/airflow/utils/dates.py.
+var cronPresets = map[string]string{
+ "@hourly": "0 * * * *",
+ "@daily": "0 0 * * *",
+ "@weekly": "0 0 * * 0",
+ "@monthly": "0 0 1 * *",
+ "@quarterly": "0 0 1 */3 *",
+ "@yearly": "0 0 1 1 *",
+}
+
+// croniterAliases are the presets that croniter, which reads a cron
expression for Airflow, accepts
+// in upper or lower case, such as "@Daily". Python passes such a preset to
croniter unchanged
+// unless the preset is a key of cronPresets, and a Go Dag does the same.
+var croniterAliases = map[string]bool{
+ "@midnight": true,
+ "@hourly": true,
+ "@daily": true,
+ "@weekly": true,
+ "@monthly": true,
+ "@yearly": true,
+ "@annually": true,
+}
+
+// cronItem matches one comma-separated item of a field of a cron expression.
An item is one value,
+// or values joined by -, / or #. A value is one of these:
+// - a number, with or without an L or a W before or after it, such as 15W
or L5
+// - one of *, ?, L, W and R
+// - R with a range, such as R(0-30)
+// - three letters, as in a month or weekday name such as JAN or MON
+//
+// cronItem checks only the shape of an item. It takes some items that
croniter rejects, such as a
+// minute of 61 or a weekday of 5L. croniter rejects those only when Airflow
tries to schedule the
+// Dag.
+var cronItem = regexp.MustCompile(`^(?i:` + cronValue + `(?:[-/#]` + cronValue
+ `)*)$`)
+
+const cronValue =
`(?:[0-9]+[LW]?|[LW][0-9]+|[*?LWR]|R\([0-9]+-[0-9]+\)|[A-Z]{3})`
+
+// checkSchedule returns an error unless the Schedule is empty, a preset, or
shaped like a cron
+// expression of five to seven fields. Airflow does not check a cron
expression when it loads a
+// serialized Dag. Without this check, a Schedule such as "every day" would
reach Airflow, and
+// Airflow would never schedule the Dag, with only a line in its log to say
why.
+func checkSchedule(schedule string) error {
+ switch schedule {
+ case "", "@once", "@continuous":
+ return nil
+ }
+ if _, ok := cronPresets[schedule]; ok ||
croniterAliases[strings.ToLower(schedule)] {
+ return nil
+ }
+ fields := strings.Fields(schedule)
+ shaped := len(fields) >= 5 && len(fields) <= 7
+ for _, field := range fields {
+ for item := range strings.SplitSeq(field, ",") {
+ shaped = shaped && cronItem.MatchString(item)
+ }
+ }
+ if !shaped {
+ return fmt.Errorf(
+ "airflow.DagSpec.Schedule is %q, which is not a cron
expression or a preset; set a "+
+ "cron expression such as \"0 3 * * *\", a
preset such as \"@daily\", \"@once\" or "+
+ "\"@continuous\", or no Schedule for a Dag that
runs only when something triggers it",
+ schedule,
+ )
+ }
+ return nil
+}
+
+// serializeTimetable returns the timetable that Python's DAG builds from the
same schedule string.
+// Dag has already checked the Schedule with checkSchedule.
+func serializeTimetable(schedule string) map[string]any {
+ switch schedule {
+ case "":
+ return map[string]any{"__type": nullTimetable, "__var":
map[string]any{}}
+ case "@once":
+ return map[string]any{"__type": onceTimetable, "__var":
map[string]any{}}
+ case "@continuous":
+ return map[string]any{"__type": continuousTimetable, "__var":
map[string]any{}}
+ }
+ expression := schedule
+ if preset, ok := cronPresets[schedule]; ok {
+ expression = preset
+ }
+ return map[string]any{
+ "__type": cronTriggerTimetable,
+ "__var": map[string]any{
+ "expression": expression,
+ "timezone": "UTC",
+ "interval": 0.0,
+ "run_immediately": false,
+ },
+ }
+}
+
+// fieldRule says how the serializer writes one field of DagSpec, TaskSpec or
TaskGroupSpec. Each of
+// the three structs has a table with an entry for every field. When the
generator adds a field to a
+// struct, TestSpecRulesCoverEveryField fails until the table gets an entry
for it.
+type fieldRule struct {
+ // key is the property of the serialization schema that the field sets.
A field with no key, such
+ // as DagSpec.Schedule, has code of its own that writes it, which the
comment on its entry names.
+ key string
+ // schemaDefault is the default that the serialization schema gives the
property, or nil when the
+ // schema has no default for it. writeSpecFields leaves out a value
equal to the default, because
+ // Airflow reads a missing property as its schema default.
+ schemaDefault any
+ // set marks a list that Python keeps in a set, so Python writes the
list sorted and without
+ // repeats. writeSpecFields writes it the same way.
+ set bool
+ // skip marks a field that has a key but that writeSpecFields never
writes. The comment on its
+ // entry says why.
+ skip bool
+}
+
+var dagSpecRules = map[string]fieldRule{
+ "Catchup": {key: "catchup"},
+ "DagDisplayName": {key: "dag_display_name"},
+ "DagrunTimeout": {key: "dagrun_timeout"},
+ "Description": {key: "description"},
+ "DisableBundleVersioning": {key: "disable_bundle_versioning"},
+ "DocMD": {key: "doc_md"},
+ "EndDate": {key: "end_date"},
+ "FailFast": {key: "fail_fast", schemaDefault: false},
+ "IsPausedUponCreation": {key: "is_paused_upon_creation"},
+ "MaxActiveRuns": {key: "max_active_runs"},
+ "MaxActiveTasks": {key: "max_active_tasks"},
+ "MaxConsecutiveFailedDagRuns": {key: "max_consecutive_failed_dag_runs"},
+ // serializeTaskLocked writes Queue as the queue of each Go task whose
TaskSpec sets no Queue.
+ "Queue": {},
+ "RenderTemplateAsNativeObj": {key: "render_template_as_native_obj",
schemaDefault: false},
+ // serializeTimetable writes Schedule as the timetable.
+ "Schedule": {},
+ "StartDate": {key: "start_date"},
+ "Tags": {key: "tags", set: true},
+}
+
+var taskSpecRules = map[string]fieldRule{
+ "TaskDisplayName": {key: "_task_display_name"},
+ "DependsOnPast": {key: "depends_on_past", schemaDefault: false},
+ "DoXComPush": {key: "do_xcom_push", schemaDefault: true},
+ "DocMD": {key: "doc_md"},
+ // Python writes email_on_failure and email_on_retry only for an
operator that has an email
+ // recipient, and a TaskSpec has no field for a recipient.
+ "EmailOnFailure": {key: "email_on_failure", schemaDefault:
true, skip: true},
+ "EmailOnRetry": {key: "email_on_retry", schemaDefault:
true, skip: true},
+ "EndDate": {key: "end_date"},
+ "ExecutionTimeout": {key: "execution_timeout"},
+ "Executor": {key: "executor"},
+ "IgnoreFirstDependsOnPast": {key: "ignore_first_depends_on_past",
schemaDefault: false},
+ "MapIndexTemplate": {key: "map_index_template"},
+ "MaxActiveTisPerDag": {key: "max_active_tis_per_dag"},
+ "MaxActiveTisPerDagrun": {key: "max_active_tis_per_dagrun"},
+ "MaxRetryDelay": {key: "max_retry_delay"},
+ "Owner": {key: "owner", schemaDefault: "airflow"},
+ "Pool": {key: "pool", schemaDefault:
"default_pool"},
+ "PoolSlots": {key: "pool_slots", schemaDefault: 1},
+ "PriorityWeight": {key: "priority_weight", schemaDefault: 1},
+ "Queue": {key: "queue", schemaDefault: "default"},
+ "Retries": {key: "retries", schemaDefault: 0},
+ "RetryDelay": {key: "retry_delay", schemaDefault: 300.0},
+ "RetryExponentialBackoff": {key: "retry_exponential_backoff",
schemaDefault: 0},
+ "StartDate": {key: "start_date"},
+ // serializeTaskLocked writes the task_id of the TaskRef, where the
group_ids of the groups that
+ // hold the task prefix it.
+ "TaskID": {key: "task_id", skip: true},
+ "TriggerRule": {key: "trigger_rule", schemaDefault:
"all_success"},
+ "WaitForDownstream": {key: "wait_for_downstream", schemaDefault: false},
+ "WaitForPastDependsBeforeSkipping": {
+ key: "wait_for_past_depends_before_skipping",
+ schemaDefault: false,
+ },
+ "WeightRule": {key: "weight_rule", schemaDefault: "downstream"},
+}
+
+// The schema gives no property of a task group a default. serializeTaskGroup
starts each group
+// from the defaults of Python's TaskGroup instead.
+var taskGroupSpecRules = map[string]fieldRule{
+ "DocMD": {key: "doc_md"},
+ "GroupDisplayName": {key: "group_display_name"},
+ "PrefixGroupID": {key: "prefix_group_id"},
+ "Tooltip": {key: "tooltip"},
+ "UIColor": {key: "ui_color"},
+ "UIFgColor": {key: "ui_fgcolor"},
+}
+
+// writeSpecFields writes into data each field of spec that has a key in
rules. It leaves out a
+// field that is unset or that holds its schema default. A field is unset when
it holds its zero
+// value. A pointer field is unset only when it is nil, so a pointer field can
set the zero value of
+// the type it points to, such as false or 0.
+func writeSpecFields(data map[string]any, spec any, rules
map[string]fieldRule) {
+ value := reflect.ValueOf(spec)
+ for i := range value.NumField() {
+ name := value.Type().Field(i).Name
+ field, ok := rules[name]
+ if !ok {
+ panic(fmt.Sprintf(
+ "airflow: the serializer has no rule for
%s.%s", value.Type().Name(), name,
+ ))
+ }
+ if field.key == "" || field.skip {
+ continue
+ }
+ encoded, set := encodeSpecValue(value.Field(i), field.set)
+ if !set || (field.schemaDefault != nil && isSameJSON(encoded,
field.schemaDefault)) {
+ continue
+ }
+ data[field.key] = encoded
+ }
+}
+
+// encodeSpecValue returns the value of a spec field in the form that Python's
serializer writes for
+// the property, or false when the field is unset. set says that Python keeps
the list in a set.
+func encodeSpecValue(value reflect.Value, set bool) (any, bool) {
+ if value.Kind() == reflect.Pointer {
+ if value.IsNil() {
+ return nil, false
+ }
+ value = value.Elem()
+ } else if value.IsZero() {
+ return nil, false
+ }
+ switch v := value.Interface().(type) {
+ case time.Time:
+ // The zero Time moved to a location with In is not the zero
value of time.Time, so value.IsZero
+ // above misses it.
+ if v.IsZero() {
+ return nil, false
+ }
+ return encodeTime(v), true
+ case time.Duration:
+ return encodeDuration(v), true
+ }
+ switch value.Kind() {
+ case reflect.String:
+ return value.String(), true
+ case reflect.Bool:
+ return value.Bool(), true
+ case reflect.Int:
+ return int(value.Int()), true
+ case reflect.Float64:
+ return value.Float(), true
+ case reflect.Slice:
+ if value.Type().Elem().Kind() == reflect.String {
+ items := make([]string, value.Len())
+ for i := range items {
+ items[i] = value.Index(i).String()
+ }
+ if set {
+ slices.Sort(items)
+ items = slices.Compact(items)
+ }
+ return items, true
+ }
+ }
+ panic(fmt.Sprintf("airflow: the serializer cannot write a spec field of
type %s", value.Type()))
+}
+
+// isSameJSON reports whether two encoded values are the same JSON value. Like
Python, it takes 1
+// and 1.0 for the same number.
+func isSameJSON(a, b any) bool {
+ if x, ok := jsonNumber(a); ok {
+ y, ok := jsonNumber(b)
+ return ok && x == y
+ }
+ return a == b
+}
+
+func jsonNumber(value any) (float64, bool) {
+ switch v := value.(type) {
+ case int:
+ return float64(v), true
+ case float64:
+ return v, true
+ }
+ return 0, false
+}
+
+// encodeTime returns t as Python's serializer writes a datetime: the seconds
since the Unix epoch,
+// to the microsecond. A Python datetime holds nothing finer than a
microsecond.
+func encodeTime(t time.Time) float64 { return float64(t.UnixMicro()) / 1e6 }
+
+// checkTime returns an error for a time that a Python datetime cannot hold,
which is a time whose
+// year in UTC is not from 1 to 9999. Airflow would reject a serialized Dag
with such a time when it
+// loads the Dag. field names the field that holds t. The zero Time passes,
because it means that
+// the field is unset.
+func checkTime(field string, t time.Time) error {
+ if year := t.UTC().Year(); !t.IsZero() && (year < 1 || year > 9999) {
+ return fmt.Errorf(
+ "%s is %s in UTC; Airflow takes a time only from year 1
to year 9999",
+ field, t.UTC().Format(time.RFC3339Nano),
+ )
+ }
+ return nil
+}
+
+// encodeDuration returns d as Python's serializer writes a timedelta: a
number of seconds, to the
+// microsecond. A Python timedelta holds nothing finer than a microsecond.
+func encodeDuration(d time.Duration) float64 { return
float64(d.Microseconds()) / 1e6 }
+
+// serialize returns the serialized Dag that Airflow stores for d. A
DagFileParsingResult carries it
+// as the data of one entry of serialized_dags. Python's
DagSerialization.to_dict returns the same
+// shape for a Dag authored in Python. fileloc is the path of the file that
declares the Dag, and
+// relativeFileloc is that path relative to the root of its Dag bundle.
+//
+// serialize panics unless d is registered. Registration expands each edge to
or from a task group
+// into edges between tasks, which the downstream_task_ids of a serialized Dag
hold.
+func (d *DagRef) serialize(fileloc, relativeFileloc string) map[string]any {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+
+ if !d.registered {
+ panic(fmt.Sprintf(
+ "airflow: Dag %q is not registered, so its group edges
are not expanded yet", d.dagID,
+ ))
+ }
+ dag := map[string]any{
+ "dag_id": d.dagID,
+ "fileloc": fileloc,
+ "relative_fileloc": relativeFileloc,
+ // A Go Dag has no timezone of its own. Airflow reads its cron
expression in UTC, the timezone
+ // that serializeTimetable writes.
+ "timezone": "UTC",
+ "timetable": serializeTimetable(d.spec.Schedule),
+ "tasks": d.serializeTasksLocked(),
+ "dag_dependencies": d.serializeDagDependenciesLocked(),
+ "task_group": d.serializeTaskGroupsLocked(),
+ "edge_info": d.serializeEdgeInfoLocked(),
+ // Python's serializer writes params, deadline and
allowed_run_types for every Dag. A Go Dag
+ // cannot set any of them yet.
+ "params": []any{},
+ "deadline": nil,
+ "allowed_run_types": nil,
+ }
+ // writeSpecFields leaves out each field that the Dag does not set.
That includes max_active_tasks,
+ // max_active_runs, max_consecutive_failed_dag_runs, catchup and
disable_bundle_versioning, which a
+ // Python Dag takes from the Airflow config when it does not set them.
A bundle cannot read the
+ // config, so Airflow fills those five in from its own config when it
receives the Dag.
+ writeSpecFields(dag, d.spec, dagSpecRules)
+ return map[string]any{"__version": dagSerializationVersion, "dag": dag}
+}
+
+// serializeTasksLocked writes the tasks of d in the order they were added, as
Python writes the
+// tasks of a Dag. The caller holds d.mu.
+func (d *DagRef) serializeTasksLocked() []any {
+ tasks := make([]any, len(d.tasks))
+ for i, task := range d.tasks {
+ tasks[i] = d.serializeTaskLocked(task)
+ }
+ return tasks
+}
+
+// serializeTaskLocked writes one task of d. The caller holds d.mu.
+func (d *DagRef) serializeTaskLocked(task *TaskRef) map[string]any {
+ data := map[string]any{"task_id": task.taskID}
+ spec := task.spec
+ if task.triggerDagRun != nil {
+ writeTriggerDagRun(data, *task.triggerDagRun)
+ } else {
+ data["task_type"] = goTaskType
+ data["_task_module"] = goTaskModule
+ data["language"] = goLanguage
+ // Python writes template_fields for every operator. A Go task
has no template fields, so the list
+ // is empty.
+ data["template_fields"] = []string{}
+ // is_stub makes the API server send the task its arguments
from _arg_bindings when the
+ // task runs, as it does for a @task.stub task of a Python Dag.
+ data["is_stub"] = true
+ if len(task.inputs) > 0 {
+ data["_arg_bindings"] =
serializeArgBindings(task.inputs)
+ }
+ if task.ifRef != nil {
Review Comment:
Heads up for the merge with #74321: that PR renames this field from `ifRef`
(`*IfRef`) to `decider` (an interface). This check needs to become
`task.decider != nil`, and `_can_skip_downstream` needs the same type-switch
handling `markRegistered` uses for `IfRef`/`SwitchRef`. Otherwise this won't
compile, or worse, gets "fixed" by only checking `*IfRef` and Switch tasks
silently lose `_can_skip_downstream`.
##########
go-sdk/airflow/trigger_dag_run.go:
##########
@@ -161,5 +172,63 @@ func copyConf(conf map[string]any) (map[string]any, error)
{
if err := dec.Decode(&copied); err != nil {
return nil, err
}
+ if _, err := resolveNumbers(copied, ""); err != nil {
+ return nil, err
+ }
return copied, nil
}
+
+// resolveNumbers replaces each json.Number in value with an int64, a uint64
or a float64, and
+// returns value. It changes a map or a slice in value in place. path names
value in an error, as
+// in ["rows"][2].
+func resolveNumbers(value any, path string) (any, error) {
+ switch v := value.(type) {
+ case json.Number:
+ return resolveNumber(v, path)
+ case map[string]any:
+ // The keys are sorted, so that a conf with more than one bad
number always gets the same
+ // error.
+ for _, key := range slices.Sorted(maps.Keys(v)) {
+ resolved, err := resolveNumbers(v[key],
path+"["+strconv.Quote(key)+"]")
+ if err != nil {
+ return nil, err
+ }
+ v[key] = resolved
+ }
+ case []any:
+ for i, item := range v {
+ resolved, err := resolveNumbers(item,
fmt.Sprintf("%s[%d]", path, i))
+ if err != nil {
+ return nil, err
+ }
+ v[i] = resolved
+ }
+ }
+ return value, nil
+}
+
+func resolveNumber(n json.Number, path string) (any, error) {
+ if i, err := n.Int64(); err == nil {
+ return i, nil
+ }
+ if u, err := strconv.ParseUint(n.String(), 10, 64); err == nil {
+ return u, nil
+ }
+ f, err := n.Float64()
+ // encoding/json writes a float64 such as 1e20 with neither a point nor
an exponent, so a number
+ // written like an integer can come from a float64. Such a number is
read as a float64 only when
+ // the float64 formats back to the same digits. Otherwise the number is
an integer that does not
+ // fit in 64 bits.
+ if !strings.ContainsAny(n.String(), ".eE") &&
Review Comment:
The round-trip check doesn't catch all 64-bit overflow:
`strconv.FormatFloat(1e20, 'f', -1, 64)` happens to equal
`"100000000000000000000"`, so `Conf` holding a number like 10^20 is silently
accepted as `float64` instead of erroring "does not fit in 64 bits" as the doc
promises. Checking magnitude directly (e.g. via `math/big`) instead of relying
on the round-trip text match would catch this.
##########
go-sdk/airflow/serialize.go:
##########
@@ -0,0 +1,779 @@
+// 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.
+
+package airflow
+
+import (
+ "cmp"
+ "fmt"
+ "reflect"
+ "regexp"
+ "slices"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// A serialized Dag is the Dag JSON that Airflow stores for a Dag and that its
scheduler reads.
+// Airflow core owns the format.
airflow-core/src/airflow/serialization/schema.json describes it,
+// and Python's DagSerialization writes it for a Dag authored in Python. The
serializer in this file
+// does not write the same bytes as Python. Python picks the fields to leave
out with a
+// client_defaults table that a bundle never gets, so Python writes some
fields at their schema
+// default, such as retry_delay, that this serializer leaves out. Airflow
reads a missing field as
+// its schema default, so it loads the same Dag from either serialization.
+// scripts/ci/lang_sdk_serialization/compare.py checks that for the Dags in
test_dags.yaml in the
+// same directory.
+
+// dagSerializationVersion is the __version of a serialized Dag. It is the
SERIALIZER_VERSION that
+// Airflow core's DagSerialization checks when it loads a serialized Dag. The
supervisor schema has
+// versions named by date, which Airflow negotiates with each runtime, but the
Dag JSON that a
+// bundle returns has only this number.
+const dagSerializationVersion = 3
+
+// A task that runs a Go function gets this task_type and _task_module, where
a task of a Python Dag
+// names the class and the module of its operator. Airflow never imports
_task_module, so the two
+// can name the coordinator that runs the task instead, as the TypeScript SDK
does for its tasks.
+// Nothing in Airflow reads language. It tells a reader of the serialized Dag
which SDK wrote the
+// task.
+const (
+ goTaskType = "GoOperator"
+ goTaskModule = "airflow.sdk.coordinators.executable"
+ goLanguage = "go"
+)
+
+// Python's serializer writes these values for a TriggerDagRunOperator. The
operator class is in
+//
providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py.
+const (
+ triggerDagRunTaskType = "TriggerDagRunOperator"
+ triggerDagRunTaskModule =
"airflow.providers.standard.operators.trigger_dagrun"
+ triggerDagRunUIColor = "#ffefeb"
+)
+
+var triggerDagRunTemplateFields = []string{
+ "trigger_dag_id",
+ "trigger_run_id",
+ "logical_date",
+ "conf",
+ "wait_for_completion",
+ "skip_when_already_exists",
+}
+
+// The ui_color and ui_fgcolor that Python's TaskGroup uses when the Dag
author sets neither.
+const (
+ defaultGroupUIColor = "CornflowerBlue"
+ defaultGroupUIFgColor = "#000"
+)
+
+// A Schedule maps to one of these timetables. Python's DAG picks the same
ones for a schedule
+// string while [scheduler] create_cron_data_intervals is false, which is its
default. When the
+// option is true, Python picks CronDataIntervalTimetable for a cron
expression instead. A bundle
+// cannot read the option, so a Go Dag always gets CronTriggerTimetable
+// (https://github.com/apache/airflow/issues/67938).
+const (
+ nullTimetable = "airflow.timetables.simple.NullTimetable"
+ onceTimetable = "airflow.timetables.simple.OnceTimetable"
+ continuousTimetable = "airflow.timetables.simple.ContinuousTimetable"
+ cronTriggerTimetable = "airflow.timetables.trigger.CronTriggerTimetable"
+)
+
+// cronPresets maps each cron preset to the expression that Python records for
it. It holds the same
+// entries as cron_presets in airflow-core/src/airflow/utils/dates.py.
+var cronPresets = map[string]string{
+ "@hourly": "0 * * * *",
+ "@daily": "0 0 * * *",
+ "@weekly": "0 0 * * 0",
+ "@monthly": "0 0 1 * *",
+ "@quarterly": "0 0 1 */3 *",
+ "@yearly": "0 0 1 1 *",
+}
+
+// croniterAliases are the presets that croniter, which reads a cron
expression for Airflow, accepts
+// in upper or lower case, such as "@Daily". Python passes such a preset to
croniter unchanged
+// unless the preset is a key of cronPresets, and a Go Dag does the same.
+var croniterAliases = map[string]bool{
+ "@midnight": true,
+ "@hourly": true,
+ "@daily": true,
+ "@weekly": true,
+ "@monthly": true,
+ "@yearly": true,
+ "@annually": true,
+}
+
+// cronItem matches one comma-separated item of a field of a cron expression.
An item is one value,
+// or values joined by -, / or #. A value is one of these:
+// - a number, with or without an L or a W before or after it, such as 15W
or L5
+// - one of *, ?, L, W and R
+// - R with a range, such as R(0-30)
+// - three letters, as in a month or weekday name such as JAN or MON
+//
+// cronItem checks only the shape of an item. It takes some items that
croniter rejects, such as a
+// minute of 61 or a weekday of 5L. croniter rejects those only when Airflow
tries to schedule the
+// Dag.
+var cronItem = regexp.MustCompile(`^(?i:` + cronValue + `(?:[-/#]` + cronValue
+ `)*)$`)
+
+const cronValue =
`(?:[0-9]+[LW]?|[LW][0-9]+|[*?LWR]|R\([0-9]+-[0-9]+\)|[A-Z]{3})`
+
+// checkSchedule returns an error unless the Schedule is empty, a preset, or
shaped like a cron
+// expression of five to seven fields. Airflow does not check a cron
expression when it loads a
+// serialized Dag. Without this check, a Schedule such as "every day" would
reach Airflow, and
+// Airflow would never schedule the Dag, with only a line in its log to say
why.
+func checkSchedule(schedule string) error {
+ switch schedule {
+ case "", "@once", "@continuous":
+ return nil
+ }
+ if _, ok := cronPresets[schedule]; ok ||
croniterAliases[strings.ToLower(schedule)] {
+ return nil
+ }
+ fields := strings.Fields(schedule)
+ shaped := len(fields) >= 5 && len(fields) <= 7
+ for _, field := range fields {
+ for item := range strings.SplitSeq(field, ",") {
+ shaped = shaped && cronItem.MatchString(item)
+ }
+ }
+ if !shaped {
+ return fmt.Errorf(
+ "airflow.DagSpec.Schedule is %q, which is not a cron
expression or a preset; set a "+
+ "cron expression such as \"0 3 * * *\", a
preset such as \"@daily\", \"@once\" or "+
+ "\"@continuous\", or no Schedule for a Dag that
runs only when something triggers it",
+ schedule,
+ )
+ }
+ return nil
+}
+
+// serializeTimetable returns the timetable that Python's DAG builds from the
same schedule string.
+// Dag has already checked the Schedule with checkSchedule.
+func serializeTimetable(schedule string) map[string]any {
+ switch schedule {
+ case "":
+ return map[string]any{"__type": nullTimetable, "__var":
map[string]any{}}
+ case "@once":
+ return map[string]any{"__type": onceTimetable, "__var":
map[string]any{}}
+ case "@continuous":
+ return map[string]any{"__type": continuousTimetable, "__var":
map[string]any{}}
+ }
+ expression := schedule
+ if preset, ok := cronPresets[schedule]; ok {
+ expression = preset
+ }
+ return map[string]any{
+ "__type": cronTriggerTimetable,
+ "__var": map[string]any{
+ "expression": expression,
+ "timezone": "UTC",
+ "interval": 0.0,
+ "run_immediately": false,
+ },
+ }
+}
+
+// fieldRule says how the serializer writes one field of DagSpec, TaskSpec or
TaskGroupSpec. Each of
+// the three structs has a table with an entry for every field. When the
generator adds a field to a
+// struct, TestSpecRulesCoverEveryField fails until the table gets an entry
for it.
+type fieldRule struct {
+ // key is the property of the serialization schema that the field sets.
A field with no key, such
+ // as DagSpec.Schedule, has code of its own that writes it, which the
comment on its entry names.
+ key string
+ // schemaDefault is the default that the serialization schema gives the
property, or nil when the
+ // schema has no default for it. writeSpecFields leaves out a value
equal to the default, because
+ // Airflow reads a missing property as its schema default.
+ schemaDefault any
+ // set marks a list that Python keeps in a set, so Python writes the
list sorted and without
+ // repeats. writeSpecFields writes it the same way.
+ set bool
+ // skip marks a field that has a key but that writeSpecFields never
writes. The comment on its
+ // entry says why.
+ skip bool
+}
+
+var dagSpecRules = map[string]fieldRule{
+ "Catchup": {key: "catchup"},
+ "DagDisplayName": {key: "dag_display_name"},
+ "DagrunTimeout": {key: "dagrun_timeout"},
+ "Description": {key: "description"},
+ "DisableBundleVersioning": {key: "disable_bundle_versioning"},
+ "DocMD": {key: "doc_md"},
+ "EndDate": {key: "end_date"},
+ "FailFast": {key: "fail_fast", schemaDefault: false},
+ "IsPausedUponCreation": {key: "is_paused_upon_creation"},
+ "MaxActiveRuns": {key: "max_active_runs"},
+ "MaxActiveTasks": {key: "max_active_tasks"},
+ "MaxConsecutiveFailedDagRuns": {key: "max_consecutive_failed_dag_runs"},
+ // serializeTaskLocked writes Queue as the queue of each Go task whose
TaskSpec sets no Queue.
+ "Queue": {},
+ "RenderTemplateAsNativeObj": {key: "render_template_as_native_obj",
schemaDefault: false},
+ // serializeTimetable writes Schedule as the timetable.
+ "Schedule": {},
+ "StartDate": {key: "start_date"},
+ "Tags": {key: "tags", set: true},
+}
+
+var taskSpecRules = map[string]fieldRule{
+ "TaskDisplayName": {key: "_task_display_name"},
+ "DependsOnPast": {key: "depends_on_past", schemaDefault: false},
+ "DoXComPush": {key: "do_xcom_push", schemaDefault: true},
+ "DocMD": {key: "doc_md"},
+ // Python writes email_on_failure and email_on_retry only for an
operator that has an email
+ // recipient, and a TaskSpec has no field for a recipient.
+ "EmailOnFailure": {key: "email_on_failure", schemaDefault:
true, skip: true},
Review Comment:
`EmailOnFailure`/`EmailOnRetry` are public, settable `*bool` fields that
`writeSpecFields` discards via `skip: true`, with nothing in the field's doc
comment warning that setting them has no effect. Worth a doc note, or wiring
them through. (`schemaDefault: true` on both is also dead code, `skip`
short-circuits before it's read.)
--
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]
