apache/flink · error · InvalidProgramException

Input type of GroupCombine must be one of composite types or

Error message

Input type of GroupCombine must be one of composite types or atomic types.

What it means

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.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupCombineOperatorBase.java:103

     * @return The secondary order.
     */
    public Ordering getGroupOrder() {
        return this.groupOrder;
    }

    private TypeComparator<IN> getTypeComparator(
            TypeInformation<IN> typeInfo,
            int[] sortColumns,
            boolean[] sortOrderings,
            ExecutionConfig executionConfig) {
        if (typeInfo instanceof CompositeType) {
            return ((CompositeType<IN>) typeInfo)
                    .createComparator(sortColumns, sortOrderings, 0, executionConfig);
        } else if (typeInfo instanceof AtomicType) {
            return ((AtomicType<IN>) typeInfo).createComparator(sortOrderings[0], executionConfig);
        }

        throw new InvalidProgramException(
                "Input type of GroupCombine must be one of composite types or atomic types.");
    }

    // --------------------------------------------------------------------------------------------

    @Override
    protected List<OUT> executeOnCollections(
            List<IN> inputData, RuntimeContext ctx, ExecutionConfig executionConfig)
            throws Exception {
        GroupCombineFunction<IN, OUT> function = this.userFunction.getUserCodeObject();

        UnaryOperatorInformation<IN, OUT> operatorInfo = getOperatorInfo();
        TypeInformation<IN> inputType = operatorInfo.getInputType();

        int[] keyColumns = getKeyColumns(0);
        int[] sortColumns = keyColumns;
        boolean[] sortOrderings = new boolean[sortColumns.length];

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Ensure the input DataSet has a composite type (Tuple, POJO, Row) or atomic type (String, Integer, etc.) before calling combineGroup.
2. Map non-standard types (arrays, maps) to Tuples or POJOs first.
3. Register a proper TypeInformation via TypeInfoFactory for custom types.
4. Verify the type via dataSet.getType() before the combine operation.

Example fix

// before
DataSet<Map<String, Integer>> maps = env.fromElements(Map.of("a", 1));
maps.combineGroup(myCombiner); // throws: Map type is neither composite nor atomic

// after
DataSet<Tuple2<String, Integer>> tuples = maps
    .flatMap(m -> m.entrySet().stream())
    .map(e -> Tuple2.of(e.getKey(), e.getValue()));
tuples.combineGroup(myCombiner); // ok: Tuple is CompositeType

void validateGroupCombineType(TypeInformation<?> type) {
    if (!(type instanceof CompositeType) && !(type instanceof AtomicType)) {
        throw new InvalidProgramException(
            "Type " + type + " is neither CompositeType nor AtomicType; cannot group combine.");
    }
}

static boolean isGroupCombineSupportedType(TypeInformation<?> type) {
    return type instanceof CompositeType || type instanceof AtomicType;
}

try {
    dataSet.combineGroup(myCombiner);
} catch (InvalidProgramException e) {
    if (e.getMessage().contains("composite types or atomic types")) {
        // transform to Tuple or POJO
        throw new RuntimeException("Unsupported combine input type", e);
    }
    throw e;
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Input type of coGroup must be one of composite types or atom
• Input types of coGroup must be composite types.
• Operator producing the next version of the partial solution
• No termination condition is set (neither fix number of itera
• The filter factor cannot be smaller than zero.

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