added test case (without fix) for a pact compiler bug with ALL_GROUP-reducer.

nephele/nephele-common/src/main/java/eu/stratosphere/nephele/fs/FileSystem.java

@@ -190,7 +190,7 @@ public abstract class FileSystem {
 		synchronized (SYNCHRONIZATION_OBJECT) {
 
 			if (uri.getScheme() == null) {
-				throw new IOException("FileSystem: Scheme is null");
+				throw new IOException("FileSystem: Scheme is null. file:// or hdfs:// are schemes.");
 			}
 
 			final FSKey key = new FSKey(uri.getScheme(), uri.getAuthority());

pact/pact-common/src/main/java/eu/stratosphere/pact/common/io/DelimitedInputFormat.java

@@ -72,7 +72,7 @@ public abstract class DelimitedInputFormat extends FileInputFormat {
 	 */
 	private static int MAX_SAMPLE_LEN;
 	
-	protected static final void loadGloablConfigParams() {
+	protected static final void loadGlobalConfigParams() {
 		int maxSamples = GlobalConfiguration.getInteger(PactConfigConstants.DELIMITED_FORMAT_MAX_LINE_SAMPLES_KEY,
 				PactConfigConstants.DEFAULT_DELIMITED_FORMAT_MAX_LINE_SAMPLES);
 		int minSamples = GlobalConfiguration.getInteger(PactConfigConstants.DELIMITED_FORMAT_MIN_LINE_SAMPLES_KEY,
@@ -111,7 +111,7 @@ public abstract class DelimitedInputFormat extends FileInputFormat {
 		MAX_SAMPLE_LEN = maxLen;
 	}
 	
-	static { loadGloablConfigParams(); }
+	static { loadGlobalConfigParams(); }
 	
 	// ------------------------------------- Config Keys ------------------------------------------
 	

pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/ReduceCompilerBugTest.java

+package eu.stratosphere.pact.compiler;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+import eu.stratosphere.pact.common.contract.FileDataSink;
+import eu.stratosphere.pact.common.contract.FileDataSource;
+import eu.stratosphere.pact.common.contract.ReduceContract;
+import eu.stratosphere.pact.common.plan.Plan;
+import eu.stratosphere.pact.compiler.plan.candidate.OptimizedPlan;
+import eu.stratosphere.pact.compiler.plantranslate.NepheleJobGraphGenerator;
+import eu.stratosphere.pact.compiler.util.DummyInputFormat;
+import eu.stratosphere.pact.compiler.util.DummyOutputFormat;
+import eu.stratosphere.pact.compiler.util.IdentityReduce;
+
+
+/**
+ * This test case has been created to validate a bug that occurred when
+ * the ReduceContract was used without a grouping key.
+ */
+public class ReduceCompilerBugTest extends CompilerTestBase  {
+
+	@Test
+	public void testReduce() {
+		// construct the plan
+		FileDataSource source = new FileDataSource(DummyInputFormat.class, IN_FILE, "Source");
+		ReduceContract reduce1 = ReduceContract.builder(IdentityReduce.class).name("Reduce1").input(source).build();
+		FileDataSink sink = new FileDataSink(DummyOutputFormat.class, OUT_FILE, "Sink");
+		sink.setInput(reduce1);
+		Plan plan = new Plan(sink, "Test Temp Task");
+		plan.setDefaultParallelism(DEFAULT_PARALLELISM);
+		
+		
+		try {
+			OptimizedPlan oPlan = compileNoStats(plan);
+			NepheleJobGraphGenerator jobGen = new NepheleJobGraphGenerator();
+			jobGen.compileJobGraph(oPlan);
+		} catch(CompilerException ce) {
+			ce.printStackTrace();
+			fail("The pact compiler is unable to compile this plan correctly");
+		}
+	}
+}