feat(bigquery/storage/managedwriter/adapt): add schema -> proto support

bigquery/storage/managedwriter/adapt/error.go

@@ -0,0 +1,36 @@
+// Copyright 2021 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package adapt
+
+import "fmt"
+
+type ConversionError struct {
+	Location string
+	Details  error
+}
+
+func (ce *ConversionError) Error() string {
+	if ce.Location == "" {
+		return ce.Details.Error()
+	}
+	return fmt.Sprintf("Location: %s, Error: %v", ce.Location, ce.Details)
+}
+
+func newConversionError(loc string, err error) *ConversionError {
+	return &ConversionError{
+		Location: loc,
+		Details:  err,
+	}
+}

bigquery/storage/managedwriter/adapt/protoconversion.go

@@ -116,7 +116,7 @@ func StorageSchemaToDescriptor(inSchema *storagepb.TableSchema, scope string) (p
 // internal implementation of the conversion code.
 func storageSchemaToDescriptorInternal(inSchema *storagepb.TableSchema, scope string, cache *dependencyCache, allowWrapperTypes bool) (protoreflect.Descriptor, error) {
 	if inSchema == nil {
-		return nil, fmt.Errorf("no input schema provided")
+		return nil, newConversionError(scope, fmt.Errorf("no input schema was provided"))
 	}
 
 	var fields []*descriptorpb.FieldDescriptorProto
@@ -147,7 +147,7 @@ func storageSchemaToDescriptorInternal(inSchema *storagepb.TableSchema, scope st
 				// construct field descriptor for the message
 				fdp, err := tableFieldSchemaToFieldDescriptorProto(f, fNumber, string(foundDesc.FullName()), allowWrapperTypes)
 				if err != nil {
-					return nil, fmt.Errorf("couldn't convert to FieldDescriptorProto (%s): %v", currentScope, err)
+					return nil, newConversionError(scope, fmt.Errorf("couldn't convert field to FieldDescriptorProto: %v", err))
 				}
 				fields = append(fields, fdp)
 			} else {
@@ -157,25 +157,25 @@ func storageSchemaToDescriptorInternal(inSchema *storagepb.TableSchema, scope st
 				}
 				desc, err := storageSchemaToDescriptorInternal(ts, currentScope, cache, allowWrapperTypes)
 				if err != nil {
-					return nil, fmt.Errorf("couldn't compile (%s): %v", currentScope, err)
+					return nil, newConversionError(currentScope, fmt.Errorf("couldn't convert message: %v", err))
 				}
 				// Now that we have the submessage definition, we append it both to the local dependencies, as well
 				// as inserting it into the cache for possible reuse elsewhere.
 				deps = append(deps, desc.ParentFile())
 				err = cache.add(ts, desc)
 				if err != nil {
-					return nil, fmt.Errorf("failed to add descriptor to cache: %v", err)
+					return nil, newConversionError(currentScope, fmt.Errorf("failed to add descriptor to dependency cache: %v", err))
 				}
 				fdp, err := tableFieldSchemaToFieldDescriptorProto(f, fNumber, currentScope, allowWrapperTypes)
 				if err != nil {
-					return nil, fmt.Errorf("couldn't compute field schema (%s): %v", currentScope, err)
+					return nil, newConversionError(currentScope, fmt.Errorf("couldn't compute field schema : %v", err))
 				}
 				fields = append(fields, fdp)
 			}
 		} else {
 			fd, err := tableFieldSchemaToFieldDescriptorProto(f, fNumber, currentScope, allowWrapperTypes)
 			if err != nil {
-				return nil, err
+				return nil, newConversionError(currentScope, err)
 			}
 			fields = append(fields, fd)
 		}