apache/flink · error · UnsupportedOperationException

Local output sorting does not support type {inputType} yet.

Error message

Local output sorting does not support type {inputType} yet.

What it means

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.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java:203

            throws Exception {
        OutputFormat<IN> format = this.formatWrapper.getUserCodeObject();
        TypeInformation<IN> inputType = getInput().getOperatorInfo().getOutputType();

        if (this.localOrdering != null) {
            int[] sortColumns = this.localOrdering.getFieldPositions();
            boolean[] sortOrderings = this.localOrdering.getFieldSortDirections();

            final TypeComparator<IN> sortComparator;
            if (inputType instanceof CompositeType) {
                sortComparator =
                        ((CompositeType<IN>) inputType)
                                .createComparator(sortColumns, sortOrderings, 0, executionConfig);
            } else if (inputType instanceof AtomicType) {
                sortComparator =
                        ((AtomicType<IN>) inputType)
                                .createComparator(sortOrderings[0], executionConfig);
            } else {
                throw new UnsupportedOperationException(
                        "Local output sorting does not support type " + inputType + " yet.");
            }

            Collections.sort(
                    inputData,
                    new Comparator<IN>() {
                        @Override
                        public int compare(IN o1, IN o2) {
                            return sortComparator.compare(o1, o2);
                        }
                    });
        }

        if (format instanceof InitializeOnMaster) {
            ((InitializeOnMaster) format).initializeGlobal(1);
        }
        format.configure(this.parameters);

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Remove the localOrdering directive from the sink, or move sorting upstream into a prior operator that operates on a sortable type.
2. Change the sink's input type to a Tuple/POJO/Row (CompositeType) or an AtomicType so a comparator can be built.
3. If you must test sorting, use the real distributed runtime (createLocalEnvironment with the actual executor) instead of collection execution.

Example fix

// before
DataSet<MyRawType> ds = ...;
ds.write(out, localOrderingFor(MyRawType.class));
// after
DataSet<Tuple2<String,Integer>> ds = ...;
ds.sortLocalBy(0).write(out);

TypeInformation<IN> t = ds.getType();
boolean sortable = t instanceof CompositeType || t instanceof AtomicType;
if (localOrdering != null && !sortable) {
    throw new IllegalArgumentException("Cannot locally sort type " + t + " in collection mode");
}

static <T> boolean isLocallySortable(TypeInformation<T> t) {
    return t instanceof CompositeType || t instanceof AtomicType;
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Field expression must be equal to '*' or '_' for atomic type
• The data sink {} has no input.
• Specifying keys via field positions is only valid for tuple
• This type ({ffd.getType()}) cannot be used as key.
• This type ({field.getType()}) cannot be used as key.

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