tvalentyn commented on code in PR #39595:
URL: https://github.com/apache/beam/pull/39595#discussion_r3728784782
##########
sdks/go/container/tools/pipeline_options.go:
##########
@@ -42,3 +46,179 @@ func MakePipelineOptionsFileAndEnvVar(options string) error
{
os.Setenv("PIPELINE_OPTIONS_FILE", f.Name())
return nil
}
+
+// PipelineOptions represents parsed pipeline options as a normalized map.
+type PipelineOptions struct {
+ options map[string]any
+ experiments map[string]string
+}
+
+// ParseOptionsFromProto creates normalized PipelineOptions directly from a
protobuf Struct.
+func ParseOptionsFromProto(opt *structpb.Struct, sdkNamespace string)
*PipelineOptions {
+ if opt == nil {
+ return &PipelineOptions{options: make(map[string]any),
experiments: make(map[string]string)}
+ }
+ raw := opt.AsMap()
+ flat := make(map[string]any)
+
+ // 1. Extract nested options if present (Dataflow runner uses this
structure)
+ if optsVal, ok := raw["options"]; ok {
+ if optsMap, ok := optsVal.(map[string]any); ok {
+ for k, v := range optsMap {
+ flat[k] = v
+ }
+ }
+ }
+
+ // 2. Extract standard URN keys (Portable runners use this structure)
+ for k, v := range raw {
+ if k == "options" || k == "display_data" {
+ continue
+ }
+ if strings.HasPrefix(k, "beam:option:") && strings.HasSuffix(k,
":v1") {
+ name := strings.TrimPrefix(k, "beam:option:")
+ name = strings.TrimSuffix(name, ":v1")
+ flat[name] = v
+ }
+ }
+
+ // 3. Promote specified SDK namespace options (Highest precedence, may
overwrite earlier entries).
+ // Beam Go SDK uses this structure.
+ if sdkNamespace != "" {
+ sdkURN := fmt.Sprintf("beam:option:%s:v1", sdkNamespace)
+ if sdkVal, ok := raw[sdkURN]; ok {
+ if urnMap, ok := sdkVal.(map[string]any); ok {
+ if nestedOpts, ok :=
urnMap["options"].(map[string]any); ok {
+ for nk, nv := range nestedOpts {
+ flat[nk] = nv
+ }
+ }
+ }
+ }
+ }
+
+ po := &PipelineOptions{
+ options: flat,
+ experiments: make(map[string]string),
+ }
+ if exps, err := po.GetStringSlice("experiments"); err == nil {
+ po.experiments = parseExperiments(exps)
+ }
+ return po
+}
+
+func parseExperiments(slice []string) map[string]string {
+ res := make(map[string]string)
+ for _, item := range slice {
+ if strings.Contains(item, "=") {
+ parts := strings.SplitN(item, "=", 2)
+ res[parts[0]] = parts[1]
+ } else {
+ res[item] = ""
+ }
+ }
+ return res
+}
+
+// HasOption returns true if the option is defined and not nil.
+func (po *PipelineOptions) HasOption(name string) bool {
+ val, ok := po.options[name]
+ return ok && val != nil
+}
+
+// GetString returns the value of an option as a string.
+func (po *PipelineOptions) GetString(name string) (string, error) {
+ val, ok := po.options[name]
+ if !ok || val == nil {
+ return "", fmt.Errorf("option %q not defined", name)
+ }
+ if str, ok := val.(string); ok {
+ return str, nil
+ }
+ return "", fmt.Errorf("option %q: expected string, got type %T", name,
val)
+}
+
+// GetStringSlice returns the value of an option as a string slice.
+func (po *PipelineOptions) GetStringSlice(name string) ([]string, error) {
+ val, ok := po.options[name]
+ if !ok || val == nil {
+ return nil, fmt.Errorf("option %q not defined", name)
+ }
+ if slice, ok := val.([]any); ok {
+ var res []string
+ for _, item := range slice {
+ if str, ok := item.(string); ok {
+ res = append(res, str)
+ } else {
+ return nil, fmt.Errorf("option %q: expected
string slice element, got type %T", name, item)
+ }
+ }
+ return res, nil
+ }
+ if str, ok := val.(string); ok {
+ // Go SDK models multi-value list flags (like experiments or
dataflow_service_options)
+ // as comma-separated string flags rather than JSON arrays.
+ if str == "" {
+ return nil, nil
+ }
+ return strings.Split(str, ","), nil
+ }
+ return nil, fmt.Errorf("option %q: expected string slice, got type %T",
name, val)
+}
+
+// GetInt returns the value of an option as an integer.
+func (po *PipelineOptions) GetInt(name string) (int, error) {
Review Comment:
found one such option.
--
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]
