apache/flink · error · InvalidProgramException

Grouping is only possible on composite types.

Error message

Grouping is only possible on composite types.

What it means

Thrown by ReduceOperatorBase.executeOnCollections when the local (collection) reduce input type is not a CompositeType, so key columns cannot be extracted for grouping; grouping requires composite types.

Solutions

1. Address the cause reported by the error message: Grouping is only possible on composite types.
2. Verify the inputs, configuration values, and classpath/dependency setup related to this operation, then retry.

Example fix

Correct the condition described ("Grouping is only possible on composite types.") and rerun the job or command.

When it happens

Trigger: Triggered at runtime when the operation fails because: Grouping is only possible on composite types.

Common situations: Commonly caused by misconfiguration, missing dependencies or files, unsupported types or operations, or invalid user input leading to: Grouping is only possible on composite types.

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

Appendix: source

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

    @Override
    protected List<T> executeOnCollections(
            List<T> inputData, RuntimeContext ctx, ExecutionConfig executionConfig)
            throws Exception {
        // make sure we can handle empty inputs
        if (inputData.isEmpty()) {
            return Collections.emptyList();
        }

        ReduceFunction<T> function = this.userFunction.getUserCodeObject();

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

        int[] inputColumns = getKeyColumns(0);

        if (!(inputType instanceof CompositeType) && inputColumns.length > 1) {
            throw new InvalidProgramException("Grouping is only possible on composite types.");
        }

        FunctionUtils.setFunctionRuntimeContext(function, ctx);
        FunctionUtils.openFunction(function, DefaultOpenContext.INSTANCE);

        TypeSerializer<T> serializer =
                getOperatorInfo()
                        .getInputType()
                        .createSerializer(executionConfig.getSerializerConfig());

        if (inputColumns.length > 0) {
            boolean[] inputOrderings = new boolean[inputColumns.length];
            TypeComparator<T> inputComparator =
                    inputType instanceof AtomicType
                            ? ((AtomicType<T>) inputType).createComparator(false, executionConfig)
                            : ((CompositeType<T>) inputType)
                                    .createComparator(
                                            inputColumns, inputOrderings, 0, executionConfig);

View on GitHub (pinned to 2f3c205e92)