diff --git a/CHANGES.md b/CHANGES.md
index 354f971acc6c..9b01a8e035d3 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -74,6 +74,7 @@
 * (Go) Pipeline-level `--resource_hints` are now forwarded to expansion services, so they apply to cross-language transforms as they do in the Java and Python SDKs. The `max_active_bundles_per_worker` hint name is now also accepted by `--resource_hints` ([#23893](https://github.com/apache/beam/issues/23893)).
 * ReadFromBigQuery now supports Lakehouse runtime catalog (BigLake metastore) tables with `method=DIRECT_READ`, using 4-part `project.catalog.namespace.table` identifiers. Previously `project:catalog.namespace.table` was silently mis-parsed and tables that report no `numBytes` failed to split (Python) ([#39597](https://github.com/apache/beam/issues/39597)).
 * (Python) Added `AwsOptions` with `--aws_*` pipeline options (e.g. `--aws_access_key_id`, `--aws_region_name`, `--aws_endpoint_url`) for configuring AWS service clients such as S3 ([#40528](https://github.com/apache/beam/issues/40528)).
+* (Go) Added `schema.RegisterLogicalTypeConversion` and `schema.RegisterLogicalTypeCoder` to register portable logical types in one call, and Go types for the standard `beam:logical_type:date:v1` and `beam:logical_type:micros_instant:v1` logical types, so schemas from other SDKs with these types can be used in cross-language pipelines ([#39684](https://github.com/apache/beam/issues/39684)).
 
 ## Breaking Changes
 
diff --git a/sdks/go/pkg/beam/core/graph/coder/row.go b/sdks/go/pkg/beam/core/graph/coder/row.go
index f958d84154c1..0c2a6872dd69 100644
--- a/sdks/go/pkg/beam/core/graph/coder/row.go
+++ b/sdks/go/pkg/beam/core/graph/coder/row.go
@@ -226,3 +226,19 @@ func ReadSimpleRowHeader(fields int, r io.Reader) error {
 	}
 	return nil
 }
+
+// RowFieldEncoderForType returns an encoding function for values of the given
+// type, using the encoding of that type as a field of a beam schema row.
+//
+// Returns an error if the given type cannot be a field of a beam schema row.
+func RowFieldEncoderForType(rt reflect.Type) (func(any, io.Writer) error, error) {
+	return defaultEnc.fieldEncoderForType(rt)
+}
+
+// RowFieldDecoderForType returns a decoding function for values of the given
+// type, using the encoding of that type as a field of a beam schema row.
+//
+// Returns an error if the given type cannot be a field of a beam schema row.
+func RowFieldDecoderForType(rt reflect.Type) (func(io.Reader) (any, error), error) {
+	return defaultDec.fieldDecoderForType(rt)
+}
diff --git a/sdks/go/pkg/beam/core/graph/coder/row_decoder.go b/sdks/go/pkg/beam/core/graph/coder/row_decoder.go
index 90938c237fe3..2402cafe0e2b 100644
--- a/sdks/go/pkg/beam/core/graph/coder/row_decoder.go
+++ b/sdks/go/pkg/beam/core/graph/coder/row_decoder.go
@@ -95,6 +95,23 @@ func (b *RowDecoderBuilder) decoderForType(t reflect.Type) (func(io.Reader) (any
 	if t.Kind() == reflect.Ptr {
 		isPtr = true
 		t = t.Elem()
+		// A provider for the value type also handles a top level pointer.
+		f, addr, err := b.customFunc(t)
+		if err != nil {
+			return nil, err
+		}
+		if f != nil && !addr {
+			vt := t
+			return func(r io.Reader) (any, error) {
+				v, err := f(r)
+				if err != nil {
+					return nil, err
+				}
+				rv := reflect.New(vt)
+				rv.Elem().Set(reflect.ValueOf(v))
+				return rv.Interface(), nil
+			}, nil
+		}
 	}
 	dec, err := b.decoderForStructReflect(t)
 	if err != nil {
@@ -435,3 +452,23 @@ type typeDecoderFieldReflect struct {
 	// of the field value (i.e. &foo.bar) and not the field value (i.e. foo.bar).
 	addr bool
 }
+
+// fieldDecoderForType returns a decoder for values of t, using the encoding
+// of t as a row field.
+func (b *RowDecoderBuilder) fieldDecoderForType(t reflect.Type) (func(io.Reader) (any, error), error) {
+	decf, err := b.decoderForSingleTypeReflect(t)
+	if err != nil {
+		return nil, err
+	}
+	return func(r io.Reader) (any, error) {
+		rv := reflect.New(t)
+		v := rv.Elem()
+		if decf.addr {
+			v = rv
+		}
+		if err := decf.decode(v, r); err != nil {
+			return nil, err
+		}
+		return rv.Elem().Interface(), nil
+	}, nil
+}
diff --git a/sdks/go/pkg/beam/core/graph/coder/row_encoder.go b/sdks/go/pkg/beam/core/graph/coder/row_encoder.go
index 6f5c4c04e81c..05d3e85b2d98 100644
--- a/sdks/go/pkg/beam/core/graph/coder/row_encoder.go
+++ b/sdks/go/pkg/beam/core/graph/coder/row_encoder.go
@@ -363,3 +363,24 @@ func (b *RowEncoderBuilder) encoderForStructReflect(t reflect.Type) (func(reflec
 		return nil
 	}, nil
 }
+
+// fieldEncoderForType returns an encoder for values of t, using the encoding
+// of t as a row field.
+func (b *RowEncoderBuilder) fieldEncoderForType(t reflect.Type) (func(any, io.Writer) error, error) {
+	encf, err := b.encoderForSingleTypeReflect(t)
+	if err != nil {
+		return nil, err
+	}
+	return func(v any, w io.Writer) error {
+		rv := reflect.ValueOf(v)
+		if !rv.IsValid() || (rv.Kind() == reflect.Ptr && rv.IsNil()) {
+			return errors.Errorf("cannot encode a nil value as a row field of type %v", t)
+		}
+		if encf.addr {
+			p := reflect.New(t)
+			p.Elem().Set(rv)
+			rv = p
+		}
+		return encf.encode(rv, w)
+	}, nil
+}
diff --git a/sdks/go/pkg/beam/core/graph/coder/row_test.go b/sdks/go/pkg/beam/core/graph/coder/row_test.go
index c2a925efd306..8b6782457d81 100644
--- a/sdks/go/pkg/beam/core/graph/coder/row_test.go
+++ b/sdks/go/pkg/beam/core/graph/coder/row_test.go
@@ -945,3 +945,108 @@ func TestReflectionRowCoderGeneration_Int16BigEndian(t *testing.T) {
 		t.Fatalf("dec(enc(%v)) = %v\ndiff (-want, +got): %v", want, got, d)
 	}
 }
+
+func TestRowFieldCoderForType(t *testing.T) {
+	type pair struct {
+		A int64
+		B string
+	}
+	tests := []struct {
+		value     any
+		wantBytes []byte
+	}{
+		{value: int64(999), wantBytes: []byte{0xe7, 0x07}},
+		{value: int16(999), wantBytes: []byte{0x03, 0xe7}},
+		{value: "abc", wantBytes: []byte{0x03, 'a', 'b', 'c'}},
+		{value: []byte{1, 2}, wantBytes: []byte{0x02, 1, 2}},
+		{value: true, wantBytes: []byte{0x01}},
+		{value: []int32{1, 2}, wantBytes: []byte{0x00, 0x00, 0x00, 0x02, 0x01, 0x02}},
+		{value: pair{A: 7, B: "x"}, wantBytes: []byte{0x02, 0x00, 0x07, 0x01, 'x'}},
+	}
+	for _, test := range tests {
+		rt := reflect.TypeOf(test.value)
+		t.Run(rt.String(), func(t *testing.T) {
+			enc, err := RowFieldEncoderForType(rt)
+			if err != nil {
+				t.Fatalf("RowFieldEncoderForType(%v) = %v, want nil error", rt, err)
+			}
+			var buf bytes.Buffer
+			if err := enc(test.value, &buf); err != nil {
+				t.Fatalf("enc(%v) = %v, want nil error", test.value, err)
+			}
+			if got := buf.Bytes(); !bytes.Equal(got, test.wantBytes) {
+				t.Fatalf("enc(%v) = %v, want %v", test.value, got, test.wantBytes)
+			}
+			dec, err := RowFieldDecoderForType(rt)
+			if err != nil {
+				t.Fatalf("RowFieldDecoderForType(%v) = %v, want nil error", rt, err)
+			}
+			got, err := dec(bytes.NewBuffer(test.wantBytes))
+			if err != nil {
+				t.Fatalf("dec(%v) = %v, want nil error", test.wantBytes, err)
+			}
+			if d := cmp.Diff(test.value, got); d != "" {
+				t.Fatalf("dec(enc(%v)) = %v\ndiff (-want, +got): %v", test.value, got, d)
+			}
+		})
+	}
+	if _, err := RowFieldEncoderForType(reflect.TypeOf(func() {})); err == nil {
+		t.Errorf("RowFieldEncoderForType(func) = nil error, want error")
+	}
+	enc, err := RowFieldEncoderForType(reflect.TypeOf(int64(0)))
+	if err != nil {
+		t.Fatalf("RowFieldEncoderForType(int64) = %v, want nil error", err)
+	}
+	if err := enc(nil, &bytes.Buffer{}); err == nil {
+		t.Errorf("enc(nil) = nil error, want error")
+	}
+	ptrEnc, err := RowFieldEncoderForType(reflect.TypeOf((*int64)(nil)))
+	if err != nil {
+		t.Fatalf("RowFieldEncoderForType(*int64) = %v, want nil error", err)
+	}
+	if err := ptrEnc((*int64)(nil), &bytes.Buffer{}); err == nil {
+		t.Errorf("enc((*int64)(nil)) = nil error, want error")
+	}
+	if _, err := RowFieldDecoderForType(reflect.TypeOf(func() {})); err == nil {
+		t.Errorf("RowFieldDecoderForType(func) = nil error, want error")
+	}
+}
+
+// A provider for a value type also encodes and decodes a top level pointer to
+// that type.
+func TestRowCoder_TopLevelPointerWithValueProvider(t *testing.T) {
+	type tenths struct{ V int64 }
+	rt := reflect.TypeOf(tenths{})
+	var encB RowEncoderBuilder
+	encB.Register(rt, func(reflect.Type) (func(any, io.Writer) error, error) {
+		return func(v any, w io.Writer) error { return EncodeVarInt(v.(tenths).V*10, w) }, nil
+	})
+	var decB RowDecoderBuilder
+	decB.Register(rt, func(reflect.Type) (func(io.Reader) (any, error), error) {
+		return func(r io.Reader) (any, error) {
+			v, err := DecodeVarInt(r)
+			return tenths{V: v / 10}, err
+		}, nil
+	})
+	prt := reflect.PointerTo(rt)
+	enc, err := encB.Build(prt)
+	if err != nil {
+		t.Fatalf("RowEncoderBuilder.Build(%v) = %v, want nil error", prt, err)
+	}
+	dec, err := decB.Build(prt)
+	if err != nil {
+		t.Fatalf("RowDecoderBuilder.Build(%v) = %v, want nil error", prt, err)
+	}
+	want := &tenths{V: 42}
+	var buf bytes.Buffer
+	if err := enc(want, &buf); err != nil {
+		t.Fatalf("enc(%v) = %v, want nil error", want, err)
+	}
+	got, err := dec(&buf)
+	if err != nil {
+		t.Fatalf("dec = %v, want nil error", err)
+	}
+	if d := cmp.Diff(want, got); d != "" {
+		t.Errorf("dec(enc(%v)) diff (-want, +got): %v", want, d)
+	}
+}
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go b/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
index ee08bbb8f549..7a711d55a8a4 100644
--- a/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go
@@ -17,9 +17,11 @@ package schema
 
 import (
 	"fmt"
+	"io"
 	"reflect"
 	"sync"
 
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
 	pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
 )
@@ -224,4 +226,115 @@ func preRegLogicalTypes(r *Registry) {
 
 func init() {
 	preRegLogicalTypes(defaultRegistry)
+	registerStandardLogicalTypes(defaultRegistry)
+}
+
+// RegisterLogicalTypeConversion registers a portable logical type identified
+// by urn. Values of GoT are stored in schema rows as values of StorageT,
+// converted with toStorage and fromStorage, and encoded with the row field
+// encoding of StorageT. StorageT must have a schema representation: an atomic
+// type, a struct of such types, or a registered logical type. GoT must be a
+// concrete type. A struct kind GoT with a non row StorageT can be used as a
+// field of a row, but not as a top level schema type.
+//
+// RegisterLogicalTypeConversion must be called before beam.Init(), and
+// conventionally is called in a package init() function.
+func RegisterLogicalTypeConversion[GoT, StorageT any](urn string, toStorage func(GoT) (StorageT, error), fromStorage func(StorageT) (GoT, error)) {
+	registerLogicalTypeConversion(defaultRegistry, ToLogicalType(urn, typeOf[GoT](), typeOf[StorageT]()), toStorage, fromStorage)
+}
+
+// RegisterParameterizedLogicalTypeConversion is RegisterLogicalTypeConversion
+// for a logical type with an argument, such as a timestamp and its precision.
+// Each argument value is registered with its own Go type, and a schema field
+// with the urn and that argument maps to GoT. See ToLogicalTypeWithArgument
+// for the supported argument types.
+func RegisterParameterizedLogicalTypeConversion[GoT, StorageT any](urn string, argument any, toStorage func(GoT) (StorageT, error), fromStorage func(StorageT) (GoT, error)) {
+	registerLogicalTypeConversion(defaultRegistry, ToLogicalTypeWithArgument(urn, typeOf[GoT](), typeOf[StorageT](), argument), toStorage, fromStorage)
+}
+
+// RegisterLogicalTypeCoder registers a portable logical type identified by
+// urn whose wire encoding differs from the row field encoding of StorageT.
+// StorageT is only used as the schema representation of the logical type.
+// enc and dec produce and consume the encoded bytes of a value directly, in
+// the wire encoding that the urn defines. GoT must be a concrete type.
+// StorageT cannot be inferred, so both type parameters are given explicitly,
+// for example
+// RegisterLogicalTypeCoder[Money, []byte]("example:money:v1", encodeMoney, decodeMoney).
+//
+// RegisterLogicalTypeCoder must be called before beam.Init(), and
+// conventionally is called in a package init() function.
+func RegisterLogicalTypeCoder[GoT, StorageT any](urn string, enc func(GoT, io.Writer) error, dec func(io.Reader) (GoT, error)) {
+	registerLogicalTypeCoder(defaultRegistry, ToLogicalType(urn, typeOf[GoT](), typeOf[StorageT]()), enc, dec)
+}
+
+func typeOf[T any]() reflect.Type {
+	return reflect.TypeOf((*T)(nil)).Elem()
+}
+
+// registerLogicalTypeConversion registers lt with the registry and registers
+// row coder providers for its Go type, built from the row field coder of the
+// storage type and the conversion functions.
+func registerLogicalTypeConversion[GoT, StorageT any](r *Registry, lt LogicalType, toStorage func(GoT) (StorageT, error), fromStorage func(StorageT) (GoT, error)) {
+	st := lt.StorageType()
+	checkLogicalTypeKinds(lt)
+	r.RegisterLogicalType(lt)
+	coder.RegisterSchemaProviders(lt.GoType(),
+		func(reflect.Type) (func(any, io.Writer) error, error) {
+			enc, err := coder.RowFieldEncoderForType(st)
+			if err != nil {
+				return nil, err
+			}
+			return func(v any, w io.Writer) error {
+				s, err := toStorage(v.(GoT))
+				if err != nil {
+					return err
+				}
+				return enc(s, w)
+			}, nil
+		},
+		func(reflect.Type) (func(io.Reader) (any, error), error) {
+			dec, err := coder.RowFieldDecoderForType(st)
+			if err != nil {
+				return nil, err
+			}
+			return func(rd io.Reader) (any, error) {
+				s, err := dec(rd)
+				if err != nil {
+					return nil, err
+				}
+				return fromStorage(s.(StorageT))
+			}, nil
+		})
+}
+
+// registerLogicalTypeCoder registers lt with the registry and registers enc
+// and dec as the row coder providers for its Go type.
+func registerLogicalTypeCoder[GoT any](r *Registry, lt LogicalType, enc func(GoT, io.Writer) error, dec func(io.Reader) (GoT, error)) {
+	checkLogicalTypeKinds(lt)
+	r.RegisterLogicalType(lt)
+	coder.RegisterSchemaProviders(lt.GoType(),
+		func(reflect.Type) (func(any, io.Writer) error, error) {
+			return func(v any, w io.Writer) error {
+				return enc(v.(GoT), w)
+			}, nil
+		},
+		func(reflect.Type) (func(io.Reader) (any, error), error) {
+			return func(rd io.Reader) (any, error) {
+				return dec(rd)
+			}, nil
+		})
+}
+
+// checkLogicalTypeKinds panics when the Go type of lt is an interface or its
+// storage type is a pointer. The row coder matches an interface by all of its
+// implementations, and the schema matches only the interface itself. The row
+// coders decide nullability from the Go type of a field, so a pointer storage
+// type behind a non pointer Go type would fail at encoding time.
+func checkLogicalTypeKinds(lt LogicalType) {
+	if lt.GoType().Kind() == reflect.Interface {
+		panic(fmt.Sprintf("LogicalType[%v] has the interface Go type %v, Go types must be concrete", lt.ID(), lt.GoType()))
+	}
+	if lt.StorageType().Kind() == reflect.Ptr {
+		panic(fmt.Sprintf("LogicalType[%v] has the pointer storage type %v, storage types must not be pointers", lt.ID(), lt.StorageType()))
+	}
 }
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go
new file mode 100644
index 000000000000..0deadd8817a2
--- /dev/null
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types.go
@@ -0,0 +1,136 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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
+//
+//    http://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 schema
+
+import (
+	"time"
+
+	"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"
+)
+
+// URNs of the standard logical types defined in schema.proto.
+const (
+	// URNDate identifies the Date logical type.
+	URNDate = "beam:logical_type:date:v1"
+	// URNMicrosInstant identifies the MicrosInstant logical type.
+	URNMicrosInstant = "beam:logical_type:micros_instant:v1"
+)
+
+// Date is a calendar date without a time zone or time of day, the
+// beam:logical_type:date:v1 logical type. It is stored as the number of days
+// since 1970-01-01 in an INT64 field. Time and String normalize out of range
+// fields as time.Date does. Encoding fails unless the fields form a valid
+// date with a year from -999999999 to 999999999, and decoding fails for a day
+// count outside that range.
+type Date struct {
+	Year  int
+	Month time.Month
+	Day   int
+}
+
+// DateOf returns the Date of t in t's location.
+func DateOf(t time.Time) Date {
+	y, m, d := t.Date()
+	return Date{Year: y, Month: m, Day: d}
+}
+
+// Time returns the start of the date in UTC.
+func (d Date) Time() time.Time {
+	return time.Date(d.Year, d.Month, d.Day, 0, 0, 0, 0, time.UTC)
+}
+
+// String returns the date in ISO 8601 format, such as 2006-01-02.
+func (d Date) String() string {
+	return d.Time().Format("2006-01-02")
+}
+
+const secondsPerDay = 24 * 60 * 60
+
+// minDateDays and maxDateDays bound the supported days since the epoch.
+var (
+	minDateDays = time.Date(-999999999, time.January, 1, 0, 0, 0, 0, time.UTC).Unix() / secondsPerDay
+	maxDateDays = time.Date(999999999, time.December, 31, 0, 0, 0, 0, time.UTC).Unix() / secondsPerDay
+)
+
+func dateToStorage(d Date) (int64, error) {
+	t := d.Time()
+	// Out of range fields can overflow in time.Date, so only a date that
+	// round trips is encoded.
+	if DateOf(t) != d {
+		return 0, errors.Errorf("%d-%02d-%02d is not a valid date", d.Year, d.Month, d.Day)
+	}
+	days := t.Unix() / secondsPerDay
+	if days < minDateDays || days > maxDateDays {
+		return 0, errors.Errorf("date %v is out of range", d)
+	}
+	return days, nil
+}
+
+func dateFromStorage(days int64) (Date, error) {
+	if days < minDateDays || days > maxDateDays {
+		return Date{}, errors.Errorf("%v days since the epoch is out of the supported date range", days)
+	}
+	return DateOf(time.Unix(days*secondsPerDay, 0).UTC()), nil
+}
+
+// MicrosInstant is a timestamp with microsecond precision, the
+// beam:logical_type:micros_instant:v1 logical type. It is stored as a row of
+// the seconds and the microseconds since the epoch. Encoding a value with
+// sub microsecond precision fails; truncate such values with time.Time.Truncate
+// first.
+type MicrosInstant time.Time
+
+// Time returns the instant as a time.Time.
+func (m MicrosInstant) Time() time.Time {
+	return time.Time(m)
+}
+
+// microsInstantStorage is the storage type of MicrosInstant.
+type microsInstantStorage struct {
+	Seconds int64 `beam:"seconds"`
+	Micros  int64 `beam:"micros"`
+}
+
+func microsInstantToStorage(m MicrosInstant) (microsInstantStorage, error) {
+	t := time.Time(m)
+	nanos := t.Nanosecond()
+	if nanos%1000 != 0 {
+		return microsInstantStorage{}, errors.Errorf("MicrosInstant %v has sub microsecond precision", t)
+	}
+	return microsInstantStorage{Seconds: t.Unix(), Micros: int64(nanos / 1000)}, nil
+}
+
+func microsInstantFromStorage(s microsInstantStorage) (MicrosInstant, error) {
+	if s.Micros < 0 || s.Micros > 999999 {
+		return MicrosInstant{}, errors.Errorf("MicrosInstant micros %v is out of range [0, 999999]", s.Micros)
+	}
+	return MicrosInstant(time.Unix(s.Seconds, s.Micros*1000).UTC()), nil
+}
+
+// registerStandardLogicalTypes registers the standard logical types with the
+// registry.
+func registerStandardLogicalTypes(r *Registry) {
+	registerLogicalTypeConversion(r, ToLogicalType(URNDate, typeOf[Date](), reflectx.Int64), dateToStorage, dateFromStorage)
+	registerLogicalTypeConversion(r, ToLogicalType(URNMicrosInstant, typeOf[MicrosInstant](), typeOf[microsInstantStorage]()), microsInstantToStorage, microsInstantFromStorage)
+}
+
+func init() {
+	// Pipeline graphs serialize the standard logical types by name.
+	runtime.RegisterType(typeOf[Date]())
+	runtime.RegisterType(typeOf[MicrosInstant]())
+}
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go
new file mode 100644
index 000000000000..02a2540ab626
--- /dev/null
+++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/standard_logical_types_test.go
@@ -0,0 +1,391 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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
+//
+//    http://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 schema
+
+import (
+	"bytes"
+	"encoding/binary"
+	"io"
+	"math"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+	pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/protobuf/encoding/prototext"
+	"google.golang.org/protobuf/testing/protocmp"
+)
+
+func atomicType(at pipepb.AtomicType) *pipepb.FieldType {
+	return &pipepb.FieldType{TypeInfo: &pipepb.FieldType_AtomicType{AtomicType: at}}
+}
+
+func logicalFieldType(urn string, representation *pipepb.FieldType) *pipepb.FieldType {
+	return &pipepb.FieldType{
+		TypeInfo: &pipepb.FieldType_LogicalType{
+			LogicalType: &pipepb.LogicalType{
+				Urn:            urn,
+				Representation: representation,
+			},
+		},
+	}
+}
+
+// standardRow uses the standard logical types as fields, with the same field
+// names as the micros_instant case of standard_coders.yaml.
+type standardRow struct {
+	F_timestamp MicrosInstant `beam:"f_timestamp"`
+	F_string    string        `beam:"f_string"`
+	F_int       int64         `beam:"f_int"`
+	F_date      Date          `beam:"f_date"`
+}
+
+func TestStandardLogicalTypes_Schema(t *testing.T) {
+	rt := reflect.TypeOf(standardRow{})
+	want := &pipepb.Schema{
+		Fields: []*pipepb.Field{
+			{Name: "f_timestamp", Type: logicalFieldType(URNMicrosInstant, &pipepb.FieldType{
+				TypeInfo: &pipepb.FieldType_RowType{RowType: &pipepb.RowType{Schema: &pipepb.Schema{
+					Fields: []*pipepb.Field{
+						{Name: "seconds", Type: atomicType(pipepb.AtomicType_INT64)},
+						{Name: "micros", Type: atomicType(pipepb.AtomicType_INT64)},
+					},
+				}}},
+			})},
+			{Name: "f_string", Type: atomicType(pipepb.AtomicType_STRING)},
+			{Name: "f_int", Type: atomicType(pipepb.AtomicType_INT64)},
+			{Name: "f_date", Type: logicalFieldType(URNDate, atomicType(pipepb.AtomicType_INT64))},
+		},
+	}
+	got, err := FromType(rt)
+	if err != nil {
+		t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+	}
+	if d := cmp.Diff(want, got, protocmp.Transform(), protocmp.IgnoreFields(&pipepb.Schema{}, "id")); d != "" {
+		t.Errorf("FromType(%v) diff (-want, +got): %v", rt, d)
+	}
+	// A schema from another SDK has a different id, and must still map to the
+	// standard Go types.
+	want.Id = "standard-logical-types-test"
+	gotType, err := ToType(want)
+	if err != nil {
+		t.Fatalf("ToType(%v) = %v, want nil error", prototext.Format(want), err)
+	}
+	for i, wantField := range []reflect.Type{typeOf[MicrosInstant](), typeOf[string](), typeOf[int64](), typeOf[Date]()} {
+		if got := gotType.Field(i).Type; got != wantField {
+			t.Errorf("ToType field %d type = %v, want %v", i, got, wantField)
+		}
+	}
+}
+
+func TestStandardLogicalTypes_RowEncoding(t *testing.T) {
+	// The bytes are the micros_instant example of standard_coders.yaml, with a
+	// trailing date field appended: 2020-08-13 is 18487 days since the epoch.
+	want := standardRow{
+		F_timestamp: MicrosInstant(time.Date(2020, 8, 13, 14, 14, 14, 123456000, time.UTC)),
+		F_string:    "2020-08-13T14:14:14.123456Z",
+		F_int:       1597328054123456,
+		F_date:      Date{Year: 2020, Month: time.August, Day: 13},
+	}
+	wantBytes := []byte("\x04\x00\x02\x00\xb6\x95\xd5\xf9\x05\xc0\xc4\x07\x1b2020-08-13T14:14:14.123456Z\xc0\xf7\x85\xda\xae\x98\xeb\x02\xb7\x90\x01")
+	rt := reflect.TypeOf(want)
+	enc, err := coder.RowEncoderForStruct(rt)
+	if err != nil {
+		t.Fatalf("RowEncoderForStruct(%v) = %v, want nil error", rt, err)
+	}
+	var buf bytes.Buffer
+	if err := enc(want, &buf); err != nil {
+		t.Fatalf("enc(%v) = %v, want nil error", want, err)
+	}
+	if got := buf.Bytes(); !bytes.Equal(got, wantBytes) {
+		t.Fatalf("enc(%v) = %q, want %q", want, got, wantBytes)
+	}
+	dec, err := coder.RowDecoderForStruct(rt)
+	if err != nil {
+		t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt, err)
+	}
+	got, err := dec(bytes.NewBuffer(wantBytes))
+	if err != nil {
+		t.Fatalf("dec(%q) = %v, want nil error", wantBytes, err)
+	}
+	if d := cmp.Diff(want, got, cmp.Comparer(func(a, b MicrosInstant) bool { return a.Time().Equal(b.Time()) })); d != "" {
+		t.Errorf("dec(enc(%v)) diff (-want, +got): %v", want, d)
+	}
+}
+
+func TestMicrosInstant_SubMicrosecondPrecision(t *testing.T) {
+	enc, err := coder.RowEncoderForStruct(reflect.TypeOf(standardRow{}))
+	if err != nil {
+		t.Fatalf("RowEncoderForStruct = %v, want nil error", err)
+	}
+	row := standardRow{F_timestamp: MicrosInstant(time.Unix(0, 1500))}
+	if err := enc(row, &bytes.Buffer{}); err == nil {
+		t.Errorf("enc(%v) = nil error, want error for sub microsecond precision", row)
+	}
+}
+
+func TestDate_Storage(t *testing.T) {
+	tests := []struct {
+		date Date
+		days int64
+	}{
+		{Date{1970, time.January, 1}, 0},
+		{Date{1969, time.December, 31}, -1},
+		{Date{2020, time.August, 13}, 18487},
+		{Date{1600, time.February, 29}, -135081},
+		{Date{9999, time.December, 31}, 2932896},
+	}
+	for _, test := range tests {
+		days, err := dateToStorage(test.date)
+		if err != nil {
+			t.Fatalf("dateToStorage(%v) = %v, want nil error", test.date, err)
+		}
+		if days != test.days {
+			t.Errorf("dateToStorage(%v) = %v, want %v", test.date, days, test.days)
+		}
+		date, err := dateFromStorage(test.days)
+		if err != nil {
+			t.Fatalf("dateFromStorage(%v) = %v, want nil error", test.days, err)
+		}
+		if date != test.date {
+			t.Errorf("dateFromStorage(%v) = %v, want %v", test.days, date, test.date)
+		}
+	}
+	if got, want := DateOf(time.Date(2020, 8, 13, 23, 59, 59, 0, time.UTC)).String(), "2020-08-13"; got != want {
+		t.Errorf("DateOf(...).String() = %v, want %v", got, want)
+	}
+}
+
+func TestRegisterLogicalTypeConversion(t *testing.T) {
+	type celsius float64
+	type reading struct {
+		Temperature celsius
+	}
+	// Registering with the default registry mirrors what users do in init().
+	RegisterLogicalTypeConversion[celsius, float64]("schema.test:celsius:v1",
+		func(c celsius) (float64, error) { return float64(c) + 273.15, nil },
+		func(k float64) (celsius, error) { return celsius(k - 273.15), nil })
+
+	rt := reflect.TypeOf(reading{})
+	schm, err := FromType(rt)
+	if err != nil {
+		t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+	}
+	wantField := logicalFieldType("schema.test:celsius:v1", atomicType(pipepb.AtomicType_DOUBLE))
+	if d := cmp.Diff(wantField, schm.GetFields()[0].GetType(), protocmp.Transform()); d != "" {
+		t.Errorf("FromType(%v) field diff (-want, +got): %v", rt, d)
+	}
+
+	enc, err := coder.RowEncoderForStruct(rt)
+	if err != nil {
+		t.Fatalf("RowEncoderForStruct(%v) = %v, want nil error", rt, err)
+	}
+	var buf bytes.Buffer
+	if err := enc(reading{Temperature: 10}, &buf); err != nil {
+		t.Fatalf("enc = %v, want nil error", err)
+	}
+	// 283.15 as a big endian double, after the row header.
+	wantBytes := []byte{0x01, 0x00, 0x40, 0x71, 0xb2, 0x66, 0x66, 0x66, 0x66, 0x66}
+	if got := buf.Bytes(); !bytes.Equal(got, wantBytes) {
+		t.Fatalf("enc(reading{10}) = %v, want %v", got, wantBytes)
+	}
+	dec, err := coder.RowDecoderForStruct(rt)
+	if err != nil {
+		t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt, err)
+	}
+	got, err := dec(&buf)
+	if err != nil {
+		t.Fatalf("dec = %v, want nil error", err)
+	}
+	if d := cmp.Diff(reading{Temperature: 10}, got, cmp.Comparer(func(a, b celsius) bool { return a-b < 1e-9 && b-a < 1e-9 })); d != "" {
+		t.Errorf("dec(enc(reading{10})) diff (-want, +got): %v", d)
+	}
+}
+
+// fixedCode is a user logical type with an argument, registered with the
+// exported parameterized registration.
+type fixedCode string
+
+func TestRegisterParameterizedLogicalTypeConversion(t *testing.T) {
+	RegisterParameterizedLogicalTypeConversion[fixedCode, string]("schema.test:fixed_code:v1", int32(4),
+		func(c fixedCode) (string, error) { return string(c), nil },
+		func(s string) (fixedCode, error) { return fixedCode(s), nil })
+	rt := reflect.TypeOf(struct{ Code fixedCode }{})
+	schm, err := FromType(rt)
+	if err != nil {
+		t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+	}
+	want := logicalFieldType("schema.test:fixed_code:v1", atomicType(pipepb.AtomicType_STRING))
+	want.GetLogicalType().ArgumentType = int32FieldType()
+	want.GetLogicalType().Argument = int32FieldValue(4)
+	if d := cmp.Diff(want, schm.GetFields()[0].GetType(), protocmp.Transform()); d != "" {
+		t.Errorf("FromType(%v) field diff (-want, +got): %v", rt, d)
+	}
+	schm.Id = "fixed-code-from-elsewhere"
+	got, err := ToType(schm)
+	if err != nil {
+		t.Fatalf("ToType = %v, want nil error", err)
+	}
+	if ft := got.Field(0).Type; ft != typeOf[fixedCode]() {
+		t.Errorf("ToType field type = %v, want %v", ft, typeOf[fixedCode]())
+	}
+	enc, err := coder.RowEncoderForStruct(rt)
+	if err != nil {
+		t.Fatalf("RowEncoderForStruct(%v) = %v, want nil error", rt, err)
+	}
+	var buf bytes.Buffer
+	if err := enc(struct{ Code fixedCode }{"ABCD"}, &buf); err != nil {
+		t.Fatalf("enc = %v, want nil error", err)
+	}
+	if got, want := buf.Bytes(), []byte("\x01\x00\x04ABCD"); !bytes.Equal(got, want) {
+		t.Errorf("enc = %q, want %q", got, want)
+	}
+}
+
+// money is a user logical type with its own wire encoding: the cents as a
+// big endian int32 inside a BYTES representation.
+type money int32
+
+func TestRegisterLogicalTypeCoder(t *testing.T) {
+	RegisterLogicalTypeCoder[money, []byte]("schema.test:money:v1",
+		func(m money, w io.Writer) error {
+			var b [4]byte
+			binary.BigEndian.PutUint32(b[:], uint32(m))
+			return coder.EncodeBytes(b[:], w)
+		},
+		func(r io.Reader) (money, error) {
+			b, err := coder.DecodeBytes(r)
+			if err != nil {
+				return 0, err
+			}
+			return money(int32(binary.BigEndian.Uint32(b))), nil
+		})
+	rt := reflect.TypeOf(struct{ Price money }{})
+	schm, err := FromType(rt)
+	if err != nil {
+		t.Fatalf("FromType(%v) = %v, want nil error", rt, err)
+	}
+	want := logicalFieldType("schema.test:money:v1", atomicType(pipepb.AtomicType_BYTES))
+	if d := cmp.Diff(want, schm.GetFields()[0].GetType(), protocmp.Transform()); d != "" {
+		t.Errorf("FromType(%v) field diff (-want, +got): %v", rt, d)
+	}
+	enc, err := coder.RowEncoderForStruct(rt)
+	if err != nil {
+		t.Fatalf("RowEncoderForStruct(%v) = %v, want nil error", rt, err)
+	}
+	var buf bytes.Buffer
+	if err := enc(struct{ Price money }{-2}, &buf); err != nil {
+		t.Fatalf("enc = %v, want nil error", err)
+	}
+	wantBytes := []byte{0x01, 0x00, 0x04, 0xff, 0xff, 0xff, 0xfe}
+	if got := buf.Bytes(); !bytes.Equal(got, wantBytes) {
+		t.Fatalf("enc = %v, want %v", got, wantBytes)
+	}
+	dec, err := coder.RowDecoderForStruct(rt)
+	if err != nil {
+		t.Fatalf("RowDecoderForStruct(%v) = %v, want nil error", rt, err)
+	}
+	got, err := dec(&buf)
+	if err != nil {
+		t.Fatalf("dec = %v, want nil error", err)
+	}
+	if d := cmp.Diff(struct{ Price money }{-2}, got); d != "" {
+		t.Errorf("dec(enc) diff (-want, +got): %v", d)
+	}
+}
+
+func TestRegisterLogicalTypeConversion_PointerStorage(t *testing.T) {
+	type celsiusPtr float64
+	defer func() {
+		if recover() == nil {
+			t.Errorf("RegisterLogicalTypeConversion with a pointer storage type did not panic")
+		}
+	}()
+	RegisterLogicalTypeConversion[celsiusPtr, *float64]("schema.test:celsius_ptr:v1",
+		func(c celsiusPtr) (*float64, error) { f := float64(c); return &f, nil },
+		func(f *float64) (celsiusPtr, error) { return celsiusPtr(*f), nil })
+}
+
+func TestRegisterLogicalType_InterfaceGoType(t *testing.T) {
+	type textual interface{ Text() string }
+	t.Run("conversion", func(t *testing.T) {
+		defer func() {
+			if recover() == nil {
+				t.Errorf("RegisterLogicalTypeConversion with an interface Go type did not panic")
+			}
+		}()
+		RegisterLogicalTypeConversion[textual, string]("schema.test:textual:v1",
+			func(v textual) (string, error) { return v.Text(), nil },
+			func(s string) (textual, error) { return nil, nil })
+	})
+	t.Run("coder", func(t *testing.T) {
+		defer func() {
+			if recover() == nil {
+				t.Errorf("RegisterLogicalTypeCoder with an interface Go type did not panic")
+			}
+		}()
+		RegisterLogicalTypeCoder[textual, string]("schema.test:textual_coder:v1",
+			func(v textual, w io.Writer) error { return nil },
+			func(r io.Reader) (textual, error) { return nil, nil })
+	})
+}
+
+func TestDate_StorageRange(t *testing.T) {
+	for _, days := range []int64{minDateDays, maxDateDays} {
+		date, err := dateFromStorage(days)
+		if err != nil {
+			t.Fatalf("dateFromStorage(%v) = %v, want nil error", days, err)
+		}
+		if got, err := dateToStorage(date); err != nil || got != days {
+			t.Errorf("dateToStorage(%v) = %v, %v, want %v, nil", date, got, err, days)
+		}
+	}
+	for _, days := range []int64{minDateDays - 1, maxDateDays + 1, 1 << 62, math.MinInt64} {
+		if date, err := dateFromStorage(days); err == nil {
+			t.Errorf("dateFromStorage(%v) = %v, want error", days, date)
+		}
+	}
+	for _, date := range []Date{
+		{-1000000000, time.December, 31},
+		{1000000000, time.January, 1},
+		{2020, time.February, 30},
+		{1970, time.January, math.MaxInt/4 + 2},
+		{math.MaxInt / 2, time.January, 1},
+	} {
+		if days, err := dateToStorage(date); err == nil {
+			t.Errorf("dateToStorage(%d-%02d-%02d) = %v, want error", date.Year, date.Month, date.Day, days)
+		}
+	}
+}
+
+func TestMicrosInstant_StorageRange(t *testing.T) {
+	for _, micros := range []int64{-1, 1000000, 1 << 61} {
+		s := microsInstantStorage{Seconds: 0, Micros: micros}
+		if m, err := microsInstantFromStorage(s); err == nil {
+			t.Errorf("microsInstantFromStorage(%+v) = %v, want error", s, m.Time())
+		}
+	}
+	s := microsInstantStorage{Seconds: -1, Micros: 999999}
+	m, err := microsInstantFromStorage(s)
+	if err != nil {
+		t.Fatalf("microsInstantFromStorage(%+v) = %v, want nil error", s, err)
+	}
+	if want := time.Unix(-1, 999999000); !m.Time().Equal(want) {
+		t.Errorf("microsInstantFromStorage(%+v) = %v, want %v", s, m.Time(), want)
+	}
+}
diff --git a/sdks/go/pkg/beam/core/runtime/graphx/serialize_test.go b/sdks/go/pkg/beam/core/runtime/graphx/serialize_test.go
index 265d936a71ec..a6aceee4409d 100644
--- a/sdks/go/pkg/beam/core/runtime/graphx/serialize_test.go
+++ b/sdks/go/pkg/beam/core/runtime/graphx/serialize_test.go
@@ -23,6 +23,7 @@ import (
 	"testing"
 
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/schema"
 	v1pb "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/v1"
 )
 
@@ -88,4 +89,19 @@ func TestEncodeType(t *testing.T) {
 			t.Errorf("expected error about wrapping in a struct, got %q", err.Error())
 		}
 	})
+	t.Run("StandardLogicalTypes", func(t *testing.T) {
+		rt := reflect.TypeOf(struct {
+			D schema.Date
+			M schema.MicrosInstant
+		}{})
+		pbT, err := encodeType(rt)
+		if err != nil {
+			t.Fatalf("got error = %v, want nil", err)
+		}
+		for _, f := range pbT.GetFields() {
+			if got, want := f.GetType().GetKind(), v1pb.Type_EXTERNAL; got != want {
+				t.Errorf("got field %v kind == %v, want %v", f.GetName(), got, want)
+			}
+		}
+	})
 }
diff --git a/sdks/go/test/regression/coders/fromyaml/fromyaml.go b/sdks/go/test/regression/coders/fromyaml/fromyaml.go
index 466f361b0138..08000db877da 100644
--- a/sdks/go/test/regression/coders/fromyaml/fromyaml.go
+++ b/sdks/go/test/regression/coders/fromyaml/fromyaml.go
@@ -29,16 +29,19 @@ import (
 	"runtime/debug"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/schema"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
 	pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
 	"github.com/google/go-cmp/cmp"
 	"golang.org/x/text/encoding/charmap"
+	"google.golang.org/protobuf/proto"
 	yaml "gopkg.in/yaml.v2"
 )
 
@@ -49,11 +52,11 @@ var unimplementedCoders = map[string]bool{
 }
 
 var filteredCases = []struct{ filter, reason string }{
-	{"logical", "BEAM-9615: Support logical types"},
 	{"30ea5a25-dcd8-4cdb-abeb-5332d15ab4b9", "https://github.com/apache/beam/issues/21206: Support encoding position."},
-	{"80be749a-5700-4ede-89d8-dd9a4433a3f8", "https://github.com/apache/beam/issues/19817: Support millis_instant."},
-	{"800c44ae-a1b7-4def-bbf6-6217cca89ec4", "https://github.com/apache/beam/issues/19817: Support decimal."},
-	{"f0ffb3a4-f46f-41ca-a942-85e3e939452a", "https://github.com/apache/beam/issues/23526: Support char/varchar, binary/varbinary."},
+	{"beam:logical_type:millis_instant:v1", "https://github.com/apache/beam/issues/39684: Support millis_instant."},
+	{"beam:logical_type:decimal:v1", "https://github.com/apache/beam/issues/39684: Support decimal."},
+	{"beam:logical_type:fixed_char:v1", "https://github.com/apache/beam/issues/39684: Support char/varchar, binary/varbinary."},
+	{"beam:logical_type:timestamp:v1", "https://github.com/apache/beam/issues/39684: Support timestamp."},
 }
 
 // Coder is a representation a serialized beam coder.
@@ -164,6 +167,7 @@ func (s *Spec) testStandardCoder() (err error) {
 		}
 		if d := cmp.Diff(recoded, string(out.Bytes())); d != "" {
 			log.Printf("Encoding error: diff(-want,+got): %v\n", d)
+			encFails++
 		}
 	}
 	if decFails+encFails > 0 {
@@ -177,6 +181,15 @@ var cmpOpts = []cmp.Option{
 	cmp.Transformer("bytes2string", func(in []byte) (out string) {
 		return string(in)
 	}),
+	cmp.Comparer(func(a, b schema.MicrosInstant) bool {
+		return a.Time().Equal(b.Time())
+	}),
+}
+
+// latin1Bytes returns the bytes that a yaml string denotes. The yaml strings
+// escape bytes above 0x7f as code points, which yaml decodes to UTF-8.
+func latin1Bytes(s string) ([]byte, error) {
+	return charmap.ISO8859_1.NewEncoder().Bytes([]byte(s))
 }
 
 func diff(c Coder, elem *exec.FullValue, eg yaml.MapItem) bool {
@@ -301,29 +314,12 @@ func diff(c Coder, elem *exec.FullValue, eg yaml.MapItem) bool {
 		}
 		return pass
 	case "beam:coder:row:v1":
-		fs := eg.Value.(yaml.MapSlice)
-		var rfs []reflect.StructField
-		// There are only 2 pointer examples, but they reuse field names,
-		// so we key off the proto hash to know which example we're handling.
-		ptrEg := strings.Contains(c.Payload, "51ace21c7393")
-		for _, rf := range fs {
-			name := rf.Key.(string)
-			t := nameToType[name]
-			if ptrEg {
-				t = reflect.PtrTo(t)
-			}
-			rfs = append(rfs, reflect.StructField{
-				Name: strings.ToUpper(name[:1]) + name[1:],
-				Type: t,
-				Tag:  reflect.StructTag(fmt.Sprintf("beam:\"%v\"", name)),
-			})
-		}
-		rv := reflect.New(reflect.StructOf(rfs)).Elem()
-		for i, rf := range fs {
-			setField(rv, i, rf.Value)
+		row, err := rowFromYAML(c.Payload, eg.Value.(yaml.MapSlice))
+		if err != nil {
+			log.Printf("unable to build the expected row: %v\n", err)
+			return false
 		}
-
-		got, want = elem.Elm, rv.Interface()
+		got, want = elem.Elm, row
 	case "beam:coder:timer:v1":
 		pass := true
 		tm := elem.Elm.(exec.TimerRecv)
@@ -411,73 +407,137 @@ func diffPane(eg any, got typex.PaneInfo) bool {
 	return pass
 }
 
-// standard_coders.yaml uses the name for type indication, except for nullability.
-var nameToType = map[string]reflect.Type{
-	"str":     reflectx.String,
-	"i32":     reflectx.Int32,
-	"f64":     reflectx.Float64,
-	"arr":     reflect.SliceOf(reflectx.String),
-	"f_bool":  reflectx.Bool,
-	"f_bytes": reflect.PtrTo(reflectx.ByteSlice),
-	"f_map":   reflect.MapOf(reflectx.String, reflect.PtrTo(reflectx.Int64)),
-	"f_float": reflectx.Float32,
+// rowFromYAML builds the expected value of a beam:coder:row:v1 example from
+// the schema in the coder payload and the field values of the example.
+// The Go type is the one the Go SDK derives from the schema, so logical type
+// fields have their registered Go types.
+func rowFromYAML(payload string, fields yaml.MapSlice) (any, error) {
+	b, err := latin1Bytes(payload)
+	if err != nil {
+		return nil, err
+	}
+	var s pipepb.Schema
+	if err := proto.Unmarshal(b, &s); err != nil {
+		return nil, fmt.Errorf("unmarshalling the row schema: %v", err)
+	}
+	rt, err := schema.ToType(&s)
+	if err != nil {
+		return nil, fmt.Errorf("converting the row schema to a type: %v", err)
+	}
+	rv := reflect.New(rt).Elem()
+	if err := setStruct(rv, fields); err != nil {
+		return nil, err
+	}
+	return rv.Interface(), nil
+}
+
+// setStruct sets the fields of the struct value rv from the yaml mapping.
+func setStruct(rv reflect.Value, fields yaml.MapSlice) error {
+	for _, f := range fields {
+		name := f.Key.(string)
+		fv, ok := fieldByName(rv, name)
+		if !ok {
+			return fmt.Errorf("no field %q in %v", name, rv.Type())
+		}
+		if err := setValue(fv, f.Value); err != nil {
+			return fmt.Errorf("field %q: %v", name, err)
+		}
+	}
+	return nil
 }
 
-func setField(rv reflect.Value, i int, v any) {
+// fieldByName returns the field of rv with the given schema field name, which
+// is either the Go field name or the name in the beam struct tag.
+func fieldByName(rv reflect.Value, name string) (reflect.Value, bool) {
+	rt := rv.Type()
+	for i := 0; i < rt.NumField(); i++ {
+		sf := rt.Field(i)
+		if sf.Name == name {
+			return rv.Field(i), true
+		}
+		if tag, ok := sf.Tag.Lookup("beam"); ok && strings.Split(tag, ",")[0] == name {
+			return rv.Field(i), true
+		}
+	}
+	return reflect.Value{}, false
+}
+
+// setValue sets rv from a yaml example value. A nil value leaves rv at its
+// zero value, which is nil for nullable fields.
+func setValue(rv reflect.Value, v any) error {
 	if v == nil {
-		return
+		return nil
 	}
-	rf := rv.Field(i)
-	if rf.Kind() == reflect.Ptr {
-		// Ensure it's initialized.
-		rf.Set(reflect.New(rf.Type().Elem()))
-		rf = rf.Elem()
+	if rv.Kind() == reflect.Ptr {
+		rv.Set(reflect.New(rv.Type().Elem()))
+		rv = rv.Elem()
 	}
-	switch rf.Kind() {
+	switch rv.Type() {
+	case reflect.TypeOf(schema.MicrosInstant{}):
+		var seconds, micros int64
+		for _, f := range v.(yaml.MapSlice) {
+			switch f.Key.(string) {
+			case "seconds":
+				seconds = int64(f.Value.(int))
+			case "micros":
+				micros = int64(f.Value.(int))
+			}
+		}
+		rv.Set(reflect.ValueOf(schema.MicrosInstant(time.Unix(seconds, micros*1000).UTC())))
+		return nil
+	}
+	switch rv.Kind() {
 	case reflect.String:
-		rf.SetString(v.(string))
-	case reflect.Int32:
-		rf.SetInt(int64(v.(int)))
+		rv.SetString(v.(string))
+	case reflect.Int16, reflect.Int32, reflect.Int64:
+		rv.SetInt(int64(v.(int)))
 	case reflect.Float32:
 		c, err := strconv.ParseFloat(v.(string), 32)
 		if err != nil {
-			panic(err)
+			return err
 		}
-		rf.SetFloat(c)
+		rv.SetFloat(c)
 	case reflect.Float64:
 		c, err := strconv.ParseFloat(v.(string), 64)
 		if err != nil {
-			panic(err)
+			return err
 		}
-		rf.SetFloat(c)
+		rv.SetFloat(c)
+	case reflect.Bool:
+		rv.SetBool(v.(bool))
 	case reflect.Slice:
-		if rf.Type() == reflectx.ByteSlice {
-			rf.Set(reflect.ValueOf([]byte(v.(string))))
-			break
+		if rv.Type() == reflectx.ByteSlice {
+			rv.SetBytes([]byte(v.(string)))
+			return nil
 		}
-		// Value is a []any with string values.
-		var arr []string
-		for _, a := range v.([]any) {
-			arr = append(arr, a.(string))
+		items := v.([]any)
+		sv := reflect.MakeSlice(rv.Type(), len(items), len(items))
+		for i, item := range items {
+			if err := setValue(sv.Index(i), item); err != nil {
+				return err
+			}
 		}
-		rf.Set(reflect.ValueOf(arr))
-	case reflect.Bool:
-		rf.SetBool(v.(bool))
+		rv.Set(sv)
 	case reflect.Map:
-		// only f_map presently, which is always map[string]*int64
-		rm := reflect.MakeMap(rf.Type())
-		for _, a := range v.(yaml.MapSlice) {
-			rk := reflect.ValueOf(a.Key.(string))
-			rv := reflect.Zero(rf.Type().Elem())
-			if a.Value != nil {
-				rv = reflect.New(reflectx.Int64)
-				rv.Elem().SetInt(int64(a.Value.(int)))
+		mv := reflect.MakeMap(rv.Type())
+		for _, entry := range v.(yaml.MapSlice) {
+			key := reflect.New(rv.Type().Key()).Elem()
+			if err := setValue(key, entry.Key); err != nil {
+				return err
+			}
+			val := reflect.New(rv.Type().Elem()).Elem()
+			if err := setValue(val, entry.Value); err != nil {
+				return err
 			}
-			rm.SetMapIndex(rk, rv)
+			mv.SetMapIndex(key, val)
 		}
-		rf.Set(rm)
-
+		rv.Set(mv)
+	case reflect.Struct:
+		return setStruct(rv, v.(yaml.MapSlice))
+	default:
+		return fmt.Errorf("unsupported field type %v", rv.Type())
 	}
+	return nil
 }
 
 func (s *Spec) parseCoder(c Coder) string {
@@ -486,10 +546,14 @@ func (s *Spec) parseCoder(c Coder) string {
 	for _, comp := range c.Components {
 		compIDs = append(compIDs, s.parseCoder(comp))
 	}
+	payload, err := latin1Bytes(c.Payload)
+	if err != nil {
+		panic(fmt.Sprintf("invalid coder payload %q: %v", c.Payload, err))
+	}
 	s.coderPBs[id] = &pipepb.Coder{
 		Spec: &pipepb.FunctionSpec{
 			Urn:     c.Urn,
-			Payload: []byte(c.Payload),
+			Payload: payload,
 		},
 		ComponentCoderIds: compIDs,
 	}