sdks/go/test/regression/coders/fromyaml/fromyaml.go - beam - Git at Google

 // 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.

 // fromyaml generates a resource file from the standard_coders.yaml
 // file for use in these coder regression tests.
 //
 // It expects to be run in it's test directory, or via it's go test.
 package main

 import (
 	"bytes"
 	"fmt"
 	"log"
 	"math"
 	"os"
 	"reflect"
 	"runtime/debug"
 	"strconv"
 	"strings"

 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
 	pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
 	"github.com/google/go-cmp/cmp"
 	"golang.org/x/text/encoding/charmap"
 	yaml "gopkg.in/yaml.v2"
 )

 var unimplementedCoders = map[string]bool{
 	"beam:coder:param_windowed_value:v1": true,
 	"beam:coder:sharded_key:v1":          true,
 	"beam:coder:custom_window:v1":        true,
 }

 var filteredCases = []struct{ filter, reason string }{
 	{"logical", "BEAM-9615: Support logical types"},
 	{"30ea5a25-dcd8-4cdb-abeb-5332d15ab4b9", "https://github.com/apache/beam/issues/21206: Support encoding position."},
 	{"80be749a-5700-4ede-89d8-dd9a4433a3f8", "https://github.com/apache/beam/issues/19817: Support millis_instant."},
 	{"800c44ae-a1b7-4def-bbf6-6217cca89ec4", "https://github.com/apache/beam/issues/19817: Support decimal."},
 	{"f0ffb3a4-f46f-41ca-a942-85e3e939452a", "https://github.com/apache/beam/issues/23526: Support char/varchar, binary/varbinary."},
 }

 // Coder is a representation a serialized beam coder.
 type Coder struct {
 	Urn              string  `yaml:"urn,omitempty"`
 	Payload          string  `yaml:"payload,omitempty"`
 	Components       []Coder `yaml:"components,omitempty"`
 	NonDeterministic bool    `yaml:"non_deterministic,omitempty"`
 }

 type logger interface {
 	Errorf(string, ...any)
 	Logf(string, ...any)
 }

 // Spec is a set of conditions that a coder must pass.
 type Spec struct {
 	Coder    Coder         `yaml:"coder,omitempty"`
 	Nested   *bool         `yaml:"nested,omitempty"`
 	Examples yaml.MapSlice `yaml:"examples,omitempty"`
 	Log      logger

 	id       int // for generating coder ids.
 	coderPBs map[string]*pipepb.Coder
 }

 func (s *Spec) nextID() string {
 	ret := fmt.Sprintf("%d", s.id)
 	s.id++
 	return ret
 }

 func (s *Spec) testStandardCoder() (err error) {
 	if unimplementedCoders[s.Coder.Urn] {
 		log.Printf("skipping unimplemented coder urn: %v", s.Coder.Urn)
 		return nil
 	}
 	if s.Coder.Urn == "beam:coder:state_backed_iterable:v1" {
 		log.Printf("skipping unimplemented test coverage for beam:coder:state_backed_iterable:v1. https://github.com/apache/beam/issues/21324")
 		return nil
 	}
 	for _, c := range filteredCases {
 		if strings.Contains(s.Coder.Payload, c.filter) {
 			log.Printf("skipping coder case. Unsupported in the Go SDK for now: %v Payload: %v", c.reason, s.Coder.Payload)
 			return nil
 		}
 	}

 	// Construct the coder proto equivalents.

 	// Only nested tests need to be run, since nestedness is a pre-portability
 	// concept.
 	// For legacy Java reasons, the row coder examples are all marked nested: false
 	// so we need to check that before skipping unnested tests.
 	if s.Coder.Urn != "beam:coder:row:v1" && s.Nested != nil && !*s.Nested {
 		log.Printf("skipping unnested coder spec: %v\n", s.Coder)
 		return nil
 	}

 	s.coderPBs = make(map[string]*pipepb.Coder)
 	id := s.parseCoder(s.Coder)
 	b := graphx.NewCoderUnmarshaller(s.coderPBs)
 	underTest, err := b.Coder(id)
 	if err != nil {
 		return fmt.Errorf("unable to create coder: %v", err)
 	}

 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panicked on coder %v || %v:\n\t%v :\n%s", underTest, s.Coder, e, debug.Stack())
 		}
 	}()

 	var decFails, encFails int
 	for _, eg := range s.Examples {

 		// Test Decoding
 		// Ideally we'd use the beam package coders, but KVs make that complicated.
 		// This can be cleaned up once a type parametered beam.KV type exists.
 		dec := exec.MakeElementDecoder(underTest)
 		encoded := eg.Key.(string)
 		var elem exec.FullValue

 		// What I would have expected.
 		//		r := charmap.ISO8859_1.NewDecoder().Reader(strings.NewReader(encoded))
 		recoded, err := charmap.ISO8859_1.NewEncoder().String(encoded)
 		if err != nil {
 			return err
 		}
 		r := strings.NewReader(recoded)
 		if err := dec.DecodeTo(r, &elem); err != nil {
 			return fmt.Errorf("err decoding %q: %v", encoded, err)
 		}
 		if !diff(s.Coder, &elem, eg) {
 			decFails++
 			continue
 		}

 		// Test Encoding
 		if s.Coder.NonDeterministic {
 			// Skip verifying nondeterministic encodings.
 			continue
 		}
 		enc := exec.MakeElementEncoder(underTest)
 		var out bytes.Buffer
 		if err := enc.Encode(&elem, &out); err != nil {
 			return err
 		}
 		if d := cmp.Diff(recoded, string(out.Bytes())); d != "" {
 			log.Printf("Encoding error: diff(-want,+got): %v\n", d)
 		}
 	}
 	if decFails+encFails > 0 {
 		return fmt.Errorf("failed to decode %v times, and encode %v times", decFails, encFails)
 	}

 	return nil
 }

 var cmpOpts = []cmp.Option{
 	cmp.Transformer("bytes2string", func(in []byte) (out string) {
 		return string(in)
 	}),
 }

 func diff(c Coder, elem *exec.FullValue, eg yaml.MapItem) bool {
 	var got, want any
 	switch c.Urn {
 	case "beam:coder:bytes:v1":
 		got = string(elem.Elm.([]byte))
 		switch egv := eg.Value.(type) {
 		case string:
 			want = egv
 		case []byte:
 			want = string(egv)
 		}
 	case "beam:coder:varint:v1":
 		got, want = elem.Elm.(int64), int64(eg.Value.(int))
 	case "beam:coder:double:v1":
 		got = elem.Elm.(float64)
 		switch v := eg.Value.(string); v {
 		case "NaN":
 			// Do the NaN comparison here since NaN by definition != NaN.
 			if math.IsNaN(got.(float64)) {
 				want, got = 1, 1
 			} else {
 				want = math.NaN()
 			}
 		case "-Infinity":
 			want = math.Inf(-1)
 		case "Infinity":
 			want = math.Inf(1)
 		default:
 			want, _ = strconv.ParseFloat(v, 64)
 		}

 	case "beam:coder:kv:v1":
 		v := eg.Value.(yaml.MapSlice)
 		pass := true
 		if !diff(c.Components[0], &exec.FullValue{Elm: elem.Elm}, v[0]) {
 			pass = false
 		}
 		if !diff(c.Components[1], &exec.FullValue{Elm: elem.Elm2}, v[1]) {
 			pass = false
 		}
 		return pass

 	case "beam:coder:nullable:v1":
 		if elem.Elm == nil || eg.Value == nil {
 			got, want = elem.Elm, eg.Value
 		} else {
 			got = string(elem.Elm.([]byte))
 			switch egv := eg.Value.(type) {
 			case string:
 				want = egv
 			case []byte:
 				want = string(egv)
 			}
 		}

 	case "beam:coder:iterable:v1":
 		pass := true
 		gotrv := reflect.ValueOf(elem.Elm)
 		wantrv := reflect.ValueOf(eg.Value)
 		if gotrv.Len() != wantrv.Len() {
 			log.Printf("Lengths don't match. got %v, want %v;  %v, %v", gotrv.Len(), wantrv.Len(), gotrv, wantrv)
 			return false
 		}
 		for i := 0; i < wantrv.Len(); i++ {
 			if !diff(c.Components[0],
 				&exec.FullValue{Elm: gotrv.Index(i).Interface()},
 				yaml.MapItem{Value: wantrv.Index(i).Interface()}) {
 				pass = false
 			}

 		}
 		return pass
 	case "beam:coder:interval_window:v1":
 		var a, b int
 		val := eg.Value
 		if is, ok := eg.Value.([]any); ok {
 			val = is[0]
 		}
 		v := val.(yaml.MapSlice)

 		a = v[0].Value.(int)
 		b = v[1].Value.(int)
 		end := mtime.FromMilliseconds(int64(a))
 		start := end - mtime.Time(int64(b))
 		want = window.IntervalWindow{Start: start, End: end}
 		// If this is nested in an iterable, windows won't be populated.
 		if len(elem.Windows) == 0 {
 			got = elem.Elm
 		} else {
 			got = elem.Windows[0]
 		}

 	case "beam:coder:global_window:v1":
 		want = window.GlobalWindow{}
 		// If this is nested in an iterable, windows won't be populated.
 		if len(elem.Windows) == 0 {
 			got = window.GlobalWindow(elem.Elm.(struct{}))
 		} else {
 			got = elem.Windows[0]
 		}
 	case "beam:coder:windowed_value:v1", "beam:coder:param_windowed_value:v1":
 		// elem contains all the information, but we need to compare the element+timestamp
 		// separately from the windows, to avoid repeated expected value parsing logic.
 		pass := true
 		vs := eg.Value.(yaml.MapSlice)
 		if !diff(c.Components[0], elem, vs[0]) {
 			pass = false
 		}
 		if d := cmp.Diff(
 			mtime.FromMilliseconds(int64(vs[1].Value.(int))),
 			elem.Timestamp, cmpOpts...); d != "" {

 			pass = false
 		}
 		if !diff(c.Components[1], elem, vs[3]) {
 			pass = false
 		}
 		if !diffPane(vs[2].Value, elem.Pane) {
 			pass = false
 		}
 		return pass
 	case "beam:coder:row:v1":
 		fs := eg.Value.(yaml.MapSlice)
 		var rfs []reflect.StructField
 		// There are only 2 pointer examples, but they reuse field names,
 		// so we key off the proto hash to know which example we're handling.
 		ptrEg := strings.Contains(c.Payload, "51ace21c7393")
 		for _, rf := range fs {
 			name := rf.Key.(string)
 			t := nameToType[name]
 			if ptrEg {
 				t = reflect.PtrTo(t)
 			}
 			rfs = append(rfs, reflect.StructField{
 				Name: strings.ToUpper(name[:1]) + name[1:],
 				Type: t,
 				Tag:  reflect.StructTag(fmt.Sprintf("beam:\"%v\"", name)),
 			})
 		}
 		rv := reflect.New(reflect.StructOf(rfs)).Elem()
 		for i, rf := range fs {
 			setField(rv, i, rf.Value)
 		}

 		got, want = elem.Elm, rv.Interface()
 	case "beam:coder:timer:v1":
 		pass := true
 		tm := elem.Elm.(exec.TimerRecv)
 		fs := eg.Value.(yaml.MapSlice)
 		for _, item := range fs {

 			switch item.Key.(string) {
 			case "userKey":
 				if want := item.Value.(string); want != tm.Key.Elm.(string) {
 					pass = false
 				}
 			case "dynamicTimerTag":
 				if want := item.Value.(string); want != tm.Tag {
 					pass = false
 				}
 			case "windows":
 				if v, ok := item.Value.([]any); ok {
 					for i, val := range v {
 						if val.(string) == "global" && fmt.Sprintf("%s", tm.Windows[i]) == "[*]" {
 							continue
 						} else if val.(string) != fmt.Sprintf("%s", tm.Windows[i]) {
 							pass = false
 						}
 					}
 				}
 			case "clearBit":
 				if want := item.Value.(bool); want != tm.Clear {
 					pass = false
 				}
 			case "fireTimestamp":
 				if want := item.Value.(int); want != int(tm.FireTimestamp) {
 					pass = false
 				}
 			case "holdTimestamp":
 				if want := item.Value.(int); want != int(tm.HoldTimestamp) {
 					pass = false
 				}
 			case "pane":
 				pass = diffPane(item.Value, tm.Pane)
 			}
 		}
 		return pass
 	default:
 		got, want = elem.Elm, eg.Value
 	}
 	if d := cmp.Diff(want, got, cmpOpts...); d != "" {
 		log.Printf("Decoding error: diff(-want,+got): %v\n", d)
 		return false
 	}
 	return true
 }

 func diffPane(eg any, got typex.PaneInfo) bool {
 	pass := true
 	paneTiming := map[typex.PaneTiming]string{
 		typex.PaneUnknown: "UNKNOWN",
 		typex.PaneEarly:   "EARLY",
 		typex.PaneLate:    "LATE",
 		typex.PaneOnTime:  "ONTIME",
 	}
 	for _, item := range eg.(yaml.MapSlice) {
 		switch item.Key.(string) {
 		case "is_first":
 			if want := item.Value.(bool); want != got.IsFirst {
 				pass = false
 			}
 		case "is_last":
 			if want := item.Value.(bool); want != got.IsLast {
 				pass = false
 			}
 		case "timing":
 			if want := item.Value.(string); want != paneTiming[got.Timing] {
 				pass = false
 			}
 		case "index":
 			if want := item.Value.(int); want != int(got.Index) {
 				pass = false
 			}
 		case "on_time_index":
 			if want := item.Value.(int); want != int(got.NonSpeculativeIndex) {
 				pass = false
 			}
 		}
 	}
 	return pass
 }

 // standard_coders.yaml uses the name for type indication, except for nullability.
 var nameToType = map[string]reflect.Type{
 	"str":     reflectx.String,
 	"i32":     reflectx.Int32,
 	"f64":     reflectx.Float64,
 	"arr":     reflect.SliceOf(reflectx.String),
 	"f_bool":  reflectx.Bool,
 	"f_bytes": reflect.PtrTo(reflectx.ByteSlice),
 	"f_map":   reflect.MapOf(reflectx.String, reflect.PtrTo(reflectx.Int64)),
 	"f_float": reflectx.Float32,
 }

 func setField(rv reflect.Value, i int, v any) {
 	if v == nil {
 		return
 	}
 	rf := rv.Field(i)
 	if rf.Kind() == reflect.Ptr {
 		// Ensure it's initialized.
 		rf.Set(reflect.New(rf.Type().Elem()))
 		rf = rf.Elem()
 	}
 	switch rf.Kind() {
 	case reflect.String:
 		rf.SetString(v.(string))
 	case reflect.Int32:
 		rf.SetInt(int64(v.(int)))
 	case reflect.Float32:
 		c, err := strconv.ParseFloat(v.(string), 32)
 		if err != nil {
 			panic(err)
 		}
 		rf.SetFloat(c)
 	case reflect.Float64:
 		c, err := strconv.ParseFloat(v.(string), 64)
 		if err != nil {
 			panic(err)
 		}
 		rf.SetFloat(c)
 	case reflect.Slice:
 		if rf.Type() == reflectx.ByteSlice {
 			rf.Set(reflect.ValueOf([]byte(v.(string))))
 			break
 		}
 		// Value is a []any with string values.
 		var arr []string
 		for _, a := range v.([]any) {
 			arr = append(arr, a.(string))
 		}
 		rf.Set(reflect.ValueOf(arr))
 	case reflect.Bool:
 		rf.SetBool(v.(bool))
 	case reflect.Map:
 		// only f_map presently, which is always map[string]*int64
 		rm := reflect.MakeMap(rf.Type())
 		for _, a := range v.(yaml.MapSlice) {
 			rk := reflect.ValueOf(a.Key.(string))
 			rv := reflect.Zero(rf.Type().Elem())
 			if a.Value != nil {
 				rv = reflect.New(reflectx.Int64)
 				rv.Elem().SetInt(int64(a.Value.(int)))
 			}
 			rm.SetMapIndex(rk, rv)
 		}
 		rf.Set(rm)

 	}
 }

 func (s *Spec) parseCoder(c Coder) string {
 	id := s.nextID()
 	var compIDs []string
 	for _, comp := range c.Components {
 		compIDs = append(compIDs, s.parseCoder(comp))
 	}
 	s.coderPBs[id] = &pipepb.Coder{
 		Spec: &pipepb.FunctionSpec{
 			Urn:     c.Urn,
 			Payload: []byte(c.Payload),
 		},
 		ComponentCoderIds: compIDs,
 	}
 	return id
 }

 // Simple logger to run as main program.
 type logLogger struct{}

 func (*logLogger) Errorf(format string, v ...any) {
 	log.Printf(format, v...)
 }

 func (*logLogger) Logf(format string, v ...any) {
 	log.Printf(format, v...)
 }

 const yamlPath = "../../../../../../model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml"

 func main() {
 	data, err := os.ReadFile(yamlPath)
 	if err != nil {
 		log.Fatalf("Couldn't read %v: %v", yamlPath, err)
 	}
 	specs := bytes.Split(data, []byte("\n---\n"))
 	var failures bool
 	var l logLogger
 	for _, data := range specs {
 		cs := Spec{Log: &l}
 		if err := yaml.Unmarshal(data, &cs); err != nil {
 			failures = true
 			l.Logf("unable to parse yaml: %v %q", err, data)
 			continue
 		}
 		if err := cs.testStandardCoder(); err != nil {
 			failures = true
 			l.Logf("Failed \"%v\": %v", cs.Coder, err)
 		}
 	}
 	if !failures {
 		log.Println("PASS")
 	}
 }
	// 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.

	// fromyaml generates a resource file from the standard_coders.yaml
	// file for use in these coder regression tests.
	//
	// It expects to be run in it's test directory, or via it's go test.
	package main

	import (
	"bytes"
	"fmt"
	"log"
	"math"
	"os"
	"reflect"
	"runtime/debug"
	"strconv"
	"strings"

	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
	pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
	"github.com/google/go-cmp/cmp"
	"golang.org/x/text/encoding/charmap"
	yaml "gopkg.in/yaml.v2"
	)

	var unimplementedCoders = map[string]bool{
	"beam:coder:param_windowed_value:v1": true,
	"beam:coder:sharded_key:v1": true,
	"beam:coder:custom_window:v1": true,
	}

	var filteredCases = []struct{ filter, reason string }{
	{"logical", "BEAM-9615: Support logical types"},
	{"30ea5a25-dcd8-4cdb-abeb-5332d15ab4b9", "https://github.com/apache/beam/issues/21206: Support encoding position."},
	{"80be749a-5700-4ede-89d8-dd9a4433a3f8", "https://github.com/apache/beam/issues/19817: Support millis_instant."},
	{"800c44ae-a1b7-4def-bbf6-6217cca89ec4", "https://github.com/apache/beam/issues/19817: Support decimal."},
	{"f0ffb3a4-f46f-41ca-a942-85e3e939452a", "https://github.com/apache/beam/issues/23526: Support char/varchar, binary/varbinary."},
	}

	// Coder is a representation a serialized beam coder.
	type Coder struct {
	Urn string `yaml:"urn,omitempty"`
	Payload string `yaml:"payload,omitempty"`
	Components []Coder `yaml:"components,omitempty"`
	NonDeterministic bool `yaml:"non_deterministic,omitempty"`
	}

	type logger interface {
	Errorf(string, ...any)
	Logf(string, ...any)
	}

	// Spec is a set of conditions that a coder must pass.
	type Spec struct {
	Coder Coder `yaml:"coder,omitempty"`
	Nested *bool `yaml:"nested,omitempty"`
	Examples yaml.MapSlice `yaml:"examples,omitempty"`
	Log logger

	id int // for generating coder ids.
	coderPBs map[string]*pipepb.Coder
	}

	func (s *Spec) nextID() string {
	ret := fmt.Sprintf("%d", s.id)
	s.id++
	return ret
	}

	func (s *Spec) testStandardCoder() (err error) {
	if unimplementedCoders[s.Coder.Urn] {
	log.Printf("skipping unimplemented coder urn: %v", s.Coder.Urn)
	return nil
	}
	if s.Coder.Urn == "beam:coder:state_backed_iterable:v1" {
	log.Printf("skipping unimplemented test coverage for beam:coder:state_backed_iterable:v1. https://github.com/apache/beam/issues/21324")
	return nil
	}
	for _, c := range filteredCases {
	if strings.Contains(s.Coder.Payload, c.filter) {
	log.Printf("skipping coder case. Unsupported in the Go SDK for now: %v Payload: %v", c.reason, s.Coder.Payload)
	return nil
	}
	}

	// Construct the coder proto equivalents.

	// Only nested tests need to be run, since nestedness is a pre-portability
	// concept.
	// For legacy Java reasons, the row coder examples are all marked nested: false
	// so we need to check that before skipping unnested tests.
	if s.Coder.Urn != "beam:coder:row:v1" && s.Nested != nil && !*s.Nested {
	log.Printf("skipping unnested coder spec: %v\n", s.Coder)
	return nil
	}

	s.coderPBs = make(map[string]*pipepb.Coder)
	id := s.parseCoder(s.Coder)
	b := graphx.NewCoderUnmarshaller(s.coderPBs)
	underTest, err := b.Coder(id)
	if err != nil {
	return fmt.Errorf("unable to create coder: %v", err)
	}

	defer func() {
	if e := recover(); e != nil {
	err = fmt.Errorf("panicked on coder %v \|\| %v:\n\t%v :\n%s", underTest, s.Coder, e, debug.Stack())
	}
	}()

	var decFails, encFails int
	for _, eg := range s.Examples {

	// Test Decoding
	// Ideally we'd use the beam package coders, but KVs make that complicated.
	// This can be cleaned up once a type parametered beam.KV type exists.
	dec := exec.MakeElementDecoder(underTest)
	encoded := eg.Key.(string)
	var elem exec.FullValue

	// What I would have expected.
	// r := charmap.ISO8859_1.NewDecoder().Reader(strings.NewReader(encoded))
	recoded, err := charmap.ISO8859_1.NewEncoder().String(encoded)
	if err != nil {
	return err
	}
	r := strings.NewReader(recoded)
	if err := dec.DecodeTo(r, &elem); err != nil {
	return fmt.Errorf("err decoding %q: %v", encoded, err)
	}
	if !diff(s.Coder, &elem, eg) {
	decFails++
	continue
	}

	// Test Encoding
	if s.Coder.NonDeterministic {
	// Skip verifying nondeterministic encodings.
	continue
	}
	enc := exec.MakeElementEncoder(underTest)
	var out bytes.Buffer
	if err := enc.Encode(&elem, &out); err != nil {
	return err
	}
	if d := cmp.Diff(recoded, string(out.Bytes())); d != "" {
	log.Printf("Encoding error: diff(-want,+got): %v\n", d)
	}
	}
	if decFails+encFails > 0 {
	return fmt.Errorf("failed to decode %v times, and encode %v times", decFails, encFails)
	}

	return nil
	}

	var cmpOpts = []cmp.Option{
	cmp.Transformer("bytes2string", func(in []byte) (out string) {
	return string(in)
	}),
	}

	func diff(c Coder, elem *exec.FullValue, eg yaml.MapItem) bool {
	var got, want any
	switch c.Urn {
	case "beam:coder:bytes:v1":
	got = string(elem.Elm.([]byte))
	switch egv := eg.Value.(type) {
	case string:
	want = egv
	case []byte:
	want = string(egv)
	}
	case "beam:coder:varint:v1":
	got, want = elem.Elm.(int64), int64(eg.Value.(int))
	case "beam:coder:double:v1":
	got = elem.Elm.(float64)
	switch v := eg.Value.(string); v {
	case "NaN":
	// Do the NaN comparison here since NaN by definition != NaN.
	if math.IsNaN(got.(float64)) {
	want, got = 1, 1
	} else {
	want = math.NaN()
	}
	case "-Infinity":
	want = math.Inf(-1)
	case "Infinity":
	want = math.Inf(1)
	default:
	want, _ = strconv.ParseFloat(v, 64)
	}

	case "beam:coder:kv:v1":
	v := eg.Value.(yaml.MapSlice)
	pass := true
	if !diff(c.Components[0], &exec.FullValue{Elm: elem.Elm}, v[0]) {
	pass = false
	}
	if !diff(c.Components[1], &exec.FullValue{Elm: elem.Elm2}, v[1]) {
	pass = false
	}
	return pass

	case "beam:coder:nullable:v1":
	if elem.Elm == nil \|\| eg.Value == nil {
	got, want = elem.Elm, eg.Value
	} else {
	got = string(elem.Elm.([]byte))
	switch egv := eg.Value.(type) {
	case string:
	want = egv
	case []byte:
	want = string(egv)
	}
	}

	case "beam:coder:iterable:v1":
	pass := true
	gotrv := reflect.ValueOf(elem.Elm)
	wantrv := reflect.ValueOf(eg.Value)
	if gotrv.Len() != wantrv.Len() {
	log.Printf("Lengths don't match. got %v, want %v; %v, %v", gotrv.Len(), wantrv.Len(), gotrv, wantrv)
	return false
	}
	for i := 0; i < wantrv.Len(); i++ {
	if !diff(c.Components[0],
	&exec.FullValue{Elm: gotrv.Index(i).Interface()},
	yaml.MapItem{Value: wantrv.Index(i).Interface()}) {
	pass = false
	}

	}
	return pass
	case "beam:coder:interval_window:v1":
	var a, b int
	val := eg.Value
	if is, ok := eg.Value.([]any); ok {
	val = is[0]
	}
	v := val.(yaml.MapSlice)

	a = v[0].Value.(int)
	b = v[1].Value.(int)
	end := mtime.FromMilliseconds(int64(a))
	start := end - mtime.Time(int64(b))
	want = window.IntervalWindow{Start: start, End: end}
	// If this is nested in an iterable, windows won't be populated.
	if len(elem.Windows) == 0 {
	got = elem.Elm
	} else {
	got = elem.Windows[0]
	}

	case "beam:coder:global_window:v1":
	want = window.GlobalWindow{}
	// If this is nested in an iterable, windows won't be populated.
	if len(elem.Windows) == 0 {
	got = window.GlobalWindow(elem.Elm.(struct{}))
	} else {
	got = elem.Windows[0]
	}
	case "beam:coder:windowed_value:v1", "beam:coder:param_windowed_value:v1":
	// elem contains all the information, but we need to compare the element+timestamp
	// separately from the windows, to avoid repeated expected value parsing logic.
	pass := true
	vs := eg.Value.(yaml.MapSlice)
	if !diff(c.Components[0], elem, vs[0]) {
	pass = false
	}
	if d := cmp.Diff(
	mtime.FromMilliseconds(int64(vs[1].Value.(int))),
	elem.Timestamp, cmpOpts...); d != "" {

	pass = false
	}
	if !diff(c.Components[1], elem, vs[3]) {
	pass = false
	}
	if !diffPane(vs[2].Value, elem.Pane) {
	pass = false
	}
	return pass
	case "beam:coder:row:v1":
	fs := eg.Value.(yaml.MapSlice)
	var rfs []reflect.StructField
	// There are only 2 pointer examples, but they reuse field names,
	// so we key off the proto hash to know which example we're handling.
	ptrEg := strings.Contains(c.Payload, "51ace21c7393")
	for _, rf := range fs {
	name := rf.Key.(string)
	t := nameToType[name]
	if ptrEg {
	t = reflect.PtrTo(t)
	}
	rfs = append(rfs, reflect.StructField{
	Name: strings.ToUpper(name[:1]) + name[1:],
	Type: t,
	Tag: reflect.StructTag(fmt.Sprintf("beam:\"%v\"", name)),
	})
	}
	rv := reflect.New(reflect.StructOf(rfs)).Elem()
	for i, rf := range fs {
	setField(rv, i, rf.Value)
	}

	got, want = elem.Elm, rv.Interface()
	case "beam:coder:timer:v1":
	pass := true
	tm := elem.Elm.(exec.TimerRecv)
	fs := eg.Value.(yaml.MapSlice)
	for _, item := range fs {

	switch item.Key.(string) {
	case "userKey":
	if want := item.Value.(string); want != tm.Key.Elm.(string) {
	pass = false
	}
	case "dynamicTimerTag":
	if want := item.Value.(string); want != tm.Tag {
	pass = false
	}
	case "windows":
	if v, ok := item.Value.([]any); ok {
	for i, val := range v {
	if val.(string) == "global" && fmt.Sprintf("%s", tm.Windows[i]) == "[*]" {
	continue
	} else if val.(string) != fmt.Sprintf("%s", tm.Windows[i]) {
	pass = false
	}
	}
	}
	case "clearBit":
	if want := item.Value.(bool); want != tm.Clear {
	pass = false
	}
	case "fireTimestamp":
	if want := item.Value.(int); want != int(tm.FireTimestamp) {
	pass = false
	}
	case "holdTimestamp":
	if want := item.Value.(int); want != int(tm.HoldTimestamp) {
	pass = false
	}
	case "pane":
	pass = diffPane(item.Value, tm.Pane)
	}
	}
	return pass
	default:
	got, want = elem.Elm, eg.Value
	}
	if d := cmp.Diff(want, got, cmpOpts...); d != "" {
	log.Printf("Decoding error: diff(-want,+got): %v\n", d)
	return false
	}
	return true
	}

	func diffPane(eg any, got typex.PaneInfo) bool {
	pass := true
	paneTiming := map[typex.PaneTiming]string{
	typex.PaneUnknown: "UNKNOWN",
	typex.PaneEarly: "EARLY",
	typex.PaneLate: "LATE",
	typex.PaneOnTime: "ONTIME",
	}
	for _, item := range eg.(yaml.MapSlice) {
	switch item.Key.(string) {
	case "is_first":
	if want := item.Value.(bool); want != got.IsFirst {
	pass = false
	}
	case "is_last":
	if want := item.Value.(bool); want != got.IsLast {
	pass = false
	}
	case "timing":
	if want := item.Value.(string); want != paneTiming[got.Timing] {
	pass = false
	}
	case "index":
	if want := item.Value.(int); want != int(got.Index) {
	pass = false
	}
	case "on_time_index":
	if want := item.Value.(int); want != int(got.NonSpeculativeIndex) {
	pass = false
	}
	}
	}
	return pass
	}

	// standard_coders.yaml uses the name for type indication, except for nullability.
	var nameToType = map[string]reflect.Type{
	"str": reflectx.String,
	"i32": reflectx.Int32,
	"f64": reflectx.Float64,
	"arr": reflect.SliceOf(reflectx.String),
	"f_bool": reflectx.Bool,
	"f_bytes": reflect.PtrTo(reflectx.ByteSlice),
	"f_map": reflect.MapOf(reflectx.String, reflect.PtrTo(reflectx.Int64)),
	"f_float": reflectx.Float32,
	}

	func setField(rv reflect.Value, i int, v any) {
	if v == nil {
	return
	}
	rf := rv.Field(i)
	if rf.Kind() == reflect.Ptr {
	// Ensure it's initialized.
	rf.Set(reflect.New(rf.Type().Elem()))
	rf = rf.Elem()
	}
	switch rf.Kind() {
	case reflect.String:
	rf.SetString(v.(string))
	case reflect.Int32:
	rf.SetInt(int64(v.(int)))
	case reflect.Float32:
	c, err := strconv.ParseFloat(v.(string), 32)
	if err != nil {
	panic(err)
	}
	rf.SetFloat(c)
	case reflect.Float64:
	c, err := strconv.ParseFloat(v.(string), 64)
	if err != nil {
	panic(err)
	}
	rf.SetFloat(c)
	case reflect.Slice:
	if rf.Type() == reflectx.ByteSlice {
	rf.Set(reflect.ValueOf([]byte(v.(string))))
	break
	}
	// Value is a []any with string values.
	var arr []string
	for _, a := range v.([]any) {
	arr = append(arr, a.(string))
	}
	rf.Set(reflect.ValueOf(arr))
	case reflect.Bool:
	rf.SetBool(v.(bool))
	case reflect.Map:
	// only f_map presently, which is always map[string]*int64
	rm := reflect.MakeMap(rf.Type())
	for _, a := range v.(yaml.MapSlice) {
	rk := reflect.ValueOf(a.Key.(string))
	rv := reflect.Zero(rf.Type().Elem())
	if a.Value != nil {
	rv = reflect.New(reflectx.Int64)
	rv.Elem().SetInt(int64(a.Value.(int)))
	}
	rm.SetMapIndex(rk, rv)
	}
	rf.Set(rm)

	}
	}

	func (s *Spec) parseCoder(c Coder) string {
	id := s.nextID()
	var compIDs []string
	for _, comp := range c.Components {
	compIDs = append(compIDs, s.parseCoder(comp))
	}
	s.coderPBs[id] = &pipepb.Coder{
	Spec: &pipepb.FunctionSpec{
	Urn: c.Urn,
	Payload: []byte(c.Payload),
	},
	ComponentCoderIds: compIDs,
	}
	return id
	}

	// Simple logger to run as main program.
	type logLogger struct{}

	func (*logLogger) Errorf(format string, v ...any) {
	log.Printf(format, v...)
	}

	func (*logLogger) Logf(format string, v ...any) {
	log.Printf(format, v...)
	}

	const yamlPath = "../../../../../../model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml"

	func main() {
	data, err := os.ReadFile(yamlPath)
	if err != nil {
	log.Fatalf("Couldn't read %v: %v", yamlPath, err)
	}
	specs := bytes.Split(data, []byte("\n---\n"))
	var failures bool
	var l logLogger
	for _, data := range specs {
	cs := Spec{Log: &l}
	if err := yaml.Unmarshal(data, &cs); err != nil {
	failures = true
	l.Logf("unable to parse yaml: %v %q", err, data)
	continue
	}
	if err := cs.testStandardCoder(); err != nil {
	failures = true
	l.Logf("Failed \"%v\": %v", cs.Coder, err)
	}
	}
	if !failures {
	log.Println("PASS")
	}
	}