apache/flink · error · InvalidProgramException

Input types of coGroup must be composite types.

Error message

Input types of coGroup must be composite types.

What it means

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.

Source

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

                        result,
                        getOperatorInfo()
                                .getOutputType()
                                .createSerializer(executionConfig.getSerializerConfig()));

        function.coGroup(iterator1, iterator2, resultCollector);

        FunctionUtils.closeFunction(function);

        return result;
    }

    private <T> TypeComparator<T> getTypeComparator(
            ExecutionConfig executionConfig,
            TypeInformation<T> inputType,
            int[] inputKeys,
            boolean[] inputSortDirections) {
        if (!(inputType instanceof CompositeType)) {
            throw new InvalidProgramException("Input types of coGroup must be composite types.");
        }

        return ((CompositeType<T>) inputType)
                .createComparator(inputKeys, inputSortDirections, 0, executionConfig);
    }

    public static class SimpleListIterable<IN> implements Iterable<IN> {
        private List<IN> values;
        private TypeSerializer<IN> serializer;
        private boolean copy;

        public SimpleListIterable(
                List<IN> values, final TypeComparator<IN> comparator, TypeSerializer<IN> serializer)
                throws IOException {
            this.values = values;
            this.serializer = serializer;

            Collections.sort(

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Ensure input DataSets for raw coGroup have composite types (Tuple, POJO, Row).
2. If your data is an atomic type, wrap it in a Tuple first: map(x -> Tuple1.of(x)).
3. Use the regular coGroup operator (which accepts atomic types) if you do not need raw coGroup semantics.
4. Verify the type with dataSet.getType() instanceof CompositeType before raw coGroup.

Example fix

// before
DataSet<String> strings = env.fromElements("a", "b");
// raw coGroup on atomic type
DataSet<Tuple2<String, String>> result = strings.coGroup(other)
    .where(0).equalTo(0).with(myCoGroupFunction); // may route through CoGroupRaw

// after
DataSet<Tuple1<String>> tuples = strings.map(Tuple1::of);
DataSet<Tuple2<String, String>> result = tuples.coGroup(otherTuples)
    .where(0).equalTo(0).with(myCoGroupFunction); // ok: Tuple is CompositeType

void validateRawCoGroupType(TypeInformation<?> type) {
    if (!(type instanceof CompositeType)) {
        throw new InvalidProgramException(
            "Type " + type + " is not a CompositeType; raw coGroup requires composite types.");
    }
}

static boolean isRawCoGroupSupportedType(TypeInformation<?> type) {
    return type instanceof CompositeType;
}

try {
    // raw coGroup operation
    result = executeRawCoGroup(data1, data2, fn);
} catch (InvalidProgramException e) {
    if (e.getMessage().contains("must be composite types")) {
        // wrap atomic type in Tuple1
        DataSet<Tuple1<String>> wrapped = data1.map(Tuple1::of);
        result = executeRawCoGroup(wrapped, data2Wrapped, fn);
    }
}

When it happens

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

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

Related errors

• Input type of coGroup must be one of composite types or atom
• Input type of GroupCombine must be one of composite types or
• 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/be28f6a5c09f6714. Report an issue: GitHub.