What causes "failed to map row %T" in apache/beam?

Make the element type's exported fields match the target table's column names (case-insensitive, per mapFields).

failed to map row %T

Error message

failed to map row %T

What it means

Raised when the per-row mapper fails to convert an element of the input PCollection into the []any row slice used for the batch INSERT. It wraps the underlying mapper error and includes the Go type of the offending value (%T). It fires for non-Writer elements whose reflection-based mapping (newWriterRowMapper) cannot produce a value for the discovered table columns.

Solutions

Make the element type's exported fields match the target table's column names (case-insensitive, per mapFields).
Implement the Writer interface (SaveData) on your element type to control column-to-value mapping explicitly.
Compare the discovered table columns (from the probe SELECT) with your struct fields; add/rename fields accordingly.
Read the wrapped inner error — it names the specific field or mapping failure.

Example fix

// before
type User struct { Name string; City string } // no matching column "city" in table
// after
type User struct { Name string; City string `beam:"city"` } // or align field names with table columns

Defensive patterns

Strategy: validation

Validate before calling

// element type check before calling Write
func checkMappableToColumns(v any, columns []string) error {
    if _, ok := v.(databaseio.Writer); ok { return nil }
    t := reflect.TypeOf(v)
    if t.Kind() != reflect.Struct { return fmt.Errorf("element %T is not a struct or Writer", v) }
    fields := map[string]bool{}
    for i := 0; i < t.NumField(); i++ { fields[strings.ToLower(t.Field(i).Name)] = true }
    for _, c := range columns {
        if !fields[strings.ToLower(c)] { return fmt.Errorf("no field maps to column %q", c) }
    }
    return nil
}

Type guard

func isWriter(v any) bool { _, ok := v.(databaseio.Writer); return ok }
func isStructWithColumns(v any, columns []string) bool {
    t := reflect.TypeOf(v)
    if t.Kind() != reflect.Struct { return false }
    for i := 0; i < t.NumField(); i++ {
        f := strings.ToLower(t.Field(i).Name)
        match := false
        for _, c := range columns { if strings.ToLower(c) == f { match = true } }
        if !match { return false }
    }
    return true
}

Try / catch

if err != nil {
    return errors.Wrapf(err, "element of type %T could not be mapped to table columns; check field names vs table schema", val)
}

Prevention

Keep struct field names aligned with table column names (case-insensitive).
Implement the Writer/SaveData interface when you need explicit control over column mapping.
Re-validate element types after any table schema migration.
Write a unit test mapping one element through the mapper before running the full pipeline.

When it happens

Trigger: Emitting elements into databaseio.Write whose type does not implement the Writer interface and whose fields cannot be mapped onto the table columns (mismatched or unexported struct fields per mapFields); a SaveData() error on a Writer-implementing element; nil or unexpected element types reaching the DoFn.

Common situations: Struct element type whose fields don't match the table column names (renamed column, renamed field); using a non-struct custom type instead of a struct; schema drift after the table was altered; passing pointer/alias types with no matching exported fields.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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

Appendix: source

Thrown at sdks/go/pkg/beam/io/databaseio/database.go:205

	if err != nil {
		return err
	}
	var val beam.X
	for iter(&val) {
		var row []any
		var data map[string]any
		if writer, ok := val.(Writer); ok {
			if data, err = writer.SaveData(); err == nil {
				row = make([]any, len(columns))
				for i, column := range columns {
					row[i] = data[column]
				}
			}
		} else {
			row, err = mapper(reflect.ValueOf(val))
		}
		if err != nil {
			return errors.Wrapf(err, "failed to map row %T", val)
		}
		if err = writer.add(row); err != nil {
			return err
		}
		if err := writer.writeBatchIfNeeded(ctx, db); err != nil {
			return err
		}
	}

	if err := writer.writeIfNeeded(ctx, db); err != nil {
		return err
	}

	log.Infof(ctx, "written %v row(s) into %v", writer.totalCount, f.Table)
	return nil
}

View on GitHub (pinned to 12126d8942)