This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 3ab18d061df [Go SDK] Add portable logical type registration and
standard date and micros_instant types (#40154)
3ab18d061df is described below
commit 3ab18d061df31be46c619da107e66cfb3ccbd996
Author: Elia Liu <[email protected]>
AuthorDate: Tue Oct 20 05:48:44 2026 +1100
[Go SDK] Add portable logical type registration and standard date and
micros_instant types (#40154)
* [Go SDK] Add portable logical type registration and standard date and
micros_instant types
schema.RegisterLogicalTypeConversion registers a logical type identified
by a URN together with the conversions between its Go type and its
storage type, and derives the row coder from the storage type.
schema.RegisterLogicalTypeCoder does the same for logical types whose
wire encoding differs from the row encoding of the storage type. The
standard beam:logical_type:date:v1 and beam:logical_type:micros_instant:v1
logical types are registered by default as schema.Date and
schema.MicrosInstant.
The standard coder regression test now builds the expected row from the
schema in the coder payload, decodes payload bytes above 0x7f, counts
encoding mismatches as failures, and runs the micros_instant case.
Part of #39684.
* [Go SDK] Replace deprecated reflect.PtrTo in row coder test
* [Go SDK] Handle the zero Date, add MicrosInstant String and Equals, check
duplicate registrations
The zero Date now stands for the epoch, MicrosInstant has String and Equals
methods, and RegisterLogicalType panics when a Go type or an identifier and
argument are already registered for a different logical type.
---
CHANGES.md | 1 +
sdks/go/pkg/beam/core/graph/coder/row.go | 16 +
sdks/go/pkg/beam/core/graph/coder/row_decoder.go | 37 ++
sdks/go/pkg/beam/core/graph/coder/row_encoder.go | 21 +
sdks/go/pkg/beam/core/graph/coder/row_test.go | 105 +++++
.../core/runtime/graphx/schema/logicaltypes.go | 142 ++++++-
.../beam/core/runtime/graphx/schema/schema_test.go | 30 ++
.../graphx/schema/standard_logical_types.go | 156 ++++++++
.../graphx/schema/standard_logical_types_test.go | 421 +++++++++++++++++++++
.../pkg/beam/core/runtime/graphx/serialize_test.go | 16 +
.../go/test/regression/coders/fromyaml/fromyaml.go | 210 ++++++----
11 files changed, 1079 insertions(+), 76 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index c49c27c929c..d50b14cadd1 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -80,6 +80,7 @@
* (Python) Added `AwsOptions` with `--aws_*` pipeline options (e.g.
`--aws_access_key_id`, `--aws_region_name`, `--aws_endpoint_url`) that set
defaults for all AWS service clients. The existing `--s3_*` options in
`S3Options` override these defaults for the S3 client only
([#40528](https://github.com/apache/beam/issues/40528)).
* (Python) Added `--aws_credentials_secret_manager` so AWS credentials
(`--aws_access_key_id`, `--aws_secret_access_key`, `--aws_session_token` and
their `--s3_*` overrides) can be secret specs resolved through a secret manager
such as `GoogleCloudSecretManager`, instead of plain-text values
([#40528](https://github.com/apache/beam/issues/40528)).
* (Python) Added `AwsSecret` (`AwsSecretsManager` / `type:awssecret`) to
retrieve secrets from AWS Secrets Manager, configured from `AwsOptions`
(`--aws_*`) with optional per-secret connection overrides
([#40528](https://github.com/apache/beam/issues/40528)).
+* (Go) Added `schema.RegisterLogicalTypeConversion` and
`schema.RegisterLogicalTypeCoder` to register portable logical types in one
call, and Go types for the standard `beam:logical_type:date:v1` and
`beam:logical_type:micros_instant:v1` logical types, so schemas from other SDKs
with these types can be used in cross-language pipelines
([#39684](https://github.com/apache/beam/issues/39684)).
## Breaking Changes
diff --git a/sdks/go/pkg/beam/core/graph/coder/row.go
b/sdks/go/pkg/beam/core/graph/coder/row.go
index f958d84154c..0c2a6872dd6 100644
--- a/sdks/go/pkg/beam/core/graph/coder/row.go
+++ b/sdks/go/pkg/beam/core/graph/coder/row.go
@@ -226,3 +226,19 @@ func ReadSimpleRowHeader(fields int, r io.Reader) error {
}
return nil
}
+
+// RowFieldEncoderForType returns an encoding function for values of the given
+// type, using the encoding of that type as a field of a beam schema row.
+//
+// Returns an error if the given type cannot be a field of a beam schema row.
+func RowFieldEncoderForType(rt reflect.Type) (func(any, io.Writer) error,
error) {
+ return defaultEnc.fieldEncoderForType(rt)
+}
+
+// RowFieldDecoderForType returns a decoding function for values of the given
+// type, using the encoding of that type as a field of a beam schema row.
+//
+// Returns an error if the given type cannot be a field of a beam schema row.
+func RowFieldDecoderForType(rt reflect.Type) (func(io.Reader) (any, error),
error) {
+ return defaultDec.fieldDecoderForType(rt)
+}
diff --git a/sdks/go/pkg/beam/core/graph/coder/row_decoder.go
b/sdks/go/pkg/beam/core/graph/coder/row_decoder.go
index 90938c237fe..2402cafe0e2 100644
--- a/sdks/go/pkg/beam/core/graph/coder/row_decoder.go
+++ b/sdks/go/pkg/beam/core/graph/coder/row_decoder.go
@@ -95,6 +95,23 @@ func (b *RowDecoderBuilder) decoderForType(t reflect.Type)
(func(io.Reader) (any
if t.Kind() == reflect.Ptr {
isPtr = true
t = t.Elem()
+ // A provider for the value type also handles a top level
pointer.
+ f, addr, err := b.customFunc(t)
+ if err != nil {
+ return nil, err
+ }
+ if f != nil && !addr {
+ vt := t
+ return func(r io.Reader) (any, error) {
+ v, err := f(r)
+ if err != nil {
+ return nil, err
+ }
+ rv := reflect.New(vt)
+ rv.Elem().Set(reflect.ValueOf(v))
+ return rv.Interface(), nil
+ }, nil
+ }
}
dec, err := b.decoderForStructReflect(t)
if err != nil {
@@ -435,3 +452,23 @@ type typeDecoderFieldReflect struct {
// of the field value (i.e. &foo.bar) and not the field value (i.e.
foo.bar).
addr bool
}
+
+// fieldDecoderForType returns a decoder for values of t, using the encoding
+// of t as a row field.
+func (b *RowDecoderBuilder) fieldDecoderForType(t reflect.Type)
(func(io.Reader) (any, error), error) {
+ decf, err := b.decoderForSingleTypeReflect(t)
+ if err != nil {
+ return nil, err
+ }
+ return func(r io.Reader) (any, error) {
+ rv := reflect.New(t)
+ v := rv.Elem()
+ if decf.addr {
+ v = rv
+ }
+ if err := decf.decode(v, r); err != nil {
+ return nil, err
+ }
+ return rv.Elem().Interface(), nil
+ }, nil
+}
diff --git a/sdks/go/pkg/beam/core/graph/coder/row_encoder.go
b/sdks/go/pkg/beam/core/graph/coder/row_encoder.go
index 6f5c4c04e81..05d3e85b2d9 100644
--- a/sdks/go/pkg/beam/core/graph/coder/row_encoder.go
+++ b/sdks/go/pkg/beam/core/graph/coder/row_encoder.go
@@ -363,3 +363,24 @@ func (b *RowEncoderBuilder) encoderForStructReflect(t
reflect.Type) (func(reflec
return nil
}, nil
}
+
+// fieldEncoderForType returns an encoder for values of t, using the encoding
+// of t as a row field.
+func (b *RowEncoderBuilder) fieldEncoderForType(t reflect.Type) (func(any,
io.Writer) error, error) {
+ encf, err := b.encoderForSingleTypeReflect(t)
+ if err != nil {
+ return nil, err
+ }
+ return func(v any, w io.Writer) error {
+ rv := reflect.ValueOf(v)
+ if !rv.IsValid() || (rv.Kind() == reflect.Ptr && rv.IsNil()) {
+ return errors.Errorf("cannot encode a nil value as a
row field of type %v", t)
+ }
+ if encf.addr {
+ p := reflect.New(t)
+ p.Elem().Set(rv)
+ rv = p
+ }
+ return encf.encode(rv, w)
+ }, nil
+}
diff --git a/sdks/go/pkg/beam/core/graph/coder/row_test.go
b/sdks/go/pkg/beam/core/graph/coder/row_test.go
index c2a925efd30..8b6782457d8 100644
--- a/sdks/go/pkg/beam/core/graph/coder/row_test.go
+++ b/sdks/go/pkg/beam/core/graph/coder/row_test.go
@@ -945,3 +945,108 @@ func TestReflectionRowCoderGeneration_Int16BigEndian(t
*testing.T) {
t.Fatalf("dec(enc(%v)) = %v\ndiff (-want, +got): %v", want,
got, d)
}
}
+
+func TestRowFieldCoderForType(t *testing.T) {
+ type pair struct {
+ A int64
+ B string
+ }
+ tests := []struct {
+ value any
+ wantBytes []byte
+ }{
+ {value: int64(999), wantBytes: []byte{0xe7, 0x07}},
+ {value: int16(999), wantBytes: []byte{0x03, 0xe7}},
+ {value: "abc", wantBytes: []byte{0x03, 'a', 'b', 'c'}},
+ {value: []byte{1, 2}, wantBytes: []byte{0x02, 1, 2}},
+ {value: true, wantBytes: []byte{0x01}},
+ {value: []int32{1, 2}, wantBytes: []byte{0x00, 0x00, 0x00,
0x02, 0x01, 0x02}},
+ {value: pair{A: 7, B: "x"}, wantBytes: []byte{0x02, 0x00, 0x07,
0x01, 'x'}},
+ }
+ for _, test := range tests {
+ rt := reflect.TypeOf(test.value)
+ t.Run(rt.String(), func(t *testing.T) {
+ enc, err := RowFieldEncoderForType(rt)
+ if err != nil {
+ t.Fatalf("RowFieldEncoderForType(%v) = %v, want
nil error", rt, err)
+ }
+ var buf bytes.Buffer
+ if err := enc(test.value, &buf); err != nil {
+ t.Fatalf("enc(%v) = %v, want nil error",
test.value, err)
+ }
+ if got := buf.Bytes(); !bytes.Equal(got,
test.wantBytes) {
+ t.Fatalf("enc(%v) = %v, want %v", test.value,
got, test.wantBytes)
+ }
+ dec, err := RowFieldDecoderForType(rt)
+ if err != nil {
+ t.Fatalf("RowFieldDecoderForType(%v) = %v, want
nil error", rt, err)
+ }
+ got, err := dec(bytes.NewBuffer(test.wantBytes))
+ if err != nil {
+ t.Fatalf("dec(%v) = %v, want nil error",
test.wantBytes, err)
+ }
+ if d := cmp.Diff(test.value, got); d != "" {
+ t.Fatalf("dec(enc(%v)) = %v\ndiff (-want,
+got): %v", test.value, got, d)
+ }
+ })
+ }
+ if _, err := RowFieldEncoderForType(reflect.TypeOf(func() {})); err ==
nil {
+ t.Errorf("RowFieldEncoderForType(func) = nil error, want error")
+ }
+ enc, err := RowFieldEncoderForType(reflect.TypeOf(int64(0)))
+ if err != nil {
+ t.Fatalf("RowFieldEncoderForType(int64) = %v, want nil error",
err)
+ }
+ if err := enc(nil, &bytes.Buffer{}); err == nil {
+ t.Errorf("enc(nil) = nil error, want error")
+ }
+ ptrEnc, err := RowFieldEncoderForType(reflect.TypeOf((*int64)(nil)))
+ if err != nil {
+ t.Fatalf("RowFieldEncoderForType(*int64) = %v, want nil error",
err)
+ }
+ if err := ptrEnc((*int64)(nil), &bytes.Buffer{}); err == nil {
+ t.Errorf("enc((*int64)(nil)) = nil error, want error")
+ }
+ if _, err := RowFieldDecoderForType(reflect.TypeOf(func() {})); err ==
nil {
+ t.Errorf("RowFieldDecoderForType(func) = nil error, want error")
+ }
+}
+
+// A provider for a value type also encodes and decodes a top level pointer to
+// that type.
+func TestRowCoder_TopLevelPointerWithValueProvider(t *testing.T) {
+ type tenths struct{ V int64 }
+ rt := reflect.TypeOf(tenths{})
+ var encB RowEncoderBuilder
+ encB.Register(rt, func(reflect.Type) (func(any, io.Writer) error,
error) {
+ return func(v any, w io.Writer) error { return
EncodeVarInt(v.(tenths).V*10, w) }, nil
+ })
+ var decB RowDecoderBuilder
+ decB.Register(rt, func(reflect.Type) (func(io.Reader) (any, error),
error) {
+ return func(r io.Reader) (any, error) {
+ v, err := DecodeVarInt(r)
+ return tenths{V: v / 10}, err
+ }, nil
+ })
+ prt := reflect.PointerTo(rt)
+ enc, err := encB.Build(prt)
+ if err != nil {
+ t.Fatalf("RowEncoderBuilder.Build(%v) = %v, want nil error",
prt, err)
+ }
+ dec, err := decB.Build(prt)
+ if err != nil {
+ t.Fatalf("RowDecoderBuilder.Build(%v) = %v, want nil error",
prt, err)
+ }
+ want := &tenths{V: 42}
+ var buf bytes.Buffer
+ if err := enc(want, &buf); err != nil {
+ t.Fatalf("enc(%v) = %v, want nil error", want, err)
+ }
+ got, err := dec(&buf)
+ if err != nil {
+ t.Fatalf("dec = %v, want nil error", err)
+ }
+ if d := cmp.Diff(want, got); d != "" {
+ t.Errorf("dec(enc(%v)) diff (-want, +got): %v", want, d)
+ }
+}
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
b/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
index ee08bbb8f54..bf54fddbc72 100644
--- a/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
@@ -17,9 +17,11 @@ package schema
import (
"fmt"
+ "io"
"reflect"
"sync"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
)
@@ -32,7 +34,9 @@ var (
// A logical type is a type that has distinct representations and storage.
//
// RegisterLogicalType will panic if the storage type of the LogicalType
-// instance is not a valid storage type.
+// instance is not a valid storage type, or if its Go type or its identifier
+// and argument are already registered for a different logical type.
+// Registering the same logical type again has no effect.
func RegisterLogicalType(lt LogicalType) {
defaultRegistry.RegisterLogicalType(lt)
}
@@ -94,7 +98,8 @@ func NewRegistry() *Registry {
}
}
-// RegisterLogicalType a single logical type.
+// RegisterLogicalType registers a single logical type. See the package level
+// RegisterLogicalType for the conditions it panics on.
func (r *Registry) RegisterLogicalType(lt LogicalType) {
// Validates that the storage type has known handling.
st := lt.StorageType()
@@ -113,7 +118,12 @@ func (r *Registry) RegisterLogicalType(lt LogicalType) {
panic(fmt.Sprintf("LogicalType[%v] has the argument %v
which does not compare equal to itself", lt.ID(), lt.argV))
}
}
- // TODO add duplication checks.
+ if prev, ok := r.logicalTypeByGoType[lt.GoType()]; ok &&
!prev.equals(lt) {
+ panic(fmt.Sprintf("Go type %v is already registered as %v ->
%v, cannot register it as %v -> %v", lt.GoType(), prev.describe(),
prev.StorageType(), lt.describe(), lt.StorageType()))
+ }
+ if prev, ok := r.logicalTypes[lt.key()]; ok && !prev.equals(lt) {
+ panic(fmt.Sprintf("%v is already registered for Go type %v,
cannot register it for Go type %v", lt.describe(), prev.GoType(), lt.GoType()))
+ }
r.logicalTypeByGoType[lt.GoType()] = lt
r.logicalTypes[lt.key()] = lt
}
@@ -213,6 +223,21 @@ func (l *LogicalType) key() logicalTypeKey {
return logicalTypeKey{identifier: l.identifier, argument:
l.argV.Interface()}
}
+// equals reports whether l and o have the same identifier, argument, Go type
+// and storage type.
+func (l *LogicalType) equals(o LogicalType) bool {
+ return l.goT == o.goT && l.storageT == o.storageT && l.argT == o.argT
&& l.key() == o.key()
+}
+
+// describe returns the identifier of the logical type and its argument, if
+// any, for messages.
+func (l *LogicalType) describe() string {
+ if l.argT == nil {
+ return fmt.Sprintf("LogicalType[%v]", l.identifier)
+ }
+ return fmt.Sprintf("LogicalType[%v, %v]", l.identifier,
l.argV.Interface())
+}
+
func preRegLogicalTypes(r *Registry) {
r.RegisterLogicalType(ToLogicalType("int", reflectx.Int,
reflectx.Int64))
r.RegisterLogicalType(ToLogicalType("int8", reflectx.Int8,
reflectx.Int64))
@@ -224,4 +249,115 @@ func preRegLogicalTypes(r *Registry) {
func init() {
preRegLogicalTypes(defaultRegistry)
+ registerStandardLogicalTypes(defaultRegistry)
+}
+
+// RegisterLogicalTypeConversion registers a portable logical type identified
+// by urn. Values of GoT are stored in schema rows as values of StorageT,
+// converted with toStorage and fromStorage, and encoded with the row field
+// encoding of StorageT. StorageT must have a schema representation: an atomic
+// type, a struct of such types, or a registered logical type. GoT must be a
+// concrete type. A struct kind GoT with a non row StorageT can be used as a
+// field of a row, but not as a top level schema type.
+//
+// RegisterLogicalTypeConversion must be called before beam.Init(), and
+// conventionally is called in a package init() function.
+func RegisterLogicalTypeConversion[GoT, StorageT any](urn string, toStorage
func(GoT) (StorageT, error), fromStorage func(StorageT) (GoT, error)) {
+ registerLogicalTypeConversion(defaultRegistry, ToLogicalType(urn,
typeOf[GoT](), typeOf[StorageT]()), toStorage, fromStorage)
+}
+
+// RegisterParameterizedLogicalTypeConversion is RegisterLogicalTypeConversion
+// for a logical type with an argument, such as a timestamp and its precision.
+// Each argument value is registered with its own Go type, and a schema field
+// with the urn and that argument maps to GoT. See ToLogicalTypeWithArgument
+// for the supported argument types.
+func RegisterParameterizedLogicalTypeConversion[GoT, StorageT any](urn string,
argument any, toStorage func(GoT) (StorageT, error), fromStorage func(StorageT)
(GoT, error)) {
+ registerLogicalTypeConversion(defaultRegistry,
ToLogicalTypeWithArgument(urn, typeOf[GoT](), typeOf[StorageT](), argument),
toStorage, fromStorage)
+}
+
+// RegisterLogicalTypeCoder registers a portable logical type identified by
+// urn whose wire encoding differs from the row field encoding of StorageT.
+// StorageT is only used as the schema representation of the logical type.
+// enc and dec produce and consume the encoded bytes of a value directly, in
+// the wire encoding that the urn defines. GoT must be a concrete type.
+// StorageT cannot be inferred, so both type parameters are given explicitly,
+// for example
+// RegisterLogicalTypeCoder[Money, []byte]("example:money:v1", encodeMoney,
decodeMoney).
+//
+// RegisterLogicalTypeCoder must be called before beam.Init(), and
+// conventionally is called in a package init() function.
+func RegisterLogicalTypeCoder[GoT, StorageT any](urn string, enc func(GoT,
io.Writer) error, dec func(io.Reader) (GoT, error)) {
+ registerLogicalTypeCoder(defaultRegistry, ToLogicalType(urn,
typeOf[GoT](), typeOf[StorageT]()), enc, dec)
+}
+
+func typeOf[T any]() reflect.Type {
+ return reflect.TypeOf((*T)(nil)).Elem()
+}
+
+// registerLogicalTypeConversion registers lt with the registry and registers
+// row coder providers for its Go type, built from the row field coder of the
+// storage type and the conversion functions.
+func registerLogicalTypeConversion[GoT, StorageT any](r *Registry, lt
LogicalType, toStorage func(GoT) (StorageT, error), fromStorage func(StorageT)
(GoT, error)) {
+ st := lt.StorageType()
+ checkLogicalTypeKinds(lt)
+ r.RegisterLogicalType(lt)
+ coder.RegisterSchemaProviders(lt.GoType(),
+ func(reflect.Type) (func(any, io.Writer) error, error) {
+ enc, err := coder.RowFieldEncoderForType(st)
+ if err != nil {
+ return nil, err
+ }
+ return func(v any, w io.Writer) error {
+ s, err := toStorage(v.(GoT))
+ if err != nil {
+ return err
+ }
+ return enc(s, w)
+ }, nil
+ },
+ func(reflect.Type) (func(io.Reader) (any, error), error) {
+ dec, err := coder.RowFieldDecoderForType(st)
+ if err != nil {
+ return nil, err
+ }
+ return func(rd io.Reader) (any, error) {
+ s, err := dec(rd)
+ if err != nil {
+ return nil, err
+ }
+ return fromStorage(s.(StorageT))
+ }, nil
+ })
+}
+
+// registerLogicalTypeCoder registers lt with the registry and registers enc
+// and dec as the row coder providers for its Go type.
+func registerLogicalTypeCoder[GoT any](r *Registry, lt LogicalType, enc
func(GoT, io.Writer) error, dec func(io.Reader) (GoT, error)) {
+ checkLogicalTypeKinds(lt)
+ r.RegisterLogicalType(lt)
+ coder.RegisterSchemaProviders(lt.GoType(),
+ func(reflect.Type) (func(any, io.Writer) error, error) {
+ return func(v any, w io.Writer) error {
+ return enc(v.(GoT), w)
+ }, nil
+ },
+ func(reflect.Type) (func(io.Reader) (any, error), error) {
+ return func(rd io.Reader) (any, error) {
+ return dec(rd)
+ }, nil
+ })
+}
+
+// checkLogicalTypeKinds panics when the Go type of lt is an interface or its
+// storage type is a pointer. The row coder matches an interface by all of its
+// implementations, and the schema matches only the interface itself. The row
+// coders decide nullability from the Go type of a field, so a pointer storage
+// type behind a non pointer Go type would fail at encoding time.
+func checkLogicalTypeKinds(lt LogicalType) {
+ if lt.GoType().Kind() == reflect.Interface {
+ panic(fmt.Sprintf("LogicalType[%v] has the interface Go type
%v, Go types must be concrete", lt.ID(), lt.GoType()))
+ }
+ if lt.StorageType().Kind() == reflect.Ptr {
+ panic(fmt.Sprintf("LogicalType[%v] has the pointer storage type
%v, storage types must not be pointers", lt.ID(), lt.StorageType()))
+ }
}
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/schema_test.go
b/sdks/go/pkg/beam/core/runtime/graphx/schema/schema_test.go
index 08916e3898d..09caa1417e7 100644
--- a/sdks/go/pkg/beam/core/runtime/graphx/schema/schema_test.go
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/schema_test.go
@@ -1163,3 +1163,33 @@ func TestRegisterLogicalType_Validation(t *testing.T) {
})
}
}
+
+func TestRegisterLogicalType_Duplicate(t *testing.T) {
+ reg := NewRegistry()
+ lt := ToLogicalType("schema.anyString", anyStringType, reflectx.String)
+ reg.RegisterLogicalType(lt)
+ // Registering the same logical type again has no effect.
+ reg.RegisterLogicalType(lt)
+
+ for name, dup := range map[string]LogicalType{
+ "identifier for another Go type":
ToLogicalType("schema.anyString", parameterizedStringType, reflectx.String),
+ "Go type with another identifier":
ToLogicalType("schema.otherString", anyStringType, reflectx.String),
+ "Go type with another storage type":
ToLogicalType("schema.anyString", anyStringType, reflectx.ByteSlice),
+ "Go type with an argument":
ToLogicalTypeWithArgument("schema.anyString", anyStringType, reflectx.String,
int32(5)),
+ } {
+ t.Run(name, func(t *testing.T) {
+ defer func() {
+ if recover() == nil {
+ t.Errorf("RegisterLogicalType did not
panic for a duplicate %v", name)
+ }
+ }()
+ reg.RegisterLogicalType(dup)
+ })
+ }
+ if got, ok := reg.lookupLogicalType("schema.anyString",
reflect.Value{}); !ok || got.GoType() != anyStringType {
+ t.Errorf("lookupLogicalType(schema.anyString) = %v, %v, want
%v, true", got.GoType(), ok, anyStringType)
+ }
+ if _, ok := reg.logicalTypeByGoType[parameterizedStringType]; ok {
+ t.Errorf("logicalTypeByGoType has %v after a rejected
registration", parameterizedStringType)
+ }
+}
diff --git
a/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go
b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go
new file mode 100644
index 00000000000..dcc12a45e69
--- /dev/null
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go
@@ -0,0 +1,156 @@
+// 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 schema
+
+import (
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
+)
+
+// URNs of the standard logical types defined in schema.proto.
+const (
+ // URNDate identifies the Date logical type.
+ URNDate = "beam:logical_type:date:v1"
+ // URNMicrosInstant identifies the MicrosInstant logical type.
+ URNMicrosInstant = "beam:logical_type:micros_instant:v1"
+)
+
+// Date is a calendar date without a time zone or time of day, the
+// beam:logical_type:date:v1 logical type. It is stored as the number of days
+// since 1970-01-01 in an INT64 field. The zero Date is the epoch, 1970-01-01.
+// Time and String normalize out of range fields as time.Date does. Encoding
+// fails unless the fields form a valid date with a year from -999999999 to
+// 999999999, and decoding fails for a day count outside that range.
+type Date struct {
+ Year int
+ Month time.Month
+ Day int
+}
+
+// DateOf returns the Date of t in t's location.
+func DateOf(t time.Time) Date {
+ y, m, d := t.Date()
+ return Date{Year: y, Month: m, Day: d}
+}
+
+// epochDate is the date that the zero Date stands for.
+var epochDate = Date{Year: 1970, Month: time.January, Day: 1}
+
+// Time returns the start of the date in UTC.
+func (d Date) Time() time.Time {
+ if d == (Date{}) {
+ d = epochDate
+ }
+ return time.Date(d.Year, d.Month, d.Day, 0, 0, 0, 0, time.UTC)
+}
+
+// String returns the date in ISO 8601 format, such as 2006-01-02.
+func (d Date) String() string {
+ return d.Time().Format("2006-01-02")
+}
+
+const secondsPerDay = 24 * 60 * 60
+
+// minDateDays and maxDateDays bound the supported days since the epoch.
+var (
+ minDateDays = time.Date(-999999999, time.January, 1, 0, 0, 0, 0,
time.UTC).Unix() / secondsPerDay
+ maxDateDays = time.Date(999999999, time.December, 31, 0, 0, 0, 0,
time.UTC).Unix() / secondsPerDay
+)
+
+func dateToStorage(d Date) (int64, error) {
+ if d == (Date{}) {
+ d = epochDate
+ }
+ t := d.Time()
+ // Out of range fields can overflow in time.Date, so only a date that
+ // round trips is encoded.
+ if DateOf(t) != d {
+ return 0, errors.Errorf("%d-%02d-%02d is not a valid date",
d.Year, d.Month, d.Day)
+ }
+ days := t.Unix() / secondsPerDay
+ if days < minDateDays || days > maxDateDays {
+ return 0, errors.Errorf("date %v is out of range", d)
+ }
+ return days, nil
+}
+
+func dateFromStorage(days int64) (Date, error) {
+ if days < minDateDays || days > maxDateDays {
+ return Date{}, errors.Errorf("%v days since the epoch is out of
the supported date range", days)
+ }
+ return DateOf(time.Unix(days*secondsPerDay, 0).UTC()), nil
+}
+
+// MicrosInstant is a timestamp with microsecond precision, the
+// beam:logical_type:micros_instant:v1 logical type. It is stored as a row of
+// the seconds and the microseconds since the epoch. Encoding a value with
+// sub microsecond precision fails; truncate such values with
time.Time.Truncate
+// first.
+type MicrosInstant time.Time
+
+// Time returns the instant as a time.Time.
+func (m MicrosInstant) Time() time.Time {
+ return time.Time(m)
+}
+
+// String returns the instant in RFC 3339 format with microseconds, such as
+// 2020-08-13T14:14:14.123456Z.
+func (m MicrosInstant) String() string {
+ return time.Time(m).Format("2006-01-02T15:04:05.000000Z07:00")
+}
+
+// Equals reports whether m and o are the same instant.
+func (m MicrosInstant) Equals(o MicrosInstant) bool {
+ return time.Time(m).Equal(time.Time(o))
+}
+
+// microsInstantStorage is the storage type of MicrosInstant.
+type microsInstantStorage struct {
+ Seconds int64 `beam:"seconds"`
+ Micros int64 `beam:"micros"`
+}
+
+func microsInstantToStorage(m MicrosInstant) (microsInstantStorage, error) {
+ t := time.Time(m)
+ nanos := t.Nanosecond()
+ if nanos%1000 != 0 {
+ return microsInstantStorage{}, errors.Errorf("MicrosInstant %v
has sub microsecond precision", t)
+ }
+ return microsInstantStorage{Seconds: t.Unix(), Micros: int64(nanos /
1000)}, nil
+}
+
+func microsInstantFromStorage(s microsInstantStorage) (MicrosInstant, error) {
+ if s.Micros < 0 || s.Micros > 999999 {
+ return MicrosInstant{}, errors.Errorf("MicrosInstant micros %v
is out of range [0, 999999]", s.Micros)
+ }
+ return MicrosInstant(time.Unix(s.Seconds, s.Micros*1000).UTC()), nil
+}
+
+// registerStandardLogicalTypes registers the standard logical types with the
+// registry.
+func registerStandardLogicalTypes(r *Registry) {
+ registerLogicalTypeConversion(r, ToLogicalType(URNDate, typeOf[Date](),
reflectx.Int64), dateToStorage, dateFromStorage)
+ registerLogicalTypeConversion(r, ToLogicalType(URNMicrosInstant,
typeOf[MicrosInstant](), typeOf[microsInstantStorage]()),
microsInstantToStorage, microsInstantFromStorage)
+}
+
+func init() {
+ // Pipeline graphs serialize the standard logical types by name.
+ runtime.RegisterType(typeOf[Date]())
+ runtime.RegisterType(typeOf[MicrosInstant]())
+}
diff --git
a/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go
b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go
new file mode 100644
index 00000000000..88b86be1248
--- /dev/null
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go
@@ -0,0 +1,421 @@
+// 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 schema
+
+import (
+ "bytes"
+ "encoding/binary"
+ "io"
+ "math"
+ "reflect"
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+ pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
+ "github.com/google/go-cmp/cmp"
+ "google.golang.org/protobuf/encoding/prototext"
+ "google.golang.org/protobuf/testing/protocmp"
+)
+
+func atomicType(at pipepb.AtomicType) *pipepb.FieldType {
+ return &pipepb.FieldType{TypeInfo:
&pipepb.FieldType_AtomicType{AtomicType: at}}
+}
+
+func logicalFieldType(urn string, representation *pipepb.FieldType)
*pipepb.FieldType {
+ return &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_LogicalType{
+ LogicalType: &pipepb.LogicalType{
+ Urn: urn,
+ Representation: representation,
+ },
+ },
+ }
+}
+
+// standardRow uses the standard logical types as fields, with the same field
+// names as the micros_instant case of standard_coders.yaml.
+type standardRow struct {
+ F_timestamp MicrosInstant `beam:"f_timestamp"`
+ F_string string `beam:"f_string"`
+ F_int int64 `beam:"f_int"`
+ F_date Date `beam:"f_date"`
+}
+
+func TestStandardLogicalTypes_Schema(t *testing.T) {
+ rt := reflect.TypeOf(standardRow{})
+ want := &pipepb.Schema{
+ Fields: []*pipepb.Field{
+ {Name: "f_timestamp", Type:
logicalFieldType(URNMicrosInstant, &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_RowType{RowType:
&pipepb.RowType{Schema: &pipepb.Schema{
+ Fields: []*pipepb.Field{
+ {Name: "seconds", Type:
atomicType(pipepb.AtomicType_INT64)},
+ {Name: "micros", Type:
atomicType(pipepb.AtomicType_INT64)},
+ },
+ }}},
+ })},
+ {Name: "f_string", Type:
atomicType(pipepb.AtomicType_STRING)},
+ {Name: "f_int", Type:
atomicType(pipepb.AtomicType_INT64)},
+ {Name: "f_date", Type: logicalFieldType(URNDate,
atomicType(pipepb.AtomicType_INT64))},
+ },
+ }
+ got, err := FromType(rt)
+ if err != nil {
+ t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+ }
+ if d := cmp.Diff(want, got, protocmp.Transform(),
protocmp.IgnoreFields(&pipepb.Schema{}, "id")); d != "" {
+ t.Errorf("FromType(%v) diff (-want, +got): %v", rt, d)
+ }
+ // A schema from another SDK has a different id, and must still map to
the
+ // standard Go types.
+ want.Id = "standard-logical-types-test"
+ gotType, err := ToType(want)
+ if err != nil {
+ t.Fatalf("ToType(%v) = %v, want nil error",
prototext.Format(want), err)
+ }
+ for i, wantField := range []reflect.Type{typeOf[MicrosInstant](),
typeOf[string](), typeOf[int64](), typeOf[Date]()} {
+ if got := gotType.Field(i).Type; got != wantField {
+ t.Errorf("ToType field %d type = %v, want %v", i, got,
wantField)
+ }
+ }
+}
+
+func TestStandardLogicalTypes_RowEncoding(t *testing.T) {
+ // The bytes are the micros_instant example of standard_coders.yaml,
with a
+ // trailing date field appended: 2020-08-13 is 18487 days since the
epoch.
+ want := standardRow{
+ F_timestamp: MicrosInstant(time.Date(2020, 8, 13, 14, 14, 14,
123456000, time.UTC)),
+ F_string: "2020-08-13T14:14:14.123456Z",
+ F_int: 1597328054123456,
+ F_date: Date{Year: 2020, Month: time.August, Day: 13},
+ }
+ wantBytes :=
[]byte("\x04\x00\x02\x00\xb6\x95\xd5\xf9\x05\xc0\xc4\x07\x1b2020-08-13T14:14:14.123456Z\xc0\xf7\x85\xda\xae\x98\xeb\x02\xb7\x90\x01")
+ rt := reflect.TypeOf(want)
+ enc, err := coder.RowEncoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowEncoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ var buf bytes.Buffer
+ if err := enc(want, &buf); err != nil {
+ t.Fatalf("enc(%v) = %v, want nil error", want, err)
+ }
+ if got := buf.Bytes(); !bytes.Equal(got, wantBytes) {
+ t.Fatalf("enc(%v) = %q, want %q", want, got, wantBytes)
+ }
+ dec, err := coder.RowDecoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ got, err := dec(bytes.NewBuffer(wantBytes))
+ if err != nil {
+ t.Fatalf("dec(%q) = %v, want nil error", wantBytes, err)
+ }
+ if d := cmp.Diff(want, got, cmp.Comparer(MicrosInstant.Equals)); d !=
"" {
+ t.Errorf("dec(enc(%v)) diff (-want, +got): %v", want, d)
+ }
+}
+
+func TestMicrosInstant_SubMicrosecondPrecision(t *testing.T) {
+ enc, err := coder.RowEncoderForStruct(reflect.TypeOf(standardRow{}))
+ if err != nil {
+ t.Fatalf("RowEncoderForStruct = %v, want nil error", err)
+ }
+ row := standardRow{F_timestamp: MicrosInstant(time.Unix(0, 1500))}
+ if err := enc(row, &bytes.Buffer{}); err == nil {
+ t.Errorf("enc(%v) = nil error, want error for sub microsecond
precision", row)
+ }
+}
+
+func TestDate_Storage(t *testing.T) {
+ tests := []struct {
+ date Date
+ days int64
+ }{
+ {Date{1970, time.January, 1}, 0},
+ {Date{1969, time.December, 31}, -1},
+ {Date{2020, time.August, 13}, 18487},
+ {Date{1600, time.February, 29}, -135081},
+ {Date{9999, time.December, 31}, 2932896},
+ }
+ for _, test := range tests {
+ days, err := dateToStorage(test.date)
+ if err != nil {
+ t.Fatalf("dateToStorage(%v) = %v, want nil error",
test.date, err)
+ }
+ if days != test.days {
+ t.Errorf("dateToStorage(%v) = %v, want %v", test.date,
days, test.days)
+ }
+ date, err := dateFromStorage(test.days)
+ if err != nil {
+ t.Fatalf("dateFromStorage(%v) = %v, want nil error",
test.days, err)
+ }
+ if date != test.date {
+ t.Errorf("dateFromStorage(%v) = %v, want %v",
test.days, date, test.date)
+ }
+ }
+ if got, want := DateOf(time.Date(2020, 8, 13, 23, 59, 59, 0,
time.UTC)).String(), "2020-08-13"; got != want {
+ t.Errorf("DateOf(...).String() = %v, want %v", got, want)
+ }
+}
+
+func TestRegisterLogicalTypeConversion(t *testing.T) {
+ type celsius float64
+ type reading struct {
+ Temperature celsius
+ }
+ // Registering with the default registry mirrors what users do in
init().
+ RegisterLogicalTypeConversion[celsius,
float64]("schema.test:celsius:v1",
+ func(c celsius) (float64, error) { return float64(c) + 273.15,
nil },
+ func(k float64) (celsius, error) { return celsius(k - 273.15),
nil })
+
+ rt := reflect.TypeOf(reading{})
+ schm, err := FromType(rt)
+ if err != nil {
+ t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+ }
+ wantField := logicalFieldType("schema.test:celsius:v1",
atomicType(pipepb.AtomicType_DOUBLE))
+ if d := cmp.Diff(wantField, schm.GetFields()[0].GetType(),
protocmp.Transform()); d != "" {
+ t.Errorf("FromType(%v) field diff (-want, +got): %v", rt, d)
+ }
+
+ enc, err := coder.RowEncoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowEncoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ var buf bytes.Buffer
+ if err := enc(reading{Temperature: 10}, &buf); err != nil {
+ t.Fatalf("enc = %v, want nil error", err)
+ }
+ // 283.15 as a big endian double, after the row header.
+ wantBytes := []byte{0x01, 0x00, 0x40, 0x71, 0xb2, 0x66, 0x66, 0x66,
0x66, 0x66}
+ if got := buf.Bytes(); !bytes.Equal(got, wantBytes) {
+ t.Fatalf("enc(reading{10}) = %v, want %v", got, wantBytes)
+ }
+ dec, err := coder.RowDecoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ got, err := dec(&buf)
+ if err != nil {
+ t.Fatalf("dec = %v, want nil error", err)
+ }
+ if d := cmp.Diff(reading{Temperature: 10}, got, cmp.Comparer(func(a, b
celsius) bool { return a-b < 1e-9 && b-a < 1e-9 })); d != "" {
+ t.Errorf("dec(enc(reading{10})) diff (-want, +got): %v", d)
+ }
+}
+
+// fixedCode is a user logical type with an argument, registered with the
+// exported parameterized registration.
+type fixedCode string
+
+func TestRegisterParameterizedLogicalTypeConversion(t *testing.T) {
+ RegisterParameterizedLogicalTypeConversion[fixedCode,
string]("schema.test:fixed_code:v1", int32(4),
+ func(c fixedCode) (string, error) { return string(c), nil },
+ func(s string) (fixedCode, error) { return fixedCode(s), nil })
+ rt := reflect.TypeOf(struct{ Code fixedCode }{})
+ schm, err := FromType(rt)
+ if err != nil {
+ t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+ }
+ want := logicalFieldType("schema.test:fixed_code:v1",
atomicType(pipepb.AtomicType_STRING))
+ want.GetLogicalType().ArgumentType = int32FieldType()
+ want.GetLogicalType().Argument = int32FieldValue(4)
+ if d := cmp.Diff(want, schm.GetFields()[0].GetType(),
protocmp.Transform()); d != "" {
+ t.Errorf("FromType(%v) field diff (-want, +got): %v", rt, d)
+ }
+ schm.Id = "fixed-code-from-elsewhere"
+ got, err := ToType(schm)
+ if err != nil {
+ t.Fatalf("ToType = %v, want nil error", err)
+ }
+ if ft := got.Field(0).Type; ft != typeOf[fixedCode]() {
+ t.Errorf("ToType field type = %v, want %v", ft,
typeOf[fixedCode]())
+ }
+ enc, err := coder.RowEncoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowEncoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ var buf bytes.Buffer
+ if err := enc(struct{ Code fixedCode }{"ABCD"}, &buf); err != nil {
+ t.Fatalf("enc = %v, want nil error", err)
+ }
+ if got, want := buf.Bytes(), []byte("\x01\x00\x04ABCD");
!bytes.Equal(got, want) {
+ t.Errorf("enc = %q, want %q", got, want)
+ }
+}
+
+// money is a user logical type with its own wire encoding: the cents as a
+// big endian int32 inside a BYTES representation.
+type money int32
+
+func TestRegisterLogicalTypeCoder(t *testing.T) {
+ RegisterLogicalTypeCoder[money, []byte]("schema.test:money:v1",
+ func(m money, w io.Writer) error {
+ var b [4]byte
+ binary.BigEndian.PutUint32(b[:], uint32(m))
+ return coder.EncodeBytes(b[:], w)
+ },
+ func(r io.Reader) (money, error) {
+ b, err := coder.DecodeBytes(r)
+ if err != nil {
+ return 0, err
+ }
+ return money(int32(binary.BigEndian.Uint32(b))), nil
+ })
+ rt := reflect.TypeOf(struct{ Price money }{})
+ schm, err := FromType(rt)
+ if err != nil {
+ t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+ }
+ want := logicalFieldType("schema.test:money:v1",
atomicType(pipepb.AtomicType_BYTES))
+ if d := cmp.Diff(want, schm.GetFields()[0].GetType(),
protocmp.Transform()); d != "" {
+ t.Errorf("FromType(%v) field diff (-want, +got): %v", rt, d)
+ }
+ enc, err := coder.RowEncoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowEncoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ var buf bytes.Buffer
+ if err := enc(struct{ Price money }{-2}, &buf); err != nil {
+ t.Fatalf("enc = %v, want nil error", err)
+ }
+ wantBytes := []byte{0x01, 0x00, 0x04, 0xff, 0xff, 0xff, 0xfe}
+ if got := buf.Bytes(); !bytes.Equal(got, wantBytes) {
+ t.Fatalf("enc = %v, want %v", got, wantBytes)
+ }
+ dec, err := coder.RowDecoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ got, err := dec(&buf)
+ if err != nil {
+ t.Fatalf("dec = %v, want nil error", err)
+ }
+ if d := cmp.Diff(struct{ Price money }{-2}, got); d != "" {
+ t.Errorf("dec(enc) diff (-want, +got): %v", d)
+ }
+}
+
+func TestRegisterLogicalTypeConversion_PointerStorage(t *testing.T) {
+ type celsiusPtr float64
+ defer func() {
+ if recover() == nil {
+ t.Errorf("RegisterLogicalTypeConversion with a pointer
storage type did not panic")
+ }
+ }()
+ RegisterLogicalTypeConversion[celsiusPtr,
*float64]("schema.test:celsius_ptr:v1",
+ func(c celsiusPtr) (*float64, error) { f := float64(c); return
&f, nil },
+ func(f *float64) (celsiusPtr, error) { return celsiusPtr(*f),
nil })
+}
+
+func TestRegisterLogicalType_InterfaceGoType(t *testing.T) {
+ type textual interface{ Text() string }
+ t.Run("conversion", func(t *testing.T) {
+ defer func() {
+ if recover() == nil {
+ t.Errorf("RegisterLogicalTypeConversion with an
interface Go type did not panic")
+ }
+ }()
+ RegisterLogicalTypeConversion[textual,
string]("schema.test:textual:v1",
+ func(v textual) (string, error) { return v.Text(), nil
},
+ func(s string) (textual, error) { return nil, nil })
+ })
+ t.Run("coder", func(t *testing.T) {
+ defer func() {
+ if recover() == nil {
+ t.Errorf("RegisterLogicalTypeCoder with an
interface Go type did not panic")
+ }
+ }()
+ RegisterLogicalTypeCoder[textual,
string]("schema.test:textual_coder:v1",
+ func(v textual, w io.Writer) error { return nil },
+ func(r io.Reader) (textual, error) { return nil, nil })
+ })
+}
+
+func TestDate_StorageRange(t *testing.T) {
+ for _, days := range []int64{minDateDays, maxDateDays} {
+ date, err := dateFromStorage(days)
+ if err != nil {
+ t.Fatalf("dateFromStorage(%v) = %v, want nil error",
days, err)
+ }
+ if got, err := dateToStorage(date); err != nil || got != days {
+ t.Errorf("dateToStorage(%v) = %v, %v, want %v, nil",
date, got, err, days)
+ }
+ }
+ for _, days := range []int64{minDateDays - 1, maxDateDays + 1, 1 << 62,
math.MinInt64} {
+ if date, err := dateFromStorage(days); err == nil {
+ t.Errorf("dateFromStorage(%v) = %v, want error", days,
date)
+ }
+ }
+ for _, date := range []Date{
+ {-1000000000, time.December, 31},
+ {1000000000, time.January, 1},
+ {2020, time.February, 30},
+ {1970, time.January, math.MaxInt/4 + 2},
+ {math.MaxInt / 2, time.January, 1},
+ } {
+ if days, err := dateToStorage(date); err == nil {
+ t.Errorf("dateToStorage(%d-%02d-%02d) = %v, want
error", date.Year, date.Month, date.Day, days)
+ }
+ }
+}
+
+func TestMicrosInstant_StorageRange(t *testing.T) {
+ for _, micros := range []int64{-1, 1000000, 1 << 61} {
+ s := microsInstantStorage{Seconds: 0, Micros: micros}
+ if m, err := microsInstantFromStorage(s); err == nil {
+ t.Errorf("microsInstantFromStorage(%+v) = %v, want
error", s, m.Time())
+ }
+ }
+ s := microsInstantStorage{Seconds: -1, Micros: 999999}
+ m, err := microsInstantFromStorage(s)
+ if err != nil {
+ t.Fatalf("microsInstantFromStorage(%+v) = %v, want nil error",
s, err)
+ }
+ if want := time.Unix(-1, 999999000); !m.Time().Equal(want) {
+ t.Errorf("microsInstantFromStorage(%+v) = %v, want %v", s,
m.Time(), want)
+ }
+}
+
+func TestDate_Zero(t *testing.T) {
+ var d Date
+ if got, want := d.Time(), time.Date(1970, time.January, 1, 0, 0, 0, 0,
time.UTC); !got.Equal(want) {
+ t.Errorf("Date{}.Time() = %v, want %v", got, want)
+ }
+ if got, want := d.String(), "1970-01-01"; got != want {
+ t.Errorf("Date{}.String() = %v, want %v", got, want)
+ }
+ if days, err := dateToStorage(d); err != nil || days != 0 {
+ t.Errorf("dateToStorage(Date{}) = %v, %v, want 0, nil", days,
err)
+ }
+}
+
+func TestMicrosInstant_StringAndEquals(t *testing.T) {
+ utc := MicrosInstant(time.Date(2020, time.August, 13, 14, 14, 14,
123456000, time.UTC))
+ if got, want := utc.String(), "2020-08-13T14:14:14.123456Z"; got !=
want {
+ t.Errorf("String() = %v, want %v", got, want)
+ }
+ zoned := MicrosInstant(utc.Time().In(time.FixedZone("UTC+2", 2*60*60)))
+ if got, want := zoned.String(), "2020-08-13T16:14:14.123456+02:00"; got
!= want {
+ t.Errorf("String() = %v, want %v", got, want)
+ }
+ if !utc.Equals(zoned) {
+ t.Errorf("%v.Equals(%v) = false, want true", utc, zoned)
+ }
+ if later := MicrosInstant(utc.Time().Add(time.Microsecond));
utc.Equals(later) {
+ t.Errorf("%v.Equals(%v) = true, want false", utc, later)
+ }
+}
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/serialize_test.go
b/sdks/go/pkg/beam/core/runtime/graphx/serialize_test.go
index 265d936a71e..a6aceee4409 100644
--- a/sdks/go/pkg/beam/core/runtime/graphx/serialize_test.go
+++ b/sdks/go/pkg/beam/core/runtime/graphx/serialize_test.go
@@ -23,6 +23,7 @@ import (
"testing"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/schema"
v1pb "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/v1"
)
@@ -88,4 +89,19 @@ func TestEncodeType(t *testing.T) {
t.Errorf("expected error about wrapping in a struct,
got %q", err.Error())
}
})
+ t.Run("StandardLogicalTypes", func(t *testing.T) {
+ rt := reflect.TypeOf(struct {
+ D schema.Date
+ M schema.MicrosInstant
+ }{})
+ pbT, err := encodeType(rt)
+ if err != nil {
+ t.Fatalf("got error = %v, want nil", err)
+ }
+ for _, f := range pbT.GetFields() {
+ if got, want := f.GetType().GetKind(),
v1pb.Type_EXTERNAL; got != want {
+ t.Errorf("got field %v kind == %v, want %v",
f.GetName(), got, want)
+ }
+ }
+ })
}
diff --git a/sdks/go/test/regression/coders/fromyaml/fromyaml.go
b/sdks/go/test/regression/coders/fromyaml/fromyaml.go
index 466f361b013..08000db877d 100644
--- a/sdks/go/test/regression/coders/fromyaml/fromyaml.go
+++ b/sdks/go/test/regression/coders/fromyaml/fromyaml.go
@@ -29,16 +29,19 @@ import (
"runtime/debug"
"strconv"
"strings"
+ "time"
"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/runtime/exec"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/schema"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
"github.com/google/go-cmp/cmp"
"golang.org/x/text/encoding/charmap"
+ "google.golang.org/protobuf/proto"
yaml "gopkg.in/yaml.v2"
)
@@ -49,11 +52,11 @@ var unimplementedCoders = map[string]bool{
}
var filteredCases = []struct{ filter, reason string }{
- {"logical", "BEAM-9615: Support logical types"},
{"30ea5a25-dcd8-4cdb-abeb-5332d15ab4b9",
"https://github.com/apache/beam/issues/21206: Support encoding position."},
- {"80be749a-5700-4ede-89d8-dd9a4433a3f8",
"https://github.com/apache/beam/issues/19817: Support millis_instant."},
- {"800c44ae-a1b7-4def-bbf6-6217cca89ec4",
"https://github.com/apache/beam/issues/19817: Support decimal."},
- {"f0ffb3a4-f46f-41ca-a942-85e3e939452a",
"https://github.com/apache/beam/issues/23526: Support char/varchar,
binary/varbinary."},
+ {"beam:logical_type:millis_instant:v1",
"https://github.com/apache/beam/issues/39684: Support millis_instant."},
+ {"beam:logical_type:decimal:v1",
"https://github.com/apache/beam/issues/39684: Support decimal."},
+ {"beam:logical_type:fixed_char:v1",
"https://github.com/apache/beam/issues/39684: Support char/varchar,
binary/varbinary."},
+ {"beam:logical_type:timestamp:v1",
"https://github.com/apache/beam/issues/39684: Support timestamp."},
}
// Coder is a representation a serialized beam coder.
@@ -164,6 +167,7 @@ func (s *Spec) testStandardCoder() (err error) {
}
if d := cmp.Diff(recoded, string(out.Bytes())); d != "" {
log.Printf("Encoding error: diff(-want,+got): %v\n", d)
+ encFails++
}
}
if decFails+encFails > 0 {
@@ -177,6 +181,15 @@ var cmpOpts = []cmp.Option{
cmp.Transformer("bytes2string", func(in []byte) (out string) {
return string(in)
}),
+ cmp.Comparer(func(a, b schema.MicrosInstant) bool {
+ return a.Time().Equal(b.Time())
+ }),
+}
+
+// latin1Bytes returns the bytes that a yaml string denotes. The yaml strings
+// escape bytes above 0x7f as code points, which yaml decodes to UTF-8.
+func latin1Bytes(s string) ([]byte, error) {
+ return charmap.ISO8859_1.NewEncoder().Bytes([]byte(s))
}
func diff(c Coder, elem *exec.FullValue, eg yaml.MapItem) bool {
@@ -301,29 +314,12 @@ func diff(c Coder, elem *exec.FullValue, eg yaml.MapItem)
bool {
}
return pass
case "beam:coder:row:v1":
- fs := eg.Value.(yaml.MapSlice)
- var rfs []reflect.StructField
- // There are only 2 pointer examples, but they reuse field
names,
- // so we key off the proto hash to know which example we're
handling.
- ptrEg := strings.Contains(c.Payload, "51ace21c7393")
- for _, rf := range fs {
- name := rf.Key.(string)
- t := nameToType[name]
- if ptrEg {
- t = reflect.PtrTo(t)
- }
- rfs = append(rfs, reflect.StructField{
- Name: strings.ToUpper(name[:1]) + name[1:],
- Type: t,
- Tag:
reflect.StructTag(fmt.Sprintf("beam:\"%v\"", name)),
- })
- }
- rv := reflect.New(reflect.StructOf(rfs)).Elem()
- for i, rf := range fs {
- setField(rv, i, rf.Value)
+ row, err := rowFromYAML(c.Payload, eg.Value.(yaml.MapSlice))
+ if err != nil {
+ log.Printf("unable to build the expected row: %v\n",
err)
+ return false
}
-
- got, want = elem.Elm, rv.Interface()
+ got, want = elem.Elm, row
case "beam:coder:timer:v1":
pass := true
tm := elem.Elm.(exec.TimerRecv)
@@ -411,73 +407,137 @@ func diffPane(eg any, got typex.PaneInfo) bool {
return pass
}
-// standard_coders.yaml uses the name for type indication, except for
nullability.
-var nameToType = map[string]reflect.Type{
- "str": reflectx.String,
- "i32": reflectx.Int32,
- "f64": reflectx.Float64,
- "arr": reflect.SliceOf(reflectx.String),
- "f_bool": reflectx.Bool,
- "f_bytes": reflect.PtrTo(reflectx.ByteSlice),
- "f_map": reflect.MapOf(reflectx.String,
reflect.PtrTo(reflectx.Int64)),
- "f_float": reflectx.Float32,
+// rowFromYAML builds the expected value of a beam:coder:row:v1 example from
+// the schema in the coder payload and the field values of the example.
+// The Go type is the one the Go SDK derives from the schema, so logical type
+// fields have their registered Go types.
+func rowFromYAML(payload string, fields yaml.MapSlice) (any, error) {
+ b, err := latin1Bytes(payload)
+ if err != nil {
+ return nil, err
+ }
+ var s pipepb.Schema
+ if err := proto.Unmarshal(b, &s); err != nil {
+ return nil, fmt.Errorf("unmarshalling the row schema: %v", err)
+ }
+ rt, err := schema.ToType(&s)
+ if err != nil {
+ return nil, fmt.Errorf("converting the row schema to a type:
%v", err)
+ }
+ rv := reflect.New(rt).Elem()
+ if err := setStruct(rv, fields); err != nil {
+ return nil, err
+ }
+ return rv.Interface(), nil
+}
+
+// setStruct sets the fields of the struct value rv from the yaml mapping.
+func setStruct(rv reflect.Value, fields yaml.MapSlice) error {
+ for _, f := range fields {
+ name := f.Key.(string)
+ fv, ok := fieldByName(rv, name)
+ if !ok {
+ return fmt.Errorf("no field %q in %v", name, rv.Type())
+ }
+ if err := setValue(fv, f.Value); err != nil {
+ return fmt.Errorf("field %q: %v", name, err)
+ }
+ }
+ return nil
}
-func setField(rv reflect.Value, i int, v any) {
+// fieldByName returns the field of rv with the given schema field name, which
+// is either the Go field name or the name in the beam struct tag.
+func fieldByName(rv reflect.Value, name string) (reflect.Value, bool) {
+ rt := rv.Type()
+ for i := 0; i < rt.NumField(); i++ {
+ sf := rt.Field(i)
+ if sf.Name == name {
+ return rv.Field(i), true
+ }
+ if tag, ok := sf.Tag.Lookup("beam"); ok && strings.Split(tag,
",")[0] == name {
+ return rv.Field(i), true
+ }
+ }
+ return reflect.Value{}, false
+}
+
+// setValue sets rv from a yaml example value. A nil value leaves rv at its
+// zero value, which is nil for nullable fields.
+func setValue(rv reflect.Value, v any) error {
if v == nil {
- return
+ return nil
}
- rf := rv.Field(i)
- if rf.Kind() == reflect.Ptr {
- // Ensure it's initialized.
- rf.Set(reflect.New(rf.Type().Elem()))
- rf = rf.Elem()
+ if rv.Kind() == reflect.Ptr {
+ rv.Set(reflect.New(rv.Type().Elem()))
+ rv = rv.Elem()
}
- switch rf.Kind() {
+ switch rv.Type() {
+ case reflect.TypeOf(schema.MicrosInstant{}):
+ var seconds, micros int64
+ for _, f := range v.(yaml.MapSlice) {
+ switch f.Key.(string) {
+ case "seconds":
+ seconds = int64(f.Value.(int))
+ case "micros":
+ micros = int64(f.Value.(int))
+ }
+ }
+ rv.Set(reflect.ValueOf(schema.MicrosInstant(time.Unix(seconds,
micros*1000).UTC())))
+ return nil
+ }
+ switch rv.Kind() {
case reflect.String:
- rf.SetString(v.(string))
- case reflect.Int32:
- rf.SetInt(int64(v.(int)))
+ rv.SetString(v.(string))
+ case reflect.Int16, reflect.Int32, reflect.Int64:
+ rv.SetInt(int64(v.(int)))
case reflect.Float32:
c, err := strconv.ParseFloat(v.(string), 32)
if err != nil {
- panic(err)
+ return err
}
- rf.SetFloat(c)
+ rv.SetFloat(c)
case reflect.Float64:
c, err := strconv.ParseFloat(v.(string), 64)
if err != nil {
- panic(err)
+ return err
}
- rf.SetFloat(c)
+ rv.SetFloat(c)
+ case reflect.Bool:
+ rv.SetBool(v.(bool))
case reflect.Slice:
- if rf.Type() == reflectx.ByteSlice {
- rf.Set(reflect.ValueOf([]byte(v.(string))))
- break
+ if rv.Type() == reflectx.ByteSlice {
+ rv.SetBytes([]byte(v.(string)))
+ return nil
}
- // Value is a []any with string values.
- var arr []string
- for _, a := range v.([]any) {
- arr = append(arr, a.(string))
+ items := v.([]any)
+ sv := reflect.MakeSlice(rv.Type(), len(items), len(items))
+ for i, item := range items {
+ if err := setValue(sv.Index(i), item); err != nil {
+ return err
+ }
}
- rf.Set(reflect.ValueOf(arr))
- case reflect.Bool:
- rf.SetBool(v.(bool))
+ rv.Set(sv)
case reflect.Map:
- // only f_map presently, which is always map[string]*int64
- rm := reflect.MakeMap(rf.Type())
- for _, a := range v.(yaml.MapSlice) {
- rk := reflect.ValueOf(a.Key.(string))
- rv := reflect.Zero(rf.Type().Elem())
- if a.Value != nil {
- rv = reflect.New(reflectx.Int64)
- rv.Elem().SetInt(int64(a.Value.(int)))
+ mv := reflect.MakeMap(rv.Type())
+ for _, entry := range v.(yaml.MapSlice) {
+ key := reflect.New(rv.Type().Key()).Elem()
+ if err := setValue(key, entry.Key); err != nil {
+ return err
+ }
+ val := reflect.New(rv.Type().Elem()).Elem()
+ if err := setValue(val, entry.Value); err != nil {
+ return err
}
- rm.SetMapIndex(rk, rv)
+ mv.SetMapIndex(key, val)
}
- rf.Set(rm)
-
+ rv.Set(mv)
+ case reflect.Struct:
+ return setStruct(rv, v.(yaml.MapSlice))
+ default:
+ return fmt.Errorf("unsupported field type %v", rv.Type())
}
+ return nil
}
func (s *Spec) parseCoder(c Coder) string {
@@ -486,10 +546,14 @@ func (s *Spec) parseCoder(c Coder) string {
for _, comp := range c.Components {
compIDs = append(compIDs, s.parseCoder(comp))
}
+ payload, err := latin1Bytes(c.Payload)
+ if err != nil {
+ panic(fmt.Sprintf("invalid coder payload %q: %v", c.Payload,
err))
+ }
s.coderPBs[id] = &pipepb.Coder{
Spec: &pipepb.FunctionSpec{
Urn: c.Urn,
- Payload: []byte(c.Payload),
+ Payload: payload,
},
ComponentCoderIds: compIDs,
}
