{"record":{"id":"92a284e2609ceffc","repo":"apache/flink","slug":"the-type-is-not-comparable","errorCode":null,"errorMessage":"The type {} is not Comparable.","messagePattern":"The type (.+?) is not Comparable\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java","lineNumber":201,"sourceCode":"        } else if (ShortValue.class.isAssignableFrom(type)) {\n            return (TypeSerializer<T>) ShortValueSerializer.INSTANCE;\n        } else if (StringValue.class.isAssignableFrom(type)) {\n            return (TypeSerializer<T>) StringValueSerializer.INSTANCE;\n        } else if (CopyableValue.class.isAssignableFrom(type)) {\n            return (TypeSerializer<T>)\n                    createCopyableValueSerializer(type.asSubclass(CopyableValue.class));\n        } else {\n            return new ValueSerializer<T>(type);\n        }\n    }\n\n    @SuppressWarnings({\"unchecked\", \"rawtypes\"})\n    @Override\n    @PublicEvolving\n    public TypeComparator<T> createComparator(\n            boolean sortOrderAscending, ExecutionConfig executionConfig) {\n        if (!isKeyType()) {\n            throw new RuntimeException(\"The type \" + type.getName() + \" is not Comparable.\");\n        }\n\n        if (BooleanValue.class.isAssignableFrom(type)) {\n            return (TypeComparator<T>) new BooleanValueComparator(sortOrderAscending);\n        } else if (ByteValue.class.isAssignableFrom(type)) {\n            return (TypeComparator<T>) new ByteValueComparator(sortOrderAscending);\n        } else if (CharValue.class.isAssignableFrom(type)) {\n            return (TypeComparator<T>) new CharValueComparator(sortOrderAscending);\n        } else if (DoubleValue.class.isAssignableFrom(type)) {\n            return (TypeComparator<T>) new DoubleValueComparator(sortOrderAscending);\n        } else if (FloatValue.class.isAssignableFrom(type)) {\n            return (TypeComparator<T>) new FloatValueComparator(sortOrderAscending);\n        } else if (IntValue.class.isAssignableFrom(type)) {\n            return (TypeComparator<T>) new IntValueComparator(sortOrderAscending);\n        } else if (LongValue.class.isAssignableFrom(type)) {\n            return (TypeComparator<T>) new LongValueComparator(sortOrderAscending);\n        } else if (NullValue.class.isAssignableFrom(type)) {\n            return (TypeComparator<T>) NullValueComparator.getInstance();","sourceCodeStart":183,"sourceCodeEnd":219,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java#L183-L219","documentation":"Thrown by ValueTypeInfo.createComparator when the Value type wrapped by ValueTypeInfo is not Comparable (does not implement java.lang.Comparable). Flink needs a comparator for sorting/grouping keyed operations, and only Value types that are Comparable can serve as keys.","triggerScenarios":"Called when createComparator is invoked on a ValueTypeInfo whose type class does not implement Comparable. This happens when a custom Value type (implementing org.apache.flink.types.Value) is used as a key in groupBy, keyBy, join, or coGroup operations without implementing Comparable.","commonSituations":"Creating a custom Value type that implements Value (for custom serialization) but forgets to implement Comparable, then using it as a key in keyed operations. Using a Value type in a sorting or grouping context where comparability is required.","solutions":["Make the custom Value type implement Comparable<ValueType> (override compareTo).","Use a KeySelector to extract a Comparable key field instead of using the Value type directly as a key.","Switch to a Comparable Value type such as IntValue, StringValue, or a CopyableValue subclass.","If using the type as a key is not needed, avoid calling operations that trigger comparator creation."],"exampleFix":"// before\npublic class MyValue implements Value {\n    public void write(DataOutput out) { ... }\n    public void readFields(DataInput in) { ... }\n    // NOT Comparable -> throws when used as key\n}\n\n// after\npublic class MyValue implements Value, Comparable<MyValue> {\n    public void write(DataOutput out) { ... }\n    public void readFields(DataInput in) { ... }\n    public int compareTo(MyValue o) { return this.id - o.id; }\n}","handlingStrategy":"validation","validationCode":"// Check if the Value type is Comparable before using as a key\nValueTypeInfo<?> vti = (ValueTypeInfo<?>) typeInfo;\nif (!vti.isKeyType()) {\n    // cannot use as key; use KeySelector or switch to Comparable Value\n}","typeGuard":"static boolean isComparableValue(Class<? extends Value> valueClass) {\n    return Comparable.class.isAssignableFrom(valueClass);\n}","tryCatchPattern":"try {\n    TypeComparator<?> cmp = vti.createComparator(true, executionConfig);\n} catch (RuntimeException e) {\n    if (e.getMessage().contains(\"not Comparable\")) {\n        // implement Comparable or use KeySelector\n    } else throw e;\n}","preventionTips":["Make custom Value types implement Comparable<ValueType> if they will be used as keys.","Use KeySelector to extract a Comparable key instead of using a non-Comparable Value as a key.","Use built-in Comparable Value types (IntValue, StringValue, etc.) for keyed operations."],"tags":["value-type","comparator","comparable","key-type"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}