apache/flink · error · InvalidProgramException

Input type of GroupReduce must be one of composite types or…

Error message

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

What it means

Thrown by GroupReduceOperatorBase.getTypeComparator when the input TypeInformation is neither a CompositeType nor an AtomicType, so no TypeComparator can be created for the configured sort columns. This guards the sort-comparator setup for group-reduce operators; the input type must be a tuple/case-class/row-like composite or a comparable atomic type.

Solutions

1. Address the cause reported by the error message: Input type of GroupReduce must be one of composite types or atomic types.
2. Verify the inputs, configuration values, and classpath/dependency setup related to this operation, then retry.

Example fix

Correct the condition described ("Input type of GroupReduce must be one of composite types or atomic types.") and rerun the job or command.

When it happens

Trigger: Triggered at runtime when the operation fails because: Input type of GroupReduce must be one of composite types or atomic types.

Common situations: Commonly caused by misconfiguration, missing dependencies or files, unsupported types or operations, or invalid user input leading to: Input type of GroupReduce must be one of composite types or atomic types.

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

Appendix: source

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

    }

    public Partitioner<?> getCustomPartitioner() {
        return customPartitioner;
    }

    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 GroupReduce must be one of composite types or atomic types.");
    }

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

    @Override
    protected List<OUT> executeOnCollections(
            List<IN> inputData, RuntimeContext ctx, ExecutionConfig executionConfig)
            throws Exception {
        GroupReduceFunction<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)