jason810496 commented on code in PR #75700:
URL: https://github.com/apache/airflow/pull/75700#discussion_r4521508181
##########
go-sdk/pkg/execution/messages.go:
##########
@@ -117,7 +119,109 @@ func decodeBody(raw msgpack.RawMessage, dst any) error {
if isNilRaw(raw) {
return fmt.Errorf("empty response body")
}
- return msgpack.Unmarshal(raw, dst)
+ return unmarshalNormalized(raw, dst)
+}
+
+// unmarshalNormalized decodes raw into dst, then normalizes the numbers that
+// msgpack decoded into any-typed values, wherever they sit in dst.
+func unmarshalNormalized(raw []byte, dst any) error {
+ if err := msgpack.Unmarshal(raw, dst); err != nil {
+ return err
+ }
+ normalizeDecodedNumbers(reflect.ValueOf(dst))
+ return nil
+}
+
+// normalizeDecodedNumbers applies normalizeDecodedValue, in place, to every
+// any-typed struct field, map value and slice element reachable from v.
+func normalizeDecodedNumbers(v reflect.Value) {
+ switch v.Kind() {
+ case reflect.Pointer:
+ if !v.IsNil() {
+ normalizeDecodedNumbers(v.Elem())
+ }
+ case reflect.Interface:
+ if !v.IsNil() && v.CanSet() {
+
v.Set(reflect.ValueOf(normalizeDecodedValue(v.Interface())))
+ }
+ case reflect.Struct:
+ for i := range v.NumField() {
+ if v.Type().Field(i).IsExported() {
+ normalizeDecodedNumbers(v.Field(i))
+ }
+ }
+ case reflect.Slice, reflect.Array:
+ if !mayHoldDecodedAny(v.Type().Elem()) {
+ return
+ }
+ for i := range v.Len() {
+ normalizeDecodedNumbers(v.Index(i))
+ }
+ case reflect.Map:
+ if !mayHoldDecodedAny(v.Type().Elem()) {
+ return
+ }
+ iter := v.MapRange()
+ for iter.Next() {
+ // A map value is not addressable, so normalize a copy
and store it back.
+ elem := reflect.New(v.Type().Elem()).Elem()
+ elem.Set(iter.Value())
+ normalizeDecodedNumbers(elem)
+ v.SetMapIndex(iter.Key(), elem)
+ }
Review Comment:
Nit: reuse the `reflect.New` across loop.
```suggestion
elem := reflect.New(v.Type().Elem()).Elem()
iter := v.MapRange()
for iter.Next() {
// A map value is not addressable, so normalize a copy
and store it back.
elem.Set(iter.Value())
normalizeDecodedNumbers(elem)
v.SetMapIndex(iter.Key(), elem)
}
```
##########
go-sdk/pkg/execution/messages.go:
##########
@@ -117,7 +119,109 @@ func decodeBody(raw msgpack.RawMessage, dst any) error {
if isNilRaw(raw) {
return fmt.Errorf("empty response body")
}
- return msgpack.Unmarshal(raw, dst)
+ return unmarshalNormalized(raw, dst)
+}
+
+// unmarshalNormalized decodes raw into dst, then normalizes the numbers that
+// msgpack decoded into any-typed values, wherever they sit in dst.
+func unmarshalNormalized(raw []byte, dst any) error {
+ if err := msgpack.Unmarshal(raw, dst); err != nil {
+ return err
+ }
+ normalizeDecodedNumbers(reflect.ValueOf(dst))
+ return nil
+}
+
+// normalizeDecodedNumbers applies normalizeDecodedValue, in place, to every
+// any-typed struct field, map value and slice element reachable from v.
+func normalizeDecodedNumbers(v reflect.Value) {
+ switch v.Kind() {
+ case reflect.Pointer:
+ if !v.IsNil() {
+ normalizeDecodedNumbers(v.Elem())
+ }
+ case reflect.Interface:
+ if !v.IsNil() && v.CanSet() {
+
v.Set(reflect.ValueOf(normalizeDecodedValue(v.Interface())))
+ }
+ case reflect.Struct:
+ for i := range v.NumField() {
+ if v.Type().Field(i).IsExported() {
+ normalizeDecodedNumbers(v.Field(i))
+ }
+ }
+ case reflect.Slice, reflect.Array:
+ if !mayHoldDecodedAny(v.Type().Elem()) {
+ return
+ }
+ for i := range v.Len() {
+ normalizeDecodedNumbers(v.Index(i))
+ }
+ case reflect.Map:
+ if !mayHoldDecodedAny(v.Type().Elem()) {
+ return
+ }
+ iter := v.MapRange()
+ for iter.Next() {
+ // A map value is not addressable, so normalize a copy
and store it back.
+ elem := reflect.New(v.Type().Elem()).Elem()
+ elem.Set(iter.Value())
+ normalizeDecodedNumbers(elem)
+ v.SetMapIndex(iter.Key(), elem)
+ }
+ }
+}
+
+// mayHoldDecodedAny reports whether a value of type t can contain an any-typed
+// value, so that normalizeDecodedNumbers skips the elements of a []byte or a
+// map[string]string.
+func mayHoldDecodedAny(t reflect.Type) bool {
+ switch t.Kind() {
+ case reflect.Interface, reflect.Pointer, reflect.Struct,
+ reflect.Slice, reflect.Array, reflect.Map:
+ return true
+ default:
+ return false
+ }
+}
+
+// normalizeDecodedValue returns value with its integers as int64, or uint64
+// above math.MaxInt64, and its floats as float64, including the values nested
+// in its maps and slices, which it updates in place. msgpack decodes a number
+// into the narrowest Go type that its wire encoding names, and the supervisor
+// packs each integer in the fewest bytes, so 7, 200 and 70000 would otherwise
+// arrive as int8, uint8 and uint32. The Java SDK decodes every integer to a
+// Long in the same way.
+func normalizeDecodedValue(value any) any {
+ switch v := value.(type) {
+ case map[string]any:
+ for key, item := range v {
+ v[key] = normalizeDecodedValue(item)
+ }
+ case []any:
+ for i, item := range v {
+ v[i] = normalizeDecodedValue(item)
+ }
+ case int8:
+ return int64(v)
Review Comment:
This PR drops the `int`/`uint` cases that `normalizeDagRunConfValue` had.
```suggestion
case int:
return int64(v)
case uint:
return normalizeDecodedValue(uint64(v))
case int8:
return int64(v)
```
##########
go-sdk/pkg/execution/messages.go:
##########
@@ -117,7 +119,109 @@ func decodeBody(raw msgpack.RawMessage, dst any) error {
if isNilRaw(raw) {
return fmt.Errorf("empty response body")
}
- return msgpack.Unmarshal(raw, dst)
+ return unmarshalNormalized(raw, dst)
+}
+
+// unmarshalNormalized decodes raw into dst, then normalizes the numbers that
+// msgpack decoded into any-typed values, wherever they sit in dst.
+func unmarshalNormalized(raw []byte, dst any) error {
+ if err := msgpack.Unmarshal(raw, dst); err != nil {
+ return err
+ }
+ normalizeDecodedNumbers(reflect.ValueOf(dst))
+ return nil
+}
+
+// normalizeDecodedNumbers applies normalizeDecodedValue, in place, to every
+// any-typed struct field, map value and slice element reachable from v.
+func normalizeDecodedNumbers(v reflect.Value) {
+ switch v.Kind() {
+ case reflect.Pointer:
+ if !v.IsNil() {
+ normalizeDecodedNumbers(v.Elem())
+ }
+ case reflect.Interface:
+ if !v.IsNil() && v.CanSet() {
+
v.Set(reflect.ValueOf(normalizeDecodedValue(v.Interface())))
+ }
+ case reflect.Struct:
+ for i := range v.NumField() {
+ if v.Type().Field(i).IsExported() {
+ normalizeDecodedNumbers(v.Field(i))
+ }
+ }
+ case reflect.Slice, reflect.Array:
+ if !mayHoldDecodedAny(v.Type().Elem()) {
+ return
+ }
+ for i := range v.Len() {
+ normalizeDecodedNumbers(v.Index(i))
+ }
+ case reflect.Map:
+ if !mayHoldDecodedAny(v.Type().Elem()) {
+ return
+ }
+ iter := v.MapRange()
+ for iter.Next() {
+ // A map value is not addressable, so normalize a copy
and store it back.
+ elem := reflect.New(v.Type().Elem()).Elem()
+ elem.Set(iter.Value())
+ normalizeDecodedNumbers(elem)
+ v.SetMapIndex(iter.Key(), elem)
+ }
+ }
+}
+
+// mayHoldDecodedAny reports whether a value of type t can contain an any-typed
+// value, so that normalizeDecodedNumbers skips the elements of a []byte or a
+// map[string]string.
+func mayHoldDecodedAny(t reflect.Type) bool {
Review Comment:
Nit: AGENTS.md asks for action-verb names or an `is`/`has` predicate prefix.
Callers at L154 and L161 need the same rename.
```suggestion
// canHoldDecodedAny reports whether a value of type t can contain an
any-typed
// value, so that normalizeDecodedNumbers skips the elements of a []byte or a
// map[string]string.
func canHoldDecodedAny(t reflect.Type) bool {
```
##########
go-sdk/pkg/execution/messages_test.go:
##########
@@ -542,6 +544,93 @@ func TestAPIErrorFromFrame(t *testing.T) {
})
}
+// marshalCompactBody encodes body with each integer in the fewest bytes, as
the
+// supervisor's msgpack packer does.
+func marshalCompactBody(t *testing.T, body any) msgpack.RawMessage {
+ t.Helper()
+ var buf bytes.Buffer
+ enc := msgpack.NewEncoder(&buf)
+ enc.UseCompactInts(true)
+ require.NoError(t, enc.Encode(body))
+ return buf.Bytes()
+}
+
+func TestDecodeBodyNormalizesNumbers(t *testing.T) {
+ cases := []struct {
+ name string
+ value any
+ want any
+ }{
+ {name: "positive fixint", value: 7, want: int64(7)},
+ {name: "uint8", value: 200, want: int64(200)},
+ {name: "uint16", value: 500, want: int64(500)},
+ {name: "uint32", value: 70000, want: int64(70000)},
+ {name: "uint64 up to max int64", value: uint64(math.MaxInt64),
want: int64(math.MaxInt64)},
+ {
+ name: "uint64 above max int64",
+ value: uint64(math.MaxInt64) + 1,
+ want: uint64(math.MaxInt64) + 1,
+ },
+ {name: "negative fixint", value: -1, want: int64(-1)},
+ {name: "int8", value: -100, want: int64(-100)},
+ {name: "int16", value: -129, want: int64(-129)},
+ {name: "int32", value: -32769, want: int64(-32769)},
+ {name: "float32", value: float32(1.25), want: float64(1.25)},
+ {
+ name: "nested in maps and arrays",
+ value: map[string]any{
+ "rows": 7,
+ "batches": []any{200, "eu", nil,
map[string]any{"ratio": float32(0.5)}},
+ },
+ want: map[string]any{
+ "rows": int64(7),
+ "batches": []any{int64(200), "eu", nil,
map[string]any{"ratio": float64(0.5)}},
+ },
+ },
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ raw := marshalCompactBody(t, map[string]any{
+ "type": "XComResult",
+ "key": "return_value",
+ "value": tc.value,
+ })
+ var result genmodels.XComResult
+ require.NoError(t, decodeBody(raw, &result))
+ assert.Equal(t, tc.want, result.Value)
+ })
+ }
+}
+
+func TestDecodeStartupDetailsNormalizesConfNumbers(t *testing.T) {
Review Comment:
Tests cover the interface path (`XComResult.Value`) and
pointer/struct/`map[string]any` (`StartupDetails` conf).
It doesn't cover: the `mayHoldDecodedAny` skip, the slice path (e.g.
`DagFileParseRequest.CallbackRequests`), maps with struct values, the
`DagFileParseRequest`/`ErrorResponse` switch to `unmarshalNormalized`, and the
Go-int defaults above.
--
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]
