{"record":{"id":"be28f6a5c09f6714","repo":"apache/flink","slug":"input-types-of-cogroup-must-be-composite-types","errorCode":null,"errorMessage":"Input types of coGroup must be composite types.","messagePattern":"Input types of coGroup must be composite types\\.","errorType":"exception","errorClass":"InvalidProgramException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/base/CoGroupRawOperatorBase.java","lineNumber":248,"sourceCode":"                        result,\n                        getOperatorInfo()\n                                .getOutputType()\n                                .createSerializer(executionConfig.getSerializerConfig()));\n\n        function.coGroup(iterator1, iterator2, resultCollector);\n\n        FunctionUtils.closeFunction(function);\n\n        return result;\n    }\n\n    private <T> TypeComparator<T> getTypeComparator(\n            ExecutionConfig executionConfig,\n            TypeInformation<T> inputType,\n            int[] inputKeys,\n            boolean[] inputSortDirections) {\n        if (!(inputType instanceof CompositeType)) {\n            throw new InvalidProgramException(\"Input types of coGroup must be composite types.\");\n        }\n\n        return ((CompositeType<T>) inputType)\n                .createComparator(inputKeys, inputSortDirections, 0, executionConfig);\n    }\n\n    public static class SimpleListIterable<IN> implements Iterable<IN> {\n        private List<IN> values;\n        private TypeSerializer<IN> serializer;\n        private boolean copy;\n\n        public SimpleListIterable(\n                List<IN> values, final TypeComparator<IN> comparator, TypeSerializer<IN> serializer)\n                throws IOException {\n            this.values = values;\n            this.serializer = serializer;\n\n            Collections.sort(","sourceCodeStart":230,"sourceCodeEnd":266,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/base/CoGroupRawOperatorBase.java#L230-L266","documentation":"Thrown by CoGroupRawOperatorBase.getTypeComparator() when the input TypeInformation is not a CompositeType. Unlike the regular CoGroupOperatorBase which accepts both composite and atomic types, the raw coGroup variant strictly requires composite types because it performs field-position-based key extraction (createComparator with int[] key positions), which is only defined for composite types.","triggerScenarios":"Calling coGroupRaw (or using CoGroupRawOperatorBase directly) on a DataSet whose type is an atomic type like String, Integer, or Long, rather than a Tuple, POJO, or Row.","commonSituations":"Using the raw coGroup operator (less common, lower-level) on simple/primitive typed DataSets. Confusion between regular coGroup (accepts atomic types) and raw coGroup (requires composite types). Custom operator construction that delegates to CoGroupRawOperatorBase.","solutions":["Ensure input DataSets for raw coGroup have composite types (Tuple, POJO, Row).","If your data is an atomic type, wrap it in a Tuple first: map(x -> Tuple1.of(x)).","Use the regular coGroup operator (which accepts atomic types) if you do not need raw coGroup semantics.","Verify the type with dataSet.getType() instanceof CompositeType before raw coGroup."],"exampleFix":"// before\nDataSet<String> strings = env.fromElements(\"a\", \"b\");\n// raw coGroup on atomic type\nDataSet<Tuple2<String, String>> result = strings.coGroup(other)\n    .where(0).equalTo(0).with(myCoGroupFunction); // may route through CoGroupRaw\n\n// after\nDataSet<Tuple1<String>> tuples = strings.map(Tuple1::of);\nDataSet<Tuple2<String, String>> result = tuples.coGroup(otherTuples)\n    .where(0).equalTo(0).with(myCoGroupFunction); // ok: Tuple is CompositeType","handlingStrategy":"type-guard","validationCode":"void validateRawCoGroupType(TypeInformation<?> type) {\n    if (!(type instanceof CompositeType)) {\n        throw new InvalidProgramException(\n            \"Type \" + type + \" is not a CompositeType; raw coGroup requires composite types.\");\n    }\n}","typeGuard":"static boolean isRawCoGroupSupportedType(TypeInformation<?> type) {\n    return type instanceof CompositeType;\n}","tryCatchPattern":"try {\n    // raw coGroup operation\n    result = executeRawCoGroup(data1, data2, fn);\n} catch (InvalidProgramException e) {\n    if (e.getMessage().contains(\"must be composite types\")) {\n        // wrap atomic type in Tuple1\n        DataSet<Tuple1<String>> wrapped = data1.map(Tuple1::of);\n        result = executeRawCoGroup(wrapped, data2Wrapped, fn);\n    }\n}","preventionTips":["Use composite types (Tuple, POJO, Row) for raw coGroup operations.","Wrap atomic types in Tuple1 before raw coGroup.","Prefer the regular coGroup operator if you need atomic type support."],"tags":["cogroup-raw","type-system","comparator","invalid-program","dataset-api","flink-core"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}