Changes requested from reviews of PR#576

src/main/java/org/apache/datasketches/quantiles/DoublesByteArrayImpl.java

@@ -58,7 +58,7 @@ static byte[] toByteArray(final DoublesSketch sketch, final boolean ordered, fin
         | (ordered ? ORDERED_FLAG_MASK : 0)
         | (compact ? (COMPACT_FLAG_MASK | READ_ONLY_FLAG_MASK) : 0);
 
-    if (empty && !sketch.hasMemory()) { //empty & has Memory
+    if (empty && !sketch.hasMemory()) { //empty & !has Memory
       final byte[] outByteArr = new byte[Long.BYTES];
       final WritableMemory memOut = WritableMemory.writableWrap(outByteArr);
       final int preLongs = 1;
@@ -79,31 +79,31 @@ static byte[] toByteArray(final DoublesSketch sketch, final boolean ordered, fin
    */
   private static byte[] convertToByteArray(final DoublesSketch sketch, final int flags,
                                            final boolean ordered, final boolean compact) {
-    final int preLongs = 2;
-    final int extra = 2; // extra space for min and max quantiles
-    final int prePlusExtraBytes = (preLongs + extra) << 3;
-    final int k = sketch.getK();
-    final long n = sketch.getN();
-
-    // If not-compact, have accessor always report full levels. Then use level size to determine
-    // whether to copy data out.
-    final DoublesSketchAccessor dsa = DoublesSketchAccessor.wrap(sketch, !compact);
+    final int preLongs = sketch.isEmpty() ? 1 : 2;
 
     final int outBytes = (compact ? sketch.getCurrentCompactSerializedSizeBytes()
         : sketch.getCurrentUpdatableSerializedSizeBytes());
 
     final byte[] outByteArr = new byte[outBytes];
     final WritableMemory memOut = WritableMemory.writableWrap(outByteArr);
 
-    //insert preamble-0, N, min, max
+    //insert pre0
+    final int k = sketch.getK();
     insertPre0(memOut, preLongs, flags, k);
     if (sketch.isEmpty()) { return outByteArr; }
 
+    //insert N, min, max
+    final long n = sketch.getN();
     insertN(memOut, n);
     insertMinDouble(memOut, sketch.isEmpty() ? Double.NaN : sketch.getMinItem());
     insertMaxDouble(memOut, sketch.isEmpty() ? Double.NaN : sketch.getMaxItem());
 
-    long memOffsetBytes = prePlusExtraBytes;
+    // If not-compact, have accessor always report full levels. Then use level size to determine
+    // whether to copy data out.
+    final DoublesSketchAccessor dsa = DoublesSketchAccessor.wrap(sketch, !compact);
+
+    final int minAndMax = 2; // extra space for min and max quantiles
+    long memOffsetBytes = (preLongs + minAndMax) << 3;
 
     // might need to sort base buffer but don't want to change input sketch
     final int bbCnt = computeBaseBufferItems(k, n);

src/test/java/org/apache/datasketches/quantiles/DoublesSketchTest.java

@@ -162,13 +162,16 @@ public void directSketchShouldMoveOntoHeapEventually2() {
         break;
       }
     }
+    assertFalse(wmem.isAlive());
   }
 
   @Test
   public void checkEmptyDirect() {
     WritableMemory wmem = WritableMemory.allocateDirect(1000);
     UpdateDoublesSketch sketch = DoublesSketch.builder().build(wmem);
-    sketch.toByteArray(); //exercises a specific path
+    byte[] bytes = sketch.toByteArray(); //exercises a specific path
+    byte[] result = {1,3,8,4,-128,0,0,0};
+    assertEquals(bytes, result);
     wmem.close();
   }
 

src/test/java/org/apache/datasketches/theta/DirectQuickSelectSketchTest.java

@@ -74,26 +74,18 @@ public void checkBadSerVer() {
     }
   }
 
-  @Test//(expectedExceptions = SketchesArgumentException.class)
+  @Test(expectedExceptions = SketchesArgumentException.class)
   public void checkConstructorKtooSmall() {
     int k = 8;
-    try (WritableMemory mem = makeNativeMemory(k)) {
-      UpdateSketch.builder().setNominalEntries(k).build(mem);
-    } catch (final Exception e) {
-      if (e instanceof SketchesArgumentException) {}
-      else { throw new RuntimeException(e); }
-    }
+    WritableMemory mem = makeNativeMemory(k);
+    UpdateSketch.builder().setNominalEntries(k).build(mem);
   }
 
-  @Test//(expectedExceptions = SketchesArgumentException.class)
+  @Test(expectedExceptions = SketchesArgumentException.class)
   public void checkConstructorMemTooSmall() {
     int k = 16;
-    try (WritableMemory mem = makeNativeMemory(k/2)) {
-      UpdateSketch.builder().setNominalEntries(k).build(mem);
-    } catch (final Exception e) {
-      if (e instanceof SketchesArgumentException) {}
-      else { throw new RuntimeException(e); }
-    }
+    WritableMemory mem = makeNativeMemory(k/2);
+    UpdateSketch.builder().setNominalEntries(k).build(mem);
   }
 
   @Test(expectedExceptions = SketchesArgumentException.class)

src/test/java/org/apache/datasketches/theta/UnionImplTest.java

@@ -187,14 +187,14 @@ public void checkVer2EmptyHandling() {
   }
 
   @Test
-  public void checkMoveAndResize() {
+  public void checkMoveAndResizeOffHeap() {
     final int k = 1 << 12;
     final int u = 2 * k;
     final int bytes = Sketches.getMaxUpdateSketchBytes(k);
-    WritableMemory wmem = WritableMemory.allocateDirect(bytes / 2);
+    WritableMemory wmem = WritableMemory.allocateDirect(bytes / 2); //too small, forces new allocation on heap
     WritableMemory wmem2 = WritableMemory.allocateDirect(bytes / 2);
     final UpdateSketch sketch = Sketches.updateSketchBuilder().setNominalEntries(k).build(wmem);
-    assertTrue(sketch.isSameResource(wmem));
+    assertTrue(sketch.isSameResource(wmem)); //also testing the isSameResource function
 
     final Union union = SetOperation.builder().buildUnion(wmem2);
     assertTrue(union.isSameResource(wmem2));
@@ -205,6 +205,7 @@ public void checkMoveAndResize() {
     final Union union2 = SetOperation.builder().buildUnion(); //on-heap union
     assertFalse(union2.isSameResource(wmem2));  //obviously not
     wmem.close();
+    //note wmem2 has already been closed by the DefaultMemoryRequestServer
   }
 
   @Test