Moved RebalanceDataset correctly to water/main/java instead of /water/test/...

Added RebalanceDatasetTest testing basic rebalance dataset functionality (just rebalance prostate.hex (380 lines) into 300 chunks and check values of all rows.

Author: tomasnykodym

src/test/java/water/fvec/RebalanceDataSet.java src/main/java/water/fvec/RebalanceDataSet.java

@@ -22,11 +22,12 @@ public class RebalanceDataSet extends H2O.H2OCountedCompleter {
   Frame _out;
   final Key _jobKey;
 
-  public RebalanceDataSet(H2O.H2OCountedCompleter cmp, Job job, Frame srcFrame, Key dstKey, int nchunks){
+  public RebalanceDataSet(Frame srcFrame, Key dstKey, int nchunks) { this(srcFrame, dstKey,nchunks,null,null);}
+  public RebalanceDataSet(Frame srcFrame, Key dstKey, int nchunks, H2O.H2OCountedCompleter cmp, Key jobKey){
     super(cmp);
     _in = srcFrame;
     _nchunks = nchunks;
-    _jobKey = job == null?null:job.self();
+    _jobKey = jobKey;
     _okey = dstKey;
   }
 
@@ -84,17 +85,14 @@ public static class RebalanceTask extends MRTask2<RebalanceTask> {
       int rem = chk._len;
       while(rem > 0 && dst._len2 < chk._len){
         Chunk srcRaw = _srcVec.chunkForRow(chk._start+dst._len2);
-        if(srcRaw == null)System.out.println("missing chunk for row " + chk._start+dst._len2);
         NewChunk src = new NewChunk((srcRaw));
         src = srcRaw.inflate_impl(src);
         assert src._len2 == srcRaw._len;
         int srcFrom = (int)(chk._start+dst._len2 - src._start);
-        boolean sparse = false;
         // check if the result is sparse (not exact since we only take subset of src in general)
         if((src.sparse() && dst.sparse()) || (src._len + dst._len < NewChunk.MIN_SPARSE_RATIO*(src._len2 + dst._len2))){
           src.set_sparse(src._len);
           dst.set_sparse(dst._len);
-          sparse = true;
         }
         final int srcTo = srcFrom + rem;
         int off = srcFrom-1;

src/test/java/water/fvec/RebalanceDatasetTest.java

@@ -0,0 +1,41 @@
+package water.fvec;
+
+import org.junit.Test;
+import water.H2O;
+import water.Key;
+import water.TestUtil;
+import water.UKV;
+
+import java.io.File;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Created by tomasnykodym on 4/1/14.
+ */
+public class RebalanceDatasetTest extends TestUtil {
+  @Test public void testProstate(){
+    Key hex = Key.make("p.hex");
+    Key rebalancedKey = Key.make("rebalanced");
+    Key raw = NFSFileVec.make(new File("smalldata/logreg/prostate.csv"));
+    try{
+      ParseDataset2.parse(hex, new Key[]{raw});
+      Frame fr = UKV.get(hex);
+      RebalanceDataSet rb = new RebalanceDataSet(fr,rebalancedKey,300);
+      H2O.submitTask(rb);
+      rb.join();
+      Frame rebalanced = UKV.get(rebalancedKey);
+      assertEquals(rebalanced.numRows(),fr.numRows());
+      assertEquals(rebalanced.anyVec()._espc.length,301);
+      for(long l = 0; l < fr.numRows(); ++l)
+        for(int i = 0; i < fr.numCols(); ++i)
+          assertEquals(fr.vecs()[i].at(l),rebalanced.vecs()[i].at(l),1e-8);
+    } finally{
+      Frame fr = UKV.get(hex);
+      fr.delete();
+      fr = UKV.get(rebalancedKey);
+      fr.delete();
+    }
+    checkLeakedKeys();
+  }
+}