[Task]: Go SDK portable logical type support

[Abacn]

What needs to happen?

Logical types defined in https://github.com/apache/beam/blob/master/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto are not supported in Go SDK. This prevents Go SDK to use schema transforms in a meaningful way.

Currently, only PYTHON_CALLABLE appear to be supported,

beam/sdks/go/pkg/beam/transforms/xlang/python/external.go

Line 42 in 1318bfe

beam.RegisterSchemaProviderWithURN(pcsType, &callableSourceProvider{}, pythonCallableUrn)

however, through a different mechanism. Likely we need to first implement a framework for portable logical types in Golang, then implement each of them

Issue Priority

Priority: 2 (default / most normal work 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

[Abacn]

Some AI insights. Likely we need a series of changes to support it

---

Architectural Comparison: Go SDK vs. Python SDK

In Python (schemas.py), a LogicalType is a unified class containing:

• The URN (urn())
• The Language Type (language_type())
• The Representation Wire Type (representation_type())
• Bidirectional Converters: to_representation_type(val) and to_language_type(val)
• Argument Specs: argument_type() and argument() for parameterized types

When RowCoder encodes/decodes rows in Python, it automatically invokes to_representation_type() / to_language_type().

In Go (logicaltypes.go), LogicalType is only a static struct containing goT, storageT, argT, argV—with no conversion methods attached. Custom encoding/decoding is offloaded to RegisterSchemaProviders or SchemaProvider instances (like callableSourceProvider in external.go:L79).

---

Key Gaps in Go SDK for Portable Logical Type Support

Gap 1: Lack of Bidirectional Conversion Methods in schema.LogicalType

• Problem: schema.LogicalType in Go (logicaltypes.go) defines types (goT and storageT), but does not provide ToStorageValue(goValue) or ToGoValue(storageValue) functions.
• Impact: Automatic conversion cannot occur during RowEncoder / RowDecoder execution. Every new logical type requires writing manual BuildEncoder and BuildDecoder closures using coder.EncodeStringUTF8 or reflection, rather than specifying how to map a Go value to its underlying storage representation type.

Gap 2: Fragmented Registries across Packages

• Problem: Logical type metadata is split across three ununified registries:
  1. schema.RegisterLogicalType: Registers goT <-> storageT in schema.Registry (logicaltypes.go:L36).
  2. coder.RegisterSchemaProviders: Registers encoder/decoder functions in RowEncoderBuilder / RowDecoderBuilder (row.go:L43).
  3. beam.RegisterSchemaProviderWithURN: Ties a URN to a SchemaProvider interface (schema.go:L87).
• Impact: Adding a new logical type requires multiple calls across schema, coder, and beam packages rather than registering a single logical type object.

Gap 3: Missing Argument & Payload Support for Parameterized Types

• Proto Specification (schema.proto:L152-L165):

  message LogicalType {
    string urn = 1;
    bytes payload = 2;
    FieldType representation = 3;
    FieldType argument_type = 4;
    FieldValue argument = 5;
  }

• Problem: In schema.go:L580, argument handling is explicitly marked as unhandled:

  // TODO(BEAM-9615): Handle type Arguments.

  Neither reflectTypeToFieldType (encoding) nor fieldTypeToReflectType (decoding) sets or parses argument_type or argument.
• Impact: Parameterized logical types defined in Beam standard schemas—such as TIMESTAMP (precision arg), DECIMAL (precision/scale arg), FIXED_BYTES (length arg), and VAR_CHAR (max_length arg)—cannot be serialized or deserialized with their parameters.

Gap 4: Local Identifier Keying vs. Standard URN Lookup

• Problem: Go SDK uses identifier as the map key in registry.logicalTypes[identifier].
  • For Go's primitive type overrides (int, uint16), identifier is "int", "uint16" (logicaltypes.go:L166-L171).
  • For time.Time, identifier is "time.Time" (encoding.go:L52).
• Impact: When decoding a row schema sent from a cross-language runner or Python/Java SDK containing standard URNs (e.g. beam:logical_type:timestamp:v1 or beam:logical_type:date:v1), fieldTypeToReflectType in schema.go:L796 attempts r.logicalTypes["beam:logical_type:timestamp:v1"] and fails with:

  unknown logical type: beam:logical_type:timestamp:v1
  

Gap 5: Non-Standard Wire Representations for Standard Types (e.g., time.Time)

• Problem: In encoding.go:L52-L53, L274, Go's time.Time is registered using a custom Go-specific logical type ("time.Time") that encodes time using t.MarshalText() (string format) inside a wrapper struct struct{ EncodedBeamData string }.
• Impact: Standard Beam portable timestamps expect either beam:logical_type:micros_instant:v1 (ROW<seconds: INT64, micros: INT64>), beam:logical_type:millis_instant:v1 (INT64), or beam:logical_type:timestamp:v1. Go's time.Time cannot interoperate out of the box with cross-language rows containing standard Beam timestamps.

Gap 6: FieldValue <-> reflect.Value Conversion Framework Missing

• Problem: FieldValue protos in schema.proto are used to encode default values, option values, and logical type arguments. Go SDK only supports basic toggle options (optGoNillableUrn, optGoEmbeddedUrn) and string options (optGoLogicalUrn) (schema.go:L408-L500).
• Impact: Arguments or payloads stored inside pipepb.FieldValue cannot be automatically converted to Go reflect.Value instances.

---

Summary of Requirements for a Full Portable Logical Type Framework in Go

To bring Go SDK to parity with Python/Java for portable logical types, the following framework enhancements are needed:

1. Extend LogicalType interface/struct in logicaltypes.go to include ToStorageValue(goVal) and ToGoValue(storageVal) conversion hooks, allowing RowEncoder / RowDecoder to automatically convert values without requiring custom SchemaProvider boilerplate.
2. Support argument_type and argument (pipepb.FieldValue) in schema.go when translating LogicalType to/from pipepb.FieldType.
3. Register standard Beam URNs (beam:logical_type:timestamp:v1, beam:logical_type:date:v1, beam:logical_type:decimal:v1, etc.) by default in schema.defaultRegistry.
4. Map time.Time to beam:logical_type:micros_instant:v1 or timestamp:v1 (or provide standard aliases) for cross-language compatibility.

[lalitium]

.take-issue