This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 665e91fc7d0 Go SDK: Reject a task priority weight outside the int32
range (#76284)
665e91fc7d0 is described below
commit 665e91fc7d01c9064c42a56835006306769a28ae
Author: Y-C <[email protected]>
AuthorDate: Sun Nov 8 21:31:17 2026 +0800
Go SDK: Reject a task priority weight outside the int32 range (#76284)
---
go-sdk/airflow/dag.go | 1 +
go-sdk/airflow/enums.go | 5 ++++-
go-sdk/airflow/serialize.go | 15 +++++++++++++++
go-sdk/airflow/serialize_test.go | 30 ++++++++++++++++++++++++++++++
4 files changed, 50 insertions(+), 1 deletion(-)
diff --git a/go-sdk/airflow/dag.go b/go-sdk/airflow/dag.go
index 63c606ed45e..c639c160004 100644
--- a/go-sdk/airflow/dag.go
+++ b/go-sdk/airflow/dag.go
@@ -215,6 +215,7 @@ type TaskRef struct {
// - the TaskSpec sets TriggerRule to a value that is not a TriggerRule
constant
// - the TaskSpec sets WeightRule to a value that is not a WeightRule
constant
// - the year of the StartDate or the EndDate of the TaskSpec in UTC is not
from 1 to 9999
+// - the TaskSpec sets PriorityWeight outside the 32-bit range, from
-2147483648 to 2147483647
// - the inputs passed to Inputs do not match the parameters of fn after the
Context
// - the task_id, with the group_ids that prefix it, is longer than 250
characters, or holds a
// character other than a letter, a digit, an underscore, a dash or a dot,
as Python's
diff --git a/go-sdk/airflow/enums.go b/go-sdk/airflow/enums.go
index 709b581a36d..81c89460810 100644
--- a/go-sdk/airflow/enums.go
+++ b/go-sdk/airflow/enums.go
@@ -154,7 +154,7 @@ var (
// checkTaskSpec rejects a TaskSpec whose TriggerRule is not a TriggerRule
constant or whose
// WeightRule is not a WeightRule constant. An empty TriggerRule or WeightRule
is valid and leaves
// that rule unset. checkTaskSpec also rejects a StartDate or an EndDate that
a Python datetime
-// cannot hold.
+// cannot hold, and a PriorityWeight that the priority_weight column of
task_instance cannot hold.
func checkTaskSpec(spec TaskSpec) error {
if err := checkRule("TriggerRule", "trigger rule", spec.TriggerRule,
triggerRules); err != nil {
return err
@@ -162,6 +162,9 @@ func checkTaskSpec(spec TaskSpec) error {
if err := checkRule("WeightRule", "weight rule", spec.WeightRule,
weightRules); err != nil {
return err
}
+ if err := checkPriorityWeight(spec.PriorityWeight); err != nil {
+ return err
+ }
if err := checkTime("airflow.TaskSpec.StartDate", spec.StartDate); err
!= nil {
return err
}
diff --git a/go-sdk/airflow/serialize.go b/go-sdk/airflow/serialize.go
index 0eec1a44a3b..ec9a850ef66 100644
--- a/go-sdk/airflow/serialize.go
+++ b/go-sdk/airflow/serialize.go
@@ -20,6 +20,7 @@ package airflow
import (
"cmp"
"fmt"
+ "math"
"reflect"
"regexp"
"slices"
@@ -355,6 +356,20 @@ func checkTime(field string, t time.Time) error {
return nil
}
+// checkPriorityWeight returns an error for a weight that the 32-bit
priority_weight column of
+// task_instance cannot hold. Python's BaseOperator caps such a weight
instead. A nil weight passes,
+// because it means that the field is unset.
+func checkPriorityWeight(weight *int) error {
+ if weight != nil && (*weight < math.MinInt32 || *weight >
math.MaxInt32) {
+ return fmt.Errorf(
+ "airflow.TaskSpec.PriorityWeight is %d; Airflow takes a
priority weight only "+
+ "from %d to %d",
+ *weight, math.MinInt32, math.MaxInt32,
+ )
+ }
+ 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 }
diff --git a/go-sdk/airflow/serialize_test.go b/go-sdk/airflow/serialize_test.go
index 4a7bcf96dd8..1e8a6544659 100644
--- a/go-sdk/airflow/serialize_test.go
+++ b/go-sdk/airflow/serialize_test.go
@@ -22,8 +22,10 @@ import (
"encoding/json"
"fmt"
"maps"
+ "math"
"reflect"
"slices"
+ "strconv"
"strings"
"testing"
"time"
@@ -339,6 +341,34 @@ func
TestSerializeWritesAPointerFieldThatHoldsTheZeroValue(t *testing.T) {
)
}
+func TestTaskRejectsAPriorityWeightOutsideTheInt32Range(t *testing.T) {
+ if strconv.IntSize < 64 {
+ t.Skip("requires 64-bit int to hold a weight outside the int32
range")
+ }
+ // int64, not int, so the file still compiles where int is 32 bits.
+ for name, weight := range map[string]int64{
+ "above": math.MaxInt32 + 1,
+ "below": math.MinInt32 - 1,
+ } {
+ t.Run(name, func(t *testing.T) {
+ assert.PanicsWithValue(t,
+ fmt.Sprintf(`airflow.DagRef.Task: task "ping"
of Dag "etl": `+
+ "airflow.TaskSpec.PriorityWeight is %d;
Airflow takes a priority weight "+
+ "only from -2147483648 to 2147483647",
weight),
+ func() { Dag("etl").Task(ping,
TaskSpec{PriorityWeight: ptr(int(weight))}) },
+ )
+ })
+ }
+}
+
+func TestTaskAcceptsAPriorityWeightAtTheInt32Bounds(t *testing.T) {
+ dag := Dag("etl")
+ assert.NotPanics(t, func() {
+ dag.Task(ping, TaskSpec{TaskID: "max", PriorityWeight:
ptr(math.MaxInt32)})
+ dag.Task(ping, TaskSpec{TaskID: "min", PriorityWeight:
ptr(math.MinInt32)})
+ })
+}
+
func TestSerializeWritesTimesAndDurationsToTheMicrosecond(t *testing.T) {
dag := Dag("etl")
dag.Task(ping, TaskSpec{
