apache/flink · error · UnsupportedOperationException

Record serialization with leading normalized keys not suppor

Error message

Record serialization with leading normalized keys not supported.

What it means

RowComparator.writeWithKeyNormalization() intentionally throws UnsupportedOperationException because Row does not support embedding normalized key bytes into the serialized record stream. Normalized keys for Rows are handled only via the separate putNormalizedKey() path (writing to a sort/hash key buffer), not by rewriting the serialized Row. This is a design limitation of the Row type in the DataSet batch comparator, not a configuration error.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java:363

        int currentOffset = offset;

        for (int i = 0; i < numLeadingNormalizableKeys && bytesLeft > 0; i++) {
            int len = normalizedKeyLengths[i];
            len = bytesLeft >= len ? len : bytesLeft;

            TypeComparator<Object> comparator = comparators[i];
            Object element = record.getField(keyPositions[i]); // element can be null
            // write key
            comparator.putNormalizedKey(element, target, currentOffset, len);

            bytesLeft -= len;
            currentOffset += len;
        }
    }

    @Override
    public void writeWithKeyNormalization(Row record, DataOutputView target) throws IOException {
        throw new UnsupportedOperationException(
                "Record serialization with leading normalized keys not supported.");
    }

    @Override
    public Row readWithKeyDenormalization(Row reuse, DataInputView source) throws IOException {
        throw new UnsupportedOperationException(
                "Record deserialization with leading normalized keys not supported.");
    }

    @Override
    public boolean invertNormalizedKey() {
        return invertNormKey;
    }

    @Override
    public TypeComparator<Row> duplicate() {
        NullAwareComparator<?>[] comparatorsCopy = new NullAwareComparator<?>[comparators.length];
        for (int i = 0; i < comparators.length; i++) {

View on GitHub (pinned to 2f3c205e92)

Solutions

Avoid using Row directly as a sort/hash key in DataSet operators that require key-normalized serialization; project the key fields into a Tuple or POJO instead.
Use the Table API / SQL which handles Row keys through its own runtime comparators, bypassing RowComparator's limitation.
If you control the operator, ensure it uses putNormalizedKey() (supported) rather than writeWithKeyNormalization() (unsupported).

Example fix

// before — Row used as key in a DataSet operator needing key normalization
DataSet<Row> rows = ...;
rows.sortPartition(0, Order.ASCENDING); // may hit writeWithKeyNormalization

// after — project key into a Tuple for the keyed/sorted operation
DataSet<Tuple2<Integer, Row>> keyed = rows.map(r -> Tuple2.of((Integer) r.getField(0), r));
keyed.sortPartition(0, Order.ASCENDING);

Defensive patterns

Strategy: fallback

Validate before calling

// Before using Row as a key in DataSet, check if the comparator supports
// key normalization
public static boolean supportsKeyNormalization(TypeComparator<Row> cmp) {
    try {
        // writeWithKeyNormalization throws UnsupportedOperationException if unsupported
        return true; // RowComparator never supports it — use a Tuple key instead
    } catch (Exception e) {
        return false;
    }
}
// Practical advice: avoid Row keys in DataSet; use Tuple

Try / catch

// This is an UnsupportedOperationException — there is no recovery at runtime.
// Prevent it by design: do not use Row as a keyed type in DataSet operators
// that require key-normalized serialization.

Prevention

Do not use Row directly as a sort/group/join key in the DataSet API.
Project key fields into a Tuple before keyed operations.
Prefer the Table API / SQL for Row-based keyed operations.
Document this limitation for team members using the DataSet API.

When it happens

Trigger: Flink's sort or hash infrastructure calls writeWithKeyNormalization() on the comparator when it needs to inline normalized keys into the serialized bytes. For RowComparator this path is unsupported, so any operator that requires key-normalized serialization of Rows will fail at runtime.

Common situations: Using a Row type as a key in a DataSet operator that requires leading normalized keys in the serialized record (certain range-partition or in-memory sort configurations); framework code paths that assume every comparator supports writeWithKeyNormalization encounter a RowComparator.

Related errors

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