Forward fit new config API to plans in pact-tests

Author: Aljoscha Krettek

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/contracts/CoGroupITCase.java

@@ -181,11 +181,13 @@ protected JobGraph getJobGraph() throws Exception {
 		String pathPrefix = getFilesystemProvider().getURIPrefix() + getFilesystemProvider().getTempDirPath();
 
 		FileDataSource input_left =  new FileDataSource(CoGroupTestInFormat.class, pathPrefix + "/cogroup_left");
-		input_left.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input_left)
+			.recordDelimiter('\n');
 		input_left.setDegreeOfParallelism(config.getInteger("CoGroupTest#NoSubtasks", 1));
 
 		FileDataSource input_right =  new FileDataSource(CoGroupTestInFormat.class, pathPrefix + "/cogroup_right");
-		input_right.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input_right)
+			.recordDelimiter('\n');
 		input_right.setDegreeOfParallelism(config.getInteger("CoGroupTest#NoSubtasks", 1));
 
 		CoGroupContract testCoGrouper = CoGroupContract.builder(TestCoGrouper.class, PactString.class, 0, 0)

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/contracts/CrossITCase.java

@@ -140,12 +140,14 @@ protected JobGraph getJobGraph() throws Exception {
 
 		FileDataSource input_left = new FileDataSource(
 				ContractITCaseInputFormat.class, pathPrefix + "/cross_left");
-		input_left.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input_left)
+			.recordDelimiter('\n');
 		input_left.setDegreeOfParallelism(config.getInteger("CrossTest#NoSubtasks", 1));
 
 		FileDataSource input_right = new FileDataSource(
 				ContractITCaseInputFormat.class, pathPrefix + "/cross_right");
-		input_right.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input_right)
+			.recordDelimiter('\n');
 		input_right.setDegreeOfParallelism(config.getInteger("CrossTest#NoSubtasks", 1));
 
 		CrossContract testCross = CrossContract.builder(TestCross.class).build();

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/contracts/MapITCase.java

@@ -109,7 +109,8 @@ protected JobGraph getJobGraph() throws Exception {
 
 		FileDataSource input = new FileDataSource(
 				ContractITCaseInputFormat.class, pathPrefix+"/mapInput");
-		input.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input)
+			.recordDelimiter('\n');
 		input.setDegreeOfParallelism(config.getInteger("MapTest#NoSubtasks", 1));
 
 		MapContract testMapper = MapContract.builder(TestMapper.class).build();

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/contracts/MatchITCase.java

@@ -135,12 +135,14 @@ protected JobGraph getJobGraph() throws Exception {
 
 		FileDataSource input_left = new FileDataSource(
 				ContractITCaseInputFormat.class, pathPrefix + "/match_left");
-		input_left.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input_left)
+			.recordDelimiter('\n');
 		input_left.setDegreeOfParallelism(config.getInteger("MatchTest#NoSubtasks", 1));
 
 		FileDataSource input_right = new FileDataSource(
 				ContractITCaseInputFormat.class, pathPrefix + "/match_right");
-		input_right.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input_right)
+			.recordDelimiter('\n');
 		input_right.setDegreeOfParallelism(config.getInteger("MatchTest#NoSubtasks", 1));
 
 		MatchContract testMatcher = MatchContract.builder(TestMatcher.class, PactString.class, 0, 0)

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/contracts/ReduceITCase.java

@@ -131,7 +131,8 @@ protected JobGraph getJobGraph() throws Exception {
 
 		FileDataSource input = new FileDataSource(
 				ContractITCaseInputFormat.class, pathPrefix + "/reduceInput");
-		input.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input)
+			.recordDelimiter('\n');
 		input.setDegreeOfParallelism(config.getInteger("ReduceTest#NoSubtasks", 1));
 
 		ReduceContract testReducer = new ReduceContract.Builder(TestReducer.class, PactString.class, 0)

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/contracts/UnionITCase.java

@@ -126,12 +126,14 @@ protected JobGraph getJobGraph() throws Exception {
 
 		FileDataSource input1 = new FileDataSource(
 			ContractITCaseInputFormat.class, pathPrefix + config.getString("UnionTest#Input1Path", ""));
-		input1.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input1)
+			.recordDelimiter('\n');
 		input1.setDegreeOfParallelism(config.getInteger("UnionTest#NoSubtasks", 1));
 
 		FileDataSource input2 = new FileDataSource(
 				ContractITCaseInputFormat.class, pathPrefix + config.getString("UnionTest#Input2Path", ""));
-		input2.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input2)
+			.recordDelimiter('\n');
 		input2.setDegreeOfParallelism(config.getInteger("UnionTest#NoSubtasks", 1));
 
 		MapContract testMapper = MapContract.builder(TestMapper.class).build();

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/failingPrograms/TaskFailureITCase.java

@@ -103,7 +103,8 @@ protected JobGraph getFailingJobGraph() throws Exception {
 		// init data source 
 		FileDataSource input = new FileDataSource(
 			ContractITCaseInputFormat.class, pathPrefix+"/mapInput");
-		input.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input)
+			.recordDelimiter('\n');
 		input.setDegreeOfParallelism(config.getInteger("MapTest#NoSubtasks", 1));
 
 		// init failing map task
@@ -143,7 +144,8 @@ protected JobGraph getJobGraph() throws Exception {
 		// init data source 
 		FileDataSource input = new FileDataSource(
 			ContractITCaseInputFormat.class, pathPrefix+"/mapInput");
-		input.setParameter(DelimitedInputFormat.RECORD_DELIMITER, "\n");
+		DelimitedInputFormat.configureDelimitedFormat(input)
+			.recordDelimiter('\n');
 		input.setDegreeOfParallelism(config.getInteger("MapTest#NoSubtasks", 1));
 
 		// init (working) map task

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/pactPrograms/GlobalSortingITCase.java

@@ -180,12 +180,11 @@ public Plan getPlan(String... args) throws IllegalArgumentException {
 			FileDataSink sink =
 				new FileDataSink(RecordOutputFormat.class, output);
 			sink.setDegreeOfParallelism(noSubtasks);
-			sink.getParameters().setString(RecordOutputFormat.RECORD_DELIMITER_PARAMETER, "\n");
-			sink.getParameters().setString(RecordOutputFormat.FIELD_DELIMITER_PARAMETER, "|");
-			sink.getParameters().setBoolean(RecordOutputFormat.LENIENT_PARSING, true);
-			sink.getParameters().setInteger(RecordOutputFormat.NUM_FIELDS_PARAMETER, 1);
-			sink.getParameters().setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 0, PactInteger.class);
-			sink.getParameters().setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 0, 0);
+			RecordOutputFormat.configureRecordFormat(sink)
+				.recordDelimiter('\n')
+				.fieldDelimiter('|')
+				.lenient(true)
+				.field(PactInteger.class, 0);
 
 			sink.setGlobalOrder(new Ordering(0, PactInteger.class, Order.ASCENDING), new UniformDistribution());
 			sink.setInput(source);

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/pactPrograms/GlobalSortingMixedOrderITCase.java

@@ -205,16 +205,13 @@ public Plan getPlan(String... args) throws IllegalArgumentException {
 				.field(DecimalTextIntParser.class, 2);
 
 			FileDataSink sink = new FileDataSink(RecordOutputFormat.class, output);
-			sink.getParameters().setString(RecordOutputFormat.RECORD_DELIMITER_PARAMETER, "\n");
-			sink.getParameters().setString(RecordOutputFormat.FIELD_DELIMITER_PARAMETER, ",");
-			sink.getParameters().setBoolean(RecordOutputFormat.LENIENT_PARSING, true);
-			sink.getParameters().setInteger(RecordOutputFormat.NUM_FIELDS_PARAMETER, 3);
-			sink.getParameters().setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 0, PactInteger.class);
-			sink.getParameters().setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 1, PactInteger.class);
-			sink.getParameters().setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 2, PactInteger.class);
-			sink.getParameters().setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 0, 0);
-			sink.getParameters().setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 1, 1);
-			sink.getParameters().setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 2, 2);
+			RecordOutputFormat.configureRecordFormat(sink)
+				.recordDelimiter('\n')
+				.fieldDelimiter(',')
+				.lenient(true)
+				.field(PactInteger.class, 0)
+				.field(PactInteger.class, 1)
+				.field(PactInteger.class, 2);
 
 			sink.setGlobalOrder(
 				new Ordering(0, PactInteger.class, Order.DESCENDING)

pact/pact-tests/src/test/java/eu/stratosphere/pact/test/testPrograms/mergeOnlyJoin/MergeOnlyJoin.java

@@ -80,17 +80,11 @@ public Plan getPlan(final String... args)
 		// create DataSourceContract for Orders input
 		FileDataSource input1 = new FileDataSource(RecordInputFormat.class, input1Path, "Input 1");
 		input1.setDegreeOfParallelism(noSubtasks);
-		
-		input1.setParameter(RecordInputFormat.RECORD_DELIMITER, "\n");
-		input1.setParameter(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|");
-		input1.setParameter(RecordInputFormat.NUM_FIELDS_PARAMETER, 2);
-
-		input1.getParameters().setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX+0, DecimalTextIntParser.class);
-		input1.setParameter(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX+0, 0);
-
-		input1.getParameters().setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX+1, DecimalTextIntParser.class);
-		input1.setParameter(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX+1, 1);
-		
+		RecordInputFormat.configureRecordFormat(input1)
+			.recordDelimiter('\n')
+			.fieldDelimiter('|')
+			.field(DecimalTextIntParser.class, 0)
+			.field(DecimalTextIntParser.class, 1);
 
 		ReduceContract aggInput1 = new ReduceContract.Builder(DummyReduce.class, PactInteger.class, 0)
 			.input(input1)
@@ -102,16 +96,11 @@ public Plan getPlan(final String... args)
 		// create DataSourceContract for Orders input
 		FileDataSource input2 = new FileDataSource(RecordInputFormat.class, input2Path, "Input 2");
 		input2.setDegreeOfParallelism(noSubtasksInput2);
-		
-		input2.setParameter(RecordInputFormat.RECORD_DELIMITER, "\n");
-		input2.setParameter(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|");
-		input2.setParameter(RecordInputFormat.NUM_FIELDS_PARAMETER, 2);
-		// order id
-		input2.getParameters().setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX+0, DecimalTextIntParser.class);
-		input2.setParameter(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX+0, 0);
-		// ship prio
-		input2.getParameters().setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX+1, DecimalTextIntParser.class);
-		input2.setParameter(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX+1, 1);
+		RecordInputFormat.configureRecordFormat(input2)
+			.recordDelimiter('\n')
+			.fieldDelimiter('|')
+			.field(DecimalTextIntParser.class, 0)
+			.field(DecimalTextIntParser.class, 1);
 
 		ReduceContract aggInput2 = new ReduceContract.Builder(DummyReduce.class, PactInteger.class, 0)
 			.input(input2)
@@ -132,16 +121,13 @@ public Plan getPlan(final String... args)
 		// create DataSinkContract for writing the result
 		FileDataSink result = new FileDataSink(RecordOutputFormat.class, output, joinLiO, "Output");
 		result.setDegreeOfParallelism(noSubtasks);
-		result.getParameters().setString(RecordOutputFormat.RECORD_DELIMITER_PARAMETER, "\n");
-		result.getParameters().setString(RecordOutputFormat.FIELD_DELIMITER_PARAMETER, "|");
-		result.getParameters().setBoolean(RecordOutputFormat.LENIENT_PARSING, true);
-		result.getParameters().setInteger(RecordOutputFormat.NUM_FIELDS_PARAMETER, 3);
-		result.getParameters().setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 0, PactInteger.class);
-		result.getParameters().setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 0, 0);
-		result.getParameters().setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 1, PactInteger.class);
-		result.getParameters().setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 1, 1);
-		result.getParameters().setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 2, PactInteger.class);
-		result.getParameters().setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 2, 2);
+		RecordOutputFormat.configureRecordFormat(result)
+			.recordDelimiter('\n')
+			.fieldDelimiter('|')
+			.lenient(true)
+			.field(PactInteger.class, 0)
+			.field(PactInteger.class, 1)
+			.field(PactInteger.class, 2);
 
 		// assemble the PACT plan
 		Plan plan = new Plan(result, "Merge Only Join");