import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private <T> TypeComparator<T> buildComparatorFor(int input, ExecutionConfig executionConfig, TypeInformation<T> typeInformation) {
	TypeComparator<T> comparator;
	if (typeInformation instanceof AtomicType) {
		comparator = ((AtomicType<T>) typeInformation).createComparator(true, executionConfig);
	} else if (typeInformation instanceof CompositeType) {
		int[] keyPositions = getKeyColumns(input);
		boolean[] orders = new boolean[keyPositions.length];
		Arrays.fill(orders, true);

		comparator = ((CompositeType<T>) typeInformation).createComparator(keyPositions, orders, 0, executionConfig);
	} else {
		throw new RuntimeException("Type information for input of type " + typeInformation.getClass()
				.getCanonicalName() + " is not supported. Could not generate a comparator.");
	}
	return comparator;
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
@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;
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private <T> TypeComparatorFactory<?> createComparator(TypeInformation<T> typeInfo, FieldList keys, boolean[] sortOrder) {
	
	TypeComparator<T> comparator;
	if (typeInfo instanceof CompositeType) {
		comparator = ((CompositeType<T>) typeInfo).createComparator(keys.toArray(), sortOrder, 0, executionConfig);
	}
	else if (typeInfo instanceof AtomicType) {
		// handle grouping of atomic types
		comparator = ((AtomicType<T>) typeInfo).createComparator(sortOrder[0], executionConfig);
	}
	else {
		throw new RuntimeException("Unrecognized type: " + typeInfo);
	}

	return new RuntimeComparatorFactory<T>(comparator);
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
private static <T> TypeComparatorFactory<?> createComparator(TypeInformation<T> typeInfo, FieldList keys, boolean[] sortOrder, ExecutionConfig executionConfig) {

		TypeComparator<T> comparator;
		if (typeInfo instanceof CompositeType) {
			comparator = ((CompositeType<T>) typeInfo).createComparator(keys.toArray(), sortOrder, 0, executionConfig);
		}
		else if (typeInfo instanceof AtomicType) {
			// handle grouping of atomic types
			comparator = ((AtomicType<T>) typeInfo).createComparator(sortOrder[0], executionConfig);
		}
		else {
			throw new RuntimeException("Unrecognized type: " + typeInfo);
		}

		return new RuntimeComparatorFactory<>(comparator);
	}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
@Override
public TypeComparator<Comparable> createComparator(boolean sortOrderAscending, ExecutionConfig config) {

	if(this.fieldTypeInfo != null) {
		TypeComparator<Comparable> fieldComparator = ((AtomicType)fieldTypeInfo).createComparator(sortOrderAscending, config);
		return new WrappingFieldComparator(fieldComparator, sortOrderAscending, Comparable.class);
	}
	else {
		TypeSerializer<Comparable> serializer = this.createSerializer(config);
		if (this.fieldComparator == null) {
			return new FieldComparator(sortOrderAscending, serializer, Comparable.class);
		} else {
			return new CustomFieldComparator(sortOrderAscending, this.fieldComparator, this.createSerializer(config));
		}
	}
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
private static <T> TypeComparatorFactory<?> createComparator(TypeInformation<T> typeInfo, FieldList keys, boolean[] sortOrder) {
	
	TypeComparator<T> comparator;
	if (typeInfo instanceof CompositeType) {
		comparator = ((CompositeType<T>) typeInfo).createComparator(keys.toArray(), sortOrder, 0);
	}
	else if (typeInfo instanceof AtomicType) {
		// handle grouping of atomic types
		throw new UnsupportedOperationException("Grouping on atomic types is currently not implemented. " + typeInfo);
	}
	else {
		throw new RuntimeException("Unrecognized type: " + typeInfo);
	}

	return new RuntimeComparatorFactory<T>(comparator);
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private <T> TypeComparator<T> getTypeComparator(ExecutionConfig executionConfig, TypeInformation<T> inputType, int[] inputKeys, boolean[] inputSortDirections) {
	if (inputType instanceof CompositeType) {
		return ((CompositeType<T>) inputType).createComparator(inputKeys, inputSortDirections, 0, executionConfig);
	} else if (inputType instanceof AtomicType) {
		return ((AtomicType<T>) inputType).createComparator(inputSortDirections[0], executionConfig);
	}

	throw new InvalidProgramException("Input type of coGroup must be one of composite types or atomic types.");
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
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 GroupCombine must be one of composite types or atomic types.");
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
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.");
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
/**
 * Recursively add all fields in this tuple type. We need this in particular to get all
 * the types.
 * @param keyId
 * @param keyFields
 */
public void addAllFields(int startKeyId, List<FlatFieldDescriptor> keyFields) {
	for(int i = 0; i < this.getArity(); i++) {
		TypeInformation<?> type = this.types[i];
		if(type instanceof AtomicType) {
			keyFields.add(new FlatFieldDescriptor(startKeyId, type));
		} else if(type instanceof TupleTypeInfoBase<?>) {
			TupleTypeInfoBase<?> ttb = (TupleTypeInfoBase<?>) type;
			ttb.addAllFields(startKeyId, keyFields);
		}
		startKeyId += type.getTotalFields();
	}
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
/**
 * Generic implementation of the comparator creation. Composite types are supplying the infrastructure
 * to create the actual comparators
 * @return
 */
public TypeComparator<T> createComparator(int[] logicalKeyFields, boolean[] orders, int logicalFieldOffset) {
	initializeNewComparator(logicalKeyFields.length);
	
	for(int logicalKeyFieldIndex = 0; logicalKeyFieldIndex < logicalKeyFields.length; logicalKeyFieldIndex++) {
		int logicalKeyField = logicalKeyFields[logicalKeyFieldIndex];
		int logicalField = logicalFieldOffset; // this is the global/logical field number
		for(int localFieldId = 0; localFieldId < this.getArity(); localFieldId++) {
			TypeInformation<?> localFieldType = this.getTypeAt(localFieldId);
			
			if(localFieldType instanceof AtomicType && logicalField == logicalKeyField) {
				// we found an atomic key --> create comparator
				addCompareField(localFieldId, ((AtomicType<?>) localFieldType).createComparator(orders[logicalKeyFieldIndex]) );
			} else if(localFieldType instanceof CompositeType  && // must be a composite type
					( logicalField <= logicalKeyField //check if keyField can be at or behind the current logicalField
					&& logicalKeyField <= logicalField + (localFieldType.getTotalFields() - 1) ) // check if logical field + lookahead could contain our key
					) {
				// we found a compositeType that is containing the logicalKeyField we are looking for --> create comparator
				addCompareField(localFieldId, ((CompositeType<?>) localFieldType).createComparator(new int[] {logicalKeyField}, new boolean[] {orders[logicalKeyFieldIndex]}, logicalField));
			}
			
			// maintain logicalField
			if(localFieldType instanceof CompositeType) {
				// we need to subtract 1 because we are not accounting for the local field (not accessible for the user)
				logicalField += localFieldType.getTotalFields() - 1;
			}
			logicalField++;
		}
	}
	return getNewComparator();
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
public FlatFieldDescriptor(int keyPosition, TypeInformation<?> type) {
	if( !(type instanceof AtomicType)) {
		throw new IllegalArgumentException("A flattened field can only be an atomic type");
	}
	this.keyPosition = keyPosition;
	this.type = type;
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
public boolean isAtomicType() {
	return typeInfo instanceof AtomicType;
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
public boolean isAtomicType() {
    return typeInfo instanceof AtomicType;
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
/**
 * Generic implementation of the comparator creation. Composite types are supplying the infrastructure
 * to create the actual comparators
 * @return The comparator
 */
@PublicEvolving
public TypeComparator<T> createComparator(int[] logicalKeyFields, boolean[] orders, int logicalFieldOffset, ExecutionConfig config) {

	TypeComparatorBuilder<T> builder = createTypeComparatorBuilder();

	builder.initializeTypeComparatorBuilder(logicalKeyFields.length);

	for (int logicalKeyFieldIndex = 0; logicalKeyFieldIndex < logicalKeyFields.length; logicalKeyFieldIndex++) {
		int logicalKeyField = logicalKeyFields[logicalKeyFieldIndex];
		int logicalField = logicalFieldOffset; // this is the global/logical field number
		boolean comparatorAdded = false;

		for (int localFieldId = 0; localFieldId < this.getArity() && logicalField <= logicalKeyField && !comparatorAdded; localFieldId++) {
			TypeInformation<?> localFieldType = this.getTypeAt(localFieldId);
			
			if (localFieldType instanceof AtomicType && logicalField == logicalKeyField) {
				// we found an atomic key --> create comparator
				builder.addComparatorField(
					localFieldId,
					((AtomicType<?>) localFieldType).createComparator(
						orders[logicalKeyFieldIndex],
						config));

				comparatorAdded = true;
			}
			// must be composite type and check that the logicalKeyField is within the bounds
			// of the composite type's logical fields
			else if (localFieldType instanceof CompositeType &&
				logicalField <= logicalKeyField &&
				logicalKeyField <= logicalField + (localFieldType.getTotalFields() - 1)) {
				// we found a compositeType that is containing the logicalKeyField we are looking for --> create comparator
				builder.addComparatorField(
					localFieldId,
					((CompositeType<?>) localFieldType).createComparator(
						new int[]{logicalKeyField},
						new boolean[]{orders[logicalKeyFieldIndex]},
						logicalField,
						config)
				);

				comparatorAdded = true;
			}

			if (localFieldType instanceof CompositeType) {
				// we need to subtract 1 because we are not accounting for the local field (not accessible for the user)
				logicalField += localFieldType.getTotalFields() - 1;
			}
			
			logicalField++;
		}

		if (!comparatorAdded) {
			throw new IllegalArgumentException("Could not add a comparator for the logical" +
				"key field index " + logicalKeyFieldIndex + ".");
		}
	}

	return builder.createTypeComparator(config);
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
protected void executeOnCollections(List<IN> inputData, RuntimeContext ctx, ExecutionConfig executionConfig) 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);

	if(format instanceof RichOutputFormat){
		((RichOutputFormat<?>) format).setRuntimeContext(ctx);
	}
	format.open(0, 1);
	for (IN element : inputData) {
		format.writeRecord(element);
	}
	
	format.close();
	
	if(format instanceof FinalizeOnMaster) {
		((FinalizeOnMaster)format).finalizeGlobal(1);
	}
}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
@Override
public TypeComparator<Tuple> createComparator(int[] keyIdxs, boolean[] orders, int offset, ExecutionConfig config) {

	if(keyIdxs.length == 0) {
		throw new RuntimeException("Empty key indexes");
	}
	if(offset != 0) {
		throw new RuntimeException("Only 0 offset supported.");
	}

	// get key comparators
	TypeComparator<?>[] keyComps = new TypeComparator[keyIdxs.length];
	for(int i = 0; i < keyIdxs.length; i++) {
		keyComps[i] = ((AtomicType)this.getTypeAt(keyIdxs[i])).createComparator(orders[i], config);
	}

	if(length > 0) {
		// comparator for tuples with defined schema
		int maxKey = 0;
		for(int i = 0; i < keyIdxs.length; ++i) {
			int key = keyIdxs[i];
			maxKey = Math.max(maxKey, key);
		}

		// get field serializers up to max key
		TypeSerializer[] serializers = new TypeSerializer[maxKey + 1];
		for(int i = 0; i <= maxKey; ++i) {
			serializers[i] = this.fieldTypes.get(Integer.toString(i)).createSerializer(config);
		}

		return new DefinedTupleComparator(keyIdxs, keyComps, serializers, this.length);
	}
	else {
		// comparator for unknown tuples
		int[] cascadingKeyIdx = new int[keyIdxs.length];
		for(int i=0; i<cascadingKeyIdx.length; i++) {
			cascadingKeyIdx[i] = getCascadingPos(keyIdxs[i]);
		}

		return new UnknownTupleComparator(cascadingKeyIdx, keyComps, new FieldTypeInfo().createSerializer(config));
	}

}
 

import org.apache.flink.api.common.typeinfo.AtomicType; //导入依赖的package包/类
public ExpressionKeys(int[] groupingFields, TypeInformation<T> type, boolean allowEmpty) {
	if (!type.isTupleType()) {
		throw new InvalidProgramException("Specifying keys via field positions is only valid" +
				"for tuple data types. Type: " + type);
	}

	if (!allowEmpty && (groupingFields == null || groupingFields.length == 0)) {
		throw new IllegalArgumentException("The grouping fields must not be empty.");
	}
	// select all fields. Therefore, set all fields on this tuple level and let the logic handle the rest
	// (makes type assignment easier).
	if (groupingFields == null || groupingFields.length == 0) {
		groupingFields = new int[type.getArity()];
		for (int i = 0; i < groupingFields.length; i++) {
			groupingFields[i] = i;
		}
	} else {
		groupingFields = rangeCheckFields(groupingFields, type.getArity() -1);
	}
	TupleTypeInfoBase<?> tupleType = (TupleTypeInfoBase<?>)type;
	Preconditions.checkArgument(groupingFields.length > 0, "Grouping fields can not be empty at this point");
	
	keyFields = new ArrayList<FlatFieldDescriptor>(type.getTotalFields());
	// for each key, find the field:
	for(int j = 0; j < groupingFields.length; j++) {
		for(int i = 0; i < type.getArity(); i++) {
			TypeInformation<?> fieldType = tupleType.getTypeAt(i);
			
			if(groupingFields[j] == i) { // check if user set the key
				int keyId = countNestedElementsBefore(tupleType, i) + i;
				if(fieldType instanceof TupleTypeInfoBase) {
					TupleTypeInfoBase<?> tupleFieldType = (TupleTypeInfoBase<?>) fieldType;
					tupleFieldType.addAllFields(keyId, keyFields);
				} else {
					Preconditions.checkArgument(fieldType instanceof AtomicType, "Wrong field type");
					keyFields.add(new FlatFieldDescriptor(keyId, fieldType));
				}
				
			}
		}
	}
	keyFields = removeNullElementsFromList(keyFields);
}