apache/flink · error · InvalidProgramException

This type ({field.getType()}) cannot be used as key.

Error message

This type ({field.getType()}) cannot be used as key.

What it means

Thrown by Keys.ExpressionKeys(String[], TypeInformation) when a nested field of a composite type (Tuple/POJO/Case Class) resolves to a type that is not a key type. isKeyType() returns false for non-hashable types such as POJOs, GenericTypeInfo (opaque Java objects), arrays, or types without a proper TypeComparator. Every field that participates in a key must be individually hashable and comparable so the runtime can partition and sort by it.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java:327

                    if (keyExpr == null) {
                        throw new InvalidProgramException("Expression key may not be null.");
                    }
                    // strip off whitespace
                    keyExpr = keyExpr.trim();

                    List<FlatFieldDescriptor> flatFields = cType.getFlatFields(keyExpr);

                    if (flatFields.size() == 0) {
                        throw new InvalidProgramException(
                                "Unable to extract key from expression '"
                                        + keyExpr
                                        + "' on key "
                                        + cType);
                    }
                    // check if all nested fields can be used as keys
                    for (FlatFieldDescriptor field : flatFields) {
                        if (!field.getType().isKeyType()) {
                            throw new InvalidProgramException(
                                    "This type (" + field.getType() + ") cannot be used as key.");
                        }
                    }
                    // add flat fields to key fields
                    keyFields.addAll(flatFields);

                    String strippedKeyExpr = WILD_CARD_REGEX.matcher(keyExpr).replaceAll("");
                    if (strippedKeyExpr.isEmpty()) {
                        this.originalKeyTypes[i] = type;
                    } else {
                        this.originalKeyTypes[i] = cType.getTypeAt(strippedKeyExpr);
                    }
                }
            } else {
                if (!type.isKeyType()) {
                    throw new InvalidProgramException(
                            "This type (" + type + ") cannot be used as key.");
                }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Flatten the problematic field into individual primitive/String sub-fields and keyBy on those instead.
2. Provide a KeySelector that extracts only the hashable leaf fields from the nested type.
3. Annotate the nested class as a valid POJO or register a TypeInfoFactory so Flink can decompose it into key-type leaves.
4. If the field is genuinely a single opaque comparable, wrap it in a type Flink recognises (e.g. a Tuple1) or implement a custom TypeInfoFactory that exposes isKeyType()==true.

Example fix

// before
ds.keyBy("address")  // address is a POJO → not a key type

// after
ds.keyBy("address.zipCode", "address.street")  // primitive String fields

// Before keyBy, check each field is a key type
CompositeType<?> ct = (CompositeType<?>) typeInfo;
List<FlatFieldDescriptor> flat = ct.getFlatFields("fieldName");
for (FlatFieldDescriptor ffd : flat) {
    if (!ffd.getType().isKeyType()) {
        throw new IllegalArgumentException(
            "Field resolves to non-key type: " + ffd.getType());
    }
}

static boolean isFieldKeyType(CompositeType<?> type, String expr) {
    List<FlatFieldDescriptor> flat = type.getFlatFields(expr);
    if (flat.isEmpty()) return false;
    return flat.stream().allMatch(f -> f.getType().isKeyType());
}

try {
    ds.keyBy("fieldName");
} catch (InvalidProgramException e) {
    if (e.getMessage().contains("cannot be used as key")) {
        // fall back to KeySelector extracting leaf fields
        ds.keyBy(record -> extractLeafKey(record));
    } else throw e;
}

When it happens

Trigger: Calling keyBy("fieldName") on a DataStream whose POJO has a field of a non-key type (e.g. a nested POJO, a byte[], or a raw Object). Also triggered by constructing new ExpressionKeys<>(new String[]{"nestedField"}, type) where the resolved FlatFieldDescriptor reports isKeyType()==false.

Common situations: Grouping or joining on a POJO field that itself contains another POJO or a collection. Migrating from a Tuple of primitives to a richer domain object and forgetting that only leaf-level key types are allowed. Using a GenericTypeInfo-backed field (e.g. a third-party class Flink cannot introspect) as a key.

Related errors

• This type ({type}) cannot be used as key.
• Field expression must be equal to '*' or '_' for non-composi
• Specifying keys via field positions is only valid for tuple
• This type ({ffd.getType()}) cannot be used as key.
• Field expression must be equal to '*' or '_' for atomic type

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