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 b2b8f8bb3ae [Go SDK] Convert logical type arguments to and from schema
protos (#40153)
b2b8f8bb3ae is described below
commit b2b8f8bb3aeb428a62646b310d30614cde08679f
Author: Elia Liu <[email protected]>
AuthorDate: Tue Oct 13 02:54:06 2026 +1100
[Go SDK] Convert logical type arguments to and from schema protos (#40153)
* [Go SDK] Encode INT16 row fields as big endian to match Java and Python
The reflection based row coder used the varint encoding for int16 and
uint16 fields. Java and Python encode the INT16 schema type as 2 big
endian bytes, so rows with such fields could not be exchanged with the
other SDKs.
Fixes #40151. Part of #39684.
* [Go SDK] Convert logical type arguments to and from schema protos
Logical types registered with an argument now emit argument_type and
argument in their FieldType, and schemas from other SDKs are matched on
the URN and the argument value. A logical type registered without an
argument still matches any argument, so pass through types keep
working. Top level logical types carry the argument in a schema option.
Part of #39684.
* Update CHANGES.md
restore whitespace
---------
Co-authored-by: Jack McCluskey
<[email protected]>
---
.../beam/core/runtime/graphx/schema/fieldvalue.go | 156 ++++++++++
.../core/runtime/graphx/schema/fieldvalue_test.go | 163 ++++++++++
.../core/runtime/graphx/schema/logicaltypes.go | 71 ++++-
.../pkg/beam/core/runtime/graphx/schema/schema.go | 165 ++++++++--
.../beam/core/runtime/graphx/schema/schema_test.go | 334 +++++++++++++++++++++
5 files changed, 849 insertions(+), 40 deletions(-)
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/fieldvalue.go
b/sdks/go/pkg/beam/core/runtime/graphx/schema/fieldvalue.go
new file mode 100644
index 00000000000..a8e8ffaed09
--- /dev/null
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/fieldvalue.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 (
+ "math"
+ "reflect"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
+ pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
+)
+
+// atomicFieldType returns the schema FieldType of an atomic Go type.
+// The Go type must be one of the built in types that map directly to an
+// AtomicType: bool, uint8, int16, int32, int64, float32, float64, string
+// and []byte.
+func atomicFieldType(t reflect.Type) (*pipepb.FieldType, error) {
+ var at pipepb.AtomicType
+ if t == reflectx.ByteSlice {
+ at = pipepb.AtomicType_BYTES
+ } else if enum, ok := reflectTypeToAtomicTypeMap[t.Kind()]; ok &&
reflectKindToTypeMap[t.Kind()] == t {
+ at = enum
+ } else {
+ return nil, errors.Errorf("%v is not an atomic schema type", t)
+ }
+ return &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_AtomicType{
+ AtomicType: at,
+ },
+ }, nil
+}
+
+// atomicValueToProto converts a Go value of an atomic type to its schema
+// FieldType and FieldValue. The BYTE schema type is signed, so a uint8 is
+// converted through int8.
+func atomicValueToProto(rv reflect.Value) (*pipepb.FieldType,
*pipepb.FieldValue, error) {
+ if !rv.IsValid() {
+ return nil, nil, errors.New("invalid value")
+ }
+ ft, err := atomicFieldType(rv.Type())
+ if err != nil {
+ return nil, nil, err
+ }
+ var av *pipepb.AtomicTypeValue
+ switch ft.GetAtomicType() {
+ case pipepb.AtomicType_BYTE:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Byte{Byte: int32(int8(rv.Uint()))}}
+ case pipepb.AtomicType_INT16:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Int16{Int16: int32(rv.Int())}}
+ case pipepb.AtomicType_INT32:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Int32{Int32: int32(rv.Int())}}
+ case pipepb.AtomicType_INT64:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Int64{Int64: rv.Int()}}
+ case pipepb.AtomicType_FLOAT:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Float{Float: float32(rv.Float())}}
+ case pipepb.AtomicType_DOUBLE:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Double{Double: rv.Float()}}
+ case pipepb.AtomicType_STRING:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_String_{String_: rv.String()}}
+ case pipepb.AtomicType_BOOLEAN:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Boolean{Boolean: rv.Bool()}}
+ case pipepb.AtomicType_BYTES:
+ av = &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Bytes{Bytes: rv.Bytes()}}
+ default:
+ return nil, nil, errors.Errorf("unhandled atomic type %v",
ft.GetAtomicType())
+ }
+ return ft, &pipepb.FieldValue{FieldValue:
&pipepb.FieldValue_AtomicValue{AtomicValue: av}}, nil
+}
+
+// atomicValueFromProto converts a schema FieldValue of the atomic FieldType ft
+// to a Go value of the matching built in type.
+func atomicValueFromProto(ft *pipepb.FieldType, fv *pipepb.FieldValue)
(reflect.Value, error) {
+ at, ok := ft.GetTypeInfo().(*pipepb.FieldType_AtomicType)
+ if !ok {
+ return reflect.Value{}, errors.Errorf("field type %v is not
atomic", ft)
+ }
+ t, ok := atomicTypeToReflectType[at.AtomicType]
+ if !ok {
+ return reflect.Value{}, errors.Errorf("unknown atomic type:
%v", at.AtomicType)
+ }
+ av := fv.GetAtomicValue()
+ if av == nil {
+ return reflect.Value{}, errors.Errorf("field value %v is not
atomic", fv)
+ }
+ if got := atomicValueType(av); got != at.AtomicType {
+ return reflect.Value{}, errors.Errorf("atomic value %v does not
match field type %v", av, at.AtomicType)
+ }
+ rv := reflect.New(t).Elem()
+ switch v := av.GetValue().(type) {
+ case *pipepb.AtomicTypeValue_Byte:
+ if v.Byte < math.MinInt8 || v.Byte > math.MaxInt8 {
+ return reflect.Value{}, errors.Errorf("BYTE value %v is
out of range", v.Byte)
+ }
+ rv.SetUint(uint64(uint8(v.Byte)))
+ case *pipepb.AtomicTypeValue_Int16:
+ if v.Int16 < math.MinInt16 || v.Int16 > math.MaxInt16 {
+ return reflect.Value{}, errors.Errorf("INT16 value %v
is out of range", v.Int16)
+ }
+ rv.SetInt(int64(v.Int16))
+ case *pipepb.AtomicTypeValue_Int32:
+ rv.SetInt(int64(v.Int32))
+ case *pipepb.AtomicTypeValue_Int64:
+ rv.SetInt(v.Int64)
+ case *pipepb.AtomicTypeValue_Float:
+ rv.SetFloat(float64(v.Float))
+ case *pipepb.AtomicTypeValue_Double:
+ rv.SetFloat(v.Double)
+ case *pipepb.AtomicTypeValue_String_:
+ rv.SetString(v.String_)
+ case *pipepb.AtomicTypeValue_Boolean:
+ rv.SetBool(v.Boolean)
+ case *pipepb.AtomicTypeValue_Bytes:
+ rv.SetBytes(v.Bytes)
+ }
+ return rv, nil
+}
+
+// atomicValueType returns the atomic type of the value set in av, or
+// UNSPECIFIED when no value is set.
+func atomicValueType(av *pipepb.AtomicTypeValue) pipepb.AtomicType {
+ switch av.GetValue().(type) {
+ case *pipepb.AtomicTypeValue_Byte:
+ return pipepb.AtomicType_BYTE
+ case *pipepb.AtomicTypeValue_Int16:
+ return pipepb.AtomicType_INT16
+ case *pipepb.AtomicTypeValue_Int32:
+ return pipepb.AtomicType_INT32
+ case *pipepb.AtomicTypeValue_Int64:
+ return pipepb.AtomicType_INT64
+ case *pipepb.AtomicTypeValue_Float:
+ return pipepb.AtomicType_FLOAT
+ case *pipepb.AtomicTypeValue_Double:
+ return pipepb.AtomicType_DOUBLE
+ case *pipepb.AtomicTypeValue_String_:
+ return pipepb.AtomicType_STRING
+ case *pipepb.AtomicTypeValue_Boolean:
+ return pipepb.AtomicType_BOOLEAN
+ case *pipepb.AtomicTypeValue_Bytes:
+ return pipepb.AtomicType_BYTES
+ }
+ return pipepb.AtomicType_UNSPECIFIED
+}
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/fieldvalue_test.go
b/sdks/go/pkg/beam/core/runtime/graphx/schema/fieldvalue_test.go
new file mode 100644
index 00000000000..a2c0a1c6614
--- /dev/null
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/fieldvalue_test.go
@@ -0,0 +1,163 @@
+// 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 (
+ "math"
+ "reflect"
+ "testing"
+
+ pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
+ "github.com/google/go-cmp/cmp"
+)
+
+func TestAtomicValueProtoRoundTrip(t *testing.T) {
+ tests := []struct {
+ value any
+ want pipepb.AtomicType
+ }{
+ {true, pipepb.AtomicType_BOOLEAN},
+ {uint8(200), pipepb.AtomicType_BYTE},
+ {int16(-300), pipepb.AtomicType_INT16},
+ {int32(70000), pipepb.AtomicType_INT32},
+ {int64(-1 << 40), pipepb.AtomicType_INT64},
+ {float32(1.5), pipepb.AtomicType_FLOAT},
+ {float64(2.25), pipepb.AtomicType_DOUBLE},
+ {"text", pipepb.AtomicType_STRING},
+ {[]byte("bytes"), pipepb.AtomicType_BYTES},
+ }
+ for _, test := range tests {
+ t.Run(reflect.TypeOf(test.value).String(), func(t *testing.T) {
+ ft, fv, err :=
atomicValueToProto(reflect.ValueOf(test.value))
+ if err != nil {
+ t.Fatalf("atomicValueToProto(%v) = %v, want nil
error", test.value, err)
+ }
+ if got := ft.GetAtomicType(); got != test.want {
+ t.Errorf("atomicValueToProto(%v) field type =
%v, want %v", test.value, got, test.want)
+ }
+ got, err := atomicValueFromProto(ft, fv)
+ if err != nil {
+ t.Fatalf("atomicValueFromProto(%v, %v) = %v,
want nil error", ft, fv, err)
+ }
+ if d := cmp.Diff(test.value, got.Interface()); d != "" {
+ t.Errorf("round trip of %v: diff (-want, +got):
%v", test.value, d)
+ }
+ })
+ }
+}
+
+func TestAtomicValueToProto_Unsupported(t *testing.T) {
+ type named int32
+ for _, value := range []any{int(1), uint32(1), named(1), struct{ A
int32 }{}, []int32{1}} {
+ if _, _, err := atomicValueToProto(reflect.ValueOf(value)); err
== nil {
+ t.Errorf("atomicValueToProto(%T) = nil error, want
error", value)
+ }
+ }
+}
+
+func TestAtomicValueFromProto_Mismatch(t *testing.T) {
+ stringType := &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_AtomicType{AtomicType:
pipepb.AtomicType_STRING},
+ }
+ int32Value := &pipepb.FieldValue{
+ FieldValue: &pipepb.FieldValue_AtomicValue{
+ AtomicValue: &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Int32{Int32: 3}},
+ },
+ }
+ if _, err := atomicValueFromProto(stringType, int32Value); err == nil {
+ t.Errorf("atomicValueFromProto(STRING, int32 value) = nil
error, want error")
+ }
+ rowType := &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_RowType{RowType:
&pipepb.RowType{Schema: &pipepb.Schema{}}},
+ }
+ if _, err := atomicValueFromProto(rowType, int32Value); err == nil {
+ t.Errorf("atomicValueFromProto(ROW, int32 value) = nil error,
want error")
+ }
+ if _, err := atomicValueFromProto(stringType, &pipepb.FieldValue{});
err == nil {
+ t.Errorf("atomicValueFromProto(STRING, null value) = nil error,
want error")
+ }
+}
+
+func TestAtomicValueToProto_ByteIsSigned(t *testing.T) {
+ _, fv, err := atomicValueToProto(reflect.ValueOf(uint8(200)))
+ if err != nil {
+ t.Fatalf("atomicValueToProto(uint8(200)) = %v, want nil error",
err)
+ }
+ if got := fv.GetAtomicValue().GetByte(); got != -56 {
+ t.Errorf("atomicValueToProto(uint8(200)) byte = %v, want -56",
got)
+ }
+ byteType := &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_AtomicType{AtomicType:
pipepb.AtomicType_BYTE},
+ }
+ got, err := atomicValueFromProto(byteType, fv)
+ if err != nil {
+ t.Fatalf("atomicValueFromProto(BYTE, -56) = %v, want nil
error", err)
+ }
+ if got.Interface() != uint8(200) {
+ t.Errorf("atomicValueFromProto(BYTE, -56) = %v, want
uint8(200)", got)
+ }
+}
+
+func TestAtomicValueFromProto_Range(t *testing.T) {
+ byteType := &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_AtomicType{AtomicType:
pipepb.AtomicType_BYTE},
+ }
+ int16Type := &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_AtomicType{AtomicType:
pipepb.AtomicType_INT16},
+ }
+ byteValue := func(v int32) *pipepb.FieldValue {
+ return &pipepb.FieldValue{FieldValue:
&pipepb.FieldValue_AtomicValue{
+ AtomicValue: &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Byte{Byte: v}},
+ }}
+ }
+ int16Value := func(v int32) *pipepb.FieldValue {
+ return &pipepb.FieldValue{FieldValue:
&pipepb.FieldValue_AtomicValue{
+ AtomicValue: &pipepb.AtomicTypeValue{Value:
&pipepb.AtomicTypeValue_Int16{Int16: v}},
+ }}
+ }
+ tests := []struct {
+ name string
+ ft *pipepb.FieldType
+ fv *pipepb.FieldValue
+ want any // nil when the value is out of range.
+ }{
+ {"BYTE min", byteType, byteValue(math.MinInt8), uint8(0x80)},
+ {"BYTE max", byteType, byteValue(math.MaxInt8), uint8(0x7f)},
+ {"BYTE below min", byteType, byteValue(math.MinInt8 - 1), nil},
+ {"BYTE above max", byteType, byteValue(256), nil},
+ {"INT16 min", int16Type, int16Value(math.MinInt16),
int16(math.MinInt16)},
+ {"INT16 max", int16Type, int16Value(math.MaxInt16),
int16(math.MaxInt16)},
+ {"INT16 below min", int16Type, int16Value(math.MinInt16 - 1),
nil},
+ {"INT16 above max", int16Type, int16Value(65537), nil},
+ }
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ got, err := atomicValueFromProto(test.ft, test.fv)
+ if test.want == nil {
+ if err == nil {
+ t.Fatalf("atomicValueFromProto(%v) =
%v, want error", test.fv, got)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("atomicValueFromProto(%v) = %v, want
nil error", test.fv, err)
+ }
+ if got.Interface() != test.want {
+ t.Errorf("atomicValueFromProto(%v) = %v, want
%v", test.fv, got, test.want)
+ }
+ })
+ }
+}
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 413b1f379a9..ee08bbb8f54 100644
--- a/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
@@ -61,12 +61,12 @@ type Registry struct {
logicalTypeProviders map[reflect.Type]LogicalTypeProvider
logicalTypeInterfaces []reflect.Type
- // Maps logical type identifiers to their reflect.Type and the schema
representation.
- // the type identifier is the reflect.Type name, and included in the
proto as well.
+ // logicalTypes maps registered logical types by identifier and
argument.
+ // The identifier is included in the proto as the logical type URN.
// We don't treat all types as "logical" types.
- // ... why don't we treat all types as Logical types?
- logicalTypes map[string]LogicalType
- logicalTypeIdentifiers map[reflect.Type]string
+ logicalTypes map[logicalTypeKey]LogicalType
+ // logicalTypeByGoType maps Go types to their registered logical type.
+ logicalTypeByGoType map[reflect.Type]LogicalType
// toReconcile contains a list of types that have been registered
// but not yet processed. Registration actually happens on first
@@ -74,6 +74,13 @@ type Registry struct {
toReconcile []reflect.Type
}
+// logicalTypeKey identifies a registered logical type. argument is nil for
+// logical types without an argument, and the argument value otherwise.
+type logicalTypeKey struct {
+ identifier string
+ argument any
+}
+
// NewRegistry creates an initialized LogicalTypeRegistry.
func NewRegistry() *Registry {
return &Registry{
@@ -81,9 +88,9 @@ func NewRegistry() *Registry {
idToType: map[string]reflect.Type{},
syntheticToUser: map[reflect.Type]reflect.Type{},
- logicalTypes: map[string]LogicalType{},
- logicalTypeIdentifiers: map[reflect.Type]string{},
- logicalTypeProviders: map[reflect.Type]LogicalTypeProvider{},
+ logicalTypes: map[logicalTypeKey]LogicalType{},
+ logicalTypeByGoType: map[reflect.Type]LogicalType{},
+ logicalTypeProviders: map[reflect.Type]LogicalTypeProvider{},
}
}
@@ -98,9 +105,30 @@ func (r *Registry) RegisterLogicalType(lt LogicalType) {
if len(lt.ID()) == 0 {
panic(fmt.Sprintf("invalid logical type, bad id: %v -> %v",
lt.GoType(), lt.StorageType()))
}
+ if lt.argT != nil {
+ if _, _, err := atomicValueToProto(lt.argV); err != nil {
+ panic(fmt.Sprintf("LogicalType[%v] has an invalid
argument %v: %v", lt.ID(), lt.argV, err))
+ }
+ if !lt.argV.Comparable() || !lt.argV.Equal(lt.argV) {
+ panic(fmt.Sprintf("LogicalType[%v] has the argument %v
which does not compare equal to itself", lt.ID(), lt.argV))
+ }
+ }
// TODO add duplication checks.
- r.logicalTypeIdentifiers[lt.GoType()] = lt.ID()
- r.logicalTypes[lt.ID()] = lt
+ r.logicalTypeByGoType[lt.GoType()] = lt
+ r.logicalTypes[lt.key()] = lt
+}
+
+// lookupLogicalType returns the logical type registered for identifier and,
+// when argument is valid, that argument. A logical type registered without an
+// argument matches any argument.
+func (r *Registry) lookupLogicalType(identifier string, argument
reflect.Value) (LogicalType, bool) {
+ if argument.IsValid() && argument.Comparable() {
+ if lt, ok := r.logicalTypes[logicalTypeKey{identifier:
identifier, argument: argument.Interface()}]; ok {
+ return lt, true
+ }
+ }
+ lt, ok := r.logicalTypes[logicalTypeKey{identifier: identifier}]
+ return lt, ok
}
// RegisterLogicalTypeProvider allows registration of providers for interface
types.
@@ -162,6 +190,29 @@ func ToLogicalType(identifier string, goType, storageType
reflect.Type) LogicalT
return LogicalType{identifier: identifier, goT: goType, storageT:
storageType}
}
+// ToLogicalTypeWithArgument creates a LogicalType with an argument, for
+// parameterized logical types such as a fixed length string and its length.
+// The argument is included in the schema representation of the logical type,
+// and must be a value of a built in Go type with an atomic schema type other
+// than []byte. It is matched by equality, so a floating point NaN is not a
+// valid argument. A nil argument creates a logical type without an argument.
+func ToLogicalTypeWithArgument(identifier string, goType, storageType
reflect.Type, argument any) LogicalType {
+ lt := ToLogicalType(identifier, goType, storageType)
+ if argument != nil {
+ lt.argT = reflect.TypeOf(argument)
+ lt.argV = reflect.ValueOf(argument)
+ }
+ return lt
+}
+
+// key returns the registry key of the logical type.
+func (l *LogicalType) key() logicalTypeKey {
+ if l.argT == nil {
+ return logicalTypeKey{identifier: l.identifier}
+ }
+ return logicalTypeKey{identifier: l.identifier, argument:
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.go
b/sdks/go/pkg/beam/core/runtime/graphx/schema/schema.go
index 0d44e68285b..9bd7a287d20 100644
--- a/sdks/go/pkg/beam/core/runtime/graphx/schema/schema.go
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/schema.go
@@ -218,8 +218,7 @@ func (r *Registry) registerType(ut reflect.Type, seen
map[reflect.Type]struct{})
// Lets do some recursion to register fundamental type parts.
t := ut
- if lID, ok := r.logicalTypeIdentifiers[t]; ok {
- lt := r.logicalTypes[lID]
+ if lt, ok := r.logicalTypeByGoType[t]; ok {
r.addToMaps(lt.StorageType(), t)
return nil
}
@@ -327,16 +326,17 @@ func (r *Registry) FromType(ot reflect.Type)
(*pipepb.Schema, error) {
return r.fromType(ot)
}
-func (r *Registry) logicalTypeToFieldType(t reflect.Type) (*pipepb.FieldType,
string, error) {
+// logicalTypeToFieldType returns the storage field type and the logical type
+// of t, or a nil field type if t is not a logical type.
+func (r *Registry) logicalTypeToFieldType(t reflect.Type) (*pipepb.FieldType,
LogicalType, error) {
// Check if a logical type was registered that matches this struct type
directly
// and if so, extract the schema from it for use.
- if lID, ok := r.logicalTypeIdentifiers[t]; ok {
- lt := r.logicalTypes[lID]
+ if lt, ok := r.logicalTypeByGoType[t]; ok {
ftype, err := r.reflectTypeToFieldType(lt.StorageType())
if err != nil {
- return nil, "", errors.Wrapf(err, "unable to convert
LogicalType[%v]'s storage type %v for Go type of %v to a schema", lID,
lt.StorageType(), lt.GoType())
+ return nil, LogicalType{}, errors.Wrapf(err, "unable to
convert LogicalType[%v]'s storage type %v for Go type of %v to a schema",
lt.ID(), lt.StorageType(), lt.GoType())
}
- return ftype, lID, nil
+ return ftype, lt, nil
}
for _, lti := range r.logicalTypeInterfaces {
if !t.Implements(lti) {
@@ -345,18 +345,42 @@ func (r *Registry) logicalTypeToFieldType(t reflect.Type)
(*pipepb.FieldType, st
p := r.logicalTypeProviders[lti]
st, err := p(t)
if err != nil {
- return nil, "", errors.Wrapf(err, "unable to convert
LogicalType[%v] using provider for %v schema field", t, lti)
+ return nil, LogicalType{}, errors.Wrapf(err, "unable to
convert LogicalType[%v] using provider for %v schema field", t, lti)
}
if st == nil {
continue
}
ftype, err := r.reflectTypeToFieldType(st)
if err != nil {
- return nil, "", errors.Wrapf(err, "unable to convert
LogicalType[%v]'s storage type %v for Go type of %v to a schema", "interface",
st, t)
+ return nil, LogicalType{}, errors.Wrapf(err, "unable to
convert LogicalType[%v]'s storage type %v for Go type of %v to a schema",
"interface", st, t)
}
- return ftype, t.String(), nil
+ return ftype, ToLogicalType(t.String(), t, st), nil
}
- return nil, "", nil
+ return nil, LogicalType{}, nil
+}
+
+// logicalStorageSchema returns a copy of the storage row schema of a logical
+// type used as a top level type.
+func logicalStorageSchema(ftype *pipepb.FieldType, lt LogicalType, ot
reflect.Type) (*pipepb.Schema, error) {
+ schm := ftype.GetRowType().GetSchema()
+ if schm == nil {
+ return nil, errors.Errorf("LogicalType[%v] for %v has the
non-row storage type %v and cannot be used as a top level schema", lt.ID(), ot,
lt.StorageType())
+ }
+ return proto.Clone(schm).(*pipepb.Schema), nil
+}
+
+// appendLogicalOptions appends the options that let ToType recover the
+// logical type of a top level schema.
+func appendLogicalOptions(schm *pipepb.Schema, lt LogicalType) error {
+ schm.Options = append(schm.Options, logicalOption(lt.ID()))
+ if lt.ArgumentType() != nil {
+ opt, err := logicalArgumentOption(lt)
+ if err != nil {
+ return errors.Wrapf(err, "unable to convert the
argument of LogicalType[%v]", lt.ID())
+ }
+ schm.Options = append(schm.Options, opt)
+ }
+ return nil
}
// fromType handles if the initial type is a pointer or not WRT lookups against
@@ -367,18 +391,20 @@ func (r *Registry) fromType(ot reflect.Type)
(*pipepb.Schema, error) {
if schm, ok := r.typeToSchema[ot]; ok {
return schm, nil
}
- ftype, lID, err := r.logicalTypeToFieldType(ot)
+ ftype, lt, err := r.logicalTypeToFieldType(ot)
if err != nil {
return nil, err
}
if ftype != nil {
- schm := ftype.GetRowType().GetSchema()
- schm = proto.Clone(schm).(*pipepb.Schema)
+ schm, err := logicalStorageSchema(ftype, lt, ot)
+ if err != nil {
+ return nil, err
+ }
if ot.Kind() == reflect.Ptr {
schm.Options = append(schm.Options, optGoNillable())
}
- if lID != "" {
- schm.Options = append(schm.Options, logicalOption(lID))
+ if err := appendLogicalOptions(schm, lt); err != nil {
+ return nil, err
}
schm.Id = getUUID(ot)
r.typeToSchema[ot] = schm
@@ -413,6 +439,9 @@ const (
// optGoLogical indicates that this top level schema has a logical type
equivalent that need to be looked up.
// It has a value type of String representing the URN for the logical
type to look up.
optGoLogicalUrn = "beam:schema:go:logical:v1"
+ // optGoLogicalArgument holds the argument of the logical type of this
top level schema.
+ // Its value type is the argument type of the logical type.
+ optGoLogicalArgumentUrn = "beam:schema:go:logical_argument:v1"
)
func optGoNillable() *pipepb.Option {
@@ -498,6 +527,43 @@ func fromLogicalOption(opts []*pipepb.Option) (string,
bool) {
return lID, true
}
+// logicalArgumentOption returns the option holding the argument of a
+// parameterized logical type.
+func logicalArgumentOption(lt LogicalType) (*pipepb.Option, error) {
+ ft, fv, err := atomicValueToProto(lt.ArgumentValue())
+ if err != nil {
+ return nil, err
+ }
+ return &pipepb.Option{
+ Name: optGoLogicalArgumentUrn,
+ Type: ft,
+ Value: fv,
+ }, nil
+}
+
+// fromLogicalArgumentOption returns the logical type argument of this top
+// level type, or an invalid Value when the schema has no argument option.
+func fromLogicalArgumentOption(opts []*pipepb.Option) (reflect.Value, error) {
+ o := checkOptions(opts, optGoLogicalArgumentUrn)
+ if o == nil {
+ return reflect.Value{}, nil
+ }
+ return atomicValueFromProto(o.GetType(), o.GetValue())
+}
+
+// logicalTypeArgument returns the argument of a logical type proto as a Go
+// value. It returns an invalid Value when the logical type has no argument,
+// or when the argument is not of an atomic type.
+func logicalTypeArgument(lst *pipepb.LogicalType) (reflect.Value, error) {
+ if lst.GetArgumentType() == nil || lst.GetArgument().GetFieldValue() ==
nil {
+ return reflect.Value{}, nil
+ }
+ if _, ok :=
lst.GetArgumentType().GetTypeInfo().(*pipepb.FieldType_AtomicType); !ok {
+ return reflect.Value{}, nil
+ }
+ return atomicValueFromProto(lst.GetArgumentType(), lst.GetArgument())
+}
+
func (r *Registry) structToSchema(t reflect.Type) (*pipepb.Schema, error) {
if t.Kind() != reflect.Struct {
return nil, errors.Errorf("non struct type received in
structToSchema: %v is kind %v", t, t.Kind())
@@ -506,14 +572,18 @@ func (r *Registry) structToSchema(t reflect.Type)
(*pipepb.Schema, error) {
return schm, nil
}
- ftype, lID, err := r.logicalTypeToFieldType(t)
+ ftype, lt, err := r.logicalTypeToFieldType(t)
if err != nil {
return nil, err
}
if ftype != nil {
- schm := ftype.GetRowType().GetSchema()
- schm = proto.Clone(schm).(*pipepb.Schema)
- schm.Options = append(schm.Options, logicalOption(lID))
+ schm, err := logicalStorageSchema(ftype, lt, t)
+ if err != nil {
+ return nil, err
+ }
+ if err := appendLogicalOptions(schm, lt); err != nil {
+ return nil, err
+ }
schm.Id = getUUID(t)
r.typeToSchema[t] = schm
r.idToType[schm.GetId()] = t
@@ -566,18 +636,26 @@ func (r *Registry) structFieldToField(sf
reflect.StructField) (*pipepb.Field, er
}
func (r *Registry) reflectTypeToFieldType(ot reflect.Type) (*pipepb.FieldType,
error) {
- ftype, lID, err := r.logicalTypeToFieldType(ot)
+ ftype, lt, err := r.logicalTypeToFieldType(ot)
if err != nil {
return nil, err
}
if ftype != nil {
+ lst := &pipepb.LogicalType{
+ Urn: lt.ID(),
+ Representation: ftype,
+ }
+ if lt.ArgumentType() != nil {
+ at, av, err := atomicValueToProto(lt.ArgumentValue())
+ if err != nil {
+ return nil, errors.Wrapf(err, "unable to
convert the argument of LogicalType[%v]", lt.ID())
+ }
+ lst.ArgumentType = at
+ lst.Argument = av
+ }
return &pipepb.FieldType{
TypeInfo: &pipepb.FieldType_LogicalType{
- LogicalType: &pipepb.LogicalType{
- Urn: lID,
- Representation: ftype,
- // TODO(BEAM-9615): Handle type
Arguments.
- },
+ LogicalType: lst,
},
}, nil
}
@@ -701,8 +779,13 @@ func (r *Registry) toType(s *pipepb.Schema) (reflect.Type,
error) {
if t, ok := r.idToType[s.GetId()]; ok {
return t, nil
}
- if lID, ok := fromLogicalOption(s.GetOptions()); ok {
- if lt, ok := r.logicalTypes[lID]; ok {
+ lID, isLogical := fromLogicalOption(s.GetOptions())
+ if isLogical {
+ arg, err := fromLogicalArgumentOption(s.GetOptions())
+ if err != nil {
+ return nil, errors.Wrapf(err, "invalid argument option
for logical type %v", lID)
+ }
+ if lt, ok := r.lookupLogicalType(lID, arg); ok {
return lt.GoType(), nil
}
}
@@ -720,7 +803,11 @@ func (r *Registry) toType(s *pipepb.Schema) (reflect.Type,
error) {
}
ret := reflect.StructOf(fields)
if ut, ok := r.syntheticToUser[ret]; ok {
- ret = ut
+ // The storage row of an unregistered logical type must not
resolve to
+ // the Go type of a registered one.
+ if !isLogical || !r.isLogicalGoType(ut) {
+ ret = ut
+ }
}
if t := nillableFromOptions(s.GetOptions(), ret); t != nil {
return t, nil
@@ -728,6 +815,17 @@ func (r *Registry) toType(s *pipepb.Schema) (reflect.Type,
error) {
return ret, nil
}
+// isLogicalGoType reports whether ut is the Go type of a registered logical
+// type, after skipping a pointer as addToMaps does.
+func (r *Registry) isLogicalGoType(ut reflect.Type) bool {
+ for gt := range r.logicalTypeByGoType {
+ if reflectx.SkipPtr(gt) == ut {
+ return true
+ }
+ }
+ return false
+}
+
func (r *Registry) fieldToStructField(sf *pipepb.Field) (reflect.StructField,
error) {
name := sf.GetName()
rt, err := r.fieldTypeToReflectType(sf.GetType(), sf.Options)
@@ -794,8 +892,15 @@ func (r *Registry) fieldTypeToReflectType(sft
*pipepb.FieldType, opts []*pipepb.
case *pipepb.FieldType_LogicalType:
lst := sft.GetLogicalType()
identifier := lst.GetUrn()
- lt, ok := r.logicalTypes[identifier]
+ arg, err := logicalTypeArgument(lst)
+ if err != nil {
+ return nil, errors.Wrapf(err, "invalid argument for
logical type %v", identifier)
+ }
+ lt, ok := r.lookupLogicalType(identifier, arg)
if !ok {
+ if arg.IsValid() {
+ return nil, errors.Errorf("unknown logical
type: %v with argument %v", identifier, arg.Interface())
+ }
return nil, errors.Errorf("unknown logical type: %v",
identifier)
}
t = lt.GoType()
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 367d70e81d1..56386954629 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
@@ -19,10 +19,12 @@ package schema
import (
"fmt"
+ "math"
"reflect"
"testing"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
+ "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"
"google.golang.org/protobuf/encoding/prototext"
@@ -90,7 +92,24 @@ type nonRegisteredLogical struct {
k int32
}
+// parameterizedString is a logical type with an argument. The tests register
+// it with the argument int32(5).
+type parameterizedString string
+
+// parameterizedRow is a logical type with an argument and a row storage type,
+// so it can be used as a top level type.
+type parameterizedRow struct {
+ S int64
+}
+
+// anyString is a logical type registered without an argument.
+type anyString string
+
var (
+ parameterizedStringType =
reflect.TypeOf((*parameterizedString)(nil)).Elem()
+ parameterizedRowType =
reflect.TypeOf((*parameterizedRow)(nil)).Elem()
+ anyStringType = reflect.TypeOf((*anyString)(nil)).Elem()
+
unexportedFieldsType =
reflect.TypeOf((*unexportedFields)(nil)).Elem()
exportedFuncType = reflect.TypeOf((*exportedFunc)(nil))
anotherType = reflect.TypeOf((*anotherStruct)(nil)).Elem()
@@ -754,6 +773,53 @@ func TestSchemaConversion(t *testing.T) {
rt: reflect.TypeOf(struct {
U nonRegisteredLogical
}{}),
+ }, {
+ st: &pipepb.Schema{
+ Fields: []*pipepb.Field{
+ {
+ Name: "P",
+ Type: &pipepb.FieldType{
+ TypeInfo:
&pipepb.FieldType_LogicalType{
+ LogicalType:
&pipepb.LogicalType{
+ Urn:
"schema.parameterizedString",
+
Representation: &pipepb.FieldType{
+
TypeInfo: &pipepb.FieldType_AtomicType{
+
AtomicType: pipepb.AtomicType_STRING,
+
},
+ },
+
ArgumentType: int32FieldType(),
+
Argument: int32FieldValue(5),
+ },
+ },
+ },
+ },
+ },
+ },
+ rt: reflect.TypeOf(struct {
+ P parameterizedString
+ }{}),
+ }, {
+ st: &pipepb.Schema{
+ Options: []*pipepb.Option{
+
logicalOption("schema.parameterizedRow"),
+ {
+ Name: optGoLogicalArgumentUrn,
+ Type: int32FieldType(),
+ Value: int32FieldValue(3),
+ },
+ },
+ Fields: []*pipepb.Field{
+ {
+ Name: "S",
+ Type: &pipepb.FieldType{
+ TypeInfo:
&pipepb.FieldType_AtomicType{
+ AtomicType:
pipepb.AtomicType_INT64,
+ },
+ },
+ },
+ },
+ },
+ rt: parameterizedRowType,
},
}
@@ -764,6 +830,8 @@ func TestSchemaConversion(t *testing.T) {
preRegLogicalTypes(reg)
reg.RegisterLogicalType(ToLogicalType(exportedFuncType.Elem().String(),
exportedFuncType.Elem(), reflect.TypeOf(struct{ V int16 }{})))
reg.RegisterLogicalType(ToLogicalType(nonRegisteredLogicalType.String(),
nonRegisteredLogicalType, reflect.TypeOf(struct{ K int32 }{})))
+
reg.RegisterLogicalType(ToLogicalTypeWithArgument("schema.parameterizedString",
parameterizedStringType, reflectx.String, int32(5)))
+
reg.RegisterLogicalType(ToLogicalTypeWithArgument("schema.parameterizedRow",
parameterizedRowType, reflect.TypeOf(struct{ S int64 }{}), int32(3)))
reg.RegisterType(reflect.TypeOf((*sRegisteredType)(nil)))
reg.RegisterLogicalTypeProvider(reflect.TypeOf((*testInterface)(nil)).Elem(),
func(t reflect.Type) (reflect.Type, error) {
switch t {
@@ -829,3 +897,269 @@ func hasEmbeddedField(t reflect.Type) bool {
}
return false
}
+
+func int32FieldType() *pipepb.FieldType {
+ return &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_AtomicType{
+ AtomicType: pipepb.AtomicType_INT32,
+ },
+ }
+}
+
+func int32FieldValue(v int32) *pipepb.FieldValue {
+ return &pipepb.FieldValue{
+ FieldValue: &pipepb.FieldValue_AtomicValue{
+ AtomicValue: &pipepb.AtomicTypeValue{
+ Value: &pipepb.AtomicTypeValue_Int32{Int32: v},
+ },
+ },
+ }
+}
+
+func TestToType_LogicalTypeArguments(t *testing.T) {
+ reg := NewRegistry()
+ preRegLogicalTypes(reg)
+
reg.RegisterLogicalType(ToLogicalTypeWithArgument("schema.parameterizedString",
parameterizedStringType, reflectx.String, int32(5)))
+ reg.RegisterLogicalType(ToLogicalType("schema.anyString",
anyStringType, reflectx.String))
+
+ stringType := &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_AtomicType{
+ AtomicType: pipepb.AtomicType_STRING,
+ },
+ }
+ rowType := &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_RowType{
+ RowType: &pipepb.RowType{
+ Schema: &pipepb.Schema{
+ Fields: []*pipepb.Field{{Name:
"precision", Type: int32FieldType()}},
+ },
+ },
+ },
+ }
+ rowValue := &pipepb.FieldValue{
+ FieldValue: &pipepb.FieldValue_RowValue{
+ RowValue: &pipepb.Row{Values:
[]*pipepb.FieldValue{int32FieldValue(5)}},
+ },
+ }
+
+ tests := []struct {
+ name string
+ lt *pipepb.LogicalType
+ want reflect.Type
+ wantErr bool
+ }{{
+ name: "matching argument",
+ lt: &pipepb.LogicalType{Urn: "schema.parameterizedString",
Representation: stringType, ArgumentType: int32FieldType(), Argument:
int32FieldValue(5)},
+ want: parameterizedStringType,
+ }, {
+ name: "unknown argument",
+ lt: &pipepb.LogicalType{Urn: "schema.parameterizedString",
Representation: stringType, ArgumentType: int32FieldType(), Argument:
int32FieldValue(7)},
+ wantErr: true,
+ }, {
+ name: "missing argument",
+ lt: &pipepb.LogicalType{Urn: "schema.parameterizedString",
Representation: stringType},
+ wantErr: true,
+ }, {
+ name: "argument ignored for logical type registered without
one",
+ lt: &pipepb.LogicalType{Urn: "schema.anyString",
Representation: stringType, ArgumentType: int32FieldType(), Argument:
int32FieldValue(7)},
+ want: anyStringType,
+ }, {
+ name: "non atomic argument ignored",
+ lt: &pipepb.LogicalType{Urn: "schema.anyString",
Representation: stringType, ArgumentType: rowType, Argument: rowValue},
+ want: anyStringType,
+ }, {
+ name: "unknown urn",
+ lt: &pipepb.LogicalType{Urn: "schema.unknown",
Representation: stringType},
+ wantErr: true,
+ }}
+ for i, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ s := &pipepb.Schema{
+ Id: fmt.Sprintf("arguments-%d", i),
+ Fields: []*pipepb.Field{{
+ Name: "f",
+ Type: &pipepb.FieldType{
+ TypeInfo:
&pipepb.FieldType_LogicalType{LogicalType: test.lt},
+ },
+ }},
+ }
+ got, err := reg.ToType(s)
+ if test.wantErr {
+ if err == nil {
+ t.Fatalf("ToType(%v) = %v, want error",
prototext.Format(s), got)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("ToType(%v) = %v, want nil error",
prototext.Format(s), err)
+ }
+ if ft := got.Field(0).Type; ft != test.want {
+ t.Errorf("ToType(%v) field type = %v, want %v",
prototext.Format(s), ft, test.want)
+ }
+ })
+ }
+}
+
+// The multi types share one URN with different arguments, and multiAny is
+// registered for the URN without an argument.
+type multiOne struct{ S int64 }
+type multiTwo struct{ S int64 }
+type multiAny struct{ S int64 }
+
+// multiOnePtr is a defined pointer type, used as a logical Go type.
+type multiOnePtr *multiOne
+
+func TestLogicalTypeArgumentPrecedence(t *testing.T) {
+ reg := NewRegistry()
+ preRegLogicalTypes(reg)
+ storage := reflect.TypeOf(struct{ S int64 }{})
+ oneType := reflect.TypeOf(multiOne{})
+ twoType := reflect.TypeOf(multiTwo{})
+ anyType := reflect.TypeOf(multiAny{})
+ reg.RegisterLogicalType(ToLogicalTypeWithArgument("schema.multi",
oneType, storage, int32(1)))
+ reg.RegisterLogicalType(ToLogicalTypeWithArgument("schema.multi",
twoType, storage, int32(2)))
+ reg.RegisterLogicalType(ToLogicalType("schema.multi", anyType, storage))
+
+ rowType := &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_RowType{RowType:
&pipepb.RowType{Schema: &pipepb.Schema{
+ Fields: []*pipepb.Field{{Name: "S", Type:
&pipepb.FieldType{
+ TypeInfo:
&pipepb.FieldType_AtomicType{AtomicType: pipepb.AtomicType_INT64},
+ }}},
+ }}},
+ }
+ fieldSchema := func(id string, argument *pipepb.FieldValue)
*pipepb.Schema {
+ lt := &pipepb.LogicalType{Urn: "schema.multi", Representation:
rowType}
+ if argument != nil {
+ lt.ArgumentType = int32FieldType()
+ lt.Argument = argument
+ }
+ return &pipepb.Schema{Id: id, Fields: []*pipepb.Field{{Name:
"f", Type: &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_LogicalType{LogicalType:
lt},
+ }}}}
+ }
+ topLevelSchema := func(id string, argument *pipepb.FieldValue)
*pipepb.Schema {
+ opts := []*pipepb.Option{logicalOption("schema.multi")}
+ if argument != nil {
+ opts = append(opts, &pipepb.Option{Name:
optGoLogicalArgumentUrn, Type: int32FieldType(), Value: argument})
+ }
+ return &pipepb.Schema{Id: id, Options: opts, Fields:
rowType.GetRowType().GetSchema().GetFields()}
+ }
+
+ tests := []struct {
+ name string
+ s *pipepb.Schema
+ want reflect.Type
+ }{
+ {"field argument 1", fieldSchema("field-1",
int32FieldValue(1)), oneType},
+ {"field argument 2", fieldSchema("field-2",
int32FieldValue(2)), twoType},
+ {"field argument 3 falls back", fieldSchema("field-3",
int32FieldValue(3)), anyType},
+ {"field without argument", fieldSchema("field-none", nil),
anyType},
+ {"top level argument 2", topLevelSchema("top-2",
int32FieldValue(2)), twoType},
+ {"top level argument 3 falls back", topLevelSchema("top-3",
int32FieldValue(3)), anyType},
+ {"top level without argument", topLevelSchema("top-none", nil),
anyType},
+ }
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ got, err := reg.ToType(test.s)
+ if err != nil {
+ t.Fatalf("ToType(%v) = %v, want nil error",
prototext.Format(test.s), err)
+ }
+ if len(test.s.GetOptions()) == 0 {
+ got = got.Field(0).Type
+ }
+ if got != test.want {
+ t.Errorf("ToType(%v) = %v, want exactly %v",
prototext.Format(test.s), got, test.want)
+ }
+ })
+ }
+ for _, rt := range []reflect.Type{oneType, twoType, anyType} {
+ schm, err := reg.FromType(rt)
+ if err != nil {
+ t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+ }
+ got, err := reg.ToType(schm)
+ if err != nil {
+ t.Fatalf("ToType(FromType(%v)) = %v, want nil error",
rt, err)
+ }
+ if got != rt {
+ t.Errorf("ToType(FromType(%v)) = %v, want exactly %v",
rt, got, rt)
+ }
+ }
+}
+
+// A top level schema whose logical type is not registered keeps its storage
+// row type, even when a registered logical type has the same storage.
+func TestToType_UnregisteredTopLevelLogicalType(t *testing.T) {
+ storage := reflect.TypeOf(struct{ S int64 }{})
+ fields := []*pipepb.Field{{Name: "S", Type: &pipepb.FieldType{
+ TypeInfo: &pipepb.FieldType_AtomicType{AtomicType:
pipepb.AtomicType_INT64},
+ }}}
+ tests := []struct {
+ name string
+ opts []*pipepb.Option
+ }{
+ {"unknown argument", []*pipepb.Option{
+ logicalOption("schema.multi"),
+ {Name: optGoLogicalArgumentUrn, Type: int32FieldType(),
Value: int32FieldValue(2)},
+ }},
+ {"missing argument",
[]*pipepb.Option{logicalOption("schema.multi")}},
+ {"unknown urn",
[]*pipepb.Option{logicalOption("schema.unknown")}},
+ }
+ for _, goType := range []reflect.Type{reflect.TypeOf(multiOne{}),
reflect.TypeOf(&multiOne{}), reflect.TypeOf(multiOnePtr(nil))} {
+ reg := NewRegistry()
+ preRegLogicalTypes(reg)
+
reg.RegisterLogicalType(ToLogicalTypeWithArgument("schema.multi", goType,
storage, int32(1)))
+ reg.RegisterType(goType)
+ for _, test := range tests {
+ for _, nillable := range []bool{false, true} {
+ name := fmt.Sprintf("%v/%v/nillable=%v",
goType, test.name, nillable)
+ t.Run(name, func(t *testing.T) {
+ opts, want := test.opts, storage
+ if nillable {
+ opts, want =
append(append([]*pipepb.Option{}, opts...), optGoNillable()),
reflect.PtrTo(storage)
+ }
+ s := &pipepb.Schema{Id: "unregistered-"
+ name, Options: opts, Fields: fields}
+ got, err := reg.ToType(s)
+ if err != nil {
+ t.Fatalf("ToType(%v) = %v, want
nil error", prototext.Format(s), err)
+ }
+ if got != want {
+ t.Errorf("ToType(%v) = %v, want
%v", prototext.Format(s), got, want)
+ }
+ })
+ }
+ }
+ }
+}
+
+type bytesBacked struct{ B []byte }
+
+func TestRegisterLogicalType_Validation(t *testing.T) {
+ reg := NewRegistry()
+ preRegLogicalTypes(reg)
+ bytesBackedType := reflect.TypeOf(bytesBacked{})
+ reg.RegisterLogicalType(ToLogicalType("schema.bytesBacked",
bytesBackedType, reflectx.ByteSlice))
+ if schm, err := reg.FromType(bytesBackedType); err == nil {
+ t.Errorf("FromType(%v) = %v, want error for a top level logical
type with non row storage", bytesBackedType, prototext.Format(schm))
+ }
+ if _, err := reg.FromType(reflect.TypeOf(struct{ F bytesBacked }{}));
err != nil {
+ t.Errorf("FromType(struct{F bytesBacked}) = %v, want nil
error", err)
+ }
+
+ for name, argument := range map[string]any{
+ "NaN": math.NaN(),
+ "float32 NaN": float32(math.NaN()),
+ "[]byte": []byte("x"),
+ "int": int(1),
+ "struct": struct{ A int32 }{},
+ } {
+ t.Run(name, func(t *testing.T) {
+ defer func() {
+ if recover() == nil {
+ t.Errorf("RegisterLogicalType with
argument %v did not panic", argument)
+ }
+ }()
+
reg.RegisterLogicalType(ToLogicalTypeWithArgument("schema.invalid",
anyStringType, reflectx.String, argument))
+ })
+ }
+}
