{"record":{"id":"84814f8c15defe90","repo":"apache/flink","slug":"input-type-of-groupcombine-must-be-one-of-composit","errorCode":null,"errorMessage":"Input type of GroupCombine must be one of composite types or atomic types.","messagePattern":"Input type of GroupCombine must be one of composite types or atomic types\\.","errorType":"exception","errorClass":"InvalidProgramException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupCombineOperatorBase.java","lineNumber":103,"sourceCode":"     * @return The secondary order.\n     */\n    public Ordering getGroupOrder() {\n        return this.groupOrder;\n    }\n\n    private TypeComparator<IN> getTypeComparator(\n            TypeInformation<IN> typeInfo,\n            int[] sortColumns,\n            boolean[] sortOrderings,\n            ExecutionConfig executionConfig) {\n        if (typeInfo instanceof CompositeType) {\n            return ((CompositeType<IN>) typeInfo)\n                    .createComparator(sortColumns, sortOrderings, 0, executionConfig);\n        } else if (typeInfo instanceof AtomicType) {\n            return ((AtomicType<IN>) typeInfo).createComparator(sortOrderings[0], executionConfig);\n        }\n\n        throw new InvalidProgramException(\n                \"Input type of GroupCombine must be one of composite types or atomic types.\");\n    }\n\n    // --------------------------------------------------------------------------------------------\n\n    @Override\n    protected List<OUT> executeOnCollections(\n            List<IN> inputData, RuntimeContext ctx, ExecutionConfig executionConfig)\n            throws Exception {\n        GroupCombineFunction<IN, OUT> function = this.userFunction.getUserCodeObject();\n\n        UnaryOperatorInformation<IN, OUT> operatorInfo = getOperatorInfo();\n        TypeInformation<IN> inputType = operatorInfo.getInputType();\n\n        int[] keyColumns = getKeyColumns(0);\n        int[] sortColumns = keyColumns;\n        boolean[] sortOrderings = new boolean[sortColumns.length];\n","sourceCodeStart":85,"sourceCodeEnd":121,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupCombineOperatorBase.java#L85-L121","documentation":"Thrown by GroupCombineOperatorBase.getTypeComparator() when the input TypeInformation is neither a CompositeType nor an AtomicType. The combine operation needs a comparator to sort/group records by key positions (for composite types) or natural ordering (for atomic types); unsupported type kinds like arrays, maps, or custom non-standard TypeInfo implementations cannot provide a comparator.","triggerScenarios":"Calling combineGroup on a DataSet whose type is not composite or atomic — e.g., a DataSet of arrays, lists, maps, or a custom TypeInformation that doesn't implement either interface.","commonSituations":"Attempting to group-combine DataSets of collection types. Using custom TypeInformation implementations that extend TypeInformation directly instead of CompositeType or AtomicType. Incorrect type inference from Java generics producing a non-standard TypeInfo.","solutions":["Ensure the input DataSet has a composite type (Tuple, POJO, Row) or atomic type (String, Integer, etc.) before calling combineGroup.","Map non-standard types (arrays, maps) to Tuples or POJOs first.","Register a proper TypeInformation via TypeInfoFactory for custom types.","Verify the type via dataSet.getType() before the combine operation."],"exampleFix":"// before\nDataSet<Map<String, Integer>> maps = env.fromElements(Map.of(\"a\", 1));\nmaps.combineGroup(myCombiner); // throws: Map type is neither composite nor atomic\n\n// after\nDataSet<Tuple2<String, Integer>> tuples = maps\n    .flatMap(m -> m.entrySet().stream())\n    .map(e -> Tuple2.of(e.getKey(), e.getValue()));\ntuples.combineGroup(myCombiner); // ok: Tuple is CompositeType","handlingStrategy":"type-guard","validationCode":"void validateGroupCombineType(TypeInformation<?> type) {\n    if (!(type instanceof CompositeType) && !(type instanceof AtomicType)) {\n        throw new InvalidProgramException(\n            \"Type \" + type + \" is neither CompositeType nor AtomicType; cannot group combine.\");\n    }\n}","typeGuard":"static boolean isGroupCombineSupportedType(TypeInformation<?> type) {\n    return type instanceof CompositeType || type instanceof AtomicType;\n}","tryCatchPattern":"try {\n    dataSet.combineGroup(myCombiner);\n} catch (InvalidProgramException e) {\n    if (e.getMessage().contains(\"composite types or atomic types\")) {\n        // transform to Tuple or POJO\n        throw new RuntimeException(\"Unsupported combine input type\", e);\n    }\n    throw e;\n}","preventionTips":["Verify dataSet.getType() is CompositeType or AtomicType before combineGroup.","Map non-standard types (arrays, maps) to Tuples or POJOs.","Register custom types with a TypeInfoFactory that produces CompositeType or AtomicType."],"tags":["group-combine","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"}