[Task][Go SDK] Disallow iterators with initial timestamp values (ET,V) and (ET, K,V) -> not in the beam model.

[lostluck]

What needs to happen?

Timestamp metadata for individual elements is lost after sending values through a GroupByKey. This is not an issue, and is WAI as a consequence of the post GBK coder (WindowedValueHeader+KV<K, Iter>), as the runner has combined timestamps.

We need to disallow usage of those iterators: They don't work because they're not in the beam model.

After a GBK, the whole <K, V*> element has a single timestamp, and while side inputs are placed into the corresponding windows, they do not preserve their timestamps.

As such, it's just a painpoint waiting to happen.

The best path forward is to have a clear error when we detect these forms instructing users that this is decomissioned and not part of the model and instruct them on the model instead. The existing code that allows them here can return this error instead.

See next comment for other places that need updating as a result.

This is separate from whether the SDK supports specifying timestamp combine strategies for a window. Presently it does not, and should default to "End Of Window" for a given Post GBK element.

Issue Priority

Priority: 2

Issue Component

Component: sdk-go

[lostluck]

Discussing this with @robertwb and @reuvenlax indicates that this is something incorrect in the Go SDK, and shouldn't be included.

By the Model and the Coders, the timestamps for individual values are either combined (in the GBK case) or dropped after being used for window assignment (for Side inputs).

So ultimately, instead we need to simply fail these pipelines at construction time (with an error referring to this issue and explain this part of the model).

There are also some bad tests & documentation that need correction.

beam/sdks/go/pkg/beam/core/funcx/sideinput.go

Line 59 in 8b213c6

// func (*typex.EventTime, *int) bool returns {typex.EventTime, int}

beam/sdks/go/pkg/beam/core/funcx/sideinput_test.go

Line 34 in 273613d

{func(*typex.EventTime) bool { return false }, false}, // no values

Reflective Iterators should have the handling code removed:

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

Line 172 in 66ffc0b

case t == typex.EventTimeType:

legacy template:

beam/sdks/go/pkg/beam/core/runtime/exec/optimized/inputs.tmpl

Line 33 in 410ad76

exec.RegisterInput(reflect.TypeOf((*func (*typex.EventTime, *{{$x.Type}})bool)(nil)).Elem(), iterMakerET{{$x.Name}})