{"record":{"id":"eebb1bf6afa7c332","repo":"apache/flink","slug":"cannot-use-the-key-partitioner-for-composite-keys","errorCode":null,"errorMessage":"Cannot use the key partitioner for composite keys (more than one key field)","messagePattern":"Cannot use the key partitioner for composite keys \\(more than one key field\\)","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java","lineNumber":165,"sourceCode":"    /**\n     * Checks whether the operation is combinable.\n     *\n     * @return True, if the UDF is combinable, false if not.\n     * @see #setCombinable(boolean)\n     */\n    public boolean isCombinable() {\n        return this.combinable;\n    }\n\n    public void setCustomPartitioner(Partitioner<?> customPartitioner) {\n        if (customPartitioner != null) {\n            int[] keys = getKeyColumns(0);\n            if (keys == null || keys.length == 0) {\n                throw new IllegalArgumentException(\n                        \"Cannot use custom partitioner for a non-grouped GroupReduce (AllGroupReduce)\");\n            }\n            if (keys.length > 1) {\n                throw new IllegalArgumentException(\n                        \"Cannot use the key partitioner for composite keys (more than one key field)\");\n            }\n        }\n        this.customPartitioner = customPartitioner;\n    }\n\n    public Partitioner<?> getCustomPartitioner() {\n        return customPartitioner;\n    }\n\n    private TypeComparator<IN> getTypeComparator(\n            TypeInformation<IN> typeInfo,\n            int[] sortColumns,\n            boolean[] sortOrderings,\n            ExecutionConfig executionConfig) {\n        if (typeInfo instanceof CompositeType) {\n            return ((CompositeType<IN>) typeInfo)\n                    .createComparator(sortColumns, sortOrderings, 0, executionConfig);","sourceCodeStart":147,"sourceCodeEnd":183,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java#L147-L183","documentation":"Thrown by GroupReduceOperatorBase.setCustomPartitioner(Partitioner<?>) when the group reduce operation has more than one key field (a composite key). A custom Partitioner<T> can only partition on a single key value; when there are multiple key fields, the partitioning key is a composite object and the single-value Partitioner interface cannot handle it. Flink rejects this rather than silently using the wrong partitioning.","triggerScenarios":"Calling groupBy(field1, field2) (two or more fields) then setCustomPartitioner(partitioner) on the GroupReduceOperatorBase. Using a tuple key selector that yields a composite key, then setting a custom partitioner.","commonSituations":"Multi-field groupBy with a custom data-distribution partitioner. Attempting to use hash-based or range-based custom partitioning on a composite key where only a single Partitioner<Object> is available.","solutions":["Reduce to a single key field: call groupBy on only one field, or flatten the composite key into a single field before groupBy.","Create a computed single key field (e.g., concatenate or hash the multi-field key) and group by that.","If you need multi-field custom partitioning, implement a custom partitioner at the DataSet level using partitionCustom with a KeySelector that produces a single partition key.","Do not use setCustomPartitioner with multi-field group keys; rely on Flink's default hash/range partitioning instead."],"exampleFix":"// before\nDataSet<Tuple3<String, Integer, Double>> data = ...;\nDataSet<Out> result = data.groupBy(0, 1).reduceGroup(myReducer);\n((GroupReduceOperatorBase) op).setCustomPartitioner(myPartitioner); // throws: composite key\n\n// after\nDataSet<Out> result = data\n    .map(t -> Tuple2.of(t.f0 + \"_\" + t.f1, t)) // flatten to single key\n    .groupBy(0).reduceGroup(myReducer);\n// or partition before grouping:\ndata.partitionCustom(myPartitioner, t -> t.f0 + \"_\" + t.f1).groupBy(...).reduceGroup(...);","handlingStrategy":"validation","validationCode":"void safeSetCustomPartitioner(GroupReduceOperatorBase<?, ?, ?> op, Partitioner<?> partitioner) {\n    if (partitioner == null) { op.setCustomPartitioner(null); return; }\n    int[] keys = op.getKeyColumns(0);\n    if (keys == null || keys.length == 0) {\n        throw new IllegalArgumentException(\"Non-keyed GroupReduce; call groupBy first\");\n    }\n    if (keys.length > 1) {\n        throw new IllegalArgumentException(\"Composite key (\" + keys.length\n            + \" fields); custom partitioner supports only single-field keys\");\n    }\n    op.setCustomPartitioner(partitioner);\n}","typeGuard":"boolean hasSingleKeyField(GroupReduceOperatorBase<?, ?, ?> op) {\n    int[] keys = op.getKeyColumns(0);\n    return keys != null && keys.length == 1;\n}","tryCatchPattern":"try {\n    reduceOp.setCustomPartitioner(partitioner);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"composite keys\")) {\n        // flatten multi-field key into a single field, or use partitionCustom\n        log.error(\"Custom partitioner requires single-field key; got composite key\");\n    }\n}","preventionTips":["Use groupBy on a single field if you need a custom partitioner.","Flatten composite keys into a single field before groupBy when custom partitioning is needed.","Use partitionCustom with a KeySelector for multi-field custom partitioning instead."],"tags":["group-reduce","custom-partitioner","composite-key","dataset-api","flink-core"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}