apache/beam · error · RuntimeException

Not implemented yet

Error message

Not implemented yet ${fieldDescriptor.getMessageType().getName()}

What it means

In the proto-to-TableRow direction, message-typed fields are only decoded for known message types (e.g. numeric/decimal wrapper messages); any other message type reaches a RuntimeException 'Not implemented yet <messageName>'. It marks an unhandled nested message kind in the reverse conversion.

Solutions

Avoid unsupported logical/message-typed columns in the read path or convert them to primitives.
Add/extend the conversion case for the message type if you control the code (fork/patch or upstream PR).
Upgrade Beam — new message types get implemented over time.
Flatten the nested message into primitive columns in your schema.

Example fix

// before
field is message type MyLogicalType -> RuntimeException
// after
store as underlying primitive (e.g. STRING/BYTES) and decode in user code

Defensive patterns

Strategy: fallback

Validate before calling

boolean decodable(FieldDescriptor fd) { return fd.getType() != FieldDescriptor.Type.MESSAGE || KNOWN_MESSAGE_TYPES.contains(fd.getMessageType().getName()); }

Type guard

boolean isKnownMessageType(FieldDescriptor fd) { return fd.getMessageType() == null || Set.of("BigDecimal","DecimalWrapper","Timestamp","date").contains(fd.getMessageType().getName()); }

Try / catch

try { protoToTableRow(message); } catch (RuntimeException e) { if (e.getMessage().startsWith("Not implemented yet")) fallBackToRawStringField(e); else throw e; }

Prevention

Prefer primitive BQ column types when data must be read back through this converter
Keep Beam up to date so newly implemented message types are available
Handle custom logical types in user code instead of relying on the built-in decoder

When it happens

Trigger: Reading rows back from the Storage API proto (e.g. in CDC or test harness code) where a field's DynamicMessage has a message type other than the recognized numeric wrappers — e.g. a custom or newer logical-type message.

Common situations: Schemas containing logical types that map to custom proto messages; Beam internal decimal/timestamp wrappers handled for some but not all types; testing round-trips with hand-crafted protos.

Related errors

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

Appendix: source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java:1910

          case MESSAGE:
            // Handle the various number wrapper types.
            Message doubleMessage = (Message) fieldValue;
            if (FLOAT_VALUE_DESCRIPTOR_NAMES.contains(fieldDescriptor.getMessageType().getName())) {
              float floatValue =
                  (float)
                      doubleMessage.getField(
                          doubleMessage.getDescriptorForType().findFieldByName("value"));

              return DECIMAL_FORMAT.format(floatValue);
            } else if (DOUBLE_VALUE_DESCRIPTOR_NAMES.contains(
                fieldDescriptor.getMessageType().getName())) {
              double doubleValue =
                  (double)
                      doubleMessage.getField(
                          doubleMessage.getDescriptorForType().findFieldByName("value"));
              return DECIMAL_FORMAT.format(doubleValue);
            } else {
              throw new RuntimeException(
                  "Not implemented yet " + fieldDescriptor.getMessageType().getName());
            }
          default:
            return fieldValue.toString();
        }
      case BOOL:
        // Wrapper type.
        if (fieldDescriptor.getType().equals(FieldDescriptor.Type.MESSAGE)) {
          Message boolMessage = (Message) fieldValue;
          if (BOOL_VALUE_DESCRIPTOR_NAMES.contains(fieldDescriptor.getMessageType().getName())) {
            return boolMessage
                .getField(boolMessage.getDescriptorForType().findFieldByName("value"))
                .toString();
          } else {
            throw new RuntimeException(
                "Not implemented yet " + fieldDescriptor.getMessageType().getName());
          }
        }

View on GitHub (pinned to 12126d8942)