Merge pull request #682 from apache/Fix_getResultAndReset

Fix get result and reset

Author: leerho

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

@@ -62,6 +62,7 @@
 final class DirectUpdateDoublesSketch extends DirectUpdateDoublesSketchR {
   private MemorySegmentRequest mSegReq = null;
 
+  //**CONSTRUCTORS**
   private DirectUpdateDoublesSketch(final int k, final MemorySegment seg, final MemorySegmentRequest mSegReq) {
     super(k, seg); //Checks k
     this.mSegReq = mSegReq;
@@ -99,6 +100,11 @@ static DirectUpdateDoublesSketch newInstance(final int k, final MemorySegment ds
     return new DirectUpdateDoublesSketch(k, dstSeg, mSegReq);
   }
 
+  static HeapUpdateDoublesSketch heapify(final DirectUpdateDoublesSketch skIn) {
+    final MemorySegment segIn = skIn.getMemorySegment();
+    return HeapUpdateDoublesSketch.heapifyInstance(segIn);
+  }
+
   /**
    * Wrap this sketch around the given non-compact MemorySegment image of a DoublesSketch.
    *
@@ -130,6 +136,8 @@ static DirectUpdateDoublesSketch wrapInstance(final MemorySegment srcSeg, final
     return new DirectUpdateDoublesSketch(k, srcSeg, mSegReq);
   }
 
+  //**END CONSTRUCTORS**
+
   @Override
   public boolean isReadOnly() {
     return false;
@@ -207,6 +215,23 @@ public void reset() {
   }
 
   //Restricted overrides
+
+  @Override
+  UpdateDoublesSketch getSketchAndReset() {
+    final HeapUpdateDoublesSketch skCopy = heapify(this);
+    reset();
+    return skCopy;
+  }
+
+  @Override
+  double[] growCombinedBuffer(final int curCombBufItemCap, final int itemSpaceNeeded) {
+    seg_ = growCombinedSegBuffer(itemSpaceNeeded);
+    // copy out any data that was there
+    final double[] newCombBuf = new double[itemSpaceNeeded];
+    MemorySegment.copy(seg_, JAVA_DOUBLE_UNALIGNED, COMBINED_BUFFER, newCombBuf, 0, curCombBufItemCap);
+    return newCombBuf;
+  }
+
   //Puts
 
   @Override
@@ -242,15 +267,6 @@ void putBitPattern(final long bitPattern) {
     //intentionally a no-op, not kept on-heap, always derived.
   }
 
-  @Override
-  double[] growCombinedBuffer(final int curCombBufItemCap, final int itemSpaceNeeded) {
-    seg_ = growCombinedSegBuffer(itemSpaceNeeded);
-    // copy out any data that was there
-    final double[] newCombBuf = new double[itemSpaceNeeded];
-    MemorySegment.copy(seg_, JAVA_DOUBLE_UNALIGNED, COMBINED_BUFFER, newCombBuf, 0, curCombBufItemCap);
-    return newCombBuf;
-  }
-
   //Direct supporting methods
 
   private MemorySegment growCombinedSegBuffer(final int itemSpaceNeeded) {

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

@@ -130,12 +130,12 @@ public boolean isSameResource(final MemorySegment that) {
 
   @Override
   public void reset() {
-    throw new SketchesReadOnlyException("Call to reset() on read-only buffer");
+    throw new SketchesReadOnlyException("Call to reset() on read-only sketch");
   }
 
   @Override
   public void update(final double dataItem) {
-    throw new SketchesReadOnlyException("Call to update() on read-only buffer");
+    throw new SketchesReadOnlyException("Call to update() on read-only sketch");
   }
 
   //Restricted overrides
@@ -147,8 +147,10 @@ int getBaseBufferCount() {
   }
 
   @Override
-  int getCombinedBufferItemCapacity() {
-    return Math.max(0, (int)seg_.byteSize() - COMBINED_BUFFER) / 8;
+  long getBitPattern() {
+    final int k = getK();
+    final long n = getN();
+    return ClassicUtil.computeBitPattern(k, n);
   }
 
   @Override
@@ -159,58 +161,59 @@ int getCombinedBufferItemCapacity() {
     final int itemCap = ClassicUtil.computeCombinedBufferItemCapacity(k, n);
     final double[] combinedBuffer = new double[itemCap];
     MemorySegment.copy(seg_, JAVA_DOUBLE_UNALIGNED, COMBINED_BUFFER, combinedBuffer, 0, itemCap);
-
-
     return combinedBuffer;
   }
 
   @Override
-  long getBitPattern() {
-    final int k = getK();
-    final long n = getN();
-    return ClassicUtil.computeBitPattern(k, n);
+  int getCombinedBufferItemCapacity() {
+    return Math.max(0, (int)seg_.byteSize() - COMBINED_BUFFER) / 8;
   }
 
   @Override
   MemorySegment getMemorySegment() {
     return seg_;
   }
 
+  @Override
+  UpdateDoublesSketch getSketchAndReset() {
+    throw new SketchesReadOnlyException("Call to getSketchAndReset() on read-only sketch");
+  }
+
+  @Override
+  double[] growCombinedBuffer(final int curCombBufItemCap, final int itemSpaceNeeded) {
+    throw new SketchesReadOnlyException("Call to growCombinedBuffer() on read-only sketch");
+  }
+
   //Puts
 
   @Override
   void putMinItem(final double minQuantile) {
-    throw new SketchesReadOnlyException("Call to putMinQuantile() on read-only buffer");
+    throw new SketchesReadOnlyException("Call to putMinItem() on read-only sketch");
   }
 
   @Override
   void putMaxItem(final double maxQuantile) {
-    throw new SketchesReadOnlyException("Call to putMaxQuantile() on read-only buffer");
+    throw new SketchesReadOnlyException("Call to putMaxItem() on read-only sketch");
   }
 
   @Override
   void putN(final long n) {
-    throw new SketchesReadOnlyException("Call to putN() on read-only buffer");
+    throw new SketchesReadOnlyException("Call to putN() on read-only sketch");
   }
 
   @Override
   void putCombinedBuffer(final double[] combinedBuffer) {
-    throw new SketchesReadOnlyException("Call to putCombinedBuffer() on read-only buffer");
+    throw new SketchesReadOnlyException("Call to putCombinedBuffer() on read-only sketch");
   }
 
   @Override
   void putBaseBufferCount(final int baseBufferCount) {
-    throw new SketchesReadOnlyException("Call to putBaseBufferCount() on read-only buffer");
+    throw new SketchesReadOnlyException("Call to putBaseBufferCount() on read-only sketch");
   }
 
   @Override
   void putBitPattern(final long bitPattern) {
-    throw new SketchesReadOnlyException("Call to putBaseBufferCount() on read-only buffer");
-  }
-
-  @Override
-  double[] growCombinedBuffer(final int curCombBufItemCap, final int itemSpaceNeeded) {
-    throw new SketchesReadOnlyException("Call to growCombinedBuffer() on read-only buffer");
+    throw new SketchesReadOnlyException("Call to putBitPattern() on read-only sketch");
   }
 
   //Checks

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

@@ -143,7 +143,7 @@ public void update(final double quantile) {
   @Override
   public UpdateDoublesSketch getResultAndReset() {
     if (gadget_ == null) { return null; } //Intentionally return null here for speed.
-    final UpdateDoublesSketch ds = gadget_;
+    final UpdateDoublesSketch ds = gadget_.getSketchAndReset();
     gadget_ = null;
     return ds;
   }

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

@@ -96,7 +96,7 @@ final class HeapUpdateDoublesSketch extends UpdateDoublesSketch {
    */
   private double[] combinedBuffer_;
 
-  //**CONSTRUCTORS**********************************************************
+  //**CONSTRUCTORS**
   private HeapUpdateDoublesSketch(final int k) {
     super(k); //Checks k
   }
@@ -120,16 +120,34 @@ static HeapUpdateDoublesSketch newInstance(final int k) {
     return huds;
   }
 
+  /**
+   * Returns a copy of the given sketch
+   * @param skIn the given sketch
+   * @return a copy of the given sketch
+   */
+  static HeapUpdateDoublesSketch copy(final HeapUpdateDoublesSketch skIn) {
+    final HeapUpdateDoublesSketch skCopy = HeapUpdateDoublesSketch.newInstance(skIn.k_);
+    if (skIn.n_ > 0) {
+      skCopy.n_ = skIn.n_;
+      skCopy.combinedBuffer_ = skIn.combinedBuffer_.clone();
+      skCopy.baseBufferCount_ = skIn.baseBufferCount_;
+      skCopy.bitPattern_ = skIn.bitPattern_;
+      skCopy.minItem_ = skIn.minItem_;
+      skCopy.maxItem_ = skIn.maxItem_;
+    }
+    return skCopy;
+  }
+
   /**
    * Heapifies the given srcSeg, which must be a MemorySegment image of a DoublesSketch and may have data.
    *
-   * @param srcSeg a MemorySegment image of a sketch, which may be in compact or not compact form.
+   * @param srcSeg a MemorySegment image of a sketch, which may be in compact or updatable form.
    * @return a DoublesSketch on the Java heap.
    */
   static HeapUpdateDoublesSketch heapifyInstance(final MemorySegment srcSeg) {
     final long segCapBytes = srcSeg.byteSize();
-    if (segCapBytes < 8) {
-      throw new SketchesArgumentException("Source MemorySegment too small: " + segCapBytes + " < 8");
+    if ((segCapBytes < 8) || (segCapBytes > Integer.MAX_VALUE)) {
+      throw new SketchesArgumentException("Source MemorySegment byteSize must be >= 8 and <= Integer.MAX_VALUE: " + segCapBytes);
     }
 
     final int preLongs = extractPreLongs(srcSeg);
@@ -166,6 +184,8 @@ static HeapUpdateDoublesSketch heapifyInstance(final MemorySegment srcSeg) {
     return huds;
   }
 
+  //**END CONSTRUCTORS**
+
   @Override
   public double getMaxItem() {
     if (isEmpty()) { throw new IllegalArgumentException(QuantilesAPI.EMPTY_MSG); }
@@ -334,8 +354,8 @@ int getBaseBufferCount() {
   }
 
   @Override
-  int getCombinedBufferItemCapacity() {
-    return combinedBuffer_.length;
+  long getBitPattern() {
+    return bitPattern_;
   }
 
   @Override
@@ -344,30 +364,38 @@ int getCombinedBufferItemCapacity() {
   }
 
   @Override
-  long getBitPattern() {
-    return bitPattern_;
+  int getCombinedBufferItemCapacity() {
+    return combinedBuffer_.length;
   }
 
   @Override
   MemorySegment getMemorySegment() {
     return null;
   }
 
-  //Puts
-
   @Override
-  void putMinItem(final double minItem) {
-    minItem_ = minItem;
+  HeapUpdateDoublesSketch getSketchAndReset() {
+    final HeapUpdateDoublesSketch skCopy = HeapUpdateDoublesSketch.copy(this);
+    reset();
+    return skCopy;
+  }
+
+  @Override //the returned array is not always used
+  double[] growCombinedBuffer(final int currentSpace, final int spaceNeeded) {
+    combinedBuffer_ = Arrays.copyOf(combinedBuffer_, spaceNeeded);
+    return combinedBuffer_;
   }
 
+  //Puts
+
   @Override
-  void putMaxItem(final double maxItem) {
-    maxItem_ = maxItem;
+  void putBaseBufferCount(final int baseBufferCount) {
+    baseBufferCount_ = baseBufferCount;
   }
 
   @Override
-  void putN(final long n) {
-    n_ = n;
+  void putBitPattern(final long bitPattern) {
+    bitPattern_ = bitPattern;
   }
 
   @Override
@@ -376,19 +404,18 @@ void putCombinedBuffer(final double[] combinedBuffer) {
   }
 
   @Override
-  void putBaseBufferCount(final int baseBufferCount) {
-    baseBufferCount_ = baseBufferCount;
+  void putMaxItem(final double maxItem) {
+    maxItem_ = maxItem;
   }
 
   @Override
-  void putBitPattern(final long bitPattern) {
-    bitPattern_ = bitPattern;
+  void putMinItem(final double minItem) {
+    minItem_ = minItem;
   }
 
-  @Override //the returned array is not always used
-  double[] growCombinedBuffer(final int currentSpace, final int spaceNeeded) {
-    combinedBuffer_ = Arrays.copyOf(combinedBuffer_, spaceNeeded);
-    return combinedBuffer_;
+  @Override
+  void putN(final long n) {
+    n_ = n;
   }
 
   /**