class CsvReader {

	public <T extends Tuple> DataSource<T> tupleType(Class<T> targetType) {
		Preconditions.checkNotNull(targetType, "The target type class must not be null.");
		if (!Tuple.class.isAssignableFrom(targetType)) {
			throw new IllegalArgumentException("The target type must be a subclass of " + Tuple.class.getName());
		}
		
		@SuppressWarnings("unchecked")
		TupleTypeInfo<T> typeInfo = (TupleTypeInfo<T>) TypeExtractor.createTypeInfo(targetType);
		CsvInputFormat<T> inputFormat = new TupleCsvInputFormat<T>(path, this.lineDelimiter, this.fieldDelimiter, typeInfo, this.includedMask);
		
		Class<?>[] classes = new Class<?>[typeInfo.getArity()];
		for (int i = 0; i < typeInfo.getArity(); i++) {
			classes[i] = typeInfo.getTypeAt(i).getTypeClass();
		}
		
		configureInputFormat(inputFormat);
		return new DataSource<T>(executionContext, inputFormat, typeInfo, Utils.getCallLocationName());
	}

	public  CsvReader(Path filePath, ExecutionEnvironment executionContext);
	public  CsvReader(String filePath, ExecutionEnvironment executionContext);

}