src/main/java/org/apache/datasketches/sampling/EbppsItemsSketch.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.sampling;

 import static org.apache.datasketches.common.Util.LS;
 import static org.apache.datasketches.sampling.PreambleUtil.EBPPS_SER_VER;
 import static org.apache.datasketches.sampling.PreambleUtil.EMPTY_FLAG_MASK;
 import static org.apache.datasketches.sampling.PreambleUtil.HAS_PARTIAL_ITEM_MASK;

 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;

 import org.apache.datasketches.common.ArrayOfItemsSerDe;
 import org.apache.datasketches.common.Family;
 import org.apache.datasketches.common.SketchesArgumentException;
 import org.apache.datasketches.memory.Memory;
 import org.apache.datasketches.memory.WritableMemory;

 /**
  * An implementation of an Exact and Bounded Sampling Proportional to Size sketch.
  *
  * <p>From: "Exact PPS Sampling with Bounded Sample Size",
  * B. Hentschel, P. J. Haas, Y. Tian. Information Processing Letters, 2023.
  *
  * <p>This sketch samples data from a stream of items proportional to the weight of each item.
  * The sample guarantees the presence of an item in the result is proportional to that item's
  * portion of the total weight seen by the sketch, and returns a sample no larger than size k.
  *
  * <p>The sample may be smaller than k and the resulting size of the sample potentially includes
  * a probabilistic component, meaning the resulting sample size is not always constant.
  * @param <T> the item class type
  * @author Jon Malkin
  */
 public final class EbppsItemsSketch<T> {
   private static final int MAX_K = Integer.MAX_VALUE - 2;
   private static final int EBPPS_C_DOUBLE        = 40; // part of sample state, not preamble
   private static final int EBPPS_ITEMS_START     = 48;

   private int k_;                      // max size of sketch, in items
   private long n_;                     // total number of items processed by the sketch

   private double cumulativeWt_;        // total weight of items processed by the sketch
   private double wtMax_;               // maximum weight seen so far
   private double rho_;                 // latest scaling parameter for downsampling

   private EbppsItemsSample<T> sample_; // Object holding the current state of the sample

   final private EbppsItemsSample<T> tmp_;    // temporary storage

   /**
    * Constructor
    * @param k The maximum number of samples to retain
    */
   public EbppsItemsSketch(final int k) {
     checkK(k);
     k_ = k;
     rho_ = 1.0;
     sample_ = new EbppsItemsSample<>(k);
     tmp_ = new EbppsItemsSample<>(1);
   }

   // private copy constructor
   private EbppsItemsSketch(final EbppsItemsSketch<T> other) {
     k_ = other.k_;
     n_ = other.n_;
     rho_ = other.rho_;
     cumulativeWt_ = other.cumulativeWt_;
     wtMax_ = other.wtMax_;
     sample_ = new EbppsItemsSample<>(other.sample_);
     tmp_ = new EbppsItemsSample<>(1);
   }

   // private constructor for heapify
   private EbppsItemsSketch(final EbppsItemsSample<T> sample,
                            final int k,
                            final long n,
                            final double cumWt,
                            final double maxWt,
                            final double rho) {
     k_ = k;
     n_ = n;
     cumulativeWt_ = cumWt;
     wtMax_ = maxWt;
     rho_ = rho;
     sample_ = sample;
     tmp_ = new EbppsItemsSample<>(1);
   }

   /**
    * Returns a sketch instance of this class from the given srcMem,
    * which must be a Memory representation of this sketch class.
    *
    * @param <T>    The type of item this sketch contains
    * @param srcMem a Memory representation of a sketch of this class.
    *               <a href="{@docRoot}/resources/dictionary.html#mem">See Memory</a>
    * @param serDe  An instance of ArrayOfItemsSerDe
    * @return a sketch instance of this class
    */
   public static <T> EbppsItemsSketch<T> heapify(final Memory srcMem,
                                                 final ArrayOfItemsSerDe<T> serDe)
   {
     final int numPreLongs = PreambleUtil.getAndCheckPreLongs(srcMem);
     final int serVer = PreambleUtil.extractSerVer(srcMem);
     final int familyId = PreambleUtil.extractFamilyID(srcMem);
     final int flags = PreambleUtil.extractFlags(srcMem);
     final boolean isEmpty = (flags & EMPTY_FLAG_MASK) != 0;
     final boolean hasPartialItem = (flags & HAS_PARTIAL_ITEM_MASK) != 0;

     // Check values
     if (isEmpty) {
       if (numPreLongs != Family.EBPPS.getMinPreLongs()) {
         throw new SketchesArgumentException("Possible corruption: Must be " + Family.EBPPS.getMinPreLongs()
                 + " for an empty sketch. Found: " + numPreLongs);
       }
     } else {
       if (numPreLongs != Family.EBPPS.getMaxPreLongs()) {
         throw new SketchesArgumentException("Possible corruption: Must be "
                 + Family.EBPPS.getMaxPreLongs() + " for a non-empty sketch. Found: " + numPreLongs);
       }
     }
     if (serVer != EBPPS_SER_VER) {
         throw new SketchesArgumentException(
                 "Possible Corruption: Ser Ver must be " + EBPPS_SER_VER + ": " + serVer);
     }
     final int reqFamilyId = Family.EBPPS.getID();
     if (familyId != reqFamilyId) {
       throw new SketchesArgumentException(
               "Possible Corruption: FamilyID must be " + reqFamilyId + ": " + familyId);
     }

     final int k = PreambleUtil.extractK(srcMem);
     if (k < 1 || k > MAX_K) {
       throw new SketchesArgumentException("Possible Corruption: k must be at least 1 "
               + "and less than " + MAX_K + ". Found: " + k);
     }

     if (isEmpty) {
       return new EbppsItemsSketch<>(k);
     }

     final long n = PreambleUtil.extractN(srcMem);
     if (n < 0) {
       throw new SketchesArgumentException("Possible Corruption: n cannot be negative: " + n);
     }

     final double cumWt = PreambleUtil.extractEbppsCumulativeWeight(srcMem);
     if (cumWt < 0.0 || Double.isNaN(cumWt) || Double.isInfinite(cumWt)) {
       throw new SketchesArgumentException("Possible Corruption: cumWt must be nonnegative and finite: " + cumWt);
     }

     final double maxWt = PreambleUtil.extractEbppsMaxWeight(srcMem);
     if (maxWt < 0.0 || Double.isNaN(maxWt) || Double.isInfinite(maxWt)) {
       throw new SketchesArgumentException("Possible Corruption: maxWt must be nonnegative and finite: " + maxWt);
     }

     final double rho = PreambleUtil.extractEbppsRho(srcMem);
     if (rho < 0.0 || rho > 1.0 ||  Double.isNaN(rho) || Double.isInfinite(rho)) {
       throw new SketchesArgumentException("Possible Corruption: rho must be in [0.0, 1.0]: " + rho);
     }

     // extract C (part of sample_, not the preamble)
     // due to numeric precision issues, c may occasionally be very slightly larger than k
     final double c = srcMem.getDouble(EBPPS_C_DOUBLE);
     if (c < 0 || c >= (k + 1) || Double.isNaN(c) || Double.isInfinite(c)) {
       throw new SketchesArgumentException("Possible Corruption: c must be between 0 and k: " + c);
     }

     // extract items
     final int numTotalItems = (int) Math.ceil(c);
     final int numFullItems = (int) Math.floor(c); // floor() not strictly necessary
     final int offsetBytes = EBPPS_ITEMS_START;
     final T[] rawItems = serDe.deserializeFromMemory(
             srcMem.region(offsetBytes, srcMem.getCapacity() - offsetBytes), 0, numTotalItems);
     final List<T> itemsList = Arrays.asList(rawItems);
     final ArrayList<T> data;
     final T partialItem;
     if (hasPartialItem) {
       if (numFullItems >= numTotalItems) {
         throw new SketchesArgumentException("Possible Corruption: Expected partial item but none found");
       }

       data = new ArrayList<>(itemsList.subList(0, numFullItems));
       partialItem = itemsList.get(numFullItems); // 0-based, so last item
     } else {
       data = new ArrayList<>(itemsList);
       partialItem = null; // just to be explicit
     }

     final EbppsItemsSample<T> sample = new EbppsItemsSample<>(data, partialItem, c);

     return new EbppsItemsSketch<>(sample, k, n, cumWt, maxWt, rho);
   }

   /**
    * Updates this sketch with the given data item with weight 1.0.
    * @param item an item from a stream of items
    */
   public void update(final T item) {
     update(item, 1.0);
   }

   /**
    * Updates this sketch with the given data item with the given weight.
    * @param item an item from a stream of items
    * @param weight the weight of the item
    */
   public void update(final T item, final double weight) {
     if (weight < 0.0 || Double.isNaN(weight) || Double.isInfinite(weight)) {
       throw new SketchesArgumentException("Item weights must be nonnegative and finite. "
         + "Found: " + weight);
     }
     if (weight == 0.0) {
       return;
     }

     final double newCumWt = cumulativeWt_ + weight;
     final double newWtMax = Math.max(wtMax_, weight);
     final double newRho = Math.min(1.0 / newWtMax, k_ / newCumWt);

     if (cumulativeWt_ > 0.0) {
       sample_.downsample((newRho / rho_));
     }

     tmp_.replaceContent(item, newRho * weight);
     sample_.merge(tmp_);

     cumulativeWt_ = newCumWt;
     wtMax_ = newWtMax;
     rho_ = newRho;
     ++n_;
   }

   /* Merging
    * There is a trivial merge algorithm that involves downsampling each sketch A and B
    * as A.cum_wt / (A.cum_wt + B.cum_wt) and B.cum_wt / (A.cum_wt + B.cum_wt),
    * respectively. That merge does preserve first-order probabilities, specifically
    * the probability proportional to size property, and like all other known merge
    * algorithms distorts second-order probabilities (co-occurrences). There are
    * pathological cases, most obvious with k=2 and A.cum_wt == B.cum_wt where that
    * approach will always take exactly 1 item from A and 1 from B, meaning the
    * co-occurrence rate for two items from either sketch is guaranteed to be 0.0.
    *
    * With EBPPS, once an item is accepted into the sketch we no longer need to
    * track the item's weight: All accepted items are treated equally. As a result, we
    * can take inspiration from the reservoir sampling merge in the datasketches-java
    * library. We need to merge the smaller sketch into the larger one, swapping as
    * needed to ensure that, at which point we simply call update() with the items
    * in the smaller sketch as long as we adjust the weight appropriately.
    * Merging smaller into larger is essential to ensure that no item has a
    * contribution to C > 1.0.
    */

   /**
    * Merges the provided sketch into the current one.
    * @param other the sketch to merge into the current object
    */
   public void merge(final EbppsItemsSketch<T> other) {
     if (other.getCumulativeWeight() == 0.0) {
       return;
     } else if (other.getCumulativeWeight() > cumulativeWt_) {
       // need to swap this with other
       // make a copy of other, merge into it, and take the result
       final EbppsItemsSketch<T> copy = new EbppsItemsSketch<>(other);
       copy.internalMerge(this);
       k_ = copy.k_;
       n_ = copy.n_;
       cumulativeWt_ = copy.cumulativeWt_;
       wtMax_ = copy.wtMax_;
       rho_ = copy.rho_;
       sample_ = copy.sample_;
     } else {
       internalMerge(other);
     }
   }

   // merge implementation called exclusively from public merge()
   private void internalMerge(final EbppsItemsSketch<T> other) {
     // assumes that other.cumulativeWeight_ <= cumulativeWt_m
     // which must be checked before calling this

     final double finalCumWt = cumulativeWt_ + other.cumulativeWt_;
     final double newWtMax = Math.max(wtMax_, other.wtMax_);
     k_ = Math.min(k_, other.k_);
     final long newN = n_ + other.n_;

     // Insert other's items with the cumulative weight
     // split between the input items. We repeat the same process
     // for full items and the partial item, scaling the input
     // weight appropriately.
     // We handle all C input items, meaning we always process
     // the partial item using a scaled down weight.
     // Handling the partial item by probabilistically including
     // it as a full item would be correct on average but would
     // introduce bias for any specific merge operation.
     final double avgWt = other.cumulativeWt_ / other.getC();
     final ArrayList<T> items = other.sample_.getFullItems();
     if (items != null) {
       for (T item : items) {
         // newWtMax is pre-computed
         final double newCumWt = cumulativeWt_ + avgWt;
         final double newRho = Math.min(1.0 / newWtMax, k_ / newCumWt);

         if (cumulativeWt_ > 0.0) {
           sample_.downsample(newRho / rho_);
         }

         tmp_.replaceContent(item, newRho * avgWt);
         sample_.merge(tmp_);

         cumulativeWt_ = newCumWt;
         rho_ = newRho;
       }
     }

     // insert partial item with weight scaled by the fractional part of C
     if (other.sample_.hasPartialItem()) {
       final double otherCFrac = other.getC() % 1;
       final double newCumWt = cumulativeWt_ + (otherCFrac * avgWt);
       final double newRho = Math.min(1.0 / newWtMax, k_ / newCumWt);

       if (cumulativeWt_ > 0.0) {
         sample_.downsample(newRho / rho_);
       }

       tmp_.replaceContent(other.sample_.getPartialItem(), newRho * otherCFrac * avgWt);
       sample_.merge(tmp_);

       // cumulativeWt_ will be assigned momentarily
       rho_ = newRho;
     }

     // avoid numeric issues by setting cumulative weight to the
     // pre-computed value
     cumulativeWt_ = finalCumWt;
     n_ = newN;
   }

   /**
    * Returns a copy of the current sample. The exact size may be
    * probabilistic, differing by at most 1 item.
    * @return the current sketch sample
    */
   public ArrayList<T> getResult() { return sample_.getSample(); }

   /**
    * Provides a human-readable summary of the sketch
    * @return a summary of information in the sketch
    */
   @Override
    public String toString() {
     final StringBuilder sb = new StringBuilder();

     sb.append(LS);
     final String thisSimpleName = this.getClass().getSimpleName();
     sb.append("### ").append(thisSimpleName).append(" SUMMARY: ").append(LS);
     sb.append("   k            : ").append(k_).append(LS);
     sb.append("   n            : ").append(n_).append(LS);
     sb.append("   Cum. weight  : ").append(cumulativeWt_).append(LS);
     sb.append("   wtMax        : ").append(wtMax_).append(LS);
     sb.append("   rho          : ").append(rho_).append(LS);
     sb.append("   C            : ").append(sample_.getC()).append(LS);
     sb.append("### END SKETCH SUMMARY").append(LS);

     return sb.toString();
   }

   /**
    * Returns the configured maximum sample size.
    * @return configured maximum sample size
    */
   public int getK() { return k_; }

   /**
    * Returns the number of items processed by the sketch, regardless
    * of item weight.
    * @return count of items processed by the sketch
    */
   public long getN() { return n_; }

   /**
    * Returns the cumulative weight of items processed by the sketch.
    * @return cumulative weight of items seen
    */
   public double getCumulativeWeight() { return cumulativeWt_; }

   /**
    * Returns the expected number of samples returned upon a call to
    * getResult(). The number is a floating point value, where the
    * fractional portion represents the probability of including a
    * "partial item" from the sample.
    *
    * <p>The value C should be no larger than the sketch's configured
    * value of k, although numerical precision limitations mean it
    * may exceed k by double precision floating point error margins
    * in certain cases.
    * @return The expected number of samples returned when querying the sketch
    */
   public double getC() { return sample_.getC(); }

   /**
    * Returns true if the sketch is empty.
    * @return empty flag
    */
   public boolean isEmpty() { return n_ == 0; }

   /**
    * Resets the sketch to its default, empty state.
    */
   public void reset() {
     n_ = 0;
     cumulativeWt_ = 0.0;
     wtMax_ = 0.0;
     rho_ = 1.0;
     sample_ = new EbppsItemsSample<>(k_);
   }

   /**
    * Returns the size of a byte array representation of this sketch. May fail for polymorphic item types.
    *
    * @param serDe An instance of ArrayOfItemsSerDe
    * @return the length of a byte array representation of this sketch
    */
   public int getSerializedSizeBytes(final ArrayOfItemsSerDe<? super T> serDe) {
     if (isEmpty()) {
       return Family.EBPPS.getMinPreLongs() << 3;
     } else if (sample_.getC() < 1.0) {
       return getSerializedSizeBytes(serDe, sample_.getPartialItem().getClass());
     } else {
       return getSerializedSizeBytes(serDe, sample_.getSample().get(0).getClass());
     }
   }

   /**
    * Returns the length of a byte array representation of this sketch. Copies contents into an array of the
    * specified class for serialization to allow for polymorphic types.
    *
    * @param serDe An instance of ArrayOfItemsSerDe
    * @param clazz The class represented by &lt;T&gt;
    * @return the length of a byte array representation of this sketch
    */
   public int getSerializedSizeBytes(final ArrayOfItemsSerDe<? super T> serDe, final Class<?> clazz) {
     if (n_ == 0) {
       return Family.EBPPS.getMinPreLongs() << 3;
     }

     final int preLongs = Family.EBPPS.getMaxPreLongs();
     final byte[] itemBytes = serDe.serializeToByteArray(sample_.getAllSamples(clazz));
     // in C++, c_ is serialized as part of the sample_ and not included in the header size
     return (preLongs << 3) + Double.BYTES + itemBytes.length;
   }

   /**
    * Returns a byte array representation of this sketch. May fail for polymorphic item types.
    *
    * @param serDe An instance of ArrayOfItemsSerDe
    * @return a byte array representation of this sketch
    */
   public byte[] toByteArray(final ArrayOfItemsSerDe<? super T> serDe) {
     if (n_ == 0) {
       // null class is ok since empty -- no need to call serDe
       return toByteArray(serDe, null);
     } else if (sample_.getC() < 1.0) {
       return toByteArray(serDe, sample_.getPartialItem().getClass());
     } else {
       return toByteArray(serDe, sample_.getSample().get(0).getClass());
     }
   }

    /**
    * Returns a byte array representation of this sketch. Copies contents into an array of the
    * specified class for serialization to allow for polymorphic types.
    *
    * @param serDe An instance of ArrayOfItemsSerDe
    * @param clazz The class represented by &lt;T&gt;
    * @return a byte array representation of this sketch
    */
   public byte[] toByteArray(final ArrayOfItemsSerDe<? super T> serDe, final Class<?> clazz) {
     final int preLongs, outBytes;
     final boolean empty = n_ == 0;
     byte[] itemBytes = null; // for serialized items from sample_

     if (empty) {
       preLongs = 1;
       outBytes = 8;
     } else {
       preLongs = Family.EBPPS.getMaxPreLongs();
       itemBytes = serDe.serializeToByteArray(sample_.getAllSamples(clazz));
       // in C++, c_ is serialized as part of the sample_ and not included in the header size
       outBytes = (preLongs << 3) + Double.BYTES + itemBytes.length;
     }
     final byte[] outArr = new byte[outBytes];
     final WritableMemory mem = WritableMemory.writableWrap(outArr);

     // Common header elements
     PreambleUtil.insertPreLongs(mem, preLongs);              // Byte 0
     PreambleUtil.insertSerVer(mem, EBPPS_SER_VER);           // Byte 1
     PreambleUtil.insertFamilyID(mem, Family.EBPPS.getID());  // Byte 2
     if (empty) {
       PreambleUtil.insertFlags(mem, EMPTY_FLAG_MASK);        // Byte 3
     } else {
       PreambleUtil.insertFlags(mem, sample_.hasPartialItem() ? HAS_PARTIAL_ITEM_MASK : 0);
     }
     PreambleUtil.insertK(mem, k_);                           // Bytes 4-7

     // conditional elements
     if (!empty) {
       PreambleUtil.insertN(mem, n_);
       PreambleUtil.insertEbppsCumulativeWeight(mem, cumulativeWt_);
       PreambleUtil.insertEbppsMaxWeight(mem, wtMax_);
       PreambleUtil.insertEbppsRho(mem, rho_);

       // data from sample_ -- itemBytes includes the partial item
       mem.putDouble(EBPPS_C_DOUBLE, sample_.getC());
       mem.putByteArray(EBPPS_ITEMS_START, itemBytes, 0, itemBytes.length);
     }

     return outArr;
   }

   private static void checkK(final int k) {
     if (k <= 0 || k > MAX_K) {
       throw new SketchesArgumentException("k must be strictly positive and less than " + MAX_K);
     }
   }
 }
	/*
	* 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.sampling;

	import static org.apache.datasketches.common.Util.LS;
	import static org.apache.datasketches.sampling.PreambleUtil.EBPPS_SER_VER;
	import static org.apache.datasketches.sampling.PreambleUtil.EMPTY_FLAG_MASK;
	import static org.apache.datasketches.sampling.PreambleUtil.HAS_PARTIAL_ITEM_MASK;

	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.List;

	import org.apache.datasketches.common.ArrayOfItemsSerDe;
	import org.apache.datasketches.common.Family;
	import org.apache.datasketches.common.SketchesArgumentException;
	import org.apache.datasketches.memory.Memory;
	import org.apache.datasketches.memory.WritableMemory;

	/**
	* An implementation of an Exact and Bounded Sampling Proportional to Size sketch.
	*
	* <p>From: "Exact PPS Sampling with Bounded Sample Size",
	* B. Hentschel, P. J. Haas, Y. Tian. Information Processing Letters, 2023.
	*
	* <p>This sketch samples data from a stream of items proportional to the weight of each item.
	* The sample guarantees the presence of an item in the result is proportional to that item's
	* portion of the total weight seen by the sketch, and returns a sample no larger than size k.
	*
	* <p>The sample may be smaller than k and the resulting size of the sample potentially includes
	* a probabilistic component, meaning the resulting sample size is not always constant.
	* @param <T> the item class type
	* @author Jon Malkin
	*/
	public final class EbppsItemsSketch<T> {
	private static final int MAX_K = Integer.MAX_VALUE - 2;
	private static final int EBPPS_C_DOUBLE = 40; // part of sample state, not preamble
	private static final int EBPPS_ITEMS_START = 48;

	private int k_; // max size of sketch, in items
	private long n_; // total number of items processed by the sketch

	private double cumulativeWt_; // total weight of items processed by the sketch
	private double wtMax_; // maximum weight seen so far
	private double rho_; // latest scaling parameter for downsampling

	private EbppsItemsSample<T> sample_; // Object holding the current state of the sample

	final private EbppsItemsSample<T> tmp_; // temporary storage

	/**
	* Constructor
	* @param k The maximum number of samples to retain
	*/
	public EbppsItemsSketch(final int k) {
	checkK(k);
	k_ = k;
	rho_ = 1.0;
	sample_ = new EbppsItemsSample<>(k);
	tmp_ = new EbppsItemsSample<>(1);
	}

	// private copy constructor
	private EbppsItemsSketch(final EbppsItemsSketch<T> other) {
	k_ = other.k_;
	n_ = other.n_;
	rho_ = other.rho_;
	cumulativeWt_ = other.cumulativeWt_;
	wtMax_ = other.wtMax_;
	sample_ = new EbppsItemsSample<>(other.sample_);
	tmp_ = new EbppsItemsSample<>(1);
	}

	// private constructor for heapify
	private EbppsItemsSketch(final EbppsItemsSample<T> sample,
	final int k,
	final long n,
	final double cumWt,
	final double maxWt,
	final double rho) {
	k_ = k;
	n_ = n;
	cumulativeWt_ = cumWt;
	wtMax_ = maxWt;
	rho_ = rho;
	sample_ = sample;
	tmp_ = new EbppsItemsSample<>(1);
	}

	/**
	* Returns a sketch instance of this class from the given srcMem,
	* which must be a Memory representation of this sketch class.
	*
	* @param <T> The type of item this sketch contains
	* @param srcMem a Memory representation of a sketch of this class.
	* <a href="{@docRoot}/resources/dictionary.html#mem">See Memory</a>
	* @param serDe An instance of ArrayOfItemsSerDe
	* @return a sketch instance of this class
	*/
	public static <T> EbppsItemsSketch<T> heapify(final Memory srcMem,
	final ArrayOfItemsSerDe<T> serDe)
	{
	final int numPreLongs = PreambleUtil.getAndCheckPreLongs(srcMem);
	final int serVer = PreambleUtil.extractSerVer(srcMem);
	final int familyId = PreambleUtil.extractFamilyID(srcMem);
	final int flags = PreambleUtil.extractFlags(srcMem);
	final boolean isEmpty = (flags & EMPTY_FLAG_MASK) != 0;
	final boolean hasPartialItem = (flags & HAS_PARTIAL_ITEM_MASK) != 0;

	// Check values
	if (isEmpty) {
	if (numPreLongs != Family.EBPPS.getMinPreLongs()) {
	throw new SketchesArgumentException("Possible corruption: Must be " + Family.EBPPS.getMinPreLongs()
	+ " for an empty sketch. Found: " + numPreLongs);
	}
	} else {
	if (numPreLongs != Family.EBPPS.getMaxPreLongs()) {
	throw new SketchesArgumentException("Possible corruption: Must be "
	+ Family.EBPPS.getMaxPreLongs() + " for a non-empty sketch. Found: " + numPreLongs);
	}
	}
	if (serVer != EBPPS_SER_VER) {
	throw new SketchesArgumentException(
	"Possible Corruption: Ser Ver must be " + EBPPS_SER_VER + ": " + serVer);
	}
	final int reqFamilyId = Family.EBPPS.getID();
	if (familyId != reqFamilyId) {
	throw new SketchesArgumentException(
	"Possible Corruption: FamilyID must be " + reqFamilyId + ": " + familyId);
	}

	final int k = PreambleUtil.extractK(srcMem);
	if (k < 1 \|\| k > MAX_K) {
	throw new SketchesArgumentException("Possible Corruption: k must be at least 1 "
	+ "and less than " + MAX_K + ". Found: " + k);
	}

	if (isEmpty) {
	return new EbppsItemsSketch<>(k);
	}

	final long n = PreambleUtil.extractN(srcMem);
	if (n < 0) {
	throw new SketchesArgumentException("Possible Corruption: n cannot be negative: " + n);
	}

	final double cumWt = PreambleUtil.extractEbppsCumulativeWeight(srcMem);
	if (cumWt < 0.0 \|\| Double.isNaN(cumWt) \|\| Double.isInfinite(cumWt)) {
	throw new SketchesArgumentException("Possible Corruption: cumWt must be nonnegative and finite: " + cumWt);
	}

	final double maxWt = PreambleUtil.extractEbppsMaxWeight(srcMem);
	if (maxWt < 0.0 \|\| Double.isNaN(maxWt) \|\| Double.isInfinite(maxWt)) {
	throw new SketchesArgumentException("Possible Corruption: maxWt must be nonnegative and finite: " + maxWt);
	}

	final double rho = PreambleUtil.extractEbppsRho(srcMem);
	if (rho < 0.0 \|\| rho > 1.0 \|\| Double.isNaN(rho) \|\| Double.isInfinite(rho)) {
	throw new SketchesArgumentException("Possible Corruption: rho must be in [0.0, 1.0]: " + rho);
	}

	// extract C (part of sample_, not the preamble)
	// due to numeric precision issues, c may occasionally be very slightly larger than k
	final double c = srcMem.getDouble(EBPPS_C_DOUBLE);
	if (c < 0 \|\| c >= (k + 1) \|\| Double.isNaN(c) \|\| Double.isInfinite(c)) {
	throw new SketchesArgumentException("Possible Corruption: c must be between 0 and k: " + c);
	}

	// extract items
	final int numTotalItems = (int) Math.ceil(c);
	final int numFullItems = (int) Math.floor(c); // floor() not strictly necessary
	final int offsetBytes = EBPPS_ITEMS_START;
	final T[] rawItems = serDe.deserializeFromMemory(
	srcMem.region(offsetBytes, srcMem.getCapacity() - offsetBytes), 0, numTotalItems);
	final List<T> itemsList = Arrays.asList(rawItems);
	final ArrayList<T> data;
	final T partialItem;
	if (hasPartialItem) {
	if (numFullItems >= numTotalItems) {
	throw new SketchesArgumentException("Possible Corruption: Expected partial item but none found");
	}

	data = new ArrayList<>(itemsList.subList(0, numFullItems));
	partialItem = itemsList.get(numFullItems); // 0-based, so last item
	} else {
	data = new ArrayList<>(itemsList);
	partialItem = null; // just to be explicit
	}

	final EbppsItemsSample<T> sample = new EbppsItemsSample<>(data, partialItem, c);

	return new EbppsItemsSketch<>(sample, k, n, cumWt, maxWt, rho);
	}

	/**
	* Updates this sketch with the given data item with weight 1.0.
	* @param item an item from a stream of items
	*/
	public void update(final T item) {
	update(item, 1.0);
	}

	/**
	* Updates this sketch with the given data item with the given weight.
	* @param item an item from a stream of items
	* @param weight the weight of the item
	*/
	public void update(final T item, final double weight) {
	if (weight < 0.0 \|\| Double.isNaN(weight) \|\| Double.isInfinite(weight)) {
	throw new SketchesArgumentException("Item weights must be nonnegative and finite. "
	+ "Found: " + weight);
	}
	if (weight == 0.0) {
	return;
	}

	final double newCumWt = cumulativeWt_ + weight;
	final double newWtMax = Math.max(wtMax_, weight);
	final double newRho = Math.min(1.0 / newWtMax, k_ / newCumWt);

	if (cumulativeWt_ > 0.0) {
	sample_.downsample((newRho / rho_));
	}

	tmp_.replaceContent(item, newRho * weight);
	sample_.merge(tmp_);

	cumulativeWt_ = newCumWt;
	wtMax_ = newWtMax;
	rho_ = newRho;
	++n_;
	}

	/* Merging
	* There is a trivial merge algorithm that involves downsampling each sketch A and B
	* as A.cum_wt / (A.cum_wt + B.cum_wt) and B.cum_wt / (A.cum_wt + B.cum_wt),
	* respectively. That merge does preserve first-order probabilities, specifically
	* the probability proportional to size property, and like all other known merge
	* algorithms distorts second-order probabilities (co-occurrences). There are
	* pathological cases, most obvious with k=2 and A.cum_wt == B.cum_wt where that
	* approach will always take exactly 1 item from A and 1 from B, meaning the
	* co-occurrence rate for two items from either sketch is guaranteed to be 0.0.
	*
	* With EBPPS, once an item is accepted into the sketch we no longer need to
	* track the item's weight: All accepted items are treated equally. As a result, we
	* can take inspiration from the reservoir sampling merge in the datasketches-java
	* library. We need to merge the smaller sketch into the larger one, swapping as
	* needed to ensure that, at which point we simply call update() with the items
	* in the smaller sketch as long as we adjust the weight appropriately.
	* Merging smaller into larger is essential to ensure that no item has a
	* contribution to C > 1.0.
	*/

	/**
	* Merges the provided sketch into the current one.
	* @param other the sketch to merge into the current object
	*/
	public void merge(final EbppsItemsSketch<T> other) {
	if (other.getCumulativeWeight() == 0.0) {
	return;
	} else if (other.getCumulativeWeight() > cumulativeWt_) {
	// need to swap this with other
	// make a copy of other, merge into it, and take the result
	final EbppsItemsSketch<T> copy = new EbppsItemsSketch<>(other);
	copy.internalMerge(this);
	k_ = copy.k_;
	n_ = copy.n_;
	cumulativeWt_ = copy.cumulativeWt_;
	wtMax_ = copy.wtMax_;
	rho_ = copy.rho_;
	sample_ = copy.sample_;
	} else {
	internalMerge(other);
	}
	}

	// merge implementation called exclusively from public merge()
	private void internalMerge(final EbppsItemsSketch<T> other) {
	// assumes that other.cumulativeWeight_ <= cumulativeWt_m
	// which must be checked before calling this

	final double finalCumWt = cumulativeWt_ + other.cumulativeWt_;
	final double newWtMax = Math.max(wtMax_, other.wtMax_);
	k_ = Math.min(k_, other.k_);
	final long newN = n_ + other.n_;

	// Insert other's items with the cumulative weight
	// split between the input items. We repeat the same process
	// for full items and the partial item, scaling the input
	// weight appropriately.
	// We handle all C input items, meaning we always process
	// the partial item using a scaled down weight.
	// Handling the partial item by probabilistically including
	// it as a full item would be correct on average but would
	// introduce bias for any specific merge operation.
	final double avgWt = other.cumulativeWt_ / other.getC();
	final ArrayList<T> items = other.sample_.getFullItems();
	if (items != null) {
	for (T item : items) {
	// newWtMax is pre-computed
	final double newCumWt = cumulativeWt_ + avgWt;
	final double newRho = Math.min(1.0 / newWtMax, k_ / newCumWt);

	if (cumulativeWt_ > 0.0) {
	sample_.downsample(newRho / rho_);
	}

	tmp_.replaceContent(item, newRho * avgWt);
	sample_.merge(tmp_);

	cumulativeWt_ = newCumWt;
	rho_ = newRho;
	}
	}

	// insert partial item with weight scaled by the fractional part of C
	if (other.sample_.hasPartialItem()) {
	final double otherCFrac = other.getC() % 1;
	final double newCumWt = cumulativeWt_ + (otherCFrac * avgWt);
	final double newRho = Math.min(1.0 / newWtMax, k_ / newCumWt);

	if (cumulativeWt_ > 0.0) {
	sample_.downsample(newRho / rho_);
	}

	tmp_.replaceContent(other.sample_.getPartialItem(), newRho * otherCFrac * avgWt);
	sample_.merge(tmp_);

	// cumulativeWt_ will be assigned momentarily
	rho_ = newRho;
	}

	// avoid numeric issues by setting cumulative weight to the
	// pre-computed value
	cumulativeWt_ = finalCumWt;
	n_ = newN;
	}

	/**
	* Returns a copy of the current sample. The exact size may be
	* probabilistic, differing by at most 1 item.
	* @return the current sketch sample
	*/
	public ArrayList<T> getResult() { return sample_.getSample(); }

	/**
	* Provides a human-readable summary of the sketch
	* @return a summary of information in the sketch
	*/
	@Override
	public String toString() {
	final StringBuilder sb = new StringBuilder();

	sb.append(LS);
	final String thisSimpleName = this.getClass().getSimpleName();
	sb.append("### ").append(thisSimpleName).append(" SUMMARY: ").append(LS);
	sb.append(" k : ").append(k_).append(LS);
	sb.append(" n : ").append(n_).append(LS);
	sb.append(" Cum. weight : ").append(cumulativeWt_).append(LS);
	sb.append(" wtMax : ").append(wtMax_).append(LS);
	sb.append(" rho : ").append(rho_).append(LS);
	sb.append(" C : ").append(sample_.getC()).append(LS);
	sb.append("### END SKETCH SUMMARY").append(LS);

	return sb.toString();
	}

	/**
	* Returns the configured maximum sample size.
	* @return configured maximum sample size
	*/
	public int getK() { return k_; }

	/**
	* Returns the number of items processed by the sketch, regardless
	* of item weight.
	* @return count of items processed by the sketch
	*/
	public long getN() { return n_; }

	/**
	* Returns the cumulative weight of items processed by the sketch.
	* @return cumulative weight of items seen
	*/
	public double getCumulativeWeight() { return cumulativeWt_; }

	/**
	* Returns the expected number of samples returned upon a call to
	* getResult(). The number is a floating point value, where the
	* fractional portion represents the probability of including a
	* "partial item" from the sample.
	*
	* <p>The value C should be no larger than the sketch's configured
	* value of k, although numerical precision limitations mean it
	* may exceed k by double precision floating point error margins
	* in certain cases.
	* @return The expected number of samples returned when querying the sketch
	*/
	public double getC() { return sample_.getC(); }

	/**
	* Returns true if the sketch is empty.
	* @return empty flag
	*/
	public boolean isEmpty() { return n_ == 0; }

	/**
	* Resets the sketch to its default, empty state.
	*/
	public void reset() {
	n_ = 0;
	cumulativeWt_ = 0.0;
	wtMax_ = 0.0;
	rho_ = 1.0;
	sample_ = new EbppsItemsSample<>(k_);
	}

	/**
	* Returns the size of a byte array representation of this sketch. May fail for polymorphic item types.
	*
	* @param serDe An instance of ArrayOfItemsSerDe
	* @return the length of a byte array representation of this sketch
	*/
	public int getSerializedSizeBytes(final ArrayOfItemsSerDe<? super T> serDe) {
	if (isEmpty()) {
	return Family.EBPPS.getMinPreLongs() << 3;
	} else if (sample_.getC() < 1.0) {
	return getSerializedSizeBytes(serDe, sample_.getPartialItem().getClass());
	} else {
	return getSerializedSizeBytes(serDe, sample_.getSample().get(0).getClass());
	}
	}

	/**
	* Returns the length of a byte array representation of this sketch. Copies contents into an array of the
	* specified class for serialization to allow for polymorphic types.
	*
	* @param serDe An instance of ArrayOfItemsSerDe
	* @param clazz The class represented by <T>
	* @return the length of a byte array representation of this sketch
	*/
	public int getSerializedSizeBytes(final ArrayOfItemsSerDe<? super T> serDe, final Class<?> clazz) {
	if (n_ == 0) {
	return Family.EBPPS.getMinPreLongs() << 3;
	}

	final int preLongs = Family.EBPPS.getMaxPreLongs();
	final byte[] itemBytes = serDe.serializeToByteArray(sample_.getAllSamples(clazz));
	// in C++, c_ is serialized as part of the sample_ and not included in the header size
	return (preLongs << 3) + Double.BYTES + itemBytes.length;
	}

	/**
	* Returns a byte array representation of this sketch. May fail for polymorphic item types.
	*
	* @param serDe An instance of ArrayOfItemsSerDe
	* @return a byte array representation of this sketch
	*/
	public byte[] toByteArray(final ArrayOfItemsSerDe<? super T> serDe) {
	if (n_ == 0) {
	// null class is ok since empty -- no need to call serDe
	return toByteArray(serDe, null);
	} else if (sample_.getC() < 1.0) {
	return toByteArray(serDe, sample_.getPartialItem().getClass());
	} else {
	return toByteArray(serDe, sample_.getSample().get(0).getClass());
	}
	}

	/**
	* Returns a byte array representation of this sketch. Copies contents into an array of the
	* specified class for serialization to allow for polymorphic types.
	*
	* @param serDe An instance of ArrayOfItemsSerDe
	* @param clazz The class represented by <T>
	* @return a byte array representation of this sketch
	*/
	public byte[] toByteArray(final ArrayOfItemsSerDe<? super T> serDe, final Class<?> clazz) {
	final int preLongs, outBytes;
	final boolean empty = n_ == 0;
	byte[] itemBytes = null; // for serialized items from sample_

	if (empty) {
	preLongs = 1;
	outBytes = 8;
	} else {
	preLongs = Family.EBPPS.getMaxPreLongs();
	itemBytes = serDe.serializeToByteArray(sample_.getAllSamples(clazz));
	// in C++, c_ is serialized as part of the sample_ and not included in the header size
	outBytes = (preLongs << 3) + Double.BYTES + itemBytes.length;
	}
	final byte[] outArr = new byte[outBytes];
	final WritableMemory mem = WritableMemory.writableWrap(outArr);

	// Common header elements
	PreambleUtil.insertPreLongs(mem, preLongs); // Byte 0
	PreambleUtil.insertSerVer(mem, EBPPS_SER_VER); // Byte 1
	PreambleUtil.insertFamilyID(mem, Family.EBPPS.getID()); // Byte 2
	if (empty) {
	PreambleUtil.insertFlags(mem, EMPTY_FLAG_MASK); // Byte 3
	} else {
	PreambleUtil.insertFlags(mem, sample_.hasPartialItem() ? HAS_PARTIAL_ITEM_MASK : 0);
	}
	PreambleUtil.insertK(mem, k_); // Bytes 4-7

	// conditional elements
	if (!empty) {
	PreambleUtil.insertN(mem, n_);
	PreambleUtil.insertEbppsCumulativeWeight(mem, cumulativeWt_);
	PreambleUtil.insertEbppsMaxWeight(mem, wtMax_);
	PreambleUtil.insertEbppsRho(mem, rho_);

	// data from sample_ -- itemBytes includes the partial item
	mem.putDouble(EBPPS_C_DOUBLE, sample_.getC());
	mem.putByteArray(EBPPS_ITEMS_START, itemBytes, 0, itemBytes.length);
	}

	return outArr;
	}

	private static void checkK(final int k) {
	if (k <= 0 \|\| k > MAX_K) {
	throw new SketchesArgumentException("k must be strictly positive and less than " + MAX_K);
	}
	}
	}