pequalsnp commented on issue #28959:
URL: https://github.com/apache/beam/issues/28959#issuecomment-1768932524
I did write a coder, essentially like this (I took this from an SDF I
re-wrote):
```
beam.RegisterCoder(reflect.TypeOf(SpannerPartitionRestriction{}),
func(read SpannerPartitionRestriction) ([]byte, error) {
tid, err := read.TransactionID.MarshalBinary()
if err != nil {
return nil, err
}
ps := [][]byte{}
for _, partition := range read.Partitions {
p, err := partition.MarshalBinary()
if err != nil {
return nil, err
}
ps = append(ps, p)
}
var buf bytes.Buffer
enc := gob.NewEncoder(&buf)
err = enc.Encode(rawSpannerPartitionRestriction{Tid:
tid, Ps: ps})
if err != nil {
return nil, err
}
return buf.Bytes(), nil
},
func(raw []byte) (SpannerPartitionRestriction, error) {
buf := bytes.NewBuffer(raw)
dec := gob.NewDecoder(buf)
var rawStruct rawSpannerPartitionRestriction
err := dec.Decode(&rawStruct)
if err != nil {
return SpannerPartitionRestriction{}, err
}
tid := spanner.BatchReadOnlyTransactionID{}
err = tid.UnmarshalBinary(rawStruct.Tid)
if err != nil {
return SpannerPartitionRestriction{}, err
}
var ps []*spanner.Partition
for _, rawP := range rawStruct.Ps {
p := spanner.Partition{}
err = p.UnmarshalBinary(rawP)
if err != nil {
return SpannerPartitionRestriction{},
err
}
ps = append(ps, &p)
}
return SpannerPartitionRestriction{TransactionID: tid,
Partitions: ps}, nil
})
```
That gets rid of that error, but there were other issues. For example,
client is nil in `CreateInitialRestriction` because `Setup` hasn't been called
yet. Not sure if that's a bug in the library or intentional.
--
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]
