lostluck commented on a change in pull request #12588:
URL: https://github.com/apache/beam/pull/12588#discussion_r489607556
##########
File path: sdks/go/pkg/beam/core/runtime/exec/coder.go
##########
@@ -434,6 +600,267 @@ func convertIfNeeded(v interface{}, allocated *FullValue)
*FullValue {
return allocated
}
+type iterableEncoder struct {
+ t reflect.Type
+ enc ElementEncoder
+}
+
+func (c *iterableEncoder) Encode(val *FullValue, w io.Writer) error {
+ // Do a reflect, get the length.
+ rv := reflect.ValueOf(val.Elm)
+ size := rv.Len()
+ if err := coder.EncodeInt32((int32)(size), w); err != nil {
+ return err
+ }
+ var e FullValue
+ for i := 0; i < size; i++ {
+ e.Elm = rv.Index(i).Interface()
+ err := c.enc.Encode(&e, w)
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+type iterableDecoder struct {
+ t reflect.Type
+ dec ElementDecoder
+}
+
+func (c *iterableDecoder) DecodeTo(r io.Reader, fv *FullValue) error {
+ // (1) Read count prefixed encoded data
+
+ size, err := coder.DecodeInt32(r)
+ if err != nil {
+ return err
+ }
+ n := int(size)
+ switch {
+ case n >= 0:
+ rv, err := c.decodeToSlice(int(n), r)
Review comment:
Done.
##########
File path: sdks/go/pkg/beam/core/runtime/exec/coder.go
##########
@@ -434,6 +600,267 @@ func convertIfNeeded(v interface{}, allocated *FullValue)
*FullValue {
return allocated
}
+type iterableEncoder struct {
+ t reflect.Type
+ enc ElementEncoder
+}
+
+func (c *iterableEncoder) Encode(val *FullValue, w io.Writer) error {
+ // Do a reflect, get the length.
+ rv := reflect.ValueOf(val.Elm)
+ size := rv.Len()
+ if err := coder.EncodeInt32((int32)(size), w); err != nil {
+ return err
+ }
+ var e FullValue
+ for i := 0; i < size; i++ {
+ e.Elm = rv.Index(i).Interface()
+ err := c.enc.Encode(&e, w)
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+type iterableDecoder struct {
+ t reflect.Type
+ dec ElementDecoder
+}
+
+func (c *iterableDecoder) DecodeTo(r io.Reader, fv *FullValue) error {
+ // (1) Read count prefixed encoded data
+
+ size, err := coder.DecodeInt32(r)
+ if err != nil {
+ return err
+ }
+ n := int(size)
+ switch {
+ case n >= 0:
+ rv, err := c.decodeToSlice(int(n), r)
+ if err != nil {
+ return err
+ }
+ *fv = FullValue{Elm: rv.Interface()}
+ return nil
+ case n == -1:
+ rv := reflect.MakeSlice(c.t, 0, 0)
+ chunk, err := coder.DecodeVarInt(r)
+ if err != nil {
+ return err
+ }
+ for chunk != 0 {
+ rvi, err := c.decodeToSlice(int(chunk), r)
+ if err != nil {
+ return err
+ }
+ rv = reflect.AppendSlice(rv, rvi)
+ chunk, err = coder.DecodeVarInt(r)
+ if err != nil {
+ return err
+ }
+ }
+ *fv = FullValue{Elm: rv.Interface()}
+ }
+
+ return nil
+}
+
+func (c *iterableDecoder) decodeToSlice(n int, r io.Reader) (reflect.Value,
error) {
+ var e FullValue
+ rv := reflect.MakeSlice(c.t, n, n)
+ for i := 0; i < int(n); i++ {
Review comment:
Done.
##########
File path: sdks/go/pkg/beam/core/runtime/graphx/coder.go
##########
@@ -249,31 +256,35 @@ func (b *CoderUnmarshaller) makeCoder(c *pipepb.Coder)
(*coder.Coder, error) {
return nil, err
}
- // No payload means this coder was length prefixed by the runner
- // but is likely self describing - AKA a beam coder.
- if len(sub.GetSpec().GetPayload()) == 0 {
- return b.makeCoder(sub)
- }
// TODO(lostluck) 2018/10/17: Make this strict again, once
dataflow can use
// the portable pipeline model directly (BEAM-2885)
- if sub.GetSpec().GetUrn() != "" && sub.GetSpec().GetUrn() !=
urnCustomCoder {
- // TODO(herohde) 11/17/2017: revisit this restriction
- return nil, errors.Errorf("could not unmarshal length
prefix coder from %v, want a custom coder as a sub component but got %v", c,
sub)
- }
-
- var ref v1pb.CustomCoder
- if err :=
protox.DecodeBase64(string(sub.GetSpec().GetPayload()), &ref); err != nil {
- return nil, err
- }
- custom, err := decodeCustomCoder(&ref)
- if err != nil {
- return nil, err
+ switch u := sub.GetSpec().GetUrn(); u {
+ case "", urnCustomCoder:
+ var ref v1pb.CustomCoder
+ if err :=
protox.DecodeBase64(string(sub.GetSpec().GetPayload()), &ref); err != nil {
+ return nil, err
+ }
+ custom, err := decodeCustomCoder(&ref)
+ if err != nil {
+ return nil, err
+ }
+ custom.ID = components[0]
+ t := typex.New(custom.Type)
+ cc := &coder.Coder{Kind: coder.Custom, T: t, Custom:
custom}
+ fmt.Println("decoded customcoder", cc)
Review comment:
Done.
##########
File path: sdks/go/pkg/beam/core/runtime/exec/coder.go
##########
@@ -117,17 +176,88 @@ func MakeElementDecoder(c *coder.Coder) ElementDecoder {
return &stringDecoder{}
case coder.Custom:
- return &customDecoder{
+ dec := &customDecoder{
t: c.Custom.Type,
dec: makeDecoder(c.Custom.Dec.Fn),
}
+ fmt.Println("getting decoder", c.Custom)
+ if c.Custom.Name != "schema" {
+ return dec
+ }
+ // Custom schema coding is shorthand for using beam
infrastructure
+ // wrapped in a custom coder.
+ switch c.T.Type().Kind() {
+ case reflect.Slice:
+ return &lpDecoder{
+ dec: &iterableDecoder{
+ t: c.Custom.Type,
+ dec: dec,
+ },
+ }
+ case reflect.Array:
+ return &lpDecoder{
+ dec: &arrayDecoder{
Review comment:
That's right the encoding format for both is the same. I'll put the
comment on the array decoder itself rather than here though.
##########
File path: sdks/go/pkg/beam/core/runtime/exec/coder.go
##########
@@ -534,26 +1038,34 @@ func (*intervalWindowEncoder) EncodeSingle(elm
typex.Window, w io.Writer) error
type intervalWindowDecoder struct{}
-func (*intervalWindowDecoder) Decode(r io.Reader) ([]typex.Window, error) {
+func (d *intervalWindowDecoder) Decode(r io.Reader) ([]typex.Window, error) {
// Encoding: upper bound and duration
n, err := coder.DecodeInt32(r) // #windows
ret := make([]typex.Window, n, n)
for i := int32(0); i < n; i++ {
- end, err := coder.DecodeEventTime(r)
- if err != nil {
- return nil, err
- }
- duration, err := coder.DecodeVarUint64(r)
+ w, err := d.DecodeSingle(r)
if err != nil {
return nil, err
}
- ret[i] = window.IntervalWindow{Start:
mtime.FromMilliseconds(end.Milliseconds() - int64(duration)), End: end}
+ ret[i] = w
}
return ret, err
}
+func (*intervalWindowDecoder) DecodeSingle(r io.Reader) (typex.Window, error) {
+ end, err := coder.DecodeEventTime(r)
+ if err != nil {
+ return nil, err
+ }
+ duration, err := coder.DecodeVarInt(r)
Review comment:
Ah no difference TBH. DecodeVarInt basically just does an extra cast
from uint64 to int64, which I didn't remove.
I went back to `DecodeVarUint64` since it avoids an extra layer of if checks
and a function call.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
