sdks/go: require type registrations in Native BigQuery IO

sdks/go/pkg/beam/io/bigqueryio/bigquery.go

@@ -27,6 +27,7 @@ import (
 
 	"cloud.google.com/go/bigquery"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/util/structx"
@@ -190,13 +191,27 @@ func mustInferSchema(t reflect.Type) bigquery.Schema {
 	if t.Kind() != reflect.Struct {
 		panic(fmt.Sprintf("schema type must be struct: %v", t))
 	}
+
+	registerTypeIfNeeded(t)
+
 	schema, err := bigquery.InferSchema(reflect.Zero(t).Interface())
 	if err != nil {
 		panic(errors.Wrapf(err, "invalid schema type: %v", t))
 	}
 	return schema
 }
 
+func registerTypeIfNeeded(t reflect.Type) {
+	t = reflectx.SkipPtr(t)
+	key, ok := runtime.TypeKey(t)
+	if !ok {
+		panic(fmt.Sprintf("type %v must be a named type (not anonymous) for registration", t))
+	}
+	if _, registered := runtime.LookupType(key); !registered {
+		runtime.RegisterType(t)
+	}
+}
+
 func mustParseTable(table string) QualifiedTableName {
 	qn, err := NewQualifiedTableName(table)
 	if err != nil {

sdks/go/pkg/beam/io/bigqueryio/bigquery_test.go

@@ -16,8 +16,12 @@
 package bigqueryio
 
 import (
+	"errors"
 	"reflect"
 	"testing"
+
+	"cloud.google.com/go/bigquery"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
 )
 
 func TestNewQualifiedTableName(t *testing.T) {
@@ -76,3 +80,70 @@ func Test_constructSelectStatementPanic(t *testing.T) {
 		constructSelectStatement(typ, tagKey, table)
 	})
 }
+
+func Test_mustInferSchema(t *testing.T) {
+	type TestSchema struct {
+		Name   bigquery.NullString   `bigquery:"name"`
+		Active bigquery.NullBool     `bigquery:"active"`
+		Score  bigquery.NullFloat64  `bigquery:"score"`
+		Time   bigquery.NullDateTime `bigquery:"time"`
+	}
+
+	tests := []struct {
+		name    string
+		input   interface{}
+		wantErr bool
+		verify  func(reflect.Type) error
+	}{
+		{
+			name:    "NewType_ShouldRegisterSuccessfully",
+			input:   TestSchema{},
+			wantErr: false,
+			verify: func(t reflect.Type) error {
+				// Verify successful type registration in runtime registry.
+				if key, ok := runtime.TypeKey(t); ok {
+					if _, registered := runtime.LookupType(key); !registered {
+						return errors.New("Type was not properly registered")
+					}
+				}
+				return nil
+			},
+		},
+		{
+			name:    "AlreadyRegisteredType_ShouldNotPanic",
+			input:   TestSchema{},
+			wantErr: false,
+			verify: func(t reflect.Type) error {
+				// Verify re-registration of existing type is handled correctly.
+				mustInferSchema(t)
+				return nil
+			},
+		},
+		{
+			name:    "AnonymousStruct_ShouldPanic",
+			input:   struct{}{},
+			wantErr: true,
+			verify:  func(t reflect.Type) error { return nil },
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			defer func() {
+				r := recover()
+				if (r != nil) != tt.wantErr {
+					t.Errorf("mustInferSchema() panic = %v, wantErr %v", r, tt.wantErr)
+				}
+			}()
+
+			typ := reflect.TypeOf(tt.input)
+			mustInferSchema(typ)
+			if tt.wantErr {
+				t.Fatal("Expected panic did not occur")
+			}
+			if err := tt.verify(typ); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}