src/main/java/org/apache/datasketches/quantiles/DoublesMergeImpl.java - datasketches-java - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 package org.apache.datasketches.quantiles;

 import static org.apache.datasketches.Util.checkIfPowerOf2;
 import static org.apache.datasketches.quantiles.PreambleUtil.EMPTY_FLAG_MASK;
 import static org.apache.datasketches.quantiles.PreambleUtil.FLAGS_BYTE;

 import org.apache.datasketches.SketchesArgumentException;
 import org.apache.datasketches.memory.WritableMemory;

 /**
  * Down-sampling and merge algorithms for doubles quantiles.
  *
  * @author Lee Rhodes
  * @author Kevin Lang
  */
 final class DoublesMergeImpl {

   private DoublesMergeImpl() {}

   /**
    * Merges the source sketch into the target sketch that can have a smaller value of K.
    * However, it is required that the ratio of the two K values be a power of 2.
    * I.e., source.getK() = target.getK() * 2^(nonnegative integer).
    * The source is not modified.
    *
    * <p>Note: It is easy to prove that the following simplified code which launches multiple waves of
    * carry propagation does exactly the same amount of merging work (including the work of
    * allocating fresh buffers) as the more complicated and seemingly more efficient approach that
    * tracks a single carry propagation wave through both sketches.
    *
    * <p>This simplified code probably does do slightly more "outer loop" work, but I am pretty
    * sure that even that is within a constant factor of the more complicated code, plus the
    * total amount of "outer loop" work is at least a factor of K smaller than the total amount of
    * merging work, which is identical in the two approaches.
    *
    * <p>Note: a two-way merge that doesn't modify either of its two inputs could be implemented
    * by making a deep copy of the larger sketch and then merging the smaller one into it.
    * However, it was decided not to do this.
    *
    * @param src The source sketch
    * @param tgt The target sketch
    */
   static void mergeInto(final DoublesSketch src, final UpdateDoublesSketch tgt) {
     final int srcK = src.getK();
     final int tgtK = tgt.getK();
     final long srcN = src.getN();
     final long tgtN = tgt.getN();

     if (srcK != tgtK) {
       downSamplingMergeInto(src, tgt);
       return;
     }
     //The remainder of this code is for the case where the k's are equal

     final DoublesSketchAccessor srcSketchBuf = DoublesSketchAccessor.wrap(src);
     final long nFinal = tgtN + srcN;

     for (int i = 0; i < srcSketchBuf.numItems(); i++) { // update only the base buffer
       tgt.update(srcSketchBuf.get(i));
     }

     final int spaceNeeded = DoublesUpdateImpl.getRequiredItemCapacity(tgtK, nFinal);
     final int tgtCombBufItemCap = tgt.getCombinedBufferItemCapacity();
     if (spaceNeeded > tgtCombBufItemCap) { //copies base buffer plus current levels
       tgt.growCombinedBuffer(tgtCombBufItemCap, spaceNeeded);
     }

     final DoublesArrayAccessor scratch2KAcc = DoublesArrayAccessor.initialize(2 * tgtK);

     long srcBitPattern = src.getBitPattern();
     assert srcBitPattern == (srcN / (2L * srcK));

     final DoublesSketchAccessor tgtSketchBuf = DoublesSketchAccessor.wrap(tgt, true);
     long newTgtBitPattern = tgt.getBitPattern();

     for (int srcLvl = 0; srcBitPattern != 0L; srcLvl++, srcBitPattern >>>= 1) {
       if ((srcBitPattern & 1L) > 0L) {
         newTgtBitPattern = DoublesUpdateImpl.inPlacePropagateCarry(
                 srcLvl,
                 srcSketchBuf.setLevel(srcLvl),
                 scratch2KAcc,
                 false,
                 tgtK,
                 tgtSketchBuf,
                 newTgtBitPattern
         );
       }
     }

     if (tgt.isDirect() && (nFinal > 0)) {
       final WritableMemory mem = tgt.getMemory();
       mem.clearBits(FLAGS_BYTE, (byte) EMPTY_FLAG_MASK);
     }

     tgt.putN(nFinal);
     tgt.putBitPattern(newTgtBitPattern); // no-op if direct

     assert (tgt.getN() / (2L * tgtK)) == tgt.getBitPattern(); // internal consistency check

     double srcMax = src.getMaxValue();
     srcMax = Double.isNaN(srcMax) ? Double.NEGATIVE_INFINITY : srcMax;
     double srcMin = src.getMinValue();
     srcMin = Double.isNaN(srcMin) ? Double.POSITIVE_INFINITY : srcMin;

     double tgtMax = tgt.getMaxValue();
     tgtMax = Double.isNaN(tgtMax) ? Double.NEGATIVE_INFINITY : tgtMax;
     double tgtMin = tgt.getMinValue();
     tgtMin = Double.isNaN(tgtMin) ? Double.POSITIVE_INFINITY : tgtMin;

     tgt.putMaxValue(Math.max(srcMax, tgtMax));
     tgt.putMinValue(Math.min(srcMin, tgtMin));
   }

   /**
    * Merges the source sketch into the target sketch that can have a smaller value of K.
    * However, it is required that the ratio of the two K values be a power of 2.
    * I.e., source.getK() = target.getK() * 2^(nonnegative integer).
    * The source is not modified.
    *
    * @param src The source sketch
    * @param tgt The target sketch
    */
   //also used by DoublesSketch, DoublesUnionImpl and HeapDoublesSketchTest
   static void downSamplingMergeInto(final DoublesSketch src, final UpdateDoublesSketch tgt) {
     final int srcK = src.getK();
     final int tgtK = tgt.getK();
     final long tgtN = tgt.getN();

     if ((srcK % tgtK) != 0) {
       throw new SketchesArgumentException(
           "source.getK() must equal target.getK() * 2^(nonnegative integer).");
     }

     final int downFactor = srcK / tgtK;
     checkIfPowerOf2(downFactor, "source.getK()/target.getK() ratio");
     final int lgDownFactor = Integer.numberOfTrailingZeros(downFactor);

     if (src.isEmpty()) { return; }

     final DoublesSketchAccessor srcSketchBuf = DoublesSketchAccessor.wrap(src);
     final long nFinal = tgtN + src.getN();

     for (int i = 0; i < srcSketchBuf.numItems(); i++) { // update only the base buffer
       tgt.update(srcSketchBuf.get(i));
     }

     final int spaceNeeded = DoublesUpdateImpl.getRequiredItemCapacity(tgtK, nFinal);
     final int curCombBufCap = tgt.getCombinedBufferItemCapacity();
     if (spaceNeeded > curCombBufCap) { //copies base buffer plus current levels
       tgt.growCombinedBuffer(curCombBufCap, spaceNeeded);
     }

     //working scratch buffers
     final DoublesArrayAccessor scratch2KAcc = DoublesArrayAccessor.initialize(2 * tgtK);
     final DoublesArrayAccessor downScratchKAcc = DoublesArrayAccessor.initialize(tgtK);

     final DoublesSketchAccessor tgtSketchBuf = DoublesSketchAccessor.wrap(tgt, true);

     long srcBitPattern = src.getBitPattern();
     long newTgtBitPattern = tgt.getBitPattern();
     for (int srcLvl = 0; srcBitPattern != 0L; srcLvl++, srcBitPattern >>>= 1) {
       if ((srcBitPattern & 1L) > 0L) {
         justZipWithStride(
                 srcSketchBuf.setLevel(srcLvl),
                 downScratchKAcc,
                 tgtK,
                 downFactor
         );
         newTgtBitPattern = DoublesUpdateImpl.inPlacePropagateCarry(
                 srcLvl + lgDownFactor,    //starting level
                 downScratchKAcc,       //optSrcKBuf,
                 scratch2KAcc,          //size2KBuf,
                 false,                    //do mergeInto version
                 tgtK,
                 tgtSketchBuf,
                 newTgtBitPattern
         );

         tgt.putBitPattern(newTgtBitPattern); //off-heap is a no-op
       }
     }
     if (tgt.isDirect() && (nFinal > 0)) {
       final WritableMemory mem = tgt.getMemory();
       mem.clearBits(FLAGS_BYTE, (byte) EMPTY_FLAG_MASK);
     }
     tgt.putN(nFinal);

     assert (tgt.getN() / (2L * tgtK)) == newTgtBitPattern; // internal consistency check

     double srcMax = src.getMaxValue();
     srcMax = Double.isNaN(srcMax) ? Double.NEGATIVE_INFINITY : srcMax;
     double srcMin = src.getMinValue();
     srcMin = Double.isNaN(srcMin) ? Double.POSITIVE_INFINITY : srcMin;

     double tgtMax = tgt.getMaxValue();
     tgtMax = Double.isNaN(tgtMax) ? Double.NEGATIVE_INFINITY : tgtMax;
     double tgtMin = tgt.getMinValue();
     tgtMin = Double.isNaN(tgtMin) ? Double.POSITIVE_INFINITY : tgtMin;

     if (srcMax > tgtMax) { tgt.putMaxValue(srcMax); }
     if (srcMin < tgtMin) { tgt.putMinValue(srcMin); }
   }

   private static void justZipWithStride(
           final DoublesBufferAccessor bufA, // input
           final DoublesBufferAccessor bufC, // output
           final int kC, // number of items that should be in the output
           final int stride) {
     final int randomOffset = DoublesSketch.rand.nextInt(stride);
     for (int a = randomOffset, c = 0; c < kC; a += stride, c++ ) {
       bufC.set(c, bufA.get(a));
     }
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	package org.apache.datasketches.quantiles;

	import static org.apache.datasketches.Util.checkIfPowerOf2;
	import static org.apache.datasketches.quantiles.PreambleUtil.EMPTY_FLAG_MASK;
	import static org.apache.datasketches.quantiles.PreambleUtil.FLAGS_BYTE;

	import org.apache.datasketches.SketchesArgumentException;
	import org.apache.datasketches.memory.WritableMemory;

	/**
	* Down-sampling and merge algorithms for doubles quantiles.
	*
	* @author Lee Rhodes
	* @author Kevin Lang
	*/
	final class DoublesMergeImpl {

	private DoublesMergeImpl() {}

	/**
	* Merges the source sketch into the target sketch that can have a smaller value of K.
	* However, it is required that the ratio of the two K values be a power of 2.
	* I.e., source.getK() = target.getK() * 2^(nonnegative integer).
	* The source is not modified.
	*
	* <p>Note: It is easy to prove that the following simplified code which launches multiple waves of
	* carry propagation does exactly the same amount of merging work (including the work of
	* allocating fresh buffers) as the more complicated and seemingly more efficient approach that
	* tracks a single carry propagation wave through both sketches.
	*
	* <p>This simplified code probably does do slightly more "outer loop" work, but I am pretty
	* sure that even that is within a constant factor of the more complicated code, plus the
	* total amount of "outer loop" work is at least a factor of K smaller than the total amount of
	* merging work, which is identical in the two approaches.
	*
	* <p>Note: a two-way merge that doesn't modify either of its two inputs could be implemented
	* by making a deep copy of the larger sketch and then merging the smaller one into it.
	* However, it was decided not to do this.
	*
	* @param src The source sketch
	* @param tgt The target sketch
	*/
	static void mergeInto(final DoublesSketch src, final UpdateDoublesSketch tgt) {
	final int srcK = src.getK();
	final int tgtK = tgt.getK();
	final long srcN = src.getN();
	final long tgtN = tgt.getN();

	if (srcK != tgtK) {
	downSamplingMergeInto(src, tgt);
	return;
	}
	//The remainder of this code is for the case where the k's are equal

	final DoublesSketchAccessor srcSketchBuf = DoublesSketchAccessor.wrap(src);
	final long nFinal = tgtN + srcN;

	for (int i = 0; i < srcSketchBuf.numItems(); i++) { // update only the base buffer
	tgt.update(srcSketchBuf.get(i));
	}

	final int spaceNeeded = DoublesUpdateImpl.getRequiredItemCapacity(tgtK, nFinal);
	final int tgtCombBufItemCap = tgt.getCombinedBufferItemCapacity();
	if (spaceNeeded > tgtCombBufItemCap) { //copies base buffer plus current levels
	tgt.growCombinedBuffer(tgtCombBufItemCap, spaceNeeded);
	}

	final DoublesArrayAccessor scratch2KAcc = DoublesArrayAccessor.initialize(2 * tgtK);

	long srcBitPattern = src.getBitPattern();
	assert srcBitPattern == (srcN / (2L * srcK));

	final DoublesSketchAccessor tgtSketchBuf = DoublesSketchAccessor.wrap(tgt, true);
	long newTgtBitPattern = tgt.getBitPattern();

	for (int srcLvl = 0; srcBitPattern != 0L; srcLvl++, srcBitPattern >>>= 1) {
	if ((srcBitPattern & 1L) > 0L) {
	newTgtBitPattern = DoublesUpdateImpl.inPlacePropagateCarry(
	srcLvl,
	srcSketchBuf.setLevel(srcLvl),
	scratch2KAcc,
	false,
	tgtK,
	tgtSketchBuf,
	newTgtBitPattern
	);
	}
	}

	if (tgt.isDirect() && (nFinal > 0)) {
	final WritableMemory mem = tgt.getMemory();
	mem.clearBits(FLAGS_BYTE, (byte) EMPTY_FLAG_MASK);
	}

	tgt.putN(nFinal);
	tgt.putBitPattern(newTgtBitPattern); // no-op if direct

	assert (tgt.getN() / (2L * tgtK)) == tgt.getBitPattern(); // internal consistency check

	double srcMax = src.getMaxValue();
	srcMax = Double.isNaN(srcMax) ? Double.NEGATIVE_INFINITY : srcMax;
	double srcMin = src.getMinValue();
	srcMin = Double.isNaN(srcMin) ? Double.POSITIVE_INFINITY : srcMin;

	double tgtMax = tgt.getMaxValue();
	tgtMax = Double.isNaN(tgtMax) ? Double.NEGATIVE_INFINITY : tgtMax;
	double tgtMin = tgt.getMinValue();
	tgtMin = Double.isNaN(tgtMin) ? Double.POSITIVE_INFINITY : tgtMin;

	tgt.putMaxValue(Math.max(srcMax, tgtMax));
	tgt.putMinValue(Math.min(srcMin, tgtMin));
	}

	/**
	* Merges the source sketch into the target sketch that can have a smaller value of K.
	* However, it is required that the ratio of the two K values be a power of 2.
	* I.e., source.getK() = target.getK() * 2^(nonnegative integer).
	* The source is not modified.
	*
	* @param src The source sketch
	* @param tgt The target sketch
	*/
	//also used by DoublesSketch, DoublesUnionImpl and HeapDoublesSketchTest
	static void downSamplingMergeInto(final DoublesSketch src, final UpdateDoublesSketch tgt) {
	final int srcK = src.getK();
	final int tgtK = tgt.getK();
	final long tgtN = tgt.getN();

	if ((srcK % tgtK) != 0) {
	throw new SketchesArgumentException(
	"source.getK() must equal target.getK() * 2^(nonnegative integer).");
	}

	final int downFactor = srcK / tgtK;
	checkIfPowerOf2(downFactor, "source.getK()/target.getK() ratio");
	final int lgDownFactor = Integer.numberOfTrailingZeros(downFactor);

	if (src.isEmpty()) { return; }

	final DoublesSketchAccessor srcSketchBuf = DoublesSketchAccessor.wrap(src);
	final long nFinal = tgtN + src.getN();

	for (int i = 0; i < srcSketchBuf.numItems(); i++) { // update only the base buffer
	tgt.update(srcSketchBuf.get(i));
	}

	final int spaceNeeded = DoublesUpdateImpl.getRequiredItemCapacity(tgtK, nFinal);
	final int curCombBufCap = tgt.getCombinedBufferItemCapacity();
	if (spaceNeeded > curCombBufCap) { //copies base buffer plus current levels
	tgt.growCombinedBuffer(curCombBufCap, spaceNeeded);
	}

	//working scratch buffers
	final DoublesArrayAccessor scratch2KAcc = DoublesArrayAccessor.initialize(2 * tgtK);
	final DoublesArrayAccessor downScratchKAcc = DoublesArrayAccessor.initialize(tgtK);

	final DoublesSketchAccessor tgtSketchBuf = DoublesSketchAccessor.wrap(tgt, true);

	long srcBitPattern = src.getBitPattern();
	long newTgtBitPattern = tgt.getBitPattern();
	for (int srcLvl = 0; srcBitPattern != 0L; srcLvl++, srcBitPattern >>>= 1) {
	if ((srcBitPattern & 1L) > 0L) {
	justZipWithStride(
	srcSketchBuf.setLevel(srcLvl),
	downScratchKAcc,
	tgtK,
	downFactor
	);
	newTgtBitPattern = DoublesUpdateImpl.inPlacePropagateCarry(
	srcLvl + lgDownFactor, //starting level
	downScratchKAcc, //optSrcKBuf,
	scratch2KAcc, //size2KBuf,
	false, //do mergeInto version
	tgtK,
	tgtSketchBuf,
	newTgtBitPattern
	);

	tgt.putBitPattern(newTgtBitPattern); //off-heap is a no-op
	}
	}
	if (tgt.isDirect() && (nFinal > 0)) {
	final WritableMemory mem = tgt.getMemory();
	mem.clearBits(FLAGS_BYTE, (byte) EMPTY_FLAG_MASK);
	}
	tgt.putN(nFinal);

	assert (tgt.getN() / (2L * tgtK)) == newTgtBitPattern; // internal consistency check

	double srcMax = src.getMaxValue();
	srcMax = Double.isNaN(srcMax) ? Double.NEGATIVE_INFINITY : srcMax;
	double srcMin = src.getMinValue();
	srcMin = Double.isNaN(srcMin) ? Double.POSITIVE_INFINITY : srcMin;

	double tgtMax = tgt.getMaxValue();
	tgtMax = Double.isNaN(tgtMax) ? Double.NEGATIVE_INFINITY : tgtMax;
	double tgtMin = tgt.getMinValue();
	tgtMin = Double.isNaN(tgtMin) ? Double.POSITIVE_INFINITY : tgtMin;

	if (srcMax > tgtMax) { tgt.putMaxValue(srcMax); }
	if (srcMin < tgtMin) { tgt.putMinValue(srcMin); }
	}

	private static void justZipWithStride(
	final DoublesBufferAccessor bufA, // input
	final DoublesBufferAccessor bufC, // output
	final int kC, // number of items that should be in the output
	final int stride) {
	final int randomOffset = DoublesSketch.rand.nextInt(stride);
	for (int a = randomOffset, c = 0; c < kC; a += stride, c++ ) {
	bufC.set(c, bufA.get(a));
	}
	}

	}