{"record":{"id":"e3a27d25b65e7e7a","repo":"apache/flink","slug":"local-output-sorting-does-not-support-type-inputt","errorCode":null,"errorMessage":"Local output sorting does not support type {inputType} yet.","messagePattern":"Local output sorting does not support type (.+?) yet\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java","lineNumber":203,"sourceCode":"            throws Exception {\n        OutputFormat<IN> format = this.formatWrapper.getUserCodeObject();\n        TypeInformation<IN> inputType = getInput().getOperatorInfo().getOutputType();\n\n        if (this.localOrdering != null) {\n            int[] sortColumns = this.localOrdering.getFieldPositions();\n            boolean[] sortOrderings = this.localOrdering.getFieldSortDirections();\n\n            final TypeComparator<IN> sortComparator;\n            if (inputType instanceof CompositeType) {\n                sortComparator =\n                        ((CompositeType<IN>) inputType)\n                                .createComparator(sortColumns, sortOrderings, 0, executionConfig);\n            } else if (inputType instanceof AtomicType) {\n                sortComparator =\n                        ((AtomicType<IN>) inputType)\n                                .createComparator(sortOrderings[0], executionConfig);\n            } else {\n                throw new UnsupportedOperationException(\n                        \"Local output sorting does not support type \" + inputType + \" yet.\");\n            }\n\n            Collections.sort(\n                    inputData,\n                    new Comparator<IN>() {\n                        @Override\n                        public int compare(IN o1, IN o2) {\n                            return sortComparator.compare(o1, o2);\n                        }\n                    });\n        }\n\n        if (format instanceof InitializeOnMaster) {\n            ((InitializeOnMaster) format).initializeGlobal(1);\n        }\n        format.configure(this.parameters);\n","sourceCodeStart":185,"sourceCodeEnd":221,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java#L185-L221","documentation":"Thrown by GenericDataSinkBase.executeOnCollections() when the sink has a localOrdering (sort on output) but the input type is neither a CompositeType nor an AtomicType. The collection execution path (used by CollectionEnvironment and local tests) needs a TypeComparator to sort records; if the type system cannot provide one, sorting is impossible and the operation is rejected.","triggerScenarios":"Running a DataSet job with a sink that calls sortLocal(...) / setLocalOrdering on a type that is neither atomic (primitives/String) nor composite (Tuple/POJO/Row), e.g. a raw ObjectTypeInfo, a WritableTypeInfo in some configs, or a custom TypeInfo that does not implement either marker.","commonSituations":"Unit/integration tests using ExecutionEnvironment.createLocalEnvironment() (collection mode) on a sink with local ordering and a non-standard type; switching a sink from Tuple to a generic type without removing the sort directive.","solutions":["Remove the localOrdering directive from the sink, or move sorting upstream into a prior operator that operates on a sortable type.","Change the sink's input type to a Tuple/POJO/Row (CompositeType) or an AtomicType so a comparator can be built.","If you must test sorting, use the real distributed runtime (createLocalEnvironment with the actual executor) instead of collection execution."],"exampleFix":"// before\nDataSet<MyRawType> ds = ...;\nds.write(out, localOrderingFor(MyRawType.class));\n// after\nDataSet<Tuple2<String,Integer>> ds = ...;\nds.sortLocalBy(0).write(out);","handlingStrategy":"validation","validationCode":"TypeInformation<IN> t = ds.getType();\nboolean sortable = t instanceof CompositeType || t instanceof AtomicType;\nif (localOrdering != null && !sortable) {\n    throw new IllegalArgumentException(\"Cannot locally sort type \" + t + \" in collection mode\");\n}","typeGuard":"static <T> boolean isLocallySortable(TypeInformation<T> t) {\n    return t instanceof CompositeType || t instanceof AtomicType;\n}","tryCatchPattern":null,"preventionTips":["Prefer Tuple/POJO/Row (CompositeType) or primitive/String (AtomicType) for sinks that need local ordering.","Test sorting under the real distributed runtime, not only collection execution.","Drop the localOrdering directive if you cannot guarantee a sortable type."],"tags":["flink-core","data-sink","collection-execution","sorting","type-system"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}