apache/flink · error · InvalidProgramException

Field expression must be equal to '*' or '_' for non-composi

Error message

Field expression must be equal to '*' or '_' for non-composite types.

What it means

Thrown by Keys.ExpressionKeys(String[], TypeInformation) in the non-composite branch. For atomic (non-composite) types the only legal field expression is the select-all wildcard '*' (Java) or '_' (Scala), meaning 'use the entire value as the key'. Any other string expression is meaningless because there are no sub-fields to address on an atomic type.

Source

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

                    }
                }
            } else {
                if (!type.isKeyType()) {
                    throw new InvalidProgramException(
                            "This type (" + type + ") cannot be used as key.");
                }

                // check that all key expressions are valid
                for (String keyExpr : keyExpressions) {
                    if (keyExpr == null) {
                        throw new InvalidProgramException("Expression key may not be null.");
                    }
                    // strip off whitespace
                    keyExpr = keyExpr.trim();
                    // check that full type is addressed
                    if (!(SELECT_ALL_CHAR.equals(keyExpr)
                            || SELECT_ALL_CHAR_SCALA.equals(keyExpr))) {
                        throw new InvalidProgramException(
                                "Field expression must be equal to '"
                                        + SELECT_ALL_CHAR
                                        + "' or '"
                                        + SELECT_ALL_CHAR_SCALA
                                        + "' for non-composite types.");
                    }
                    // add full type as key
                    keyFields.add(new FlatFieldDescriptor(0, type));
                }
                this.originalKeyTypes = new TypeInformation[] {type};
            }
        }

        @Override
        public int getNumberOfKeyFields() {
            if (keyFields == null) {
                return 0;
            }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Use keyBy("*") (Java) or keyBy("_") (Scala) to key on the entire atomic value.
2. Switch to positional keyBy (keyBy(0)) if appropriate for a tuple.
3. If you intended to key on a sub-field, change the stream type to a composite (POJO/Tuple/Row) first.

Example fix

// before
DataStream<String> names = ...;
names.keyBy("length");  // String is atomic

// after
DataStream<String> names = ...;
names.keyBy("*");  // key on the whole String value

if (!(type instanceof CompositeType)) {
    for (String expr : keyExpressions) {
        String trimmed = expr.trim();
        if (!("*".equals(trimmed) || "_".equals(trimmed))) {
            throw new IllegalArgumentException(
                "Atomic type requires '*' or '_' as key expression, got: " + trimmed);
        }
    }
}

static boolean isValidAtomicKeyExpr(String expr) {
    String t = expr.trim();
    return "*".equals(t) || "_".equals(t);
}

When it happens

Trigger: Calling keyBy("someField") on a DataStream<String>, DataStream<Integer>, or any non-composite type. Passing a named field expression where the type has no fields.

Common situations: Treating a primitive/String stream like a POJO stream and naming a field that does not exist. Scala users using a Java-style '*' where '_' is idiomatic, or vice-versa (both are accepted, but a typo triggers this).

Related errors

• This type ({field.getType()}) cannot be used as key.
• This type ({type}) cannot be used as key.
• 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/e2ac5948e1e9d8d0. Report an issue: GitHub.