apache/flink · error · InvalidProgramException

Input type of coGroup must be one of composite types or atom

Error message

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

What it means

Thrown by CoGroupOperatorBase.getTypeComparator() when the input TypeInformation is neither a CompositeType (e.g., TupleTypeInfo, PojoTypeInfo, RowTypeInfo) nor an AtomicType (e.g., BasicTypeInfo). The comparator factory needs to extract key positions from composite types or use the natural ordering of atomic types; other type kinds (e.g., ObjectArrayTypeInfo, MultisetTypeInfo) are unsupported for coGroup key extraction.

Source

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

        return result;
    }

    @SuppressWarnings("unchecked")
    private <T> TypeComparator<T> getTypeComparator(
            ExecutionConfig executionConfig,
            TypeInformation<T> inputType,
            int[] inputKeys,
            boolean[] inputSortDirections) {
        if (inputType instanceof CompositeType) {
            return ((CompositeType<T>) inputType)
                    .createComparator(inputKeys, inputSortDirections, 0, executionConfig);
        } else if (inputType instanceof AtomicType) {
            return ((AtomicType<T>) inputType)
                    .createComparator(inputSortDirections[0], executionConfig);
        }

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

    private static class CoGroupSortListIterator<IN1, IN2> {

        private static enum MatchStatus {
            NONE_REMAINED,
            FIRST_REMAINED,
            SECOND_REMAINED,
            FIRST_EMPTY,
            SECOND_EMPTY
        }

        private final ListKeyGroupedIterator<IN1> iterator1;

        private final ListKeyGroupedIterator<IN2> iterator2;

        private final TypePairComparator<IN1, IN2> pairComparator;

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Ensure the input DataSets have composite types (Tuple, POJO, Row, CaseClass) or atomic types (String, Integer, etc.) before calling coGroup.
2. If your data is in arrays or lists, map them to Tuples or POJOs first.
3. Register a proper TypeInformation via a TypeInfoFactory if using custom types.
4. Check the type with dataSet.getType() and verify it is a CompositeType or AtomicType before coGroup.

Example fix

// before
DataSet<String[]> arrays = env.fromElements(new String[]{"a"}, new String[]{"b"});
arrays.coGroup(other).where(0).equalTo(0); // throws: array type is neither composite nor atomic

// after
DataSet<Tuple2<String, Integer>> tuples = arrays
    .map(arr -> Tuple2.of(arr[0], arr.length));
tuples.coGroup(other).where(0).equalTo(0); // ok: Tuple is CompositeType

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

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

try {
    dataSet1.coGroup(dataSet2).where(0).equalTo(0).with(fn);
} catch (InvalidProgramException e) {
    if (e.getMessage().contains("composite types or atomic types")) {
        // transform data to Tuple or POJO before coGroup
        throw new RuntimeException("Unsupported coGroup input type: " + e.getMessage(), e);
    }
    throw e;
}

When it happens

Trigger: Calling coGroup on a DataSet whose type is not a tuple, POJO, row, or basic type — for instance, a DataSet of arrays, maps, or lists where key extraction is attempted by field position. Using coGroup with a type that was created via a custom TypeInfo that does not extend CompositeType or AtomicType.

Common situations: Attempting to coGroup DataSets of collections or arrays. Using a non-standard TypeInformation registered via a TypeInfoFactory that doesn't extend CompositeType or AtomicType. Accidentally operating on a raw/primitive-typed dataset where a keyed coGroup requires positional field access.

Related errors

• Input types of coGroup must be composite types.
• Input type of GroupCombine must be one of composite types or
• The number of specified keys is different.
• Operator producing the next version of the partial solution
• No termination condition is set (neither fix number of itera

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