lostluck commented on code in PR #40156:
URL: https://github.com/apache/beam/pull/40156#discussion_r4497243215
##########
sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go:
##########
@@ -499,3 +500,265 @@ func TestMillisInstant_Range(t *testing.T) {
t.Error("truncated millis_instant decoded successfully")
}
}
+
+func TestPassThroughLogicalTypes(t *testing.T) {
+ nullable := func(urn string, representation pipepb.AtomicType, length
int32) *pipepb.FieldType {
+ ft := logicalFieldType(urn, atomicType(representation))
+ ft.Nullable = true
+ ft.GetLogicalType().ArgumentType = int32FieldType()
+ ft.GetLogicalType().Argument = int32FieldValue(length)
+ return ft
+ }
+ // The schema of the char, varchar, binary and varbinary case of
standard_coders.yaml.
+ s := &pipepb.Schema{
+ Id: "pass-through-logical-types-test",
+ Fields: []*pipepb.Field{
+ {Name: "f_char", Type: nullable(URNFixedChar,
pipepb.AtomicType_STRING, 5)},
+ {Name: "f_varchar", Type: nullable(URNVarChar,
pipepb.AtomicType_STRING, 10)},
+ {Name: "f_bytes", Type: nullable(URNFixedBytes,
pipepb.AtomicType_BYTES, 5)},
+ {Name: "f_varbytes", Type: nullable(URNVarBytes,
pipepb.AtomicType_BYTES, 10)},
+ },
+ }
+ rt, err := ToType(s)
+ if err != nil {
+ t.Fatalf("ToType(%v) = %v, want nil error",
prototext.Format(s), err)
+ }
+ stringPtr := reflect.PointerTo(typeOf[string]())
+ bytesPtr := reflect.PointerTo(typeOf[[]byte]())
+ for i, want := range []reflect.Type{stringPtr, stringPtr, bytesPtr,
bytesPtr} {
+ if got := rt.Field(i).Type; got != want {
+ t.Errorf("ToType field %d type = %v, want %v", i, got,
want)
+ }
+ }
+ // Decoding uses the plain string and bytes encodings, and pads nothing.
+ dec, err := coder.RowDecoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ got, err := dec(bytes.NewBufferString("\x04\x01\x06\x05null?\x04null"))
+ if err != nil {
+ t.Fatalf("dec = %v, want nil error", err)
+ }
+ rv := reflect.ValueOf(got)
+ if s := rv.Field(0).Elem().String(); s != "null?" {
+ t.Errorf("f_char = %q, want %q", s, "null?")
+ }
+ if !rv.Field(1).IsNil() || !rv.Field(2).IsNil() {
+ t.Errorf("f_varchar and f_bytes = %v, %v, want nil",
rv.Field(1), rv.Field(2))
+ }
+ if b := rv.Field(3).Elem().Bytes(); string(b) != "null" {
+ t.Errorf("f_varbytes = %q, want %q", b, "null")
+ }
+ // Plain string and bytes fields keep their atomic schema types.
+ schm, err := FromType(reflect.TypeOf(struct {
+ S string
+ B []byte
+ }{}))
+ if err != nil {
+ t.Fatalf("FromType = %v, want nil error", err)
+ }
+ for i, want := range []pipepb.AtomicType{pipepb.AtomicType_STRING,
pipepb.AtomicType_BYTES} {
+ if got := schm.GetFields()[i].GetType(); got.GetLogicalType()
!= nil || got.GetAtomicType() != want {
+ t.Errorf("FromType field %d = %v, want atomic %v", i,
got, want)
+ }
+ }
+}
+
+func TestRegisterPassThroughLogicalType(t *testing.T) {
+ r := NewRegistry()
+ const urn = "test:pass_through:v1"
+ r.RegisterPassThroughLogicalType(urn, typeOf[string]())
+ r.RegisterPassThroughLogicalType(urn, typeOf[string]())
+ for _, length := range []int32{1, 100} {
+ ft := logicalFieldType(urn,
atomicType(pipepb.AtomicType_STRING))
+ ft.GetLogicalType().ArgumentType = int32FieldType()
+ ft.GetLogicalType().Argument = int32FieldValue(length)
+ rt, err := r.ToType(&pipepb.Schema{Fields:
[]*pipepb.Field{{Name: "value", Type: ft}}})
+ if err != nil {
+ t.Fatalf("ToType(length=%v): %v", length, err)
+ }
+ if got := rt.Field(0).Type; got != typeOf[string]() {
+ t.Errorf("field type = %v, want string", got)
+ }
+ }
+ t.Run("ConflictingRegistration", func(t *testing.T) {
+ defer func() {
+ if recover() == nil {
+ t.Error("conflicting registration succeeded")
Review Comment:
It's trite, but for testing best practices don't just say what happened, say
waht should have happened. In this case, "wanted it to fail".
See https://google.github.io/styleguide/go/decisions#useful-test-failures
for guidance.
##########
sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go:
##########
@@ -499,3 +500,265 @@ func TestMillisInstant_Range(t *testing.T) {
t.Error("truncated millis_instant decoded successfully")
}
}
+
+func TestPassThroughLogicalTypes(t *testing.T) {
+ nullable := func(urn string, representation pipepb.AtomicType, length
int32) *pipepb.FieldType {
+ ft := logicalFieldType(urn, atomicType(representation))
+ ft.Nullable = true
+ ft.GetLogicalType().ArgumentType = int32FieldType()
+ ft.GetLogicalType().Argument = int32FieldValue(length)
+ return ft
+ }
+ // The schema of the char, varchar, binary and varbinary case of
standard_coders.yaml.
+ s := &pipepb.Schema{
+ Id: "pass-through-logical-types-test",
+ Fields: []*pipepb.Field{
+ {Name: "f_char", Type: nullable(URNFixedChar,
pipepb.AtomicType_STRING, 5)},
+ {Name: "f_varchar", Type: nullable(URNVarChar,
pipepb.AtomicType_STRING, 10)},
+ {Name: "f_bytes", Type: nullable(URNFixedBytes,
pipepb.AtomicType_BYTES, 5)},
+ {Name: "f_varbytes", Type: nullable(URNVarBytes,
pipepb.AtomicType_BYTES, 10)},
+ },
+ }
+ rt, err := ToType(s)
+ if err != nil {
+ t.Fatalf("ToType(%v) = %v, want nil error",
prototext.Format(s), err)
+ }
+ stringPtr := reflect.PointerTo(typeOf[string]())
+ bytesPtr := reflect.PointerTo(typeOf[[]byte]())
+ for i, want := range []reflect.Type{stringPtr, stringPtr, bytesPtr,
bytesPtr} {
+ if got := rt.Field(i).Type; got != want {
+ t.Errorf("ToType field %d type = %v, want %v", i, got,
want)
+ }
+ }
+ // Decoding uses the plain string and bytes encodings, and pads nothing.
+ dec, err := coder.RowDecoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ got, err := dec(bytes.NewBufferString("\x04\x01\x06\x05null?\x04null"))
+ if err != nil {
+ t.Fatalf("dec = %v, want nil error", err)
+ }
+ rv := reflect.ValueOf(got)
+ if s := rv.Field(0).Elem().String(); s != "null?" {
+ t.Errorf("f_char = %q, want %q", s, "null?")
+ }
+ if !rv.Field(1).IsNil() || !rv.Field(2).IsNil() {
+ t.Errorf("f_varchar and f_bytes = %v, %v, want nil",
rv.Field(1), rv.Field(2))
+ }
+ if b := rv.Field(3).Elem().Bytes(); string(b) != "null" {
+ t.Errorf("f_varbytes = %q, want %q", b, "null")
+ }
+ // Plain string and bytes fields keep their atomic schema types.
+ schm, err := FromType(reflect.TypeOf(struct {
+ S string
+ B []byte
+ }{}))
+ if err != nil {
+ t.Fatalf("FromType = %v, want nil error", err)
+ }
+ for i, want := range []pipepb.AtomicType{pipepb.AtomicType_STRING,
pipepb.AtomicType_BYTES} {
+ if got := schm.GetFields()[i].GetType(); got.GetLogicalType()
!= nil || got.GetAtomicType() != want {
+ t.Errorf("FromType field %d = %v, want atomic %v", i,
got, want)
+ }
+ }
+}
+
+func TestRegisterPassThroughLogicalType(t *testing.T) {
+ r := NewRegistry()
+ const urn = "test:pass_through:v1"
+ r.RegisterPassThroughLogicalType(urn, typeOf[string]())
+ r.RegisterPassThroughLogicalType(urn, typeOf[string]())
Review Comment:
Why is it called twice here? My read is that it just overwrites the first
one. If the intent is to validate that it overwrites/doesn't panic, please add
a comment to that effect.
##########
sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go:
##########
@@ -499,3 +500,265 @@ func TestMillisInstant_Range(t *testing.T) {
t.Error("truncated millis_instant decoded successfully")
}
}
+
+func TestPassThroughLogicalTypes(t *testing.T) {
+ nullable := func(urn string, representation pipepb.AtomicType, length
int32) *pipepb.FieldType {
+ ft := logicalFieldType(urn, atomicType(representation))
+ ft.Nullable = true
+ ft.GetLogicalType().ArgumentType = int32FieldType()
+ ft.GetLogicalType().Argument = int32FieldValue(length)
+ return ft
+ }
+ // The schema of the char, varchar, binary and varbinary case of
standard_coders.yaml.
+ s := &pipepb.Schema{
+ Id: "pass-through-logical-types-test",
+ Fields: []*pipepb.Field{
+ {Name: "f_char", Type: nullable(URNFixedChar,
pipepb.AtomicType_STRING, 5)},
+ {Name: "f_varchar", Type: nullable(URNVarChar,
pipepb.AtomicType_STRING, 10)},
+ {Name: "f_bytes", Type: nullable(URNFixedBytes,
pipepb.AtomicType_BYTES, 5)},
+ {Name: "f_varbytes", Type: nullable(URNVarBytes,
pipepb.AtomicType_BYTES, 10)},
+ },
+ }
+ rt, err := ToType(s)
+ if err != nil {
+ t.Fatalf("ToType(%v) = %v, want nil error",
prototext.Format(s), err)
+ }
+ stringPtr := reflect.PointerTo(typeOf[string]())
+ bytesPtr := reflect.PointerTo(typeOf[[]byte]())
+ for i, want := range []reflect.Type{stringPtr, stringPtr, bytesPtr,
bytesPtr} {
+ if got := rt.Field(i).Type; got != want {
+ t.Errorf("ToType field %d type = %v, want %v", i, got,
want)
+ }
+ }
+ // Decoding uses the plain string and bytes encodings, and pads nothing.
+ dec, err := coder.RowDecoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt,
err)
+ }
+ got, err := dec(bytes.NewBufferString("\x04\x01\x06\x05null?\x04null"))
+ if err != nil {
+ t.Fatalf("dec = %v, want nil error", err)
+ }
+ rv := reflect.ValueOf(got)
+ if s := rv.Field(0).Elem().String(); s != "null?" {
+ t.Errorf("f_char = %q, want %q", s, "null?")
+ }
+ if !rv.Field(1).IsNil() || !rv.Field(2).IsNil() {
+ t.Errorf("f_varchar and f_bytes = %v, %v, want nil",
rv.Field(1), rv.Field(2))
+ }
+ if b := rv.Field(3).Elem().Bytes(); string(b) != "null" {
+ t.Errorf("f_varbytes = %q, want %q", b, "null")
+ }
+ // Plain string and bytes fields keep their atomic schema types.
+ schm, err := FromType(reflect.TypeOf(struct {
+ S string
+ B []byte
+ }{}))
+ if err != nil {
+ t.Fatalf("FromType = %v, want nil error", err)
+ }
+ for i, want := range []pipepb.AtomicType{pipepb.AtomicType_STRING,
pipepb.AtomicType_BYTES} {
+ if got := schm.GetFields()[i].GetType(); got.GetLogicalType()
!= nil || got.GetAtomicType() != want {
+ t.Errorf("FromType field %d = %v, want atomic %v", i,
got, want)
+ }
+ }
+}
+
+func TestRegisterPassThroughLogicalType(t *testing.T) {
+ r := NewRegistry()
+ const urn = "test:pass_through:v1"
+ r.RegisterPassThroughLogicalType(urn, typeOf[string]())
+ r.RegisterPassThroughLogicalType(urn, typeOf[string]())
+ for _, length := range []int32{1, 100} {
+ ft := logicalFieldType(urn,
atomicType(pipepb.AtomicType_STRING))
+ ft.GetLogicalType().ArgumentType = int32FieldType()
+ ft.GetLogicalType().Argument = int32FieldValue(length)
+ rt, err := r.ToType(&pipepb.Schema{Fields:
[]*pipepb.Field{{Name: "value", Type: ft}}})
+ if err != nil {
+ t.Fatalf("ToType(length=%v): %v", length, err)
+ }
+ if got := rt.Field(0).Type; got != typeOf[string]() {
+ t.Errorf("field type = %v, want string", got)
+ }
+ }
+ t.Run("ConflictingRegistration", func(t *testing.T) {
+ defer func() {
+ if recover() == nil {
+ t.Error("conflicting registration succeeded")
+ }
+ }()
+ r.RegisterPassThroughLogicalType(urn, typeOf[[]byte]())
+ })
+ lt, ok := r.lookupLogicalType(urn, reflect.Value{})
+ if !ok || lt.GoType() != typeOf[string]() {
+ t.Errorf("registration after conflict = %v, %v", lt, ok)
+ }
+}
+
+func TestRegisterPassThroughLogicalType_PublicAPI(t *testing.T) {
+ const urn = "test:public_pass_through:v1"
+ t.Cleanup(func() { delete(defaultRegistry.logicalTypes,
logicalTypeKey{identifier: urn}) })
+ RegisterPassThroughLogicalType(urn, typeOf[string]())
+ incoming := &pipepb.Schema{Fields: []*pipepb.Field{{
+ Name: "value", Type: logicalFieldType(urn,
atomicType(pipepb.AtomicType_STRING)),
+ }}}
+ rt, err := ToType(incoming)
+ if err != nil {
+ t.Fatalf("ToType: %v", err)
+ }
+ if got := rt.Field(0).Type; got != typeOf[string]() {
+ t.Fatalf("field type = %v, want string", got)
+ }
+ dec, err := coder.RowDecoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowDecoderForStruct: %v", err)
+ }
+ wire := []byte("\x01\x00\x05hello")
+ row, err := dec(bytes.NewReader(wire))
+ if err != nil {
+ t.Fatalf("decode: %v", err)
+ }
+ if got := reflect.ValueOf(row).Field(0).String(); got != "hello" {
+ t.Errorf("decoded field = %q, want hello", got)
+ }
+ enc, err := coder.RowEncoderForStruct(rt)
+ if err != nil {
+ t.Fatalf("RowEncoderForStruct: %v", err)
+ }
+ var buf bytes.Buffer
+ if err := enc(row, &buf); err != nil {
+ t.Fatalf("encode: %v", err)
+ }
+ if got := buf.Bytes(); !bytes.Equal(got, wire) {
+ t.Errorf("encoded row = %x, want %x", got, wire)
+ }
+ outgoing, err := FromType(rt)
+ if err != nil {
+ t.Fatalf("FromType: %v", err)
+ }
+ if got := outgoing.GetFields()[0].GetType(); got.GetLogicalType() !=
nil || got.GetAtomicType() != pipepb.AtomicType_STRING {
+ t.Errorf("outgoing field = %v, want atomic STRING", got)
+ }
+}
Review Comment:
Nothing wrong with the test, but please do use a blank line or two to
visually distinguish sections of the code to improve readability. There's no
benefit for humans or robots to pack everything in like this.
For example, this test has a sequence of "call API -> check results"
sections so they can be separated that way, which makes reading easier.
eg:
```suggestion
func TestRegisterPassThroughLogicalType_PublicAPI(t *testing.T) {
const urn = "test:public_pass_through:v1"
t.Cleanup(func() { delete(defaultRegistry.logicalTypes,
logicalTypeKey{identifier: urn}) })
RegisterPassThroughLogicalType(urn, typeOf[string]())
incoming := &pipepb.Schema{Fields: []*pipepb.Field{{
Name: "value", Type: logicalFieldType(urn,
atomicType(pipepb.AtomicType_STRING)),
}}}
rt, err := ToType(incoming)
if err != nil {
t.Fatalf("ToType: %v", err)
}
if got := rt.Field(0).Type; got != typeOf[string]() {
t.Fatalf("field type = %v, want string", got)
}
dec, err := coder.RowDecoderForStruct(rt)
if err != nil {
t.Fatalf("RowDecoderForStruct: %v", err)
}
wire := []byte("\x01\x00\x05hello")
row, err := dec(bytes.NewReader(wire))
if err != nil {
t.Fatalf("decode: %v", err)
}
if got := reflect.ValueOf(row).Field(0).String(); got != "hello" {
t.Errorf("decoded field = %q, want hello", got)
}
enc, err := coder.RowEncoderForStruct(rt)
if err != nil {
t.Fatalf("RowEncoderForStruct: %v", err)
}
var buf bytes.Buffer
if err := enc(row, &buf); err != nil {
t.Fatalf("encode: %v", err)
}
if got := buf.Bytes(); !bytes.Equal(got, wire) {
t.Errorf("encoded row = %x, want %x", got, wire)
}
outgoing, err := FromType(rt)
if err != nil {
t.Fatalf("FromType: %v", err)
}
if got := outgoing.GetFields()[0].GetType(); got.GetLogicalType() !=
nil || got.GetAtomicType() != pipepb.AtomicType_STRING {
t.Errorf("outgoing field = %v, want atomic STRING", got)
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
