henry3260 commented on code in PR #75602: URL: https://github.com/apache/airflow/pull/75602#discussion_r4563281368
########## go-sdk/sdk/conf.go: ########## @@ -0,0 +1,370 @@ +// 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 sdk + +import ( + "bytes" + "encoding" + "encoding/json" + "fmt" + "maps" + "reflect" + "slices" + "strings" + "sync" +) + +// UnmarshalConf reads [DagRun.Conf] into target, which must be a non-nil +// pointer, usually to a struct with one field per conf key: +// +// type etlConf struct { +// TargetTable string +// Limit int +// Since time.Time +// } +// +// etl := etlConf{Limit: 100} +// if err := actx.DagRun().UnmarshalConf(&etl); err != nil { +// return err +// } +// +// For a run triggered with {"target_table": "sales", "since": "2024-01-31T00:00:00Z"}, +// etl.TargetTable is "sales", etl.Since is that date, and etl.Limit stays 100. +// +// Each key goes to the field whose name matches it when case and underscores +// are ignored, so target_table reaches TargetTable. A `json:` tag renames the +// field, and the tag name matches the same way. If one key matches a field +// name exactly, that key wins over keys that match it only once case and +// underscores are ignored. Objects nested in the conf follow the same rules, +// including objects in slices and objects that are map values. The keys of a +// map field stay as written, because they are data rather than field names. +// +// Keys that match no field are ignored, because other tasks in the run may +// read them. A field with no matching key keeps the value target already +// holds, so set the defaults on target before the call, as Limit does above. +// The conf carries only the values the run was triggered with: it does not +// carry the defaults of the Dag's params, so in a scheduled run every field +// keeps the value it came in with. +// +// The `arg:` tag that binds a TaskFlow argument has no effect here. Read a key +// that no field name can match, such as target-table, from [DagRun.Conf], or +// name it in a `json:` tag. +// +// A number decoded into an any, such as a field of type any or a value of a +// map[string]any or []any field, arrives as a [json.Number] rather than as the +// number type [DagRun.Conf] carries, because a float64 cannot hold every +// integer a conf can. Read it with its Int64 or Float64 method, or give the +// field a Go integer or float type to have the decoder convert it. Of those, +// only an integer type keeps every digit of a large integer. +// +// It returns an error when a value does not fit the type of its field, when +// two keys match one field only once case and underscores are ignored, or when +// two fields of a struct differ only in case or underscores. +func (d DagRun) UnmarshalConf(target any) error { + rv := reflect.ValueOf(target) + if rv.Kind() != reflect.Pointer || rv.IsNil() { + return fmt.Errorf( + "sdk.DagRun.UnmarshalConf: target must be a non-nil pointer, got %T", + target, + ) + } + matched, err := matchConfValue(d.Conf, rv.Type().Elem()) + if err != nil { + return err + } + blob, err := json.Marshal(matched) + if err != nil { + return fmt.Errorf("sdk.DagRun.UnmarshalConf: cannot encode the Dag run conf: %w", err) + } + dec := json.NewDecoder(bytes.NewReader(blob)) + // A number that reaches an any-typed destination keeps its literal, so the + // task reads an exact integer instead of the float64 encoding/json would + // otherwise decode it into. + dec.UseNumber() + if err := dec.Decode(target); err != nil { + return fmt.Errorf( + "sdk.DagRun.UnmarshalConf: cannot read the Dag run conf into %s: %w", + rv.Type().Elem(), + err, + ) + } + return nil +} + +// matchConfValue renames the keys of every conf object that becomes a struct +// when the conf is decoded into target. encoding/json matches a key to a field +// only when the two differ at most in case, so renaming the keys to the exact +// field names is what lets target_table reach TargetTable. +// +// The walk follows target rather than the conf, so it reaches objects nested +// in slices, arrays and map values, and stops at a type that decodes itself. +func matchConfValue(value any, target reflect.Type) (any, error) { + for target.Kind() == reflect.Pointer { + target = target.Elem() + } + // A type such as time.Time decodes from a value of its own shape, and its + // Go fields are not conf keys. + if implementsConfUnmarshaler(target) { + return value, nil + } + + switch target.Kind() { + case reflect.Struct: + if obj, ok := value.(map[string]any); ok { + return matchConfFields(obj, target) + } + case reflect.Map: + if obj, ok := value.(map[string]any); ok { + matched := make(map[string]any, len(obj)) + for key, elem := range obj { + elem, err := matchConfValue(elem, target.Elem()) + if err != nil { + return nil, err + } + matched[key] = elem + } + return matched, nil + } + case reflect.Slice, reflect.Array: + if list, ok := value.([]any); ok { + matched := make([]any, len(list)) + for i, elem := range list { + elem, err := matchConfValue(elem, target.Elem()) + if err != nil { + return nil, err + } + matched[i] = elem + } + return matched, nil + } + } + // A value whose shape does not fit the field, such as a string for a + // struct, passes through so that the decoder reports the mismatch. + return value, nil +} + +// matchConfFields renames each key of obj to the name of the target field the +// key matches, and drops a key that matches no field. Keys are visited in +// sorted order so that an error names the same pair of keys on every run. +func matchConfFields(obj map[string]any, target reflect.Type) (map[string]any, error) { Review Comment: You're right, and it is a real limitation rather than something I can design away. `encoding/json` matches a key to a field only when the two differ at most in case, so `target_table` can never reach `TargetTable` on its own. The only way to get there is to rename the key to the field's exact name before decoding — and to rename it I need the field set, which means re-deriving the rules for what that set is. The drift is real and I have hit two concrete cases, both of which match base and plain `encoding/json`, so neither is a regression in this PR: - an unexported embedded pointer the SDK skips still receives the key the SDK matched to a field it shadows; - a tagged field inside such a pointer beats an untagged one at the same depth. I have added both to #76902, which tracks unifying this with `binding.go`. -- 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]
