From d1c1d8bf47f7068519080fefac6eac5681d44bef Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Tue, 18 Aug 2026 08:53:37 +0000 Subject: [PATCH 1/2] Bind Python temporal and UUID arguments to their native Go types A Dag author who annotates a stub parameter `datetime`, `timedelta` or `UUID` means a moment, a span or an identifier -- not the string those become on the wire. Without this the Go task receives that string and has to parse it itself, re-deriving a contract the annotation already stated, and getting it wrong in ways the runtime could have caught. The annotation is what pydantic turns into the JSON-schema `format` the argument spec carries, so it is also what decides which Go type can receive the value: `date-time`/`date`/`time` fill a time.Time, `duration` a time.Duration, `uuid` a uuid.UUID. Reading it as the contract means a Go parameter the declared Python type cannot fill now fails before the task body runs rather than binding something the author never meant -- including a plain int declared as a time.Duration, which would otherwise be read as a count of nanoseconds. These values are rewritten into a spelling encoding/json can read by one walk over the target type, so a duration is reached wherever it sits: the argument itself, a slice element, a map value or a struct field. A single walk is what keeps those cases consistent; reaching only the top level would leave a `list[timedelta]` or a struct's `timedelta` field failing against a schema that describes them perfectly well. --- .../test_go_sdk_taskflow_binding.py | 11 + go-sdk/README.md | 27 + .../airflow-go-pack/pack_integration_test.go | 1 + go-sdk/dags/go_examples.py | 15 +- go-sdk/example/bundle/main.go | 1 + .../bundle/taskflowbinding/taskflowbinding.go | 35 ++ .../taskflowbinding/taskflowbinding_test.go | 34 ++ go-sdk/pkg/binding/binding.go | 410 ++++++++++++++- go-sdk/pkg/binding/binding_test.go | 492 +++++++++++++++++- 9 files changed, 1002 insertions(+), 24 deletions(-) diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_taskflow_binding.py b/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_taskflow_binding.py index d078261f8b1c7..6ccce4f5d171c 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_taskflow_binding.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_taskflow_binding.py @@ -83,6 +83,7 @@ def test_all_tasks_succeeded(completed_run: _CompletedRun): "via_flat_map", "via_struct_map", "via_plain_map", + "via_temporal_args", ): assert completed_run.ti_states.get(task_id) == "success", completed_run.ti_states @@ -168,3 +169,13 @@ def test_via_plain_map_decodes_dict_into_a_typed_map(completed_run: _CompletedRu """``via_plain_map`` passes one dict literal onto a Go ``map[string]string`` parameter, so it decodes with no struct involved at all.""" assert completed_run.xcom("via_plain_map") == {"team": "data", "tier": "gold"} + + +def test_via_temporal_args_binds_formatted_strings_onto_native_go_types( + completed_run: _CompletedRun, +): + assert completed_run.xcom("via_temporal_args") == { + "when": "2024-01-02T03:04:05Z", + "window_seconds": 300, + "trace_id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8", + } diff --git a/go-sdk/README.md b/go-sdk/README.md index 360e185a9a546..c890779d2b9c4 100644 --- a/go-sdk/README.md +++ b/go-sdk/README.md @@ -169,6 +169,33 @@ data parameters is rejected at registration. A sole struct parameter also falls decoding when it gets exactly one passed argument no field claims, so a task can still take an upstream object as a single argument. +### Temporal and UUID arguments + +The stub's Python annotation decides which Go type an argument binds onto. The annotation describes +what the Go task receives, not what the Dag passes: a TaskFlow call argument has to be +JSON-serializable to cross into another runtime, so pass the formatted string (`window="PT5M"`) +rather than a native `timedelta`, which fails when the Dag is serialized. + +| Python annotation | Go parameter type | +| --- | --- | +| `datetime`, `pendulum.DateTime` | `time.Time` | +| `date` | `time.Time` (midnight) | +| `time` | `time.Time` (zero date) | +| `timedelta`, `pendulum.Duration` | `time.Duration` | +| `UUID` | `uuid.UUID` | + +Any of these also binds onto a plain Go `string` if the task would rather parse the value itself — +the escape hatch for annotations the table doesn't cover, alongside any type that decodes itself +from JSON. They nest too: `list[timedelta]` binds onto `[]time.Duration`, `dict[str, datetime]` onto +`map[string]time.Time`, and a struct field is reached just as readily, with a malformed value +reported by its position, key or field name. + +The annotation is authoritative: a Go type the declared Python type cannot fill fails the task +before its body runs rather than binding a value the author never meant — including a plain `int` +declared as a Go `time.Duration`, which would otherwise be read as a count of nanoseconds. A +nullable annotation (`datetime | None`) needs a pointer on the Go side so a `None` has somewhere to +land. + ### Reading the task runtime context Declare an `sdk.TIRunContext` parameter on a task to read the identifiers and scheduling timestamps of the diff --git a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go index ace867461e2cc..f24356ba7fece 100644 --- a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go +++ b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go @@ -166,6 +166,7 @@ dags: - "via_flat_map" - "via_struct_map" - "via_plain_map" + - "via_temporal_args" ` assert.Equal(t, expectedManifest, string(metadata)) diff --git a/go-sdk/dags/go_examples.py b/go-sdk/dags/go_examples.py index d6e768a2c3721..3c9143c98f278 100644 --- a/go-sdk/dags/go_examples.py +++ b/go-sdk/dags/go_examples.py @@ -54,7 +54,8 @@ from __future__ import annotations -from datetime import timedelta +from datetime import datetime, timedelta +from uuid import UUID from airflow.sdk import dag, task @@ -165,6 +166,11 @@ def via_struct_map(payload: dict): ... def via_plain_map(labels: dict): ... +# Annotations select native Go types. +@task.stub(queue="golang") +def via_temporal_args(when: datetime, window: timedelta, trace_id: UUID): ... + + @dag(dag_id="taskflow_binding_dag") def taskflow_binding_dag(): """ @@ -184,6 +190,7 @@ def taskflow_binding_dag(): default no Go field claims. * ``via_flat_map`` / ``via_struct_map`` / ``via_plain_map``: one dict bound whole into a struct, onto a struct's map field, and into a plain Go map. + * ``via_temporal_args``: formatted strings bound to native temporal and UUID types. Each ``via_struct_*`` call mixes a ``threshold`` literal with ``make_region``'s XCom, so struct fields are proven against both argument sources. The Go @@ -205,6 +212,12 @@ def taskflow_binding_dag(): via_flat_map(config={"region": "eu-west-1", "count": 3}) via_struct_map(payload={"region": "eu-west-1", "count": 3}) via_plain_map(labels={"team": "data", "tier": "gold"}) + # Cross-runtime arguments use JSON strings. + via_temporal_args( + when="2024-01-02T03:04:05Z", + window="PT5M", + trace_id="6ba7b810-9dad-11d1-80b4-00c04fd430c8", + ) taskflow_binding_dag() diff --git a/go-sdk/example/bundle/main.go b/go-sdk/example/bundle/main.go index 6d6569889008c..4e77edd29710b 100644 --- a/go-sdk/example/bundle/main.go +++ b/go-sdk/example/bundle/main.go @@ -57,6 +57,7 @@ func (m *myBundle) RegisterDags(dagbag v1.Registry) error { bindingDag.AddTaskWithName("via_flat_map", taskflowbinding.ViaFlatMap) bindingDag.AddTaskWithName("via_struct_map", taskflowbinding.ViaStructMap) bindingDag.AddTaskWithName("via_plain_map", taskflowbinding.ViaPlainMap) + bindingDag.AddTaskWithName("via_temporal_args", taskflowbinding.ViaTemporalArgs) return nil } diff --git a/go-sdk/example/bundle/taskflowbinding/taskflowbinding.go b/go-sdk/example/bundle/taskflowbinding/taskflowbinding.go index 2e2ed740cd787..775b8fab65320 100644 --- a/go-sdk/example/bundle/taskflowbinding/taskflowbinding.go +++ b/go-sdk/example/bundle/taskflowbinding/taskflowbinding.go @@ -22,6 +22,9 @@ import ( "fmt" "log/slog" "reflect" + "time" + + "github.com/google/uuid" "github.com/apache/airflow/go-sdk/sdk" ) @@ -264,3 +267,35 @@ func ViaPlainMap( log.InfoContext(ctx, "Bound dict into a plain map", "labels", fmt.Sprint(labels)) return map[string]any{"team": labels["team"], "tier": labels["tier"]}, nil } + +// ViaTemporalArgs validates native argument binding. +func ViaTemporalArgs( + ctx sdk.TIRunContext, + log *slog.Logger, + when time.Time, + window time.Duration, + traceID uuid.UUID, +) (any, error) { + wantWhen := time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC) + if !when.Equal(wantWhen) { + return nil, fmt.Errorf("datetime bound incorrectly: when=%s, want %s", when, wantWhen) + } + if window != 5*time.Minute { + return nil, fmt.Errorf("timedelta bound incorrectly: window=%s, want 5m", window) + } + wantTrace := uuid.MustParse("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + if traceID != wantTrace { + return nil, fmt.Errorf("UUID bound incorrectly: trace_id=%s, want %s", traceID, wantTrace) + } + + log.InfoContext(ctx, "Bound temporal arguments", + "when", when.Format(time.RFC3339), + "window", window.String(), + "trace_id", traceID.String(), + ) + return map[string]any{ + "when": when.UTC().Format(time.RFC3339), + "window_seconds": window.Seconds(), + "trace_id": traceID.String(), + }, nil +} diff --git a/go-sdk/example/bundle/taskflowbinding/taskflowbinding_test.go b/go-sdk/example/bundle/taskflowbinding/taskflowbinding_test.go index 02a3d6c75ade5..6881e63b5b6b7 100644 --- a/go-sdk/example/bundle/taskflowbinding/taskflowbinding_test.go +++ b/go-sdk/example/bundle/taskflowbinding/taskflowbinding_test.go @@ -21,7 +21,9 @@ import ( "context" "log/slog" "testing" + "time" + "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -178,3 +180,35 @@ func TestViaPlainMapRejectsWrongBinding(t *testing.T) { _, err := ViaPlainMap(ctx, slog.Default(), map[string]string{"team": "wrong"}) assert.ErrorContains(t, err, "plain map bound incorrectly") } + +func TestViaTemporalArgs(t *testing.T) { + ctx := sdk.NewTIRunContext(context.Background(), sdk.TaskInstance{}, sdk.DagRun{}) + got, err := ViaTemporalArgs( + ctx, slog.Default(), + time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC), + 5*time.Minute, + uuid.MustParse("6ba7b810-9dad-11d1-80b4-00c04fd430c8"), + ) + require.NoError(t, err) + + summary, ok := got.(map[string]any) + require.True(t, ok, "ViaTemporalArgs should return a map summary, got %T", got) + assert.Equal(t, "2024-01-02T03:04:05Z", summary["when"]) + assert.Equal(t, 300.0, summary["window_seconds"]) + assert.Equal(t, "6ba7b810-9dad-11d1-80b4-00c04fd430c8", summary["trace_id"]) +} + +func TestViaTemporalArgsRejectsWrongBinding(t *testing.T) { + ctx := sdk.NewTIRunContext(context.Background(), sdk.TaskInstance{}, sdk.DagRun{}) + trace := uuid.MustParse("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + when := time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC) + + _, err := ViaTemporalArgs(ctx, slog.Default(), when.Add(time.Hour), 5*time.Minute, trace) + assert.ErrorContains(t, err, "datetime bound incorrectly") + + _, err = ViaTemporalArgs(ctx, slog.Default(), when, 300, trace) + assert.ErrorContains(t, err, "timedelta bound incorrectly") + + _, err = ViaTemporalArgs(ctx, slog.Default(), when, 5*time.Minute, uuid.Nil) + assert.ErrorContains(t, err, "UUID bound incorrectly") +} diff --git a/go-sdk/pkg/binding/binding.go b/go-sdk/pkg/binding/binding.go index 02b3018dc24ae..011e07e3f9055 100644 --- a/go-sdk/pkg/binding/binding.go +++ b/go-sdk/pkg/binding/binding.go @@ -33,9 +33,16 @@ import ( "errors" "fmt" "log/slog" + "math" "reflect" + "regexp" + "slices" + "strconv" "strings" "sync" + "time" + + "github.com/google/uuid" "github.com/apache/airflow/go-sdk/pkg/execution/genmodels" "github.com/apache/airflow/go-sdk/pkg/sdkcontext" @@ -468,7 +475,8 @@ func (p *Plan) decodeArg( "task function %s: %s: nil argument binding", p.fnName, errCtx, ) } - if err := checkValueType(arg.Schema(), targetType); err != nil { + shape, err := checkValueType(arg.Schema(), targetType) + if err != nil { return reflect.Value{}, fmt.Errorf("task function %s: %s: %w", p.fnName, errCtx, err) } var source string @@ -482,7 +490,7 @@ func (p *Plan) decodeArg( "task function %s: %s: unsupported argument binding %T", p.fnName, errCtx, arg, ) } - v, err := decodeValue(raw, targetType) + v, err := decodeValue(raw, targetType, shape) if err != nil { return reflect.Value{}, fmt.Errorf( "task function %s: %s: decoding %s into %s: %w", @@ -536,6 +544,10 @@ func structParamType(in reflect.Type) reflect.Type { if t.Kind() != reflect.Struct { return nil } + // time.Time binds positionally despite being a struct. + if _, isNative := nativeFormats[t]; isNative { + return nil + } return t } @@ -662,6 +674,52 @@ type schemaShape struct { fragment map[string]any } +func (s schemaShape) child(key string) schemaShape { + nested, ok := s.fragment[key].(map[string]any) + if !ok { + return schemaShape{} + } + return nestedShapeOf(nested) +} + +func (s schemaShape) property(name string) schemaShape { + props, ok := s.fragment["properties"].(map[string]any) + if !ok { + return schemaShape{} + } + nested, ok := props[name].(map[string]any) + if !ok { + return schemaShape{} + } + return nestedShapeOf(nested) +} + +// Preserve formats inside nullable schemas. +func nestedShapeOf(fragment map[string]any) schemaShape { + branches, isUnion := fragment["anyOf"].([]any) + if !isUnion { + return shapeOf(fragment) + } + var only map[string]any + for _, branch := range branches { + alt, isMap := branch.(map[string]any) + if !isMap { + return shapeOf(fragment) + } + if jsonType, _ := alt["type"].(string); jsonType == "null" { + continue + } + if only != nil { + return shapeOf(fragment) + } + only = alt + } + if only == nil { + return shapeOf(fragment) + } + return shapeOf(only) +} + func shapeOf(fragment map[string]any) schemaShape { jsonType, _ := fragment["type"].(string) format, _ := fragment["format"].(string) @@ -705,13 +763,22 @@ func schemaShapes( return []schemaShape{shapeOf(fragment)}, false, true } -func checkValueType(schema *genmodels.ArgValueSchema, target reflect.Type) error { +var nativeFormats = map[reflect.Type][]string{ + timeType: {"date-time", "date", "time"}, + durationType: {"duration"}, + uuidType: {"uuid"}, +} + +func checkValueType( + schema *genmodels.ArgValueSchema, + target reflect.Type, +) (matched schemaShape, err error) { shapes, nullable, ok := schemaShapes(schema) if !ok { - return nil + return schemaShape{}, nil } if nullable && !isNilableType(target) { - return fmt.Errorf( + return schemaShape{}, fmt.Errorf( "the Dag may pass null for this argument, so Go parameter type %s must be a "+ "pointer (or a slice, map or any)", target, @@ -722,20 +789,23 @@ func checkValueType(schema *genmodels.ArgValueSchema, target reflect.Type) error t = t.Elem() } if t.Kind() == reflect.Interface { - return nil + return schemaShape{}, nil } for _, shape := range shapes { if isBindableShape(shape, t) { - return nil + return shape, nil } } - return fmt.Errorf( + return schemaShape{}, fmt.Errorf( "the Dag declares %s which cannot bind to Go parameter type %s", describeShapes(shapes), target, ) } func isBindableShape(shape schemaShape, t reflect.Type) bool { + if formats, isNative := nativeFormats[t]; isNative { + return slices.Contains(formats, shape.format) + } // Self-decoding types define their own wire representation. if implementsUnmarshaler(t) { return true @@ -788,7 +858,7 @@ func describeShapes(shapes []schemaShape) string { return "JSON-schema alternatives " + strings.Join(parts, " | ") } -func decodeValue(raw any, target reflect.Type) (reflect.Value, error) { +func decodeValue(raw any, target reflect.Type, shape schemaShape) (reflect.Value, error) { out := reflect.New(target) if raw == nil { @@ -800,7 +870,12 @@ func decodeValue(raw any, target reflect.Type) (reflect.Value, error) { ) } - blob, err := json.Marshal(raw) + normalized, err := normalizeNative(raw, target, shape) + if err != nil { + return reflect.Value{}, err + } + + blob, err := json.Marshal(normalized) if err != nil { return reflect.Value{}, err } @@ -812,6 +887,317 @@ func decodeValue(raw any, target reflect.Type) (reflect.Value, error) { return out.Elem(), nil } +func normalizeNative(raw any, target reflect.Type, shape schemaShape) (any, error) { + if raw == nil { + // Reject null before encoding/json can zero the target. + if !isNilableType(target) { + return nil, fmt.Errorf( + "value is null but the type %s is not nilable", target, + ) + } + return nil, nil + } + for target.Kind() == reflect.Pointer { + target = target.Elem() + } + + if formats, isNative := nativeFormats[target]; isNative { + if shape.jsonType != "" && !slices.Contains(formats, shape.format) { + return nil, fmt.Errorf( + "the Dag declares %s which cannot bind to Go type %s", + describeShapes([]schemaShape{shape}), target, + ) + } + } + + switch target { + case durationType: + text, ok := raw.(string) + if !ok { + // Reject numbers, which Go would interpret as nanoseconds. + return nil, fmt.Errorf( + "expected an ISO-8601 duration string, got %T; a bare number would be "+ + "read as nanoseconds", + raw, + ) + } + d, err := parseISO8601Duration(text) + if err != nil { + return nil, err + } + return int64(d), nil + case timeType: + text, ok := raw.(string) + if !ok { + return nil, fmt.Errorf("expected a timestamp string, got %T", raw) + } + ts, err := parseTemporal(text, shape.format) + if err != nil { + return nil, err + } + return ts.Format(time.RFC3339Nano), nil + } + + if !containsNative(target) { + return raw, nil + } + + switch target.Kind() { + case reflect.Slice, reflect.Array: + items, ok := raw.([]any) + if !ok { + return nil, fmt.Errorf("expected an array, got %T", raw) + } + if target.Kind() == reflect.Array && len(items) != target.Len() { + return nil, fmt.Errorf( + "expected %d elements for %s, got %d", target.Len(), target, len(items), + ) + } + elemShape := shape.child("items") + out := make([]any, len(items)) + for i, item := range items { + v, err := normalizeNative(item, target.Elem(), elemShape) + if err != nil { + return nil, fmt.Errorf("element %d: %w", i, err) + } + out[i] = v + } + return out, nil + + case reflect.Map: + entries, ok := raw.(map[string]any) + if !ok { + return nil, fmt.Errorf("expected an object, got %T", raw) + } + elemShape := shape.child("additionalProperties") + out := make(map[string]any, len(entries)) + for key, item := range entries { + v, err := normalizeNative(item, target.Elem(), elemShape) + if err != nil { + return nil, fmt.Errorf("key %q: %w", key, err) + } + out[key] = v + } + return out, nil + + case reflect.Struct: + entries, ok := raw.(map[string]any) + if !ok { + // Let strict decoding report the mismatch. + return raw, nil + } + fields := nativeStructFields(target) + out := make(map[string]any, len(entries)) + for key, item := range entries { + fieldType, ok := lookupNativeField(fields, key) + if !ok { + // Preserve unknown keys for strict decoding. + out[key] = item + continue + } + v, err := normalizeNative(item, fieldType, shape.property(key)) + if err != nil { + return nil, fmt.Errorf("field %q: %w", key, err) + } + out[key] = v + } + return out, nil + } + return raw, nil +} + +func nativeStructFields(structType reflect.Type) map[string]reflect.Type { + fields := make(map[string]reflect.Type) + collectNativeFields(structType, fields, map[string]bool{}, map[reflect.Type]bool{}) + return fields +} + +func collectNativeFields( + structType reflect.Type, + into map[string]reflect.Type, + claimed map[string]bool, + seen map[reflect.Type]bool, +) { + if seen[structType] { + return + } + seen[structType] = true + + var embeddedTypes []reflect.Type + for i := range structType.NumField() { + f := structType.Field(i) + tag := f.Tag.Get("json") + if tag == "-" { + continue + } + name, _, _ := strings.Cut(tag, ",") + + embedded := f.Type + for embedded.Kind() == reflect.Pointer { + embedded = embedded.Elem() + } + // Embedded structs promote exported fields even when unexported. + if f.Anonymous && name == "" && embedded.Kind() == reflect.Struct && embedded != timeType { + embeddedTypes = append(embeddedTypes, embedded) + continue + } + if !f.IsExported() { + continue + } + if name == "" { + name = f.Name + } + if claimed[name] { + continue + } + claimed[name] = true + if containsNative(f.Type) { + into[name] = f.Type + } + } + for _, embedded := range embeddedTypes { + collectNativeFields(embedded, into, claimed, seen) + } +} + +func lookupNativeField( + fields map[string]reflect.Type, + key string, +) (reflect.Type, bool) { + if t, ok := fields[key]; ok { + return t, true + } + // Match encoding/json's case-insensitive fallback. + for name, t := range fields { + if strings.EqualFold(name, key) { + return t, true + } + } + return nil, false +} + +func containsNative(t reflect.Type) bool { + return containsNativeSeen(t, map[reflect.Type]bool{}) +} + +func containsNativeSeen(t reflect.Type, seen map[reflect.Type]bool) bool { + for t.Kind() == reflect.Pointer { + t = t.Elem() + } + if t == timeType || t == durationType { + return true + } + if implementsUnmarshaler(t) { + // Self-decoding types manage their own wire format. + return false + } + switch t.Kind() { + case reflect.Slice, reflect.Array, reflect.Map: + return containsNativeSeen(t.Elem(), seen) + case reflect.Struct: + if seen[t] { + return false + } + seen[t] = true + for i := range t.NumField() { + f := t.Field(i) + // Embedded structs may promote native fields. + if !f.IsExported() && !f.Anonymous { + continue + } + if containsNativeSeen(f.Type, seen) { + return true + } + } + } + return false +} + +// parseTemporal treats naive Python datetimes as UTC. +func parseTemporal(text, format string) (time.Time, error) { + var layouts []string + switch format { + case "date": + layouts = []string{time.DateOnly} + case "time": + layouts = []string{"15:04:05.999999999Z07:00", "15:04:05.999999999"} + default: + layouts = []string{time.RFC3339Nano, "2006-01-02T15:04:05.999999999"} + } + for _, layout := range layouts { + if ts, err := time.Parse(layout, text); err == nil { + return ts, nil + } + } + return time.Time{}, fmt.Errorf("%q is not a valid %s value", text, temporalName(format)) +} + +func temporalName(format string) string { + if format == "" { + return "date-time" + } + return format +} + +// iso8601Duration matches fixed-length ISO 8601 durations. +var iso8601Duration = regexp.MustCompile( + `^([+-])?P(?:(\d+(?:\.\d+)?)W)?(?:(\d+(?:\.\d+)?)D)?` + + `(?:T(?:(\d+(?:\.\d+)?)H)?(?:(\d+(?:\.\d+)?)M)?(?:(\d+(?:\.\d+)?)S)?)?$`, +) + +var calendarDuration = regexp.MustCompile(`^[+-]?P\d+(?:\.\d+)?[YM]`) + +func parseISO8601Duration(text string) (time.Duration, error) { + groups := iso8601Duration.FindStringSubmatch(text) + if groups == nil { + if calendarDuration.MatchString(text) { + return 0, fmt.Errorf( + "duration %q counts years or months, which have no fixed length; "+ + "express it in weeks, days, or smaller units", + text, + ) + } + return 0, fmt.Errorf( + "%q is not an ISO-8601 duration (want e.g. \"PT5M\" or \"P1DT2H3M4S\")", text, + ) + } + + units := []time.Duration{ + 7 * 24 * time.Hour, // W + 24 * time.Hour, // D + time.Hour, // H + time.Minute, // M + time.Second, // S + } + var total float64 + empty := true + for i, unit := range units { + field := groups[i+2] + if field == "" { + continue + } + n, err := strconv.ParseFloat(field, 64) + if err != nil { + return 0, fmt.Errorf("duration %q: %w", text, err) + } + total += n * float64(unit) + empty = false + } + if empty { + return 0, fmt.Errorf("duration %q carries no components", text) + } + if groups[1] == "-" { + total = -total + } + // Guard float-to-int overflow explicitly. + if total >= math.MaxInt64 || total < math.MinInt64 { + return 0, fmt.Errorf( + "duration %q does not fit in a time.Duration (max about 292 years)", text, + ) + } + return time.Duration(total), nil +} + // Struct fields are checked only if the wire value names them. func isDecodableType(inType reflect.Type) bool { if implementsUnmarshaler(inType) { @@ -847,6 +1233,10 @@ var ( jsonUnmarshalerType = reflect.TypeFor[json.Unmarshaler]() textUnmarshalerType = reflect.TypeFor[encoding.TextUnmarshaler]() + + timeType = reflect.TypeFor[time.Time]() + durationType = reflect.TypeFor[time.Duration]() + uuidType = reflect.TypeFor[uuid.UUID]() ) func isContext(inType reflect.Type) bool { diff --git a/go-sdk/pkg/binding/binding_test.go b/go-sdk/pkg/binding/binding_test.go index 8f94c9a8c111a..6be1ac53830aa 100644 --- a/go-sdk/pkg/binding/binding_test.go +++ b/go-sdk/pkg/binding/binding_test.go @@ -46,6 +46,21 @@ func argSchema(jsonType string) *genmodels.ArgValueSchema { return &s } +func temporalSchema(format string) *genmodels.ArgValueSchema { + s := genmodels.ArgValueSchema{"type": "string", "format": format} + return &s +} + +func arraySchema(items map[string]any) *genmodels.ArgValueSchema { + s := genmodels.ArgValueSchema{"type": "array", "items": items} + return &s +} + +func objectSchema(values map[string]any) *genmodels.ArgValueSchema { + s := genmodels.ArgValueSchema{"type": "object", "additionalProperties": values} + return &s +} + func anyOfSchema(branches ...map[string]any) *genmodels.ArgValueSchema { alternatives := make([]any, len(branches)) for i, b := range branches { @@ -275,6 +290,186 @@ func (s *BindingSuite) TestResolveSelfDecodingLiterals() { s.Equal(3.0, got[2].Interface()) } +func (s *BindingSuite) TestResolveTemporalLiterals() { + fn := func( + aware time.Time, naive time.Time, day time.Time, clock time.Time, window time.Duration, + ) error { + return nil + } + got, err := s.resolve(fn, []Arg{ + LiteralArg{Value: "2024-01-02T03:04:05Z", ValueSchema: temporalSchema("date-time")}, + LiteralArg{Value: "2024-01-02T03:04:05", ValueSchema: temporalSchema("date-time")}, + LiteralArg{Value: "2024-01-02", ValueSchema: temporalSchema("date")}, + LiteralArg{Value: "03:04:05", ValueSchema: temporalSchema("time")}, + LiteralArg{Value: "P1DT2H3M4S", ValueSchema: temporalSchema("duration")}, + }, &fakeXComClient{}) + s.Require().NoError(err) + s.Equal(time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC), got[0].Interface()) + s.Equal(time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC), got[1].Interface(), + "an offset-less timestamp is read as UTC") + s.Equal(time.Date(2024, 1, 2, 0, 0, 0, 0, time.UTC), got[2].Interface()) + s.Equal(time.Date(0, 1, 1, 3, 4, 5, 0, time.UTC), got[3].Interface()) + s.Equal(26*time.Hour+3*time.Minute+4*time.Second, got[4].Interface()) +} + +func (s *BindingSuite) TestSoleTemporalParamStaysFlat() { + for name, fn := range map[string]any{ + "time.Time": func(when time.Time) error { return nil }, + "*time.Time": func(when *time.Time) error { return nil }, + } { + s.Run(name, func() { + plan := analyze(s, fn) + s.False(plan.loneStruct, "a sole timestamp parameter must bind positionally") + s.Equal(1, plan.numData) + }) + } + + got, err := s.resolve( + func(when time.Time) error { return nil }, + []Arg{LiteralArg{Value: "2024-01-02T03:04:05Z", ValueSchema: temporalSchema("date-time")}}, + &fakeXComClient{}, + ) + s.Require().NoError(err) + s.Equal(time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC), got[0].Interface()) +} + +func (s *BindingSuite) TestResolveNestedTemporalContainers() { + fn := func( + windows []time.Duration, + days []time.Time, + stamps []time.Time, + byName map[string]time.Duration, + ) error { + return nil + } + got, err := s.resolve(fn, []Arg{ + LiteralArg{ + Value: []any{"PT5M", "P1DT2H"}, + ValueSchema: arraySchema(map[string]any{"type": "string", "format": "duration"}), + }, + LiteralArg{ + Value: []any{"2024-01-02"}, + ValueSchema: arraySchema(map[string]any{"type": "string", "format": "date"}), + }, + LiteralArg{ + Value: []any{"2024-01-02T03:04:05Z"}, + ValueSchema: arraySchema(map[string]any{"type": "string", "format": "date-time"}), + }, + LiteralArg{ + Value: map[string]any{"short": "PT30S"}, + ValueSchema: objectSchema(map[string]any{"type": "string", "format": "duration"}), + }, + }, &fakeXComClient{}) + s.Require().NoError(err) + s.Equal([]time.Duration{5 * time.Minute, 26 * time.Hour}, got[0].Interface()) + s.Equal([]time.Time{time.Date(2024, 1, 2, 0, 0, 0, 0, time.UTC)}, got[1].Interface()) + s.Equal([]time.Time{time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC)}, got[2].Interface()) + s.Equal(map[string]time.Duration{"short": 30 * time.Second}, got[3].Interface()) +} + +func (s *BindingSuite) TestResolveNestedTemporalReportsTheBadElement() { + fn := func(windows []time.Duration) error { return nil } + _, err := s.resolve(fn, []Arg{ + LiteralArg{ + Value: []any{"PT5M", "nope"}, + ValueSchema: arraySchema(map[string]any{"type": "string", "format": "duration"}), + }, + }, &fakeXComClient{}) + if s.Assert().Error(err) { + s.Contains(err.Error(), "element 1") + s.Contains(err.Error(), "not an ISO-8601 duration") + } +} + +func (s *BindingSuite) TestResolveNullableTemporalLiteral() { + nullable := anyOfSchema( + map[string]any{"type": "string", "format": "date-time"}, + map[string]any{"type": "null"}, + ) + fn := func(when *time.Time) error { return nil } + + got, err := s.resolve(fn, []Arg{ + LiteralArg{Value: nil, ValueSchema: nullable, FromDefault: true}, + }, &fakeXComClient{}) + s.Require().NoError(err) + s.True(got[0].IsNil(), "a null datetime arrives as a nil pointer") + + got, err = s.resolve(fn, []Arg{ + LiteralArg{Value: "2024-01-02T03:04:05Z", ValueSchema: nullable}, + }, &fakeXComClient{}) + s.Require().NoError(err) + s.Equal(time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC), got[0].Elem().Interface()) +} + +func (s *BindingSuite) TestResolveDurationRejectsNumber() { + fn := func(window time.Duration) error { return nil } + _, err := s.resolve(fn, []Arg{ + LiteralArg{Value: 300, ValueSchema: temporalSchema("duration")}, + }, &fakeXComClient{}) + if s.Assert().Error(err) { + s.Contains(err.Error(), "would be read as nanoseconds") + } +} + +func (s *BindingSuite) TestParseISO8601Duration() { + valid := map[string]time.Duration{ + "PT5M": 5 * time.Minute, + "P1DT2H3M4S": 26*time.Hour + 3*time.Minute + 4*time.Second, + "-PT1M30S": -(time.Minute + 30*time.Second), + "PT1.5S": 1500 * time.Millisecond, + "P2W": 14 * 24 * time.Hour, + "P1D": 24 * time.Hour, + "PT0S": 0, + "+PT10M": 10 * time.Minute, + "P1DT2H": 26 * time.Hour, + "PT0.000001S": time.Microsecond, + } + for text, want := range valid { + s.Run(text, func() { + got, err := parseISO8601Duration(text) + s.Require().NoError(err) + s.Equal(want, got) + }) + } + + invalid := map[string]string{ + "P1M": "no fixed length", + "P1Y": "no fixed length", + "-P2M": "no fixed length", + "300": "not an ISO-8601 duration", + "5m": "not an ISO-8601 duration", + "": "not an ISO-8601 duration", + "P": "carries no components", + "PT": "carries no components", + "PT5X": "not an ISO-8601 duration", + "P1DT2H3": "not an ISO-8601 duration", + } + for text, wantErr := range invalid { + s.Run("invalid/"+text, func() { + _, err := parseISO8601Duration(text) + if s.Assert().Error(err) { + s.Contains(err.Error(), wantErr) + } + }) + } +} + +func (s *BindingSuite) TestParseTemporalRejectsMalformedValues() { + for _, tc := range []struct{ text, format string }{ + {"not-a-date", "date-time"}, + {"2024-13-45", "date"}, + {"99:99:99", "time"}, + {"2024-01-02", "date-time"}, + } { + s.Run(tc.format+"/"+tc.text, func() { + _, err := parseTemporal(tc.text, tc.format) + if s.Assert().Error(err) { + s.Contains(err.Error(), "is not a valid") + } + }) + } +} + func (s *BindingSuite) TestResolveTypedMapParam() { fn := func(labels map[string]string) error { return nil } got, err := s.resolve(fn, []Arg{ @@ -333,15 +528,40 @@ func (s *BindingSuite) TestCheckValueTypeMatrix() { "unknown-type-skips": {argSchema("uuid"), reflect.TypeFor[string](), ""}, "any-target-skips": {argSchema("string"), reflect.TypeFor[any](), ""}, - // Pydantic encodes bytes as a string. - "binary-to-string": { - &genmodels.ArgValueSchema{"type": "string", "format": "binary"}, - reflect.TypeFor[string](), "", + "datetime-to-time": {temporalSchema("date-time"), reflect.TypeFor[time.Time](), ""}, + "datetime-to-time-ptr": {temporalSchema("date-time"), reflect.TypeFor[*time.Time](), ""}, + "date-to-time": {temporalSchema("date"), reflect.TypeFor[time.Time](), ""}, + "time-to-time": {temporalSchema("time"), reflect.TypeFor[time.Time](), ""}, + "duration-to-duration": {temporalSchema("duration"), reflect.TypeFor[time.Duration](), ""}, + "uuid-to-uuid": {temporalSchema("uuid"), reflect.TypeFor[uuid.UUID](), ""}, + "datetime-to-string": {temporalSchema("date-time"), reflect.TypeFor[string](), ""}, + + "integer-vs-time": {argSchema("integer"), reflect.TypeFor[time.Time](), "cannot bind"}, + "plain-string-vs-time": {argSchema("string"), reflect.TypeFor[time.Time](), "cannot bind"}, + "plain-string-vs-uuid": {argSchema("string"), reflect.TypeFor[uuid.UUID](), "cannot bind"}, + "date-time-vs-uuid": { + temporalSchema("date-time"), + reflect.TypeFor[uuid.UUID](), + "cannot bind", }, - "binary-vs-bytes": { - &genmodels.ArgValueSchema{"type": "string", "format": "binary"}, - reflect.TypeFor[[]byte](), "cannot bind", + "date-time-vs-slice": { + temporalSchema("date-time"), + reflect.TypeFor[[]string](), + "cannot bind", }, + "integer-vs-duration": { + argSchema("integer"), + reflect.TypeFor[time.Duration](), + "cannot bind", + }, + "duration-vs-int64": { + temporalSchema("duration"), + reflect.TypeFor[int64](), + "cannot bind", + }, + // Pydantic encodes bytes as a string. + "binary-to-string": {temporalSchema("binary"), reflect.TypeFor[string](), ""}, + "binary-vs-bytes": {temporalSchema("binary"), reflect.TypeFor[[]byte](), "cannot bind"}, // Unions match any branch but preserve nullability. "anyof-matching-branch": { @@ -354,21 +574,21 @@ func (s *BindingSuite) TestCheckValueTypeMatrix() { }, "anyof-nullable-ptr-ok": { anyOfSchema( - map[string]any{"type": "string"}, + map[string]any{"type": "string", "format": "date-time"}, map[string]any{"type": "null"}, ), - reflect.TypeFor[*string](), "", + reflect.TypeFor[*time.Time](), "", }, "anyof-nullable-needs-pointer": { anyOfSchema( - map[string]any{"type": "string"}, + map[string]any{"type": "string", "format": "date-time"}, map[string]any{"type": "null"}, ), - reflect.TypeFor[string](), "must be a pointer", + reflect.TypeFor[time.Time](), "must be a pointer", }, "anyof-nullable-wrong-type": { anyOfSchema( - map[string]any{"type": "string"}, + map[string]any{"type": "string", "format": "date-time"}, map[string]any{"type": "null"}, ), reflect.TypeFor[*int64](), "cannot bind", @@ -380,7 +600,7 @@ func (s *BindingSuite) TestCheckValueTypeMatrix() { } for name, tt := range cases { s.Run(name, func() { - err := checkValueType(tt.schema, tt.target) + _, err := checkValueType(tt.schema, tt.target) if tt.errContains == "" { s.NoError(err) } else if s.Assert().Error(err) { @@ -390,6 +610,29 @@ func (s *BindingSuite) TestCheckValueTypeMatrix() { } } +func (s *BindingSuite) TestCheckValueTypeReturnsMatchedShape() { + shape, err := checkValueType(temporalSchema("date-time"), reflect.TypeFor[time.Time]()) + s.Require().NoError(err) + s.Equal("date-time", shape.format) + + shape, err = checkValueType( + anyOfSchema( + map[string]any{"type": "string", "format": "date-time"}, + map[string]any{"type": "null"}, + ), + reflect.TypeFor[*time.Time](), + ) + s.Require().NoError(err) + s.Equal("date-time", shape.format) + + shape, err = checkValueType( + arraySchema(map[string]any{"type": "string", "format": "duration"}), + reflect.TypeFor[[]time.Duration](), + ) + s.Require().NoError(err) + s.Equal("duration", shape.child("items").format) +} + func (s *BindingSuite) TestResolveTypeMismatchFailsLoudly() { fn := func(count int) error { return nil } _, err := s.resolve( @@ -905,6 +1148,51 @@ func (s *BindingSuite) TestResolveEmbeddedStructFields() { s.Equal(0.75, input.Threshold) } +type windowConfig struct { + Name string `json:"name"` + Window time.Duration `json:"window"` + Start time.Time `json:"start"` +} + +func (s *BindingSuite) TestResolveNativeStructFields() { + schema := genmodels.ArgValueSchema{ + "type": "object", + "properties": map[string]any{ + "window": map[string]any{"type": "string", "format": "duration"}, + "start": map[string]any{"type": "string", "format": "date"}, + }, + } + fn := func(config windowConfig) error { return nil } + got, err := s.resolve(fn, []Arg{ + LiteralArg{ + Name: "config", + Value: map[string]any{ + "name": "nightly", + "window": "PT5M", + "start": "2024-01-02", + }, + ValueSchema: &schema, + }, + }, &fakeXComClient{}) + s.Require().NoError(err) + s.Equal(windowConfig{ + Name: "nightly", + Window: 5 * time.Minute, + Start: time.Date(2024, 1, 2, 0, 0, 0, 0, time.UTC), + }, got[0].Interface()) +} + +func (s *BindingSuite) TestResolveNativeContainerWithoutItemsSchema() { + fn := func(windows []time.Duration, byName map[string]time.Duration) error { return nil } + got, err := s.resolve(fn, []Arg{ + LiteralArg{Name: "windows", Value: []any{"PT5M", "PT30S"}}, + LiteralArg{Name: "by_name", Value: map[string]any{"short": "PT1S"}}, + }, &fakeXComClient{}) + s.Require().NoError(err) + s.Equal([]time.Duration{5 * time.Minute, 30 * time.Second}, got[0].Interface()) + s.Equal(map[string]time.Duration{"short": time.Second}, got[1].Interface()) +} + type money struct { Amount string } @@ -952,3 +1240,181 @@ func (s *BindingSuite) TestResolveEmptyInterfaceDataParam() { s.Require().NoError(err) s.Equal(map[string]any{"k": "v"}, got[0].Interface()) } +func (s *BindingSuite) TestParseISO8601DurationRejectsOverflow() { + _, err := parseISO8601Duration("P200000D") + if s.Assert().Error(err) { + s.Contains(err.Error(), "does not fit in a time.Duration") + } +} + +type nativeCommon struct { + Window time.Duration `json:"window"` +} + +type embeddedNativeConfig struct { + nativeCommon + Name string `json:"name"` +} + +func (s *BindingSuite) TestResolveNativeFieldPromotedFromEmbeddedStruct() { + schema := genmodels.ArgValueSchema{ + "type": "object", + "properties": map[string]any{ + "window": map[string]any{"type": "string", "format": "duration"}, + }, + } + fn := func(config embeddedNativeConfig) error { return nil } + got, err := s.resolve(fn, []Arg{ + LiteralArg{ + Name: "config", + Value: map[string]any{"name": "nightly", "window": "PT5M"}, + ValueSchema: &schema, + }, + }, &fakeXComClient{}) + s.Require().NoError(err) + config := got[0].Interface().(embeddedNativeConfig) + s.Equal("nightly", config.Name) + s.Equal(5*time.Minute, config.Window) +} + +func (s *BindingSuite) TestResolveNativeFixedArray() { + fn := func(windows [2]time.Duration) error { return nil } + got, err := s.resolve(fn, []Arg{ + LiteralArg{ + Name: "windows", + Value: []any{"PT5M", "PT30S"}, + ValueSchema: arraySchema(map[string]any{"type": "string", "format": "duration"}), + }, + }, &fakeXComClient{}) + s.Require().NoError(err) + s.Equal([2]time.Duration{5 * time.Minute, 30 * time.Second}, got[0].Interface()) + + _, err = s.resolve(fn, []Arg{ + LiteralArg{ + Name: "windows", + Value: []any{"PT5M"}, + ValueSchema: arraySchema(map[string]any{"type": "string", "format": "duration"}), + }, + }, &fakeXComClient{}) + if s.Assert().Error(err) { + s.Contains(err.Error(), "expected 2 elements") + } +} + +func (s *BindingSuite) TestParseISO8601DurationRejectsBoundaryOverflow() { + _, err := parseISO8601Duration("PT9223372036.854775808S") + if s.Assert().Error(err) { + s.Contains(err.Error(), "does not fit in a time.Duration") + } +} + +func (s *BindingSuite) TestResolveNestedTypeMismatchFailsLoudly() { + fn := func(windows []time.Duration) error { return nil } + _, err := s.resolve(fn, []Arg{ + LiteralArg{ + Name: "windows", + Value: []any{"PT5M"}, + ValueSchema: arraySchema(map[string]any{"type": "string"}), + }, + }, &fakeXComClient{}) + if s.Assert().Error(err) { + s.Contains(err.Error(), "cannot bind to Go type time.Duration") + } + + fn2 := func(config windowConfig) error { return nil } + _, err = s.resolve(fn2, []Arg{ + LiteralArg{ + Name: "config", + Value: map[string]any{"name": "nightly", "window": "PT5M"}, + ValueSchema: &genmodels.ArgValueSchema{ + "type": "object", + "properties": map[string]any{"window": map[string]any{"type": "string"}}, + }, + }, + }, &fakeXComClient{}) + if s.Assert().Error(err) { + s.Contains(err.Error(), "cannot bind to Go type time.Duration") + } +} + +func (s *BindingSuite) TestResolveNestedNullFailsLoudly() { + fn := func(windows []time.Duration) error { return nil } + _, err := s.resolve(fn, []Arg{ + LiteralArg{ + Name: "windows", + Value: []any{"PT5M", nil}, + ValueSchema: arraySchema(map[string]any{ + "anyOf": []any{ + map[string]any{"type": "string", "format": "duration"}, + map[string]any{"type": "null"}, + }, + }), + }, + }, &fakeXComClient{}) + if s.Assert().Error(err) { + s.Contains(err.Error(), "element 1") + s.Contains(err.Error(), "not nilable") + } + + fnPtr := func(windows []*time.Duration) error { return nil } + got, err := s.resolve(fnPtr, []Arg{ + LiteralArg{ + Name: "windows", + Value: []any{"PT5M", nil}, + ValueSchema: arraySchema(map[string]any{ + "anyOf": []any{ + map[string]any{"type": "string", "format": "duration"}, + map[string]any{"type": "null"}, + }, + }), + }, + }, &fakeXComClient{}) + s.Require().NoError(err) + windows := got[0].Interface().([]*time.Duration) + s.Require().Len(windows, 2) + s.Equal(5*time.Minute, *windows[0]) + s.Nil(windows[1], "a nested None arrives as a nil pointer") +} + +type shallowOuter struct { + inner + Window string `json:"window"` +} + +type inner struct { + Window time.Duration `json:"window"` +} + +func (s *BindingSuite) TestResolveShallowestFieldWinsOverEmbedded() { + fn := func(prefix string, config shallowOuter) error { return nil } + got, err := s.resolve(fn, []Arg{ + LiteralArg{Name: "prefix", Value: "p", ValueSchema: argSchema("string")}, + LiteralArg{ + Name: "config", + Value: map[string]any{"window": "PT5M"}, + ValueSchema: argSchema("object"), + }, + }, &fakeXComClient{}) + s.Require().NoError(err) + config := got[1].Interface().(shallowOuter) + s.Equal("PT5M", config.Window, "the shallow string field takes the value verbatim") + s.Zero(config.inner.Window, "the embedded duration field is not the target") +} + +func (s *BindingSuite) TestResolveNestedNullableTemporalKeepsFormat() { + nullableDate := arraySchema(map[string]any{ + "anyOf": []any{ + map[string]any{"type": "string", "format": "date"}, + map[string]any{"type": "null"}, + }, + }) + fn := func(days []*time.Time) error { return nil } + got, err := s.resolve(fn, []Arg{ + LiteralArg{Name: "days", Value: []any{"2024-01-02", nil}, ValueSchema: nullableDate}, + }, &fakeXComClient{}) + s.Require().NoError(err) + days := got[0].Interface().([]*time.Time) + s.Require().Len(days, 2) + s.Equal(time.Date(2024, 1, 2, 0, 0, 0, 0, time.UTC), *days[0]) + s.Nil(days[1]) +} From 0342c293d9bd4065e4f7502fd3c56dc0d5d32ccf Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Thu, 20 Aug 2026 13:38:03 +0000 Subject: [PATCH 2/2] Fix nested UUID format validation and cache type-tree walks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit containsNativeSeen omitted uuidType, so containers of uuid.UUID (slices, maps, struct fields) skipped format validation entirely — a schema declaring the wrong format on a nested UUID was silently accepted. Add uuidType to the identity check and add an early return in normalizeNative for native types that are self-decoding (uuid.UUID via TextUnmarshaler) so they do not fall through to the container-recursion switch. Cache containsNative and nativeStructFields results in sync.Maps keyed by reflect.Type — both depend only on static type information but were recomputed from scratch on every Resolve call. Restore a test case for anyOf(string, null) → *string that was inadvertently dropped when the nullable matrix cases were changed to exercise temporal types. --- go-sdk/pkg/binding/binding.go | 22 ++++++++++- go-sdk/pkg/binding/binding_test.go | 41 +++++++++++++++++++++ kubernetes-tests/lang_sdk/go_example/go.mod | 1 + 3 files changed, 62 insertions(+), 2 deletions(-) diff --git a/go-sdk/pkg/binding/binding.go b/go-sdk/pkg/binding/binding.go index 011e07e3f9055..f91ae86f97e1c 100644 --- a/go-sdk/pkg/binding/binding.go +++ b/go-sdk/pkg/binding/binding.go @@ -938,6 +938,11 @@ func normalizeNative(raw any, target reflect.Type, shape schemaShape) (any, erro return ts.Format(time.RFC3339Nano), nil } + // Native types handled above or self-decoding (e.g. uuid.UUID via TextUnmarshaler). + if _, isNative := nativeFormats[target]; isNative { + return raw, nil + } + if !containsNative(target) { return raw, nil } @@ -1006,9 +1011,15 @@ func normalizeNative(raw any, target reflect.Type, shape schemaShape) (any, erro return raw, nil } +var nativeStructFieldsCache sync.Map // reflect.Type → map[string]reflect.Type + func nativeStructFields(structType reflect.Type) map[string]reflect.Type { + if v, ok := nativeStructFieldsCache.Load(structType); ok { + return v.(map[string]reflect.Type) + } fields := make(map[string]reflect.Type) collectNativeFields(structType, fields, map[string]bool{}, map[reflect.Type]bool{}) + nativeStructFieldsCache.Store(structType, fields) return fields } @@ -1076,15 +1087,22 @@ func lookupNativeField( return nil, false } +var containsNativeCache sync.Map // reflect.Type → bool + func containsNative(t reflect.Type) bool { - return containsNativeSeen(t, map[reflect.Type]bool{}) + if v, ok := containsNativeCache.Load(t); ok { + return v.(bool) + } + result := containsNativeSeen(t, map[reflect.Type]bool{}) + containsNativeCache.Store(t, result) + return result } func containsNativeSeen(t reflect.Type, seen map[reflect.Type]bool) bool { for t.Kind() == reflect.Pointer { t = t.Elem() } - if t == timeType || t == durationType { + if t == timeType || t == durationType || t == uuidType { return true } if implementsUnmarshaler(t) { diff --git a/go-sdk/pkg/binding/binding_test.go b/go-sdk/pkg/binding/binding_test.go index 6be1ac53830aa..4aed4f713b777 100644 --- a/go-sdk/pkg/binding/binding_test.go +++ b/go-sdk/pkg/binding/binding_test.go @@ -381,6 +381,39 @@ func (s *BindingSuite) TestResolveNestedTemporalReportsTheBadElement() { } } +func (s *BindingSuite) TestResolveNestedUUIDContainers() { + a := uuid.MustParse("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + b := uuid.MustParse("550e8400-e29b-41d4-a716-446655440000") + + fn := func(ids []uuid.UUID, byName map[string]uuid.UUID) error { return nil } + got, err := s.resolve(fn, []Arg{ + LiteralArg{ + Value: []any{a.String(), b.String()}, + ValueSchema: arraySchema(map[string]any{"type": "string", "format": "uuid"}), + }, + LiteralArg{ + Value: map[string]any{"primary": a.String()}, + ValueSchema: objectSchema(map[string]any{"type": "string", "format": "uuid"}), + }, + }, &fakeXComClient{}) + s.Require().NoError(err) + s.Equal([]uuid.UUID{a, b}, got[0].Interface()) + s.Equal(map[string]uuid.UUID{"primary": a}, got[1].Interface()) +} + +func (s *BindingSuite) TestResolveNestedUUIDRejectsFormatMismatch() { + fn := func(ids []uuid.UUID) error { return nil } + _, err := s.resolve(fn, []Arg{ + LiteralArg{ + Value: []any{"550e8400-e29b-41d4-a716-446655440000"}, + ValueSchema: arraySchema(map[string]any{"type": "string", "format": "date-time"}), + }, + }, &fakeXComClient{}) + if s.Assert().Error(err) { + s.Contains(err.Error(), "cannot bind") + } +} + func (s *BindingSuite) TestResolveNullableTemporalLiteral() { nullable := anyOfSchema( map[string]any{"type": "string", "format": "date-time"}, @@ -579,6 +612,13 @@ func (s *BindingSuite) TestCheckValueTypeMatrix() { ), reflect.TypeFor[*time.Time](), "", }, + "anyof-nullable-string-ptr-ok": { + anyOfSchema( + map[string]any{"type": "string"}, + map[string]any{"type": "null"}, + ), + reflect.TypeFor[*string](), "", + }, "anyof-nullable-needs-pointer": { anyOfSchema( map[string]any{"type": "string", "format": "date-time"}, @@ -1240,6 +1280,7 @@ func (s *BindingSuite) TestResolveEmptyInterfaceDataParam() { s.Require().NoError(err) s.Equal(map[string]any{"k": "v"}, got[0].Interface()) } + func (s *BindingSuite) TestParseISO8601DurationRejectsOverflow() { _, err := parseISO8601Duration("P200000D") if s.Assert().Error(err) { diff --git a/kubernetes-tests/lang_sdk/go_example/go.mod b/kubernetes-tests/lang_sdk/go_example/go.mod index b5a88b0b658da..786900ce7721d 100644 --- a/kubernetes-tests/lang_sdk/go_example/go.mod +++ b/kubernetes-tests/lang_sdk/go_example/go.mod @@ -5,6 +5,7 @@ go 1.25.0 require github.com/apache/airflow/go-sdk v0.0.0 require ( + github.com/google/uuid v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/spf13/cobra v1.10.1 // indirect github.com/spf13/pflag v1.0.10 // indirect