ObjectMapper implemented

pact/pact-clients/src/main/java/eu/stratosphere/pact/client/nephele/api/Client.java

@@ -19,10 +19,10 @@ import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 
+import eu.stratosphere.nephele.client.AbstractJobResult.ReturnCode;
 import eu.stratosphere.nephele.client.JobClient;
 import eu.stratosphere.nephele.client.JobExecutionException;
 import eu.stratosphere.nephele.client.JobSubmissionResult;
-import eu.stratosphere.nephele.client.AbstractJobResult.ReturnCode;
 import eu.stratosphere.nephele.configuration.ConfigConstants;
 import eu.stratosphere.nephele.configuration.Configuration;
 import eu.stratosphere.nephele.configuration.GlobalConfiguration;

pact/pact-clients/src/main/java/eu/stratosphere/pact/client/web/GUIServletStub.java

@@ -15,16 +15,16 @@
 
 package eu.stratosphere.pact.client.web;
 
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import javax.servlet.ServletException;
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
 
 /**
  * @author Stephan Ewen (stephan.ewen@tu-berlin.de)

pact/pact-clients/src/main/java/eu/stratosphere/pact/client/web/JobSubmissionServlet.java

@@ -15,34 +15,34 @@
 
 package eu.stratosphere.pact.client.web;
 
-import java.io.File;
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.io.StringWriter;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-
-import javax.servlet.ServletException;
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import eu.stratosphere.nephele.configuration.Configuration;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import eu.stratosphere.nephele.configuration.Configuration;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
-import eu.stratosphere.pact.client.nephele.api.Client;
-import eu.stratosphere.pact.client.nephele.api.ErrorInPlanAssemblerException;
-import eu.stratosphere.pact.client.nephele.api.PactProgram;
-import eu.stratosphere.pact.client.nephele.api.ProgramInvocationException;
+import eu.stratosphere.pact.client.nephele.api.Client;
+import eu.stratosphere.pact.client.nephele.api.ErrorInPlanAssemblerException;
+import eu.stratosphere.pact.client.nephele.api.PactProgram;
+import eu.stratosphere.pact.client.nephele.api.ProgramInvocationException;
 import eu.stratosphere.pact.compiler.CompilerException;
-import eu.stratosphere.pact.compiler.jobgen.JSONGenerator;
-import eu.stratosphere.pact.compiler.plan.OptimizedPlan;
+import eu.stratosphere.pact.compiler.jobgen.JSONGenerator;
+import eu.stratosphere.pact.compiler.plan.OptimizedPlan;
 
 /**
  * @author Stephan Ewen (stephan.ewen@tu-berlin.com)

pact/pact-clients/src/main/java/eu/stratosphere/pact/client/web/JobsServlet.java

@@ -15,25 +15,25 @@
 
 package eu.stratosphere.pact.client.web;
 
-import java.io.File;
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.GregorianCalendar;
-import java.util.Iterator;
-import java.util.List;
-
-import javax.servlet.ServletConfig;
-import javax.servlet.ServletException;
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
-
-import org.apache.commons.fileupload.FileItem;
-import org.apache.commons.fileupload.FileUploadException;
-import org.apache.commons.fileupload.disk.DiskFileItemFactory;
-import org.apache.commons.fileupload.servlet.ServletFileUpload;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.GregorianCalendar;
+import java.util.Iterator;
+import java.util.List;
+
+import javax.servlet.ServletConfig;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.commons.fileupload.FileItem;
+import org.apache.commons.fileupload.FileUploadException;
+import org.apache.commons.fileupload.disk.DiskFileItemFactory;
+import org.apache.commons.fileupload.servlet.ServletFileUpload;
 
 /**
  * A servlet that accepts uploads of pact programs, returns a listing of the

pact/pact-clients/src/main/java/eu/stratosphere/pact/client/web/PactJobJSONServlet.java

@@ -15,18 +15,18 @@
 
 package eu.stratosphere.pact.client.web;
 
-import java.io.File;
-import java.io.IOException;
-import java.io.PrintWriter;
-
-import javax.servlet.ServletException;
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
-
-import eu.stratosphere.pact.client.nephele.api.PactProgram;
-import eu.stratosphere.pact.compiler.jobgen.JSONGenerator;
-import eu.stratosphere.pact.compiler.plan.OptimizedPlan;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintWriter;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import eu.stratosphere.pact.client.nephele.api.PactProgram;
+import eu.stratosphere.pact.compiler.jobgen.JSONGenerator;
+import eu.stratosphere.pact.compiler.plan.OptimizedPlan;
 
 /**
  * @author Stephan Ewen (stephan.ewen@tu-berlin.com)

pact/pact-clients/src/main/java/eu/stratosphere/pact/client/web/PlanDisplayServlet.java

@@ -15,9 +15,9 @@
 
 package eu.stratosphere.pact.client.web;
 
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.util.Map;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.Map;
 
 /**
  * @author Stephan Ewen (stephan.ewen@tu-berlin.com)

pact/pact-clients/src/main/java/eu/stratosphere/pact/testing/MockInstance.java

@@ -20,7 +20,6 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 
 import junit.framework.Assert;
-
 import eu.stratosphere.nephele.instance.AbstractInstance;
 import eu.stratosphere.nephele.instance.HardwareDescription;
 import eu.stratosphere.nephele.instance.HardwareDescriptionFactory;

pact/pact-clients/src/main/java/eu/stratosphere/pact/testing/MockTaskManager.java

@@ -20,7 +20,6 @@ import java.util.ArrayList;
 import java.util.List;
 
 import junit.framework.Assert;
-
 import eu.stratosphere.nephele.configuration.ConfigConstants;
 import eu.stratosphere.nephele.configuration.Configuration;
 import eu.stratosphere.nephele.configuration.GlobalConfiguration;

pact/pact-clients/src/test/java/eu/stratosphere/pact/client/nephele/api/ClientTest.java

@@ -14,6 +14,13 @@
  **********************************************************************************************************************/
 package eu.stratosphere.pact.client.nephele.api;
 
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.MockitoAnnotations.initMocks;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+
 import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -25,9 +32,9 @@ import org.mockito.Mock;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
+import eu.stratosphere.nephele.client.AbstractJobResult.ReturnCode;
 import eu.stratosphere.nephele.client.JobClient;
 import eu.stratosphere.nephele.client.JobSubmissionResult;
-import eu.stratosphere.nephele.client.AbstractJobResult.ReturnCode;
 import eu.stratosphere.nephele.configuration.ConfigConstants;
 import eu.stratosphere.nephele.configuration.Configuration;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
@@ -38,13 +45,6 @@ import eu.stratosphere.pact.compiler.costs.CostEstimator;
 import eu.stratosphere.pact.compiler.jobgen.JobGraphGenerator;
 import eu.stratosphere.pact.compiler.plan.OptimizedPlan;
 
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import static org.mockito.MockitoAnnotations.initMocks;
-import static org.mockito.Matchers.any;
-import static org.powermock.api.mockito.PowerMockito.whenNew;
-
 
 
 /**

pact/pact-clients/src/test/java/eu/stratosphere/pact/testing/AssertUtilTest.java

@@ -21,8 +21,6 @@ import java.util.List;
 
 import org.junit.Test;
 
-import eu.stratosphere.pact.testing.AssertUtil;
-
 /**
  * Tests {@link AssertUtil}.
  * 

pact/pact-clients/src/test/java/eu/stratosphere/pact/testing/TestPairsTest.java

@@ -20,14 +20,11 @@ import java.util.Arrays;
 import org.junit.Before;
 import org.junit.Test;
 
-
 import eu.stratosphere.pact.common.type.Key;
 import eu.stratosphere.pact.common.type.KeyValuePair;
 import eu.stratosphere.pact.common.type.Value;
 import eu.stratosphere.pact.common.type.base.PactInteger;
 import eu.stratosphere.pact.common.type.base.PactString;
-import eu.stratosphere.pact.testing.AssertUtil;
-import eu.stratosphere.pact.testing.TestPairs;
 
 /**
  * Tests {@link TestPairs}.

pact/pact-common/src/main/java/eu/stratosphere/pact/common/contract/GenericDataSink.java

@@ -15,10 +15,10 @@
 
 package eu.stratosphere.pact.common.contract;
 
+import eu.stratosphere.pact.common.io.OutputFormat;
 import eu.stratosphere.pact.common.plan.Visitor;
 import eu.stratosphere.pact.common.type.Key;
 import eu.stratosphere.pact.common.type.Value;
-import eu.stratosphere.pact.common.io.OutputFormat;
 
 /**
  * Contract for nodes which act as data sinks, storing the data they receive somewhere instead of sending it to another

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

@@ -18,9 +18,6 @@ package eu.stratosphere.pact.common.io;
 
 import java.io.IOException;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
 import eu.stratosphere.nephele.configuration.Configuration;
 import eu.stratosphere.nephele.fs.FSDataOutputStream;
 import eu.stratosphere.nephele.fs.FileSystem;

pact/pact-common/src/test/java/eu/stratosphere/pact/common/type/base/CollectionsDataTypeTest.java

@@ -29,13 +29,6 @@ import junit.framework.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import eu.stratosphere.pact.common.type.base.PactDouble;
-import eu.stratosphere.pact.common.type.base.PactInteger;
-import eu.stratosphere.pact.common.type.base.PactList;
-import eu.stratosphere.pact.common.type.base.PactMap;
-import eu.stratosphere.pact.common.type.base.PactPair;
-import eu.stratosphere.pact.common.type.base.PactString;
-
 public class CollectionsDataTypeTest {
 	private DataOutputStream out;
 

pact/pact-common/src/test/java/eu/stratosphere/pact/common/type/base/PrimitiveDataTypeTest.java

@@ -26,10 +26,6 @@ import junit.framework.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import eu.stratosphere.pact.common.type.base.PactDouble;
-import eu.stratosphere.pact.common.type.base.PactInteger;
-import eu.stratosphere.pact.common.type.base.PactString;
-
 public class PrimitiveDataTypeTest {
 
 	private DataOutputStream mOut;

pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/PactCompiler.java

@@ -58,11 +58,11 @@ import eu.stratosphere.pact.compiler.plan.MatchNode;
 import eu.stratosphere.pact.compiler.plan.OptimizedPlan;
 import eu.stratosphere.pact.compiler.plan.OptimizerNode;
 import eu.stratosphere.pact.compiler.plan.PactConnection;
+import eu.stratosphere.pact.compiler.plan.PactConnection.TempMode;
 import eu.stratosphere.pact.compiler.plan.ReduceNode;
 import eu.stratosphere.pact.compiler.plan.SingleInputNode;
 import eu.stratosphere.pact.compiler.plan.SinkJoiner;
 import eu.stratosphere.pact.compiler.plan.TwoInputNode;
-import eu.stratosphere.pact.compiler.plan.PactConnection.TempMode;
 import eu.stratosphere.pact.runtime.task.HistogramTask;
 import eu.stratosphere.pact.runtime.task.util.OutputEmitter.ShipStrategy;
 

pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/jobgen/JobGraphGenerator.java

@@ -26,13 +26,12 @@ import org.apache.commons.logging.LogFactory;
 import eu.stratosphere.nephele.configuration.Configuration;
 import eu.stratosphere.nephele.io.channels.ChannelType;
 import eu.stratosphere.nephele.io.compression.CompressionLevel;
+import eu.stratosphere.nephele.jobgraph.AbstractJobOutputVertex;
 import eu.stratosphere.nephele.jobgraph.AbstractJobVertex;
-import eu.stratosphere.nephele.jobgraph.JobInputVertex;
-import eu.stratosphere.nephele.jobgraph.JobOutputVertex;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.nephele.jobgraph.JobGraphDefinitionException;
 import eu.stratosphere.nephele.jobgraph.JobInputVertex;
-import eu.stratosphere.nephele.jobgraph.AbstractJobOutputVertex;
+import eu.stratosphere.nephele.jobgraph.JobOutputVertex;
 import eu.stratosphere.nephele.jobgraph.JobTaskVertex;
 import eu.stratosphere.pact.common.contract.GenericDataSink;
 import eu.stratosphere.pact.common.contract.GenericDataSource;
@@ -59,8 +58,8 @@ import eu.stratosphere.pact.runtime.task.ReduceTask;
 import eu.stratosphere.pact.runtime.task.SampleTask;
 import eu.stratosphere.pact.runtime.task.SelfMatchTask;
 import eu.stratosphere.pact.runtime.task.TempTask;
-import eu.stratosphere.pact.runtime.task.util.TaskConfig;
 import eu.stratosphere.pact.runtime.task.util.OutputEmitter.ShipStrategy;
+import eu.stratosphere.pact.runtime.task.util.TaskConfig;
 import eu.stratosphere.pact.runtime.task.util.TaskConfig.LocalStrategy;
 
 /**

pact/pact-compiler/src/main/java/eu/stratosphere/pact/contextcheck/ContextChecker.java

@@ -21,8 +21,8 @@ import java.util.Set;
 import eu.stratosphere.pact.common.contract.CoGroupContract;
 import eu.stratosphere.pact.common.contract.Contract;
 import eu.stratosphere.pact.common.contract.CrossContract;
-import eu.stratosphere.pact.common.contract.FileDataSinkContract;
 import eu.stratosphere.pact.common.contract.DualInputContract;
+import eu.stratosphere.pact.common.contract.FileDataSinkContract;
 import eu.stratosphere.pact.common.contract.MapContract;
 import eu.stratosphere.pact.common.contract.MatchContract;
 import eu.stratosphere.pact.common.contract.ReduceContract;

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

@@ -33,8 +33,8 @@ import eu.stratosphere.pact.common.contract.CrossContract;
 import eu.stratosphere.pact.common.contract.FileDataSinkContract;
 import eu.stratosphere.pact.common.contract.FileDataSourceContract;
 import eu.stratosphere.pact.common.contract.MapContract;
-import eu.stratosphere.pact.common.contract.ReduceContract;
 import eu.stratosphere.pact.common.contract.OutputContract.SameKeyFirst;
+import eu.stratosphere.pact.common.contract.ReduceContract;
 import eu.stratosphere.pact.common.io.TextInputFormat;
 import eu.stratosphere.pact.common.plan.Plan;
 import eu.stratosphere.pact.common.type.base.PactInteger;

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

@@ -32,8 +32,8 @@ import eu.stratosphere.nephele.instance.InstanceTypeFactory;
 import eu.stratosphere.pact.common.contract.FileDataSinkContract;
 import eu.stratosphere.pact.common.contract.FileDataSourceContract;
 import eu.stratosphere.pact.common.contract.MapContract;
-import eu.stratosphere.pact.common.contract.ReduceContract;
 import eu.stratosphere.pact.common.contract.OutputContract.UniqueKey;
+import eu.stratosphere.pact.common.contract.ReduceContract;
 import eu.stratosphere.pact.common.io.TextInputFormat;
 import eu.stratosphere.pact.common.plan.Plan;
 import eu.stratosphere.pact.common.plan.Visitor;

pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/CombiningUnilateralSortMerger.java

@@ -28,10 +28,10 @@ import org.apache.commons.logging.LogFactory;
 
 import eu.stratosphere.nephele.io.Reader;
 import eu.stratosphere.nephele.services.iomanager.Channel;
-import eu.stratosphere.nephele.services.iomanager.StreamChannelAccess;
 import eu.stratosphere.nephele.services.iomanager.ChannelWriter;
 import eu.stratosphere.nephele.services.iomanager.IOManager;
 import eu.stratosphere.nephele.services.iomanager.SerializationFactory;
+import eu.stratosphere.nephele.services.iomanager.StreamChannelAccess;
 import eu.stratosphere.nephele.services.iomanager.Writer;
 import eu.stratosphere.nephele.services.memorymanager.MemoryAllocationException;
 import eu.stratosphere.nephele.services.memorymanager.MemoryManager;

pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/LazyDelegatingIterator.java

@@ -15,10 +15,10 @@
 
 package eu.stratosphere.pact.runtime.sort;
 
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * An iterator that delegates requests to another iterator (the target

pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/UnilateralSortMerger.java

@@ -29,13 +29,13 @@ import org.apache.commons.logging.LogFactory;
 
 import eu.stratosphere.nephele.io.Reader;
 import eu.stratosphere.nephele.services.iomanager.Channel;
-import eu.stratosphere.nephele.services.iomanager.StreamChannelAccess;
 import eu.stratosphere.nephele.services.iomanager.ChannelReader;
 import eu.stratosphere.nephele.services.iomanager.ChannelWriter;
 import eu.stratosphere.nephele.services.iomanager.Deserializer;
 import eu.stratosphere.nephele.services.iomanager.IOManager;
 import eu.stratosphere.nephele.services.iomanager.RawComparator;
 import eu.stratosphere.nephele.services.iomanager.SerializationFactory;
+import eu.stratosphere.nephele.services.iomanager.StreamChannelAccess;
 import eu.stratosphere.nephele.services.memorymanager.MemoryAllocationException;
 import eu.stratosphere.nephele.services.memorymanager.MemoryManager;
 import eu.stratosphere.nephele.services.memorymanager.MemorySegment;

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/HashJoinTest.java

@@ -15,6 +15,9 @@
 
 package eu.stratosphere.pact.runtime.hash;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -37,8 +40,6 @@ import eu.stratosphere.pact.runtime.hash.HashJoin.HashBucketIterator;
 import eu.stratosphere.pact.runtime.test.util.DummyInvokable;
 import eu.stratosphere.pact.runtime.test.util.RegularlyGeneratedInputGenerator;
 import eu.stratosphere.pact.runtime.test.util.UnionIterator;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
 
 /**
  *

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/resettable/BlockResettableIteratorTest.java

@@ -20,6 +20,8 @@ import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 
+import junit.framework.Assert;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -31,9 +33,7 @@ import eu.stratosphere.nephele.services.memorymanager.MemoryManager;
 import eu.stratosphere.nephele.services.memorymanager.spi.DefaultMemoryManager;
 import eu.stratosphere.nephele.template.AbstractInvokable;
 import eu.stratosphere.pact.common.type.base.PactInteger;
-import eu.stratosphere.pact.runtime.resettable.BlockResettableIterator;
 import eu.stratosphere.pact.runtime.test.util.DummyInvokable;
-import junit.framework.Assert;
 
 public class BlockResettableIteratorTest
 {

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/resettable/SpillingResettableIteratorTest.java

@@ -20,6 +20,8 @@ import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 
+import junit.framework.Assert;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -32,9 +34,7 @@ import eu.stratosphere.nephele.services.memorymanager.MemoryManager;
 import eu.stratosphere.nephele.services.memorymanager.spi.DefaultMemoryManager;
 import eu.stratosphere.nephele.template.AbstractInvokable;
 import eu.stratosphere.pact.common.type.base.PactInteger;
-import eu.stratosphere.pact.runtime.resettable.SpillingResettableIterator;
 import eu.stratosphere.pact.runtime.test.util.DummyInvokable;
-import junit.framework.Assert;
 
 public class SpillingResettableIteratorTest {
 

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/CombiningUnilateralSortMergerITCase.java

@@ -41,10 +41,10 @@ import eu.stratosphere.pact.common.type.base.PactInteger;
 import eu.stratosphere.pact.runtime.serialization.WritableSerializationFactory;
 import eu.stratosphere.pact.runtime.test.util.DummyInvokable;
 import eu.stratosphere.pact.runtime.test.util.TestData;
-import eu.stratosphere.pact.runtime.test.util.TestData.Key;
-import eu.stratosphere.pact.runtime.test.util.TestData.Value;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator.KeyMode;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator.ValueMode;
+import eu.stratosphere.pact.runtime.test.util.TestData.Key;
+import eu.stratosphere.pact.runtime.test.util.TestData.Value;
 
 /**
  * @author Fabian Hueske

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/DeserializerComparatorTest.java

@@ -27,7 +27,6 @@ import eu.stratosphere.nephele.services.iomanager.RawComparator;
 import eu.stratosphere.nephele.services.iomanager.SerializationFactory;
 import eu.stratosphere.nephele.services.iomanager.Serializer;
 import eu.stratosphere.pact.runtime.serialization.WritableSerializationFactory;
-import eu.stratosphere.pact.runtime.sort.DeserializerComparator;
 import eu.stratosphere.pact.runtime.test.util.TestData;
 
 /**

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/SortMergeCoGroupIteratorITCase.java

@@ -43,9 +43,9 @@ import eu.stratosphere.pact.runtime.task.util.TaskConfig.LocalStrategy;
 import eu.stratosphere.pact.runtime.test.util.DummyInvokable;
 import eu.stratosphere.pact.runtime.test.util.TestData;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator;
-import eu.stratosphere.pact.runtime.test.util.TestData.RecordReaderMock;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator.KeyMode;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator.ValueMode;
+import eu.stratosphere.pact.runtime.test.util.TestData.RecordReaderMock;
 
 /**
  * @author Fabian Hueske (fabian.hueske@tu-berlin.de)

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/SortMergeMatchIteratorITCase.java

@@ -45,9 +45,9 @@ import eu.stratosphere.pact.runtime.test.util.DiscardingOutputCollector;
 import eu.stratosphere.pact.runtime.test.util.DummyInvokable;
 import eu.stratosphere.pact.runtime.test.util.TestData;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator;
-import eu.stratosphere.pact.runtime.test.util.TestData.RecordReaderMock;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator.KeyMode;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator.ValueMode;
+import eu.stratosphere.pact.runtime.test.util.TestData.RecordReaderMock;
 import eu.stratosphere.pact.runtime.test.util.UnionIterator;
 
 /**

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/task/util/HashVsSortTest.java

@@ -38,9 +38,9 @@ import eu.stratosphere.pact.runtime.test.util.DiscardingOutputCollector;
 import eu.stratosphere.pact.runtime.test.util.DummyInvokable;
 import eu.stratosphere.pact.runtime.test.util.TestData;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator;
-import eu.stratosphere.pact.runtime.test.util.TestData.RecordReaderMock;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator.KeyMode;
 import eu.stratosphere.pact.runtime.test.util.TestData.Generator.ValueMode;
+import eu.stratosphere.pact.runtime.test.util.TestData.RecordReaderMock;
 
 
 public class HashVsSortTest {

pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/TaskTestBase.java

@@ -41,8 +41,8 @@ import eu.stratosphere.pact.common.io.TextInputFormat;
 import eu.stratosphere.pact.common.stub.Stub;
 import eu.stratosphere.pact.common.type.KeyValuePair;
 import eu.stratosphere.pact.common.type.base.PactInteger;
-import eu.stratosphere.pact.runtime.task.util.TaskConfig;
 import eu.stratosphere.pact.runtime.task.util.OutputEmitter.ShipStrategy;
+import eu.stratosphere.pact.runtime.task.util.TaskConfig;
 
 @RunWith(PowerMockRunner.class)
 @PrepareForTest(LibraryCacheManager.class)

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

@@ -14,6 +14,7 @@ import eu.stratosphere.sopremo.expressions.PathExpression;
 import eu.stratosphere.sopremo.jsondatamodel.ArrayNode;
 import eu.stratosphere.sopremo.jsondatamodel.IntNode;
 import eu.stratosphere.sopremo.jsondatamodel.JsonNode;
+import eu.stratosphere.sopremo.jsondatamodel.ObjectMapper;
 import eu.stratosphere.sopremo.jsondatamodel.ObjectNode;
 import eu.stratosphere.sopremo.pact.PactJsonObject;
 
@@ -140,7 +141,7 @@ public class JsonUtil {
 	}
 
 	public static ArrayNode createArrayNode(final Object... constants) {
-		return JsonUtil.OBJECT_MAPPER.valueToTree(constants);
+		return (ArrayNode)JsonUtil.OBJECT_MAPPER.valueToTree(constants);
 	}
 
 	public static ArrayNode createCompactArray(final Object... constants) {

sopremo/sopremo-common/src/main/java/eu/stratosphere/sopremo/jsondatamodel/ArrayNode.java

@@ -28,10 +28,11 @@ public class ArrayNode extends JsonNode implements Iterable<JsonNode> {
 		return this.children.size();
 	}
 
-	public void add(JsonNode node) {
+	public ArrayNode add(JsonNode node) {
 		if (node == null)
 			node = NullNode.getInstance();
 		this._add(node);
+		return this;
 	}
 
 	public JsonNode get(final int index) {

sopremo/sopremo-common/src/main/java/eu/stratosphere/sopremo/jsondatamodel/ObjectMapper.java

+package eu.stratosphere.sopremo.jsondatamodel;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.HashMap;
+
+public class ObjectMapper {
+
+	private HashMap<Class<? extends Object>, Constructor<? extends JsonNode>> typeDict = new HashMap<Class<? extends Object>, Constructor<? extends JsonNode>>();
+
+	public ObjectMapper() {
+		try {
+			typeDict.put(Integer.class, IntNode.class.getConstructor(Integer.TYPE));
+			typeDict.put(Long.class, LongNode.class.getConstructor(Long.TYPE));
+			typeDict.put(BigInteger.class, BigIntegerNode.class.getConstructor(BigInteger.class));
+			typeDict.put(BigDecimal.class, DecimalNode.class.getConstructor(BigDecimal.class));
+			typeDict.put(Double.class, DoubleNode.class.getConstructor(Double.TYPE));
+			typeDict.put(String.class, TextNode.class.getConstructor(String.class));
+		} catch (SecurityException e) {
+			e.printStackTrace();
+		} catch (NoSuchMethodException e) {
+			e.printStackTrace();
+		}
+
+	}
+
+	public JsonNode valueToTree(Object value) {
+
+		if (value == null) {
+			return NullNode.getInstance();
+		}
+		Class<? extends Object> valueClass = value.getClass();
+		if (valueClass.isArray()) {
+			Object[] array = (Object[]) value;
+			ArrayNode arrayNode = new ArrayNode();
+			for (int i = 0; i < array.length; i++) {
+				arrayNode.add(valueToTree(array[i]));
+			}
+			return arrayNode;
+		}
+		try {
+			return typeDict.get(valueClass).newInstance(value);
+		} catch (IllegalArgumentException e) {
+			e.printStackTrace();
+		} catch (SecurityException e) {
+			e.printStackTrace();
+		} catch (InstantiationException e) {
+			e.printStackTrace();
+		} catch (IllegalAccessException e) {
+			e.printStackTrace();
+		} catch (InvocationTargetException e) {
+			e.printStackTrace();
+		}
+		return NullNode.getInstance();
+
+	}
+
+}

sopremo/sopremo-common/src/main/java/eu/stratosphere/sopremo/pact/JsonOutputFormat.java

@@ -21,6 +21,7 @@ import eu.stratosphere.nephele.configuration.Configuration;
 import eu.stratosphere.pact.common.io.FileOutputFormat;
 import eu.stratosphere.pact.common.type.KeyValuePair;
 import eu.stratosphere.pact.common.type.base.PactNull;
+import eu.stratosphere.sopremo.jsondatamodel.ObjectMapper;
 
 /**
  * Writes json files with Jackson. The incoming key/value pair consists of {@link PactNull} and a {@link PactJsonObject}

sopremo/sopremo-common/src/main/java/eu/stratosphere/sopremo/pact/PactJsonObject.java

@@ -38,7 +38,7 @@ public class PactJsonObject implements Value {
 	 * Initializes PactJsonObject with an empty {@link ObjectNode}.
 	 */
 	public PactJsonObject() {
-		this.value = JsonUtil.OBJECT_MAPPER.createObjectNode();
+		this.value = new ObjectNode();
 	}
 
 	/**

sopremo/sopremo-common/src/test/java/eu/stratosphere/sopremo/expressions/FunctionCallTest.java

@@ -33,7 +33,7 @@ public class FunctionCallTest extends EvaluableExpressionTest<FunctionCall> {
 	public void shouldCallFunction() {
 		final JsonNode result = new FunctionCall("sum", new ArrayAccess(0), new ArrayAccess(1)).evaluate(
 			createArrayNode(1, 2), this.context);
-		Assert.assertEquals(new IntNode(3), result);
+		Assert.assertEquals(new DoubleNode(3), result);
 	}
 
 	@Test

sopremo/sopremo-common/src/test/java/eu/stratosphere/sopremo/jsondatamodel/ObjectMapperTest.java

+package eu.stratosphere.sopremo.jsondatamodel;
+
+import junit.framework.Assert;
+
+import org.junit.Test;
+
+public class ObjectMapperTest {
+
+	@Test
+	public void shouldMapArray() {
+		Object[] array = { "field1", 1 };
+		JsonNode node = new ObjectMapper().valueToTree(array);
+		Assert.assertEquals(new ArrayNode().add(new TextNode("field1")).add(new IntNode(1)), node);
+	}
+
+	@Test
+	public void shouldMapNestedArray() {
+		Object[] root = new Object[2];
+		Object[] array1 = { "field1", 1 };
+		Object[] array2 = { "field2", 2 };
+		root[0] = array1;
+		root[1] = array2;
+		JsonNode node = new ObjectMapper().valueToTree(root);
+
+		Assert.assertEquals(
+			new ArrayNode().add(new ArrayNode().add(new TextNode("field1")).add(new IntNode(1)))
+				.add(new ArrayNode().add(new TextNode("field2")).add(new IntNode(2))),
+			node);
+	}
+}