Forward fitted changes to Sopremo Packages.

Author: sewen

sopremo/sopremo-base/src/main/java/eu/stratosphere/sopremo/base/Projection.java

@@ -6,7 +6,6 @@
 import eu.stratosphere.sopremo.InputCardinality;
 import eu.stratosphere.sopremo.Name;
 import eu.stratosphere.sopremo.OutputCardinality;
-import eu.stratosphere.sopremo.expressions.CachingExpression;
 import eu.stratosphere.sopremo.expressions.EvaluationExpression;
 import eu.stratosphere.sopremo.pact.JsonCollector;
 import eu.stratosphere.sopremo.pact.SopremoMap;

sopremo/sopremo-base/src/main/java/eu/stratosphere/sopremo/base/UnionAll.java

@@ -34,7 +34,7 @@ public PactModule asPactModule(EvaluationContext context) {
 		final List<JsonStream> inputs = this.getInputs();
 		final PactModule module = new PactModule(this.getName(), inputs.size(), 1);
 		// TODO: remove identity map, when Nephele can deal with direct source->sink connections
-		MapContract identityContract = new MapContract.Builder(IdentityMap.class).build();
+		MapContract identityContract = MapContract.builder(IdentityMap.class).build();
 		for (Contract input : module.getInputs())
 			identityContract.addInput(input);
 		module.getOutput(0).setInput(identityContract);

sopremo/sopremo-common/src/main/java/eu/stratosphere/sopremo/ElementaryOperator.java

@@ -224,7 +224,7 @@ public PactModule asPactModule(final EvaluationContext context) {
 		for (final List<Contract> inputs : inputLists) {
 			// assume at least one input for each contract input slot
 			if (inputs.isEmpty())
-				inputs.add(new MapContract.Builder(IdentityMap.class).build());
+				inputs.add(MapContract.builder(IdentityMap.class).build());
 			for (final Contract input : inputs)
 				if (!distinctInputs.contains(input))
 					distinctInputs.add(input);
@@ -318,27 +318,31 @@ protected Contract getContract(final Schema globalSchema) {
 		try {
 			if (contractClass == ReduceContract.class) {
 				int[] keyIndices = this.getKeyIndices(globalSchema, this.getKeyExpressions(0));
-				ReduceContract.Builder builder = new ReduceContract.Builder((Class<? extends ReduceStub>) stubClass);
+				ReduceContract.Builder builder = ReduceContract.builder((Class<? extends ReduceStub>) stubClass);
 				builder.name(this.toString());
 				PactBuilderUtil.addKeys(builder, this.getKeyClasses(globalSchema, keyIndices), keyIndices);
 				return builder.build();
 			}
 			else if (contractClass == CoGroupContract.class) {
 				int[] keyIndices1 = this.getKeyIndices(globalSchema, this.getKeyExpressions(0));
 				int[] keyIndices2 = this.getKeyIndices(globalSchema, this.getKeyExpressions(1));
-				CoGroupContract.Builder builder = new CoGroupContract.Builder((Class<? extends CoGroupStub>) stubClass);
+				Class<? extends Key>[] keyTypes = this.getCommonKeyClasses(globalSchema, keyIndices1, keyIndices2);
+				
+				CoGroupContract.Builder builder = CoGroupContract.builder((Class<? extends CoGroupStub>) stubClass,
+					keyTypes[0], keyIndices1[0], keyIndices2[0]);
 				builder.name(this.toString());
-				PactBuilderUtil.addKeys(builder, this.getCommonKeyClasses(globalSchema, keyIndices1, keyIndices2),
-						keyIndices1, keyIndices2);
+				PactBuilderUtil.addKeysExceptFirst(builder, keyTypes, keyIndices1, keyIndices2);
 				return builder.build();
 			}
 			else if (contractClass == MatchContract.class) {
 				int[] keyIndices1 = this.getKeyIndices(globalSchema, this.getKeyExpressions(0));
 				int[] keyIndices2 = this.getKeyIndices(globalSchema, this.getKeyExpressions(1));
-				MatchContract.Builder builder = new MatchContract.Builder((Class<? extends MatchStub>) stubClass);
+				Class<? extends Key>[] keyTypes = this.getCommonKeyClasses(globalSchema, keyIndices1, keyIndices2);
+				
+				MatchContract.Builder builder = MatchContract.builder((Class<? extends MatchStub>) stubClass,
+					keyTypes[0], keyIndices1[0], keyIndices2[0]);
 				builder.name(this.toString());
-				PactBuilderUtil.addKeys(builder, this.getCommonKeyClasses(globalSchema, keyIndices1, keyIndices2),
-						keyIndices1, keyIndices2);
+				PactBuilderUtil.addKeysExceptFirst(builder, keyTypes, keyIndices1, keyIndices2);
 				return builder.build();
 			}
 			return ReflectUtil.newInstance(contractClass, stubClass, this.toString());

sopremo/sopremo-common/src/main/java/eu/stratosphere/sopremo/PactBuilderUtil.java

@@ -26,7 +26,8 @@
  * 
  * @author Aljoscha Krettek
  */
-public class PactBuilderUtil {
+public class PactBuilderUtil
+{
 	public static void addKeys(ReduceContract.Builder builder, Class<? extends Key>[] keyClasses, int[] keyIndices) {
 		Preconditions.checkArgument(keyClasses.length == keyIndices.length,
 				"Lenght of keyClasses and keyIndices must match.");
@@ -35,21 +36,39 @@ public static void addKeys(ReduceContract.Builder builder, Class<? extends Key>[
 		}
 	}
 
-	public static void addKeys(CoGroupContract.Builder builder, Class<? extends Key>[] keyClasses, int[] keyIndices,
+	public static void addKeys(CoGroupContract.Builder builder, Class<? extends Key>[] keyClasses, int[] keyIndices1,
 			int[] keyIndices2) {
-		Preconditions.checkArgument(keyClasses.length == keyIndices.length,
+		Preconditions.checkArgument(keyClasses.length == keyIndices1.length && keyClasses.length == keyIndices2.length,
 				"Lenght of keyClasses and keyIndices must match.");
 		for (int i = 0; i < keyClasses.length; ++i) {
-			builder.keyField(keyClasses[i], keyIndices[i], keyIndices2[i]);
+			builder.keyField(keyClasses[i], keyIndices1[i], keyIndices2[i]);
+		}
+	}
+	
+	public static void addKeysExceptFirst(CoGroupContract.Builder builder, Class<? extends Key>[] keyClasses, int[] keyIndices1,
+			int[] keyIndices2) {
+		Preconditions.checkArgument(keyClasses.length == keyIndices1.length && keyClasses.length == keyIndices2.length,
+				"Lenght of keyClasses and keyIndices must match.");
+		for (int i = 1; i < keyClasses.length; ++i) {
+			builder.keyField(keyClasses[i], keyIndices1[i], keyIndices2[i]);
 		}
 	}
 
-	public static void addKeys(MatchContract.Builder builder, Class<? extends Key>[] keyClasses, int[] keyIndices,
+	public static void addKeys(MatchContract.Builder builder, Class<? extends Key>[] keyClasses, int[] keyIndices1,
 			int[] keyIndices2) {
-		Preconditions.checkArgument(keyClasses.length == keyIndices.length,
+		Preconditions.checkArgument(keyClasses.length == keyIndices1.length && keyClasses.length == keyIndices2.length,
 				"Lenght of keyClasses and keyIndices must match.");
 		for (int i = 0; i < keyClasses.length; ++i) {
-			builder.keyField(keyClasses[i], keyIndices[i], keyIndices2[i]);
+			builder.keyField(keyClasses[i], keyIndices1[i], keyIndices2[i]);
+		}
+	}
+
+	public static void addKeysExceptFirst(MatchContract.Builder builder, Class<? extends Key>[] keyClasses, int[] keyIndices1,
+			int[] keyIndices2) {
+		Preconditions.checkArgument(keyClasses.length == keyIndices1.length && keyClasses.length == keyIndices2.length,
+				"Lenght of keyClasses and keyIndices must match.");
+		for (int i = 1; i < keyClasses.length; ++i) {
+			builder.keyField(keyClasses[i], keyIndices1[i], keyIndices2[i]);
 		}
 	}
 }

sopremo/sopremo-common/src/main/java/eu/stratosphere/sopremo/io/JsonToCsv.java

@@ -99,7 +99,7 @@ public static FileDataSink convert(final String sourceFile, final JsonToCsv json
 		final PactModule sourceModule = new Source(sourceFile).asPactModule(jsonToCsv.getContext());
 		final Contract source = sourceModule.getOutput(0).getInputs().get(0);
 
-		final MapContract jsonToString = new MapContract.Builder(JsonToString.class).build();
+		final MapContract jsonToString = MapContract.builder(JsonToString.class).build();
 		jsonToString.getParameters().setString("separator", jsonToCsv.getSeparator());
 		SopremoUtil.serialize(jsonToString.getParameters(), SopremoUtil.CONTEXT, jsonToCsv.getContext());
 		SopremoUtil.serialize(jsonToString.getParameters(), "extractionExpressions",
@@ -176,7 +176,7 @@ public void open(final Configuration parameters) throws Exception {
 		 * eu.stratosphere.pact.common.stubs.Collector)
 		 */
 		@Override
-		public void map(final PactRecord record, final Collector out) throws Exception {
+		public void map(final PactRecord record, final Collector<PactRecord> out) throws Exception {
 			final StringBuilder string = new StringBuilder();
 
 			this.node = this.schema.recordToJson(record, this.node);

sopremo/sopremo-common/src/test/java/eu/stratosphere/sopremo/pact/CsvInputFormatTest.java

@@ -35,7 +35,7 @@ public void completeTestPassesWithExpectedValues() throws IOException {
 		final FileDataSource read = new FileDataSource(
 			CsvInputFormat.class, this.getResource("SopremoTestPlan/restaurant_short.csv"), "Input");
 
-		final MapContract map = new MapContract.Builder(IdentityMap.class).name("Map").build();
+		final MapContract map = MapContract.builder(IdentityMap.class).name("Map").build();
 		map.setInput(read);
 
 		final FileDataSink output = this.createOutput(map, SequentialOutputFormat.class);

sopremo/sopremo-common/src/test/java/eu/stratosphere/sopremo/pact/JsonInputFormatTest.java

@@ -54,7 +54,7 @@ public void completeTestPasses() throws IOException {
 		SopremoUtil.serialize(read.getParameters(), IOConstants.SCHEMA, SCHEMA);
 
 		final MapContract map =
-			new MapContract.Builder(IdentityMap.class).name("Map").build();
+			MapContract.builder(IdentityMap.class).name("Map").build();
 		map.setInput(read);
 
 		final FileDataSink output = this.createOutput(map, SequentialOutputFormat.class);
@@ -80,7 +80,7 @@ public void completeTestPassesWithExpectedValues() throws IOException {
 			JsonInputFormat.class, this.getResource("SopremoTestPlan/test.json"), "Input");
 		SopremoUtil.serialize(read.getParameters(), IOConstants.SCHEMA, SCHEMA);
 
-		final MapContract map = new MapContract.Builder(IdentityMap.class).name("Map").build();
+		final MapContract map = MapContract.builder(IdentityMap.class).name("Map").build();
 		map.setInput(read);
 
 		final FileDataSink output = this.createOutput(map,