lostluck commented on a change in pull request #14192:
URL: https://github.com/apache/beam/pull/14192#discussion_r596209027
##########
File path: sdks/go/pkg/beam/core/runtime/graphx/schema/schema.go
##########
@@ -280,82 +301,194 @@ 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) {
+ // 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]
+ 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 ftype, lID, nil
+ }
+ for _, lti := range r.logicalTypeInterfaces {
+ if !t.Implements(lti) {
+ continue
+ }
+ 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)
+ }
+ 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 ftype, t.String(), nil
+ }
+ return nil, "", nil
+}
+
func (r *Registry) fromType(ot reflect.Type) (*pipepb.Schema, error) {
- if reflectx.SkipPtr(ot).Kind() != reflect.Struct {
- return nil, errors.Errorf("cannot convert %v to schema.
FromType only converts structs to schemas", ot)
+ if schm, ok := r.typeToSchema[ot]; ok {
+ return schm, nil
}
- schm, err := r.structToSchema(ot)
+ ftype, lID, err := r.logicalTypeToFieldType(ot)
if err != nil {
return nil, err
}
- if ot.Kind() == reflect.Ptr {
- schm.Options = append(schm.Options, &pipepb.Option{
- Name: optGoNillable,
- })
+ if ftype != nil {
+ schm := ftype.GetRowType().GetSchema()
+ schm = proto.Clone(schm).(*pipepb.Schema)
+ if ot.Kind() == reflect.Ptr {
+ schm.Options = append(schm.Options, &pipepb.Option{
+ Name: optGoNillable,
+ })
+ }
+ if lID != "" {
+ schm.Options = append(schm.Options, logicalOption(lID))
+ }
+ schm.Id = r.getNextID()
+ r.typeToSchema[ot] = schm
+ r.idToType[schm.GetId()] = ot
+ return schm, nil
}
- return schm, nil
+
+ t := reflectx.SkipPtr(ot)
+
+ schm, err := r.structToSchema(t)
+ if err != nil {
+ return nil, err
+ }
+ // Cache the pointer type here with it's own id.
+ pt := reflect.PtrTo(t)
+ schm = proto.Clone(schm).(*pipepb.Schema)
+ schm.Id = r.getNextID()
+ schm.Options = append(schm.Options, &pipepb.Option{
+ Name: optGoNillable,
+ })
+ r.idToType[schm.GetId()] = pt
+ r.typeToSchema[pt] = schm
+
+ // Return whatever the original type was.
+ return r.typeToSchema[ot], nil
Review comment:
Combined addressing this comment with the earlier one by adding a
comment to fromType about it's purpose.
----------------------------------------------------------------
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]
