From 2689ec73bf1cd90581c43603e5b43bfb0a1bb84d Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Thu, 17 Sep 2026 22:58:51 +1000 Subject: [PATCH 1/3] [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. --- CHANGES.md | 1 + sdks/go/pkg/beam/core/graph/coder/row.go | 16 + .../pkg/beam/core/graph/coder/row_decoder.go | 37 ++ .../pkg/beam/core/graph/coder/row_encoder.go | 21 + sdks/go/pkg/beam/core/graph/coder/row_test.go | 105 +++++ .../runtime/graphx/schema/logicaltypes.go | 113 +++++ .../graphx/schema/standard_logical_types.go | 136 ++++++ .../schema/standard_logical_types_test.go | 391 ++++++++++++++++++ .../core/runtime/graphx/serialize_test.go | 16 + .../regression/coders/fromyaml/fromyaml.go | 210 ++++++---- 10 files changed, 973 insertions(+), 73 deletions(-) create mode 100644 sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go create mode 100644 sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go diff --git a/CHANGES.md b/CHANGES.md index 354f971acc6c..9b01a8e035d3 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -74,6 +74,7 @@ * (Go) Pipeline-level `--resource_hints` are now forwarded to expansion services, so they apply to cross-language transforms as they do in the Java and Python SDKs. The `max_active_bundles_per_worker` hint name is now also accepted by `--resource_hints` ([#23893](https://github.com/apache/beam/issues/23893)). * ReadFromBigQuery now supports Lakehouse runtime catalog (BigLake metastore) tables with `method=DIRECT_READ`, using 4-part `project.catalog.namespace.table` identifiers. Previously `project:catalog.namespace.table` was silently mis-parsed and tables that report no `numBytes` failed to split (Python) ([#39597](https://github.com/apache/beam/issues/39597)). * (Python) Added `AwsOptions` with `--aws_*` pipeline options (e.g. `--aws_access_key_id`, `--aws_region_name`, `--aws_endpoint_url`) for configuring AWS service clients such as S3 ([#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 f958d84154c1..0c2a6872dd69 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 90938c237fe3..2402cafe0e2b 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 6f5c4c04e81c..05d3e85b2d98 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 c2a925efd306..f7b030fb7b97 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.PtrTo(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 ee08bbb8f549..7a711d55a8a4 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" ) @@ -224,4 +226,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/standard_logical_types.go b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go new file mode 100644 index 000000000000..0deadd8817a2 --- /dev/null +++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go @@ -0,0 +1,136 @@ +// 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. 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} +} + +// Time returns the start of the date in UTC. +func (d Date) Time() time.Time { + 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) { + 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) +} + +// 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 000000000000..02a2540ab626 --- /dev/null +++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go @@ -0,0 +1,391 @@ +// 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(func(a, b MicrosInstant) bool { return a.Time().Equal(b.Time()) })); 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) + } +} 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 265d936a71ec..a6aceee4409d 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 466f361b0138..08000db877da 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, } From d0088adb075488a1403d57b39fd90aefa1104039 Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Tue, 13 Oct 2026 11:17:13 +1100 Subject: [PATCH 2/3] [Go SDK] Replace deprecated reflect.PtrTo in row coder test --- sdks/go/pkg/beam/core/graph/coder/row_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 f7b030fb7b97..8b6782457d81 100644 --- a/sdks/go/pkg/beam/core/graph/coder/row_test.go +++ b/sdks/go/pkg/beam/core/graph/coder/row_test.go @@ -1028,7 +1028,7 @@ func TestRowCoder_TopLevelPointerWithValueProvider(t *testing.T) { return tenths{V: v / 10}, err }, nil }) - prt := reflect.PtrTo(rt) + prt := reflect.PointerTo(rt) enc, err := encB.Build(prt) if err != nil { t.Fatalf("RowEncoderBuilder.Build(%v) = %v, want nil error", prt, err) From e7c4f6cb2fe8a7a618bf5b17dc6635cc91ab06e4 Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Sat, 17 Oct 2026 16:19:50 +1100 Subject: [PATCH 3/3] [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. --- .../runtime/graphx/schema/logicaltypes.go | 29 +++++++++++++++-- .../core/runtime/graphx/schema/schema_test.go | 30 +++++++++++++++++ .../graphx/schema/standard_logical_types.go | 28 +++++++++++++--- .../schema/standard_logical_types_test.go | 32 ++++++++++++++++++- 4 files changed, 111 insertions(+), 8 deletions(-) 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 7a711d55a8a4..bf54fddbc722 100644 --- a/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go +++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go @@ -34,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) } @@ -96,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() @@ -115,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 } @@ -215,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)) 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 08916e3898d0..09caa1417e74 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 index 0deadd8817a2..dcc12a45e69f 100644 --- 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 @@ -33,10 +33,10 @@ const ( // 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. 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. +// 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 @@ -49,8 +49,14 @@ func DateOf(t time.Time) 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) } @@ -68,6 +74,9 @@ var ( ) 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. @@ -100,6 +109,17 @@ 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"` 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 index 02a2540ab626..88b86be1248e 100644 --- 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 @@ -123,7 +123,7 @@ func TestStandardLogicalTypes_RowEncoding(t *testing.T) { if err != nil { t.Fatalf("dec(%q) = %v, want nil error", wantBytes, err) } - if d := cmp.Diff(want, got, cmp.Comparer(func(a, b MicrosInstant) bool { return a.Time().Equal(b.Time()) })); d != "" { + if d := cmp.Diff(want, got, cmp.Comparer(MicrosInstant.Equals)); d != "" { t.Errorf("dec(enc(%v)) diff (-want, +got): %v", want, d) } } @@ -389,3 +389,33 @@ func TestMicrosInstant_StorageRange(t *testing.T) { 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) + } +}