{"record":{"id":"d6e3b310fc6760b6","repo":"apache/flink","slug":"record-serialization-with-leading-normalized-keys-d6e3b3","errorCode":null,"errorMessage":"Record serialization with leading normalized keys not supported.","messagePattern":"Record serialization with leading normalized keys not supported\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java","lineNumber":363,"sourceCode":"        int currentOffset = offset;\n\n        for (int i = 0; i < numLeadingNormalizableKeys && bytesLeft > 0; i++) {\n            int len = normalizedKeyLengths[i];\n            len = bytesLeft >= len ? len : bytesLeft;\n\n            TypeComparator<Object> comparator = comparators[i];\n            Object element = record.getField(keyPositions[i]); // element can be null\n            // write key\n            comparator.putNormalizedKey(element, target, currentOffset, len);\n\n            bytesLeft -= len;\n            currentOffset += len;\n        }\n    }\n\n    @Override\n    public void writeWithKeyNormalization(Row record, DataOutputView target) throws IOException {\n        throw new UnsupportedOperationException(\n                \"Record serialization with leading normalized keys not supported.\");\n    }\n\n    @Override\n    public Row readWithKeyDenormalization(Row reuse, DataInputView source) throws IOException {\n        throw new UnsupportedOperationException(\n                \"Record deserialization with leading normalized keys not supported.\");\n    }\n\n    @Override\n    public boolean invertNormalizedKey() {\n        return invertNormKey;\n    }\n\n    @Override\n    public TypeComparator<Row> duplicate() {\n        NullAwareComparator<?>[] comparatorsCopy = new NullAwareComparator<?>[comparators.length];\n        for (int i = 0; i < comparators.length; i++) {","sourceCodeStart":345,"sourceCodeEnd":381,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java#L345-L381","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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)."],"exampleFix":"// before — Row used as key in a DataSet operator needing key normalization\nDataSet<Row> rows = ...;\nrows.sortPartition(0, Order.ASCENDING); // may hit writeWithKeyNormalization\n\n// after — project key into a Tuple for the keyed/sorted operation\nDataSet<Tuple2<Integer, Row>> keyed = rows.map(r -> Tuple2.of((Integer) r.getField(0), r));\nkeyed.sortPartition(0, Order.ASCENDING);","handlingStrategy":"fallback","validationCode":"// Before using Row as a key in DataSet, check if the comparator supports\n// key normalization\npublic static boolean supportsKeyNormalization(TypeComparator<Row> cmp) {\n    try {\n        // writeWithKeyNormalization throws UnsupportedOperationException if unsupported\n        return true; // RowComparator never supports it — use a Tuple key instead\n    } catch (Exception e) {\n        return false;\n    }\n}\n// Practical advice: avoid Row keys in DataSet; use Tuple","typeGuard":null,"tryCatchPattern":"// This is an UnsupportedOperationException — there is no recovery at runtime.\n// Prevent it by design: do not use Row as a keyed type in DataSet operators\n// that require key-normalized serialization.","preventionTips":["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."],"tags":["row","comparator","unsupported-operation","normalized-key","dataset-api"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}