Abacn commented on issue #39684: URL: https://github.com/apache/beam/issues/39684#issuecomment-5222824415
Some AI insights. Likely we need a series of changes to support it --- ### Architectural Comparison: Go SDK vs. Python SDK In Python ([`schemas.py`](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/typehints/schemas.py#L812)), 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`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go#L126)), `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`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/transforms/xlang/python/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`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go#L36)). 2. `coder.RegisterSchemaProviders`: Registers encoder/decoder functions in `RowEncoderBuilder` / `RowDecoderBuilder` ([`row.go:L43`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/graph/coder/row.go#L43)). 3. `beam.RegisterSchemaProviderWithURN`: Ties a URN to a `SchemaProvider` interface ([`schema.go:L87`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/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`](https://github.com/apache/beam/blob/master/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto#L152-L165)): ```protobuf message LogicalType { string urn = 1; bytes payload = 2; FieldType representation = 3; FieldType argument_type = 4; FieldValue argument = 5; } ``` - **Problem**: In [`schema.go:L580`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/graphx/schema/schema.go#L580), argument handling is explicitly marked as unhandled: ```go // 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`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go#L166-L171)). - For `time.Time`, `identifier` is `"time.Time"` ([`encoding.go:L52`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/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`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/graphx/schema/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`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/encoding.go#L52-L53), 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`](https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/graphx/schema/schema.go#L408)). - **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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
