Repository navigation
Conversation
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 apache#40151. Part of apache#39684.
1d937c9 to
311a8fe
Compare
|
Assigning reviewers: R: @lostluck for label go. This pull request likely touches a core component ("core" label). Please review with scrutiny. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
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 apache#39684.
… 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 apache#39684.
schema.MillisInstant encodes as the Java InstantCoder does: 8 big endian bytes of the milliseconds shifted by the minimum int64. schema.Decimal encodes as the Java BigDecimalCoder does: the scale as a varint followed by the length prefixed two's complement bytes of the unscaled value. Both cases of standard_coders.yaml now run in the Go regression test. Part of apache#39684.
…gth string and bytes logical types schema.RegisterPassThroughLogicalType maps a logical type URN to the Go type of its representation without mapping that Go type back, so a schema field with the URN decodes with the plain row encoding and the length argument is ignored. The standard fixed_char, var_char, fixed_bytes and var_bytes logical types are registered this way, and their standard_coders.yaml case now runs in the Go regression test. Part of apache#39684.
311a8fe to
6020d83
Compare
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
📋 Review Summary
This PR adds support for pass-through logical types in the Go SDK via schema.RegisterPassThroughLogicalType, enabling incoming schema fields with fixed/variable length string and byte logical type URNs (fixed_char, var_char, fixed_bytes, var_bytes) to map directly to string and []byte while preserving atomic schema output for those Go types. It also enables the corresponding standard_coders.yaml regression test case.
🔍 General Feedback
- The implementation is clean, concise, and well-aligned with the existing
Registrydesign and logical type lookup mechanics. - One edge case to address in
RegisterPassThroughLogicalTypeis rejecting pointergoTypes (similar tocheckLogicalTypeKinds), since nullable logical type fields wrapgoTypein a pointer duringToTypeand the row coder determines field nullability by checking whether the Go field is a pointer.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40156 +/- ##
============================================
+ Coverage 58.95% 58.96% +0.01%
Complexity 4768 4768
============================================
Files 1461 1461
Lines 185377 185393 +16
Branches 3585 3585
============================================
+ Hits 109296 109324 +28
+ Misses 72738 72725 -13
- Partials 3343 3344 +1
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Reminder, please take a look at this pr: @lostluck |
|
Thanks @shunping for reviewing and merging #40155! I merged master into this branch and resolved the conflicts, so the diff shows only this PR's own changes again. The release note is now in the 2.79.0 section. Locally the schema, graphx, coder and exec package tests pass, and the char, varchar, binary and varbinary case of standard_coders.yaml passes in the fromyaml regression test. @lostluck or @shunping, I would appreciate a review when you have time. Happy to adjust anything you spot. |
| r := NewRegistry() | ||
| const urn = "test:pass_through:v1" | ||
| r.RegisterPassThroughLogicalType(urn, typeOf[string]()) | ||
| r.RegisterPassThroughLogicalType(urn, typeOf[string]()) |
There was a problem hiding this 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.
| t.Run("ConflictingRegistration", func(t *testing.T) { | ||
| defer func() { | ||
| if recover() == nil { | ||
| t.Error("conflicting registration succeeded") |
There was a problem hiding this 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.
| 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) | ||
| } | ||
| } |
There was a problem hiding this 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:
| 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) | |
| } | |
| } | |
| 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) | |
| } | |
| } |
Separate the call and check sections with blank lines, say what a failing check wanted, and explain the repeated identical registration.
|
Thanks @lostluck for the review and the patient guidance. I learned from your feedback. Fixed in 783fa9b: a comment now explains the repeated identical registration, the flagged failure message says it wanted a panic and I went through the other new checks with the same guide, and the call and check sections are separated with blank lines. I also merged master again to resolve the CHANGES.md conflict. |
Part of #39684 and the Go side of #23526. This PR is independent of #40155. The predecessor PRs #40152, #40153 and #40154 have merged; this branch merges current master and retains only pass-through registration changes.
schema.RegisterPassThroughLogicalType(urn, goType)maps an incoming logical type URN to its Go representation type. Plain Go fields continue to produce atomic schema fields. Repeated identical registration is accepted; conflicting registration returns a panic consistently with the existing logical type API.The standard fixed_char, var_char, fixed_bytes and var_bytes logical types map to string and []byte. Nullable fields map to pointers. Length arguments are accepted through the existing argument fallback and are not enforced. The char, varchar, binary and varbinary shared YAML case now runs in Go.
Validation on
2933f26f483ec96eefbc2ad0bd8ceecec4faba2e:All four packages passed.
git diff upstream/master...HEAD --checkpassed. Tests cover nullable decoding, atomic output schema preservation, multiple length arguments, repeated registration and conflicting registration. Full SDK and runner integration suites were not run.Coverage follow-up: public registration is exercised through schema conversion and row decoding/encoding. Validation tests cover empty URNs and unsupported types, verify failed registrations remain unresolved, and verify a valid registration succeeds after a rejected type. Test registrations are cleaned up; repeated shuffled runs pass.
Local schema coverage: 84.0% of statements, up from 83.3%. Both the package-level and Registry RegisterPassThroughLogicalType functions have 100.0% statement coverage. This is local Go coverage; a new hosted Codecov patch result is pending.
Hosted Analyze (go) failed because the runner disk quota was exceeded. No workflow or coverage threshold changes were made.
Pointer/nullability review fix: pass-through registration validates concrete, non-pointer Go kinds with the existing checkLogicalTypeKinds before schema conversion and insertion. Regressions cover rejected pointer/interface registrations, absence of rejected URNs, recovery with a valid registration, and required/nullable string and bytes row wire encodings. The pointer regression fails on the preceding head and passes with this fix. Focused four-package tests, repeated shuffled registration tests and schema coverage pass; schema statement coverage remains 84.0%, and both registration functions remain at 100.0%.