apache/beam · error

LogicalType[%v] has an invalid StorageType %v: %v

Error message

LogicalType[%v] has an invalid StorageType %v: %v

What it means

Registry.RegisterLogicalType validates that a LogicalType's StorageType has known schema handling by calling reflectTypeToFieldType. If the storage type cannot be mapped to a schema field type, registration panics, since the logical type could never be encoded in the schema protocol.

Source

Thrown at sdks/go/pkg/beam/core/runtime/graphx/schema/logicaltypes.go:96

func NewRegistry() *Registry {
	return &Registry{
		typeToSchema:    map[reflect.Type]*pipepb.Schema{},
		idToType:        map[string]reflect.Type{},
		syntheticToUser: map[reflect.Type]reflect.Type{},

		logicalTypes:           map[string]LogicalType{},
		logicalTypeIdentifiers: map[reflect.Type]string{},
		logicalTypeProviders:   map[reflect.Type]LogicalTypeProvider{},
	}
}

// RegisterLogicalType a single logical type.
func (r *Registry) RegisterLogicalType(lt LogicalType) {
	// Validates that the storage type has known handling.
	st := lt.StorageType()
	_, err := r.reflectTypeToFieldType(st)
	if err != nil {
		panic(fmt.Sprintf("LogicalType[%v] has an invalid StorageType %v: %v", lt.ID(), st, err))
	}
	if len(lt.ID()) == 0 {
		panic(fmt.Sprintf("invalid logical type, bad id: %v -> %v", lt.GoType(), lt.StorageType()))
	}
	// TODO add duplication checks.
	r.logicalTypeIdentifiers[lt.GoType()] = lt.ID()
	r.logicalTypes[lt.ID()] = lt
}

// RegisterLogicalTypeProvider allows registration of providers for interface types.
func (r *Registry) RegisterLogicalTypeProvider(rt reflect.Type, ltp LogicalTypeProvider) {
	if rt.Kind() != reflect.Interface {
		panic(fmt.Sprintf("Logical Types must be registered with interface types. %v is not an interface type.", rt))
	}
	if rt.NumMethod() == 0 {
		panic(fmt.Sprintf("Logical Types may not be registered with empty interface types. %v has no methods.", rt))
	}
	r.logicalTypeProviders[rt] = ltp

View on GitHub (pinned to 12126d8942)

Solutions

1. Change StorageType() to return a supported primitive/known type (e.g. []byte, string, int64)
2. Check the schema package's reflectTypeToFieldType for supported kinds and register additional field type handling if available
3. Implement custom row encoding via a wrapper type with a supported storage type

Example fix

// before
func (t myLT) StorageType() reflect.Type { return reflect.TypeOf(map[string]string{}) }
// after
func (t myLT) StorageType() reflect.Type { return reflect.TypeOf("") }

st := lt.StorageType()
switch st.Kind() {
case reflect.Slice, reflect.String, reflect.Int64, reflect.Bool, reflect.Float64:
default:
    return fmt.Errorf("unsupported storage type: %v", st)
}

defer func() { if r := recover(); r != nil { err = fmt.Errorf("register logical type: %v", r) } }()

When it happens

Trigger: Registering a LogicalType whose StorageType() returns a Go type unsupported by the schema package (e.g. a struct, map, or other unhandled kind) via RegisterLogicalType, registerType, or preRegLogicalTypes.

Common situations: Implementing a custom logical type with a composite or pointer storage type; typos returning the wrong type from StorageType(); SDK versions lacking handling for a newer storage type.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• invalid logical type, bad id: %v -> %v
• unable to provide schema.LogicalType for type %v, want %v
• reconciling for ToType
• Logical Types must be registered with interface types. %v is
• Logical Types may not be registered with empty interface typ

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/2d71ca8dd8c0fa70. Report an issue: GitHub.