[Bug]: Go SDK row coder encodes INT16 schema fields as varint instead of big endian

[Eliaaazzz]

What happened?

The Go SDK reflection row coder encodes int16 and uint16 struct fields with the varint encoding: the reflect.Int16 case in sdks/go/pkg/beam/core/graph/coder/row_encoder.go shares EncodeVarInt with the other int kinds. The Java row coder uses BigEndianShortCoder for INT16 (SchemaCoderHelpers.CODER_MAP) and the Python row coder uses BigEndianShortCoder as well (apache_beam/coders/row_coder.py). Rows with an INT16 field therefore cannot be exchanged between the Go SDK and the other SDKs.

Example: for the value 999 the Go SDK writes \xe7\x07, Java and Python write \x03\xe7. The beam:logical_type:timestamp:v1 cases in standard_coders.yaml (precision 3, subseconds is INT16) show the big endian bytes.

standard_coders.yaml has no row case with a plain INT16 field, so the Go regression test in sdks/go/test/regression/coders/fromyaml does not catch this.

Fixing it changes the wire format of rows with int16 or uint16 fields produced by the Go SDK, which is a breaking change for Go pipelines that persist such rows.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Prism Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

LABELS: bug, go, P2
BODY:
Fixes #ISSUE_A. Part of #39684.

The reflection based row coder encoded int16 and uint16 struct fields with the varint encoding. Java (BigEndianShortCoder) and Python (BigEndianShortCoder) encode the INT16 schema type as 2 big endian bytes, so rows with such fields could not be exchanged with the other SDKs. The beam:logical_type:timestamp:v1 cases in standard_coders.yaml depend on this encoding for precisions below 5.

Changes:

• Add coder.EncodeInt16, DecodeInt16, EncodeUint16 and DecodeUint16.
• The row encoder and decoder use them for int16 and uint16 fields. uint16 is stored as an INT16 schema field.
• Byte level test against the Java and Python encoding.
• CHANGES.md breaking change note: rows with int16 or uint16 fields written by earlier Go SDK versions decode differently.

COMMIT MESSAGE:
[Go SDK] Encode INT16 row fields as big endian to match Java and Python

The reflection based row coder used the varint encoding for int16 and
uint16 fields. Java and Python encode the INT16 schema type as 2 big
endian bytes, so rows with such fields could not be exchanged with the
other SDKs.