docs: @Override
  	protected List<IN> executeOnCollections(List<IN> inputData, RuntimeContext runtimeContext, ExecutionConfig executionConfig) {
  
  		TypeInformation<IN> inputType = getInput().getOperatorInfo().getOutputType();
  
  		int[] sortColumns = this.partitionOrdering.getFieldPositions();
  		boolean[] sortOrderings = this.partitionOrdering.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) inputType).createComparator(sortOrderings[0], executionConfig);
  		} else {
  			throw new UnsupportedOperationException("Partition 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);
  			}
  		});
  
  		return inputData;
  	}