coder marshaling tests

apache · Nov 28, 2022 · f28028a · f28028a
1 parent 203ca05
commit f28028a
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 0 deletions.
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/coder.go b/sdks/go/pkg/beam/core/runtime/graphx/coder.go
@@ -525,6 +525,9 @@ func (b *CoderMarshaller) Add(c *coder.Coder) (string, error) {
 	case coder.String:
 		return b.internBuiltInCoder(urnStringCoder), nil
 
+	case coder.IW:
+		return b.internBuiltInCoder(urnIntervalWindow), nil
+
 	case coder.Row:
 		rt := c.T.Type()
 		s, err := schema.FromType(rt)

diff --git a/sdks/go/pkg/beam/core/runtime/graphx/coder_test.go b/sdks/go/pkg/beam/core/runtime/graphx/coder_test.go
@@ -84,6 +84,10 @@ func TestMarshalUnmarshalCoders(t *testing.T) {
 			"baz",
 			baz,
 		},
+		{
+			name: "IW",
+			c:    coder.NewIntervalWindowCoder(),
+		},
 		{
 			"W<bytes>",
 			coder.NewW(coder.NewBytes(), coder.NewGlobalWindow()),

diff --git a/sdks/go/pkg/beam/core/runtime/graphx/dataflow.go b/sdks/go/pkg/beam/core/runtime/graphx/dataflow.go
@@ -164,6 +164,9 @@ func EncodeCoderRef(c *coder.Coder) (*CoderRef, error) {
 		}
 		return &CoderRef{Type: windowedValueType, Components: []*CoderRef{elm, w}, IsWrapper: true}, nil
 
+	case coder.IW:
+		return &CoderRef{Type: intervalWindowType}, nil
+
 	case coder.Bytes:
 		return &CoderRef{Type: bytesType}, nil
 
@@ -301,6 +304,9 @@ func DecodeCoderRef(c *CoderRef) (*coder.Coder, error) {
 			return decodeDataflowCustomCoder(subC.Type)
 		}
 
+	case intervalWindowType:
+		return coder.NewIntervalWindowCoder(), nil
+
 	case windowedValueType:
 		if len(c.Components) != 2 {
 			return nil, errors.Errorf("bad windowed value: %+v", c)