apache/flink · error · PbCodegenException

Do not support flink data type:

Error message

Do not support flink data type: 

What it means

Serialization-side twin of the deserializer factory: it dispatches on LogicalType into Row/simple/Array/Map serializers; anything else (RAW, MULTISET, structured/distinct types) throws PbCodegenException at codegen time with the offending type in the message.

Source

Thrown at flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/serialize/PbCodegenSerializeFactory.java:46

import com.google.protobuf.Descriptors;

/** Codegen factory class which return {@link PbCodegenSerializer} of different data type. */
public class PbCodegenSerializeFactory {
    public static PbCodegenSerializer getPbCodegenSer(
            Descriptors.FieldDescriptor fd, LogicalType type, PbFormatContext formatContext)
            throws PbCodegenException {
        if (type instanceof RowType) {
            return new PbCodegenRowSerializer(fd.getMessageType(), (RowType) type, formatContext);
        } else if (PbFormatUtils.isSimpleType(type)) {
            return new PbCodegenSimpleSerializer(fd, type, formatContext);
        } else if (type instanceof ArrayType) {
            return new PbCodegenArraySerializer(
                    fd, ((ArrayType) type).getElementType(), formatContext);
        } else if (type instanceof MapType) {
            return new PbCodegenMapSerializer(fd, (MapType) type, formatContext);
        } else {
            throw new PbCodegenException("Do not support flink data type: " + type);
        }
    }

    public static PbCodegenSerializer getPbCodegenTopRowSer(
            Descriptors.Descriptor descriptor, RowType rowType, PbFormatContext formatContext) {
        return new PbCodegenRowSerializer(descriptor, rowType, formatContext);
    }
}

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Locate the column printed in the message and drop or cast it to a protobuf-mappable type.
2. Restrict sink column lists: INSERT INTO pb_sink (id, name) SELECT ... so unsupported intermediate columns never reach the format.
3. Validate the final query plan's output schema before submission.

Example fix

-- before
INSERT INTO pb_sink SELECT id, collect(name) FROM src GROUP BY id;
-- after
INSERT INTO pb_sink (id) SELECT id FROM src GROUP BY id;

static boolean pbSupports(LogicalType t) {
    return t instanceof RowType || t instanceof ArrayType || t instanceof MapType
            || PbFormatUtils.isSimpleType(t);
}
// validate the query's resolved output before INSERT:
for (int i = 0; i < sinkRowType.getFieldCount(); i++) {
    if (!pbSupports(sinkRowType.getTypeAt(i))) throw new IllegalArgumentException("Column " + i);
}

boolean isPbMappable(LogicalType t) {
    return t instanceof RowType || t instanceof ArrayType || t instanceof MapType
            || PbFormatUtils.isSimpleType(t);
}

When it happens

Trigger: INSERT INTO a protobuf sink whose table schema contains an unsupported LogicalType; computed columns producing RAW or MULTISET types that reach the serializer.

Common situations: Reusing a source schema for a protobuf sink after adding metadata/RAW columns; SQL views with multiset aggregations written to protobuf.

Related errors

• Do not support flink type:
• Unsupported data type in schema:
• Unsupported data type in schema:
• Illegal type for protobuf enum, only char/vachar/int/bigint
• Unsupported protobuf simple type:

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/764b2cdf09528c2c. Report an issue: GitHub.