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>(); 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; } GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo, String name);  GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo);  GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo, String name);  GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo); UserCodeWrapper<? extends T> getFormatWrapper(); @Override UserCodeWrapper<? extends T> getUserCodeWrapper(); String getStatisticsKey(); void setStatisticsKey(String statisticsKey); void setSplitDataProperties(SplitDataProperties<OUT> splitDataProperties); SplitDataProperties<OUT> getSplitDataProperties(); @Override void accept(Visitor<Operator<?>> visitor); String toString();  }
@Test public void testDataSourcePlain() { try { TestNonRichInputFormat in = new TestNonRichInputFormat(); GenericDataSourceBase<String, TestNonRichInputFormat> source = new GenericDataSourceBase<String, TestNonRichInputFormat>( in, new OperatorInformation<String>(BasicTypeInfo.STRING_TYPE_INFO), "testSource"); ExecutionConfig executionConfig = new ExecutionConfig(); executionConfig.disableObjectReuse(); List<String> resultMutableSafe = source.executeOnCollections(null, executionConfig); in.reset(); executionConfig.enableObjectReuse(); List<String> resultRegular = source.executeOnCollections(null, executionConfig); assertEquals(asList(TestIOData.NAMES), resultMutableSafe); assertEquals(asList(TestIOData.NAMES), resultRegular); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
@Test public void testDataSourceWithRuntimeContext() { try { TestRichInputFormat in = new TestRichInputFormat(); GenericDataSourceBase<String, TestRichInputFormat> source = new GenericDataSourceBase<String, TestRichInputFormat>( in, new OperatorInformation<String>(BasicTypeInfo.STRING_TYPE_INFO), "testSource"); final HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<String, Accumulator<?, ?>>(); final HashMap<String, Future<Path>> cpTasks = new HashMap<>(); final TaskInfo taskInfo = new TaskInfo("test_source", 0, 1, 0); ExecutionConfig executionConfig = new ExecutionConfig(); executionConfig.disableObjectReuse(); List<String> resultMutableSafe = source.executeOnCollections(new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap), executionConfig); in.reset(); executionConfig.enableObjectReuse(); List<String> resultRegular = source.executeOnCollections(new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap), executionConfig); assertEquals(asList(TestIOData.RICH_NAMES), resultMutableSafe); assertEquals(asList(TestIOData.RICH_NAMES), resultRegular); } catch(Exception e){ e.printStackTrace(); fail(e.getMessage()); } }