class GenericDataSourceBase extends Operator<OUT> {

	protected List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
		@SuppressWarnings("unchecked")
		InputFormat<OUT, InputSplit> inputFormat = (InputFormat<OUT, InputSplit>) this.formatWrapper.getUserCodeObject();
		inputFormat.configure(this.parameters);

		if(inputFormat instanceof RichInputFormat){
			((RichInputFormat) inputFormat).setRuntimeContext(ctx);
		}

		List<OUT> result = new ArrayList<OUT>();
		
		// splits
		InputSplit[] splits = inputFormat.createInputSplits(1);
		TypeSerializer<OUT> serializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
		
		for (InputSplit split : splits) {
			inputFormat.open(split);
			
			while (!inputFormat.reachedEnd()) {
				OUT next = inputFormat.nextRecord(serializer.createInstance());
				if (next != null) {
					result.add(serializer.copy(next));
				}
			}
			
			inputFormat.close();
		}
		
		return result;
	}

	public  GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo, String name);
	public  GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo);
	public  GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo, String name);
	public  GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo);

}