jrmccluskey commented on code in PR #39595:
URL: https://github.com/apache/beam/pull/39595#discussion_r3715493493
##########
sdks/go/container/tools/pipeline_options.go:
##########
@@ -42,3 +46,181 @@ 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, error) {
+ if opt == nil {
+ return &PipelineOptions{options: make(map[string]any),
experiments: make(map[string]string)}, nil
+ }
+ 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 {
+ if expsMap, err := parseExperiments(exps); err == nil {
+ po.experiments = expsMap
+ }
+ }
+ return po, nil
Review Comment:
This is the only place in the function where the code potentially receiving
and handling an error, but it does wind up swallowing it. If the error here is
relevant it should be surfaced, otherwise the function doesn't need to return
an error at all.
##########
sdks/go/container/tools/pipeline_options.go:
##########
@@ -42,3 +46,181 @@ 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, error) {
+ if opt == nil {
+ return &PipelineOptions{options: make(map[string]any),
experiments: make(map[string]string)}, nil
+ }
+ 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 {
+ if expsMap, err := parseExperiments(exps); err == nil {
+ po.experiments = expsMap
+ }
+ }
+ return po, nil
+}
+
+func parseExperiments(slice []string) (map[string]string, error) {
Review Comment:
Same thing here, there's never an error case to return. Is there an
interface you're programming towards here?
##########
sdks/go/container/tools/pipeline_options_test.go:
##########
@@ -56,3 +78,217 @@ func TestMakePipelineOptionsFileAndEnvVar(t *testing.T) {
}
os.Remove("pipeline_options.json")
}
+
+func TestPipelineOptions(t *testing.T) {
Review Comment:
In general the benefit to using table-driven test structures is to avoid
re-writing similar/the same testing logic repeatedly, making the test
definition here take a complex `validate` function makes the test much harder
to follow.
--
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]
