[Go SDK] Add timer coder support

sdks/go/pkg/beam/core/graph/coder/coder.go

@@ -347,17 +347,9 @@ func NewT(c *Coder, w *WindowCoder) *Coder {
 		panic("window must not be nil")
 	}
 
-	// TODO(https://github.com/apache/beam/issues/20510): Implement proper timer support.
 	return &Coder{
-		Kind: Timer,
-		T: typex.New(reflect.TypeOf((*struct {
-			Key                          []byte // elm type.
-			Tag                          string
-			Windows                      []byte // []typex.Window
-			Clear                        bool
-			FireTimestamp, HoldTimestamp int64
-			Span                         int
-		})(nil)).Elem()),
+		Kind:       Timer,
+		T:          typex.New(typex.TimersType),
 		Window:     w,
 		Components: []*Coder{c},
 	}

sdks/go/pkg/beam/core/runtime/exec/coder.go

@@ -145,6 +145,7 @@ func MakeElementEncoder(c *coder.Coder) ElementEncoder {
 	case coder.Timer:
 		return &timerEncoder{
 			elm: MakeElementEncoder(c.Components[0]),
+			win: MakeWindowEncoder(c.Window),
 		}
 
 	case coder.Row:
@@ -262,6 +263,7 @@ func MakeElementDecoder(c *coder.Coder) ElementDecoder {
 	case coder.Timer:
 		return &timerDecoder{
 			elm: MakeElementDecoder(c.Components[0]),
+			win: MakeWindowDecoder(c.Window),
 		}
 
 	case coder.Row:
@@ -890,18 +892,25 @@ func (d *paramWindowedValueDecoder) Decode(r io.Reader) (*FullValue, error) {
 
 type timerEncoder struct {
 	elm ElementEncoder
+	win WindowEncoder
 }
 
 func (e *timerEncoder) Encode(val *FullValue, w io.Writer) error {
-	return e.elm.Encode(val, w)
+	return encodeTimer(e.elm, e.win, val.Elm.(typex.TimerMap), w)
 }
 
 type timerDecoder struct {
 	elm ElementDecoder
+	win WindowDecoder
 }
 
 func (d *timerDecoder) DecodeTo(r io.Reader, fv *FullValue) error {
-	return d.elm.DecodeTo(r, fv)
+	data, err := decodeTimer(d.elm, d.win, r)
+	if err != nil {
+		return err
+	}
+	fv.Elm = data
+	return nil
 }
 
 func (d *timerDecoder) Decode(r io.Reader) (*FullValue, error) {
@@ -1208,3 +1217,91 @@ func DecodeWindowedValueHeader(dec WindowDecoder, r io.Reader) ([]typex.Window,
 
 	return ws, t, pn, nil
 }
+
+// encodeTimer encodes a typex.TimerMap into a byte stream.
+func encodeTimer(elm ElementEncoder, win WindowEncoder, tm typex.TimerMap, w io.Writer) error {
+	var b bytes.Buffer
+
+	elm.Encode(&FullValue{Elm: tm.Key}, &b)
+
+	if err := coder.EncodeStringUTF8(tm.Tag, &b); err != nil {
+		return errors.WithContext(err, "error encoding tag")
+	}
+
+	if err := win.Encode(tm.Windows, &b); err != nil {
+		return errors.WithContext(err, "error encoding window")
+	}
+	if err := coder.EncodeBool(tm.Clear, &b); err != nil {
+		return errors.WithContext(err, "error encoding clear bit")
+	}
+
+	if !tm.Clear {
+		if err := coder.EncodeEventTime(tm.FireTimestamp, &b); err != nil {
+			return errors.WithContext(err, "error encoding fire timestamp")
+		}
+		if err := coder.EncodeEventTime(tm.HoldTimestamp, &b); err != nil {
+			return errors.WithContext(err, "error encoding hold timestamp")
+		}
+		if err := coder.EncodePane(tm.Pane, &b); err != nil {
+			return errors.WithContext(err, "error encoding paneinfo")
+		}
+	}
+
+	w.Write(b.Bytes())
+	return nil
+}
+
+// decodeTimer decodes timer byte encoded with standard timer coder spec.
+func decodeTimer(dec ElementDecoder, win WindowDecoder, r io.Reader) (typex.TimerMap, error) {
+	tm := typex.TimerMap{}
+
+	fv, err := dec.Decode(r)
+	if err != nil {
+		return tm, errors.WithContext(err, "error decoding timer key")
+	}
+	tm.Key = fv.Elm.(string)
+
+	s, err := coder.DecodeStringUTF8(r)
+	if err != nil && err != io.EOF {
+		return tm, errors.WithContext(err, "error decoding timer tag")
+	} else if err == io.EOF {
+		// when tag is empty
+		tm.Tag = ""
+	}
+	tm.Tag = s
+
+	w, err := win.Decode(r)
+	if err != nil {
+		return tm, errors.WithContext(err, "error decoding timer window")
+	}
+	tm.Windows = w
+
+	c, err := coder.DecodeBool(r)
+	if err != nil {
+		return tm, errors.WithContext(err, "error decoding clear")
+	}
+	tm.Clear = c
+	if tm.Clear {
+		return tm, nil
+	}
+
+	ft, err := coder.DecodeEventTime(r)
+	if err != nil && err != io.EOF {
+		return tm, errors.WithContext(err, "error decoding ft")
+	}
+	tm.FireTimestamp = ft
+
+	ht, err := coder.DecodeEventTime(r)
+	if err != nil && err != io.EOF {
+		return tm, errors.WithContext(err, "error decoding ht")
+	}
+	tm.HoldTimestamp = ht
+
+	pn, err := coder.DecodePane(r)
+	if err != nil && err != io.EOF {
+		return tm, errors.WithContext(err, "error decoding pn")
+	}
+	tm.Pane = pn
+
+	return tm, nil
+}

sdks/go/pkg/beam/core/runtime/exec/coder_test.go

@@ -188,26 +188,6 @@ func TestIterableCoder(t *testing.T) {
 	}
 }
 
-// TODO(https://github.com/apache/beam/issues/20510): Update once proper timer support is added
-func TestTimerCoder(t *testing.T) {
-	var buf bytes.Buffer
-	tCoder := coder.NewT(coder.NewVarInt(), coder.NewGlobalWindow())
-	wantVal := &FullValue{Elm: int64(13)}
-
-	enc := MakeElementEncoder(tCoder)
-	if err := enc.Encode(wantVal, &buf); err != nil {
-		t.Fatalf("Couldn't encode value: %v", err)
-	}
-
-	dec := MakeElementDecoder(tCoder)
-	result, err := dec.Decode(&buf)
-	if err != nil {
-		t.Fatalf("Couldn't decode value: %v", err)
-	}
-
-	compareFV(t, result, wantVal)
-}
-
 type namedTypeForTest struct {
 	A, B int64
 	C    string

sdks/go/pkg/beam/core/runtime/exec/timers_test.go

@@ -0,0 +1,96 @@
+// 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.
+
+package exec
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+	"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/typex"
+)
+
+func equalTimers(a, b typex.TimerMap) bool {
+	return a.Key == b.Key && a.Tag == b.Tag && (a.FireTimestamp) == b.FireTimestamp && a.Clear == b.Clear
+}
+
+func TestTimerEncodingDecoding(t *testing.T) {
+	tc := coder.NewT(coder.NewString(), window.NewGlobalWindows().Coder())
+	ec := MakeElementEncoder(coder.SkipW(tc))
+	dec := MakeElementDecoder(coder.SkipW(tc))
+
+	tests := []struct {
+		name   string
+		tm     typex.TimerMap
+		result bool
+	}{
+		{
+			name: "all set fields",
+			tm: typex.TimerMap{
+				Key:           "Basic",
+				Tag:           "first",
+				Windows:       window.SingleGlobalWindow,
+				Clear:         false,
+				FireTimestamp: mtime.Now(),
+			},
+			result: true,
+		},
+		{
+			name: "without tag",
+			tm: typex.TimerMap{
+				Key:           "Basic",
+				Tag:           "",
+				Windows:       window.SingleGlobalWindow,
+				Clear:         false,
+				FireTimestamp: mtime.Now(),
+			},
+			result: true,
+		},
+		{
+			name: "with clear set",
+			tm: typex.TimerMap{
+				Key:           "Basic",
+				Tag:           "first",
+				Windows:       window.SingleGlobalWindow,
+				Clear:         true,
+				FireTimestamp: mtime.Now(),
+			},
+			result: false,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fv := FullValue{Elm: test.tm}
+			var buf bytes.Buffer
+			err := ec.Encode(&fv, &buf)
+			if err != nil {
+				t.Fatalf("error encoding timer: %#v, got: %v", test.tm, err)
+			}
+
+			gotFv, err := dec.Decode(&buf)
+			if err != nil {
+				t.Fatalf("failed to decode timer, got %v", err)
+			}
+
+			if got, want := gotFv.Elm.(typex.TimerMap), test.tm; test.result != equalTimers(got, want) {
+				t.Errorf("got timer %v, want %v", got, want)
+			}
+		})
+	}
+
+}

sdks/go/pkg/beam/core/runtime/graphx/coder.go

@@ -537,7 +537,19 @@ func (b *CoderMarshaller) Add(c *coder.Coder) (string, error) {
 		}
 		return b.internRowCoder(s), nil
 
-	// TODO(https://github.com/apache/beam/issues/20510): Handle coder.Timer support.
+	case coder.Timer:
+		comp := []string{}
+		if ids, err := b.AddMulti(c.Components); err != nil {
+			return "", errors.SetTopLevelMsgf(err, "failed to marshal timer coder %v", c)
+		} else {
+			comp = append(comp, ids...)
+		}
+		if id, err := b.AddWindowCoder(c.Window); err != nil {
+			return "", errors.Wrapf(err, "failed to marshal window coder %v", c)
+		} else {
+			comp = append(comp, id)
+		}
+		return b.internBuiltInCoder(urnTimerCoder, comp...), nil
 
 	default:
 		err := errors.Errorf("unexpected coder kind: %v", c.Kind)

sdks/go/pkg/beam/core/typex/class.go

@@ -234,7 +234,7 @@ func IsUniversal(t reflect.Type) bool {
 // Composite marker types: KV, CoGBK or WindowedValue.
 func IsComposite(t reflect.Type) bool {
 	switch t {
-	case KVType, CoGBKType, WindowedValueType:
+	case KVType, CoGBKType, WindowedValueType, TimersType:
 		return true
 	default:
 		return false

sdks/go/pkg/beam/core/typex/fulltype.go

@@ -139,6 +139,8 @@ func New(t reflect.Type, components ...FullType) FullType {
 				panic("Invalid to nest composites inside CoGBK")
 			}
 			return &tree{class, t, components}
+		case TimersType:
+			return &tree{class, t, components}
 		default:
 			panic(fmt.Sprintf("Unexpected composite type: %v", t))
 		}

sdks/go/pkg/beam/core/typex/special.go

@@ -36,6 +36,7 @@ var (
 
 	EventTimeType = reflect.TypeOf((*EventTime)(nil)).Elem()
 	WindowType    = reflect.TypeOf((*Window)(nil)).Elem()
+	TimersType    = reflect.TypeOf((*Timers)(nil)).Elem()
 	PaneInfoType  = reflect.TypeOf((*PaneInfo)(nil)).Elem()
 
 	KVType                 = reflect.TypeOf((*KV)(nil)).Elem()
@@ -88,6 +89,25 @@ type PaneInfo struct {
 	Index, NonSpeculativeIndex int64
 }
 
+// Timers is the actual type used for standard timer coder.
+type Timers struct {
+	Key                          []byte // elm type.
+	Tag                          string
+	Windows                      []byte // []typex.Window
+	Clear                        bool
+	FireTimestamp, HoldTimestamp mtime.Time
+	Pane                         PaneInfo
+}
+
+// TimerMap is a placeholder for timer details used in encoding/decoding.
+type TimerMap struct {
+	Key, Tag                     string
+	Windows                      []Window // []typex.Window
+	Clear                        bool
+	FireTimestamp, HoldTimestamp mtime.Time
+	Pane                         PaneInfo
+}
+
 // KV, Nullable, CoGBK, WindowedValue represent composite generic types. They are not used
 // directly in user code signatures, but only in FullTypes.