coder marshaling tests

apache · Nov 29, 2022 · 2f24f56 · 2f24f56
1 parent bf604de
commit 2f24f56
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
@@ -522,6 +522,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
@@ -85,6 +85,10 @@ func TestMarshalUnmarshalCoders(t *testing.T) {
 			name: "baz",
 			c:    baz,
 		},
+		{
+			name: "IW",
+			c:    coder.NewIntervalWindowCoder(),
+		},
 		{
 			name: "W<bytes>",
 			c:    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
@@ -176,6 +176,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
 
@@ -305,6 +308,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)