import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
/**
 * Initialize the input function to map input elements to HTM encoder input.
 * @throws Exception
 */
protected void initInputFunction() throws Exception {

    // it is premature to call getInputNetwork, because no 'key' is available
    // when the operator is first opened.
    Network network = networkFactory.createNetwork(null);
    MultiEncoder encoder = network.getEncoder();

    if(encoder == null)
        throw new IllegalArgumentException("a network encoder must be provided");

    // handle the situation where an encoder parameter map was supplied rather than a fully-baked encoder.
    if(encoder.getEncoders(encoder) == null || encoder.getEncoders(encoder).size() < 1) {
        Map<String, Map<String, Object>> encoderParams =
                (Map<String, Map<String, Object>>) network.getParameters().get(Parameters.KEY.FIELD_ENCODING_MAP);
        if(encoderParams == null || encoderParams.size() < 1) {
            throw new IllegalStateException("No field encoding map found for MultiEncoder");
        }
        encoder.addMultipleEncoders(encoderParams);
    }

    // generate the encoder input function
    final GenerateEncoderInputFunction<IN> generator = new GenerateEncoderInputFunction<>((CompositeType<IN>) inputType, encoder, executionConfig);
    inputFunction = generator.generate();
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
/**
 * Creates a JSON serialization schema for the given fields and types.
 *
 * @param rowSchema The schema of the rows to encode.
 */
public JsonRowSerializationSchema(RowTypeInfo rowSchema) {

	Preconditions.checkNotNull(rowSchema);
	String[] fieldNames = rowSchema.getFieldNames();
	TypeInformation[] fieldTypes = rowSchema.getFieldTypes();

	// check that no field is composite
	for (int i = 0; i < fieldTypes.length; i++) {
		if (fieldTypes[i] instanceof CompositeType) {
			throw new IllegalArgumentException("JsonRowSerializationSchema cannot encode rows with nested schema, " +
				"but field '" + fieldNames[i] + "' is nested: " + fieldTypes[i].toString());
		}
	}

	this.fieldNames = fieldNames;
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
public static SingleInputSemanticProperties createProjectionPropertiesSingle(int[] fields, CompositeType<?> inType) {

		Character.isJavaIdentifierStart(1);

		SingleInputSemanticProperties ssp = new SingleInputSemanticProperties();

		int[] sourceOffsets = new int[inType.getArity()];
		sourceOffsets[0] = 0;
		for (int i = 1; i < inType.getArity(); i++) {
			sourceOffsets[i] = inType.getTypeAt(i - 1).getTotalFields() + sourceOffsets[i - 1];
		}

		int targetOffset = 0;
		for (int i = 0; i < fields.length; i++) {
			int sourceOffset = sourceOffsets[fields[i]];
			int numFieldsToCopy = inType.getTypeAt(fields[i]).getTotalFields();

			for (int j = 0; j < numFieldsToCopy; j++) {
				ssp.addForwardedField(sourceOffset + j, targetOffset + j);
			}
			targetOffset += numFieldsToCopy;
		}

		return ssp;
	}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
private static TypeInformation<?> getExpressionTypeInformation(String fieldStr, TypeInformation<?> typeInfo) {
	Matcher wildcardMatcher = PATTERN_WILDCARD.matcher(fieldStr);
	if (wildcardMatcher.matches()) {
		return typeInfo;
	} else {
		Matcher expMatcher = PATTERN_FIELD.matcher(fieldStr);
		if (!expMatcher.matches()) {
			throw new InvalidFieldReferenceException("Invalid field expression \"" + fieldStr + "\".");
		}
		if (typeInfo instanceof CompositeType<?>) {
			return ((CompositeType<?>) typeInfo).getTypeAt(expMatcher.group(1));
		} else {
			throw new InvalidFieldReferenceException("Nested field expression \"" + fieldStr + "\" not possible on atomic type (" + typeInfo + ").");
		}
	}
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
private static <T> String getSerializerTree(TypeInformation<T> ti, int indent) {
	String ret = "";
	if (ti instanceof CompositeType) {
		ret += StringUtils.repeat(' ', indent) + ti.getClass().getSimpleName() + "\n";
		CompositeType<T> cti = (CompositeType<T>) ti;
		String[] fieldNames = cti.getFieldNames();
		for (int i = 0; i < cti.getArity(); i++) {
			TypeInformation<?> fieldType = cti.getTypeAt(i);
			ret += StringUtils.repeat(' ', indent + 2) + fieldNames[i] + ":" + getSerializerTree(fieldType, indent);
		}
	} else {
		if (ti instanceof GenericTypeInfo) {
			ret += StringUtils.repeat(' ', indent) + "GenericTypeInfo (" + ti.getTypeClass().getSimpleName() + ")\n";
			ret += getGenericTypeTree(ti.getTypeClass(), indent + 4);
		} else {
			ret += StringUtils.repeat(' ', indent) + ti.toString() + "\n";
		}
	}
	return ret;
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
public static void recursivelyRegisterType(TypeInformation<?> typeInfo, ExecutionConfig config, Set<Class<?>> alreadySeen) {
	if (typeInfo instanceof GenericTypeInfo) {
		GenericTypeInfo<?> genericTypeInfo = (GenericTypeInfo<?>) typeInfo;
		Serializers.recursivelyRegisterType(genericTypeInfo.getTypeClass(), config, alreadySeen);
	}
	else if (typeInfo instanceof CompositeType) {
		List<GenericTypeInfo<?>> genericTypesInComposite = new ArrayList<>();
		getContainedGenericTypes((CompositeType<?>)typeInfo, genericTypesInComposite);
		for (GenericTypeInfo<?> gt : genericTypesInComposite) {
			Serializers.recursivelyRegisterType(gt.getTypeClass(), config, alreadySeen);
		}
	}
	else if (typeInfo instanceof ObjectArrayTypeInfo) {
		ObjectArrayTypeInfo<?, ?> objectArrayTypeInfo = (ObjectArrayTypeInfo<?, ?>) typeInfo;
		recursivelyRegisterType(objectArrayTypeInfo.getComponentInfo(), config, alreadySeen);
	}
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的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.typeutils.CompositeType; //导入依赖的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.typeutils.CompositeType; //导入依赖的package包/类
public SelectorFunctionKeys(KeySelector<T, K> keyExtractor, TypeInformation<T> inputType, TypeInformation<K> keyType) {

			if (keyExtractor == null) {
				throw new NullPointerException("Key extractor must not be null.");
			}
			if (keyType == null) {
				throw new NullPointerException("Key type must not be null.");
			}
			if (!keyType.isKeyType()) {
				throw new InvalidProgramException("Return type "+keyType+" of KeySelector "+keyExtractor.getClass()+" is not a valid key type");
			}

			this.keyExtractor = keyExtractor;
			this.inputType = inputType;
			this.keyType = keyType;

			this.originalKeyTypes = new TypeInformation[] {keyType};
			if (keyType instanceof CompositeType) {
				this.keyFields = ((CompositeType<T>)keyType).getFlatFields(ExpressionKeys.SELECT_ALL_CHAR);
			}
			else {
				this.keyFields = new ArrayList<>(1);
				this.keyFields.add(new FlatFieldDescriptor(0, keyType));
			}
		}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
public static boolean isSortKey(int fieldPos, TypeInformation<?> type) {

			if (!type.isTupleType() || !(type instanceof CompositeType)) {
				throw new InvalidProgramException("Specifying keys via field positions is only valid " +
					"for tuple data types. Type: " + type);
			}
			if (type.getArity() == 0) {
				throw new InvalidProgramException("Tuple size must be greater than 0. Size: " + type.getArity());
			}

			if(fieldPos < 0 || fieldPos >= type.getArity()) {
				throw new IndexOutOfBoundsException("Tuple position is out of range: " + fieldPos);
			}

			TypeInformation<?> sortKeyType = ((CompositeType<?>)type).getTypeAt(fieldPos);
			return sortKeyType.isSortKeyType();
		}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
private static FieldExpression decomposeFieldExpression(String fieldExpression) {
	Matcher matcher = PATTERN_NESTED_FIELDS_WILDCARD.matcher(fieldExpression);
	if (!matcher.matches()) {
		throw new CompositeType.InvalidFieldReferenceException("Invalid field expression \"" + fieldExpression + "\".");
	}

	String head = matcher.group(0);
	if (head.equals(Keys.ExpressionKeys.SELECT_ALL_CHAR) || head.equals(Keys.ExpressionKeys.SELECT_ALL_CHAR_SCALA)) {
		throw new CompositeType.InvalidFieldReferenceException("No wildcards are allowed here.");
	} else {
		head = matcher.group(1);
	}

	String tail = matcher.group(3);

	return new FieldExpression(head, tail);
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
RecursiveTupleFieldAccessor(int pos, FieldAccessor<R, F> innerAccessor, TypeInformation<T> typeInfo) {
	checkNotNull(typeInfo, "typeInfo must not be null.");
	checkNotNull(innerAccessor, "innerAccessor must not be null.");

	int arity = ((TupleTypeInfo) typeInfo).getArity();
	if (pos < 0 || pos >= arity) {
		throw new CompositeType.InvalidFieldReferenceException(
			"Tried to select " + ((Integer) pos).toString() + ". field on \"" +
				typeInfo.toString() + "\", which is an invalid index.");
	}

	if (pos < 0) {
		throw new CompositeType.InvalidFieldReferenceException("Tried to select " + ((Integer) pos).toString() + ". field.");
	}

	this.pos = pos;
	this.innerAccessor = innerAccessor;
	this.fieldType = innerAccessor.fieldType;
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
RecursiveProductFieldAccessor(int pos, TypeInformation<T> typeInfo, FieldAccessor<R, F> innerAccessor, ExecutionConfig config) {
	int arity = ((TupleTypeInfoBase) typeInfo).getArity();
	if (pos < 0 || pos >= arity) {
		throw new CompositeType.InvalidFieldReferenceException(
			"Tried to select " + ((Integer) pos).toString() + ". field on \"" +
				typeInfo.toString() + "\", which is an invalid index.");
	}
	checkNotNull(typeInfo, "typeInfo must not be null.");
	checkNotNull(innerAccessor, "innerAccessor must not be null.");

	this.pos = pos;
	this.fieldType = ((TupleTypeInfoBase<T>) typeInfo).getTypeAt(pos);
	this.serializer = (TupleSerializerBase<T>) typeInfo.createSerializer(config);
	this.length = this.serializer.getArity();
	this.fields = new Object[this.length];
	this.innerAccessor = innerAccessor;
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
public static <X> KeySelector<X, Tuple> getSelectorForKeys(Keys<X> keys, TypeInformation<X> typeInfo, ExecutionConfig executionConfig) {
	if (!(typeInfo instanceof CompositeType)) {
		throw new InvalidTypesException(
				"This key operation requires a composite type such as Tuples, POJOs, or Case Classes.");
	}

	CompositeType<X> compositeType = (CompositeType<X>) typeInfo;

	int[] logicalKeyPositions = keys.computeLogicalKeyPositions();
	int numKeyFields = logicalKeyPositions.length;

	TypeInformation<?>[] typeInfos = keys.getKeyFieldTypes();
	// use ascending order here, the code paths for that are usually a slight bit faster
	boolean[] orders = new boolean[numKeyFields];
	for (int i = 0; i < numKeyFields; i++) {
		orders[i] = true;
	}

	TypeComparator<X> comparator = compositeType.createComparator(logicalKeyPositions, orders, 0, executionConfig);
	return new ComparableKeySelector<>(comparator, numKeyFields, new TupleTypeInfo<>(typeInfos));
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
public static <X, K> KeySelector<X, K> getSelectorForOneKey(
		Keys<X> keys, Partitioner<K> partitioner, TypeInformation<X> typeInfo, ExecutionConfig executionConfig) {
	if (!(typeInfo instanceof CompositeType)) {
		throw new InvalidTypesException(
				"This key operation requires a composite type such as Tuples, POJOs, case classes, etc");
	}
	if (partitioner != null) {
		keys.validateCustomPartitioner(partitioner, null);
	}

	CompositeType<X> compositeType = (CompositeType<X>) typeInfo;
	int[] logicalKeyPositions = keys.computeLogicalKeyPositions();
	if (logicalKeyPositions.length != 1) {
		throw new IllegalArgumentException("There must be exactly 1 key specified");
	}

	TypeComparator<X> comparator = compositeType.createComparator(
			logicalKeyPositions, new boolean[] { true }, 0, executionConfig);
	return new OneKeySelector<>(comparator);
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的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.typeutils.CompositeType; //导入依赖的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.typeutils.CompositeType; //导入依赖的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.typeutils.CompositeType; //导入依赖的package包/类
/**
 * Create ExpressionKeys from String-expressions
 */
public ExpressionKeys(String[] expressionsIn, TypeInformation<T> type) {
	if(!(type instanceof CompositeType<?>)) {
		throw new IllegalArgumentException("Type "+type+" is not a composite type. "
				+ "Key expressions are only supported on POJO types and Tuples. "
				+ "A type is considered a POJO if all its fields are public, or have both getters and setters defined");
	}
	CompositeType<T> cType = (CompositeType<T>) type;
	
	String[] expressions = removeDuplicates(expressionsIn);
	if(expressionsIn.length != expressions.length) {
		LOG.warn("The key expressions contained duplicates. They are now unique");
	}
	// extract the keys on their flat position
	keyFields = new ArrayList<FlatFieldDescriptor>(expressions.length);
	for (int i = 0; i < expressions.length; i++) {
		List<FlatFieldDescriptor> keys = new ArrayList<FlatFieldDescriptor>(); // use separate list to do a size check
		cType.getKey(expressions[i], 0, keys);
		if(keys.size() == 0) {
			throw new IllegalArgumentException("Unable to extract key from expression '"+expressions[i]+"' on key "+cType);
		}
		keyFields.addAll(keys);
	}
}
 

import org.apache.flink.api.common.typeutils.CompositeType; //导入依赖的package包/类
public PartitionOperator(DataSet<T> input, PartitionMethod pMethod, Keys<T> pKeys) {
	super(input, input.getType());

	if(pMethod == PartitionMethod.HASH && pKeys == null) {
		throw new IllegalArgumentException("Hash Partitioning requires keys");
	} else if(pMethod == PartitionMethod.RANGE) {
		throw new UnsupportedOperationException("Range Partitioning not yet supported");
	}
	
	if(pKeys instanceof Keys.ExpressionKeys<?> && !(input.getType() instanceof CompositeType) ) {
		throw new IllegalArgumentException("Hash Partitioning with key fields only possible on Composite-type DataSets");
	}
	
	this.pMethod = pMethod;
	this.pKeys = pKeys;
}