Merge pull request #644 from apache/compressed_iterator

use iteration in set ops, wrap compressed sketch and unpack in iterator

Author: AlexanderSaydakov

src/main/java/org/apache/datasketches/theta/AnotBimpl.java

@@ -20,8 +20,11 @@
 package org.apache.datasketches.theta;
 
 import static org.apache.datasketches.common.Util.exactLog2OfLong;
-import static org.apache.datasketches.thetacommon.HashOperations.convertToHashTable;
+import static org.apache.datasketches.thetacommon.HashOperations.checkThetaCorruption;
+import static org.apache.datasketches.thetacommon.HashOperations.continueCondition;
 import static org.apache.datasketches.thetacommon.HashOperations.hashSearch;
+import static org.apache.datasketches.thetacommon.HashOperations.hashSearchOrInsert;
+import static org.apache.datasketches.thetacommon.HashOperations.minLgHashTableSize;
 
 import java.util.Arrays;
 
@@ -124,7 +127,7 @@ public CompactSketch aNotB(final Sketch skA, final Sketch skB, final boolean dst
 
     if (skB.isEmpty()) {
       return skA.compact(dstOrdered, dstMem);
-   }
+    }
     ThetaUtil.checkSeedHashes(skB.getSeedHash(), seedHash_);
     //Both skA & skB are not empty
 
@@ -162,14 +165,12 @@ private static long[] getResultHashArr( //returns a new array
       final long[] hashArrA,
       final Sketch skB) {
 
-    //Rebuild/get hashtable of skB
+    // Rebuild or get hashtable of skB
     final long[] hashTableB; //read only
-    final long[] thetaCache = skB.getCache();
-    final int countB = skB.getRetainedEntries(true);
     if (skB instanceof CompactSketch) {
-      hashTableB = convertToHashTable(thetaCache, countB, minThetaLong, ThetaUtil.REBUILD_THRESHOLD);
+      hashTableB = convertToHashTable(skB, minThetaLong, ThetaUtil.REBUILD_THRESHOLD);
     } else {
-      hashTableB = thetaCache;
+      hashTableB = skB.getCache();
     }
 
     //build temporary result arrays of skA
@@ -191,6 +192,25 @@ private static long[] getResultHashArr( //returns a new array
     return Arrays.copyOfRange(tmpHashArrA, 0, nonMatches);
   }
 
+  private static long[] convertToHashTable(
+      final Sketch sketch,
+      final long thetaLong,
+      final double rebuildThreshold) {
+    final int lgArrLongs = minLgHashTableSize(sketch.getRetainedEntries(true), rebuildThreshold);
+    final int arrLongs = 1 << lgArrLongs;
+    final long[] hashTable = new long[arrLongs];
+    checkThetaCorruption(thetaLong);
+    final HashIterator it = sketch.iterator();
+    while (it.next()) {
+      final long hash = it.get();
+      if (continueCondition(thetaLong, hash) ) {
+        continue;
+      }
+      hashSearchOrInsert(hashTable, lgArrLongs, hash);
+    }
+    return hashTable;
+  }
+
   private void reset() {
     thetaLong_ = Long.MAX_VALUE;
     empty_ = true;

src/main/java/org/apache/datasketches/theta/CompactOperations.java

@@ -161,7 +161,7 @@ static CompactSketch memoryToCompact(
       final long hash = srcMem.getLong(srcPreLongs << 3);
       final SingleItemSketch sis = new SingleItemSketch(hash, srcSeedHash);
       if (dstMem != null) {
-        dstMem.putByteArray(0, sis.toByteArray(),0, 16);
+        dstMem.putByteArray(0, sis.toByteArray(), 0, 16);
         return new DirectCompactSketch(dstMem);
       } else { //heap
         return sis;

src/main/java/org/apache/datasketches/theta/CompactSketch.java

@@ -32,6 +32,7 @@
 import static org.apache.datasketches.theta.PreambleUtil.extractEntryBitsV4;
 import static org.apache.datasketches.theta.PreambleUtil.extractNumEntriesBytesV4;
 import static org.apache.datasketches.theta.PreambleUtil.extractThetaLongV4;
+import static org.apache.datasketches.theta.PreambleUtil.wholeBytesToHoldBits;
 import static org.apache.datasketches.theta.SingleItemSketch.otherCheckForSingleItem;
 
 import org.apache.datasketches.common.Family;
@@ -189,7 +190,8 @@ private static CompactSketch wrap(final Memory srcMem, final long seed, final bo
     if (serVer == 4) {
       // not wrapping the compressed format since currently we cannot take advantage of
       // decompression during iteration because set operations reach into memory directly
-      return heapifyV4(srcMem, seed, enforceSeed);
+      return DirectCompactCompressedSketch.wrapInstance(srcMem,
+          enforceSeed ? seedHash : (short) extractSeedHash(srcMem));
     }
     else if (serVer == 3) {
       if (PreambleUtil.isEmptyFlag(srcMem)) {
@@ -274,10 +276,6 @@ private int computeMinLeadingZeros() {
     return Long.numberOfLeadingZeros(ored);
   }
 
-  private static int wholeBytesToHoldBits(final int bits) {
-    return (bits >>> 3) + ((bits & 7) > 0 ? 1 : 0);
-  }
-
   private byte[] toByteArrayV4() {
     final int preambleLongs = isEstimationMode() ? 2 : 1;
     final int entryBits = 64 - computeMinLeadingZeros();
@@ -286,8 +284,8 @@ private byte[] toByteArrayV4() {
     // store num_entries as whole bytes since whole-byte blocks will follow (most probably)
     final int numEntriesBytes = wholeBytesToHoldBits(32 - Integer.numberOfLeadingZeros(getRetainedEntries()));
 
-    final int size = preambleLongs * Long.BYTES + numEntriesBytes + wholeBytesToHoldBits(compressedBits);
-    final byte[] bytes = new byte[size];
+    final int sizeBytes = preambleLongs * Long.BYTES + numEntriesBytes + wholeBytesToHoldBits(compressedBits);
+    final byte[] bytes = new byte[sizeBytes];
     final WritableMemory mem = WritableMemory.writableWrap(bytes);
     int offsetBytes = 0;
     mem.putByte(offsetBytes++, (byte) preambleLongs);
@@ -334,12 +332,10 @@ private byte[] toByteArrayV4() {
 
   private static CompactSketch heapifyV4(final Memory srcMem, final long seed, final boolean enforceSeed) {
     final int preLongs = extractPreLongs(srcMem);
-    final int flags = extractFlags(srcMem);
     final int entryBits = extractEntryBitsV4(srcMem);
     final int numEntriesBytes = extractNumEntriesBytesV4(srcMem);
     final short seedHash = (short) extractSeedHash(srcMem);
-    final boolean isEmpty = (flags & EMPTY_FLAG_MASK) > 0;
-    if (enforceSeed && !isEmpty) { PreambleUtil.checkMemorySeedHash(srcMem, seed); }
+    if (enforceSeed) { PreambleUtil.checkMemorySeedHash(srcMem, seed); }
     int offsetBytes = 8;
     long theta = Long.MAX_VALUE;
     if (preLongs > 1) {
@@ -374,7 +370,7 @@ private static CompactSketch heapifyV4(final Memory srcMem, final long seed, fin
       entries[i] += previous;
       previous = entries[i];
     }
-    return new HeapCompactSketch(entries, isEmpty, seedHash, numEntries, theta, true);
+    return new HeapCompactSketch(entries, false, seedHash, numEntries, theta, true);
   }
 
 }

src/main/java/org/apache/datasketches/theta/DirectCompactCompressedSketch.java

@@ -0,0 +1,140 @@
+/*
+ * 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.theta;
+
+import static org.apache.datasketches.theta.PreambleUtil.extractEntryBitsV4;
+import static org.apache.datasketches.theta.PreambleUtil.extractNumEntriesBytesV4;
+import static org.apache.datasketches.theta.PreambleUtil.extractPreLongs;
+import static org.apache.datasketches.theta.PreambleUtil.extractSeedHash;
+import static org.apache.datasketches.theta.PreambleUtil.extractThetaLongV4;
+import static org.apache.datasketches.theta.PreambleUtil.wholeBytesToHoldBits;
+
+import org.apache.datasketches.memory.Memory;
+import org.apache.datasketches.memory.WritableMemory;
+import org.apache.datasketches.thetacommon.ThetaUtil;
+
+/**
+ * An off-heap (Direct), compact, compressed, read-only sketch. It is not empty, not a single item and ordered.
+ *
+ * <p>This sketch can only be associated with a Serialization Version 4 format binary image.</p>
+ *
+ * <p>This implementation uses data in a given Memory that is owned and managed by the caller.
+ * This Memory can be off-heap, which if managed properly will greatly reduce the need for
+ * the JVM to perform garbage collection.</p>
+ */
+class DirectCompactCompressedSketch extends DirectCompactSketch {
+  /**
+   * Construct this sketch with the given memory.
+   * @param mem Read-only Memory object.
+   */
+  DirectCompactCompressedSketch(final Memory mem) {
+    super(mem);
+  }
+
+  /**
+   * Wraps the given Memory, which must be a SerVer 4 compressed CompactSketch image.
+   * Must check the validity of the Memory before calling.
+   * @param srcMem <a href="{@docRoot}/resources/dictionary.html#mem">See Memory</a>
+   * @param seedHash The update seedHash.
+   * <a href="{@docRoot}/resources/dictionary.html#seedHash">See Seed Hash</a>.
+   * @return this sketch
+   */
+  static DirectCompactCompressedSketch wrapInstance(final Memory srcMem, final short seedHash) {
+    ThetaUtil.checkSeedHashes((short) extractSeedHash(srcMem), seedHash);
+    return new DirectCompactCompressedSketch(srcMem);
+  }
+
+  //Sketch Overrides
+
+  @Override
+  public CompactSketch compact(final boolean dstOrdered, final WritableMemory dstMem) {
+    if (dstMem != null) {
+      mem_.copyTo(0, dstMem, 0, getCurrentBytes());
+      return new DirectCompactSketch(dstMem);
+    }
+    return CompactSketch.heapify(mem_);
+  }
+
+  @Override
+  public int getCurrentBytes() {
+    final int preLongs = extractPreLongs(mem_);
+    final int entryBits = extractEntryBitsV4(mem_);
+    final int numEntriesBytes = extractNumEntriesBytesV4(mem_);
+    return preLongs * Long.BYTES + numEntriesBytes + wholeBytesToHoldBits(getRetainedEntries() * entryBits);
+  }
+
+  private static final int START_PACKED_DATA_EXACT_MODE = 8;
+  private static final int START_PACKED_DATA_ESTIMATION_MODE = 16;
+  
+  @Override
+  public int getRetainedEntries(final boolean valid) { //compact is always valid
+    // number of entries is stored using variable length encoding
+    // most significant bytes with all zeros are not stored
+    // one byte in the preamble has the number of non-zero bytes used
+    final int preLongs = extractPreLongs(mem_); // if > 1 then the second long has theta
+    final int numEntriesBytes = extractNumEntriesBytesV4(mem_);
+    int offsetBytes = preLongs > 1 ? START_PACKED_DATA_ESTIMATION_MODE : START_PACKED_DATA_EXACT_MODE;
+    int numEntries = 0;
+    for (int i = 0; i < numEntriesBytes; i++) {
+      numEntries |= Byte.toUnsignedInt(mem_.getByte(offsetBytes++)) << (i << 3);
+    }
+    return numEntries;
+  }
+
+  @Override
+  public long getThetaLong() {
+    final int preLongs = extractPreLongs(mem_);
+    return (preLongs > 1) ? extractThetaLongV4(mem_) : Long.MAX_VALUE;
+  }
+
+  @Override
+  public boolean isEmpty() {
+    return false;
+  }
+
+  @Override
+  public boolean isOrdered() {
+    return true;
+  }
+
+  @Override
+  public HashIterator iterator() {
+    return new MemoryCompactCompressedHashIterator(
+      mem_,
+      (extractPreLongs(mem_) > 1 ? 16 : 8) + extractNumEntriesBytesV4(mem_),
+      extractEntryBitsV4(mem_),
+      getRetainedEntries()
+    );
+  }
+
+  //restricted methods
+
+  @Override
+  long[] getCache() {
+    final int numEntries = getRetainedEntries();
+    final long[] cache = new long[numEntries];
+    int i = 0;
+    HashIterator it = iterator();
+    while (it.next()) {
+      cache[i++] = it.get();
+    }
+    return cache;
+  }
+}

src/main/java/org/apache/datasketches/theta/DirectCompactSketch.java

@@ -86,11 +86,7 @@ public int getCurrentBytes() {
 
   @Override
   public double getEstimate() {
-    if (otherCheckForSingleItem(mem_)) { return 1; }
-    final int preLongs = extractPreLongs(mem_);
-    final int curCount = (preLongs == 1) ? 0 : extractCurCount(mem_);
-    final long thetaLong = (preLongs > 2) ? extractThetaLong(mem_) : Long.MAX_VALUE;
-    return Sketch.estimate(thetaLong, curCount);
+    return Sketch.estimate(getThetaLong(), getRetainedEntries());
   }
 
   @Override
@@ -142,10 +138,8 @@ public HashIterator iterator() {
 
   @Override
   public byte[] toByteArray() {
-    final int curCount = getRetainedEntries(true);
-    checkIllegalCurCountAndEmpty(isEmpty(), curCount);
-    final int preLongs = extractPreLongs(mem_);
-    final int outBytes = (curCount + preLongs) << 3;
+    checkIllegalCurCountAndEmpty(isEmpty(), getRetainedEntries());
+    final int outBytes = getCurrentBytes();
     final byte[] byteArrOut = new byte[outBytes];
     mem_.getByteArray(0, byteArrOut, 0, outBytes);
     return byteArrOut;

src/main/java/org/apache/datasketches/theta/IntersectionImpl.java

@@ -288,7 +288,7 @@ else if (curCount_ < 0 && sketchInEntries > 0) {
       else { //On the heap, allocate a HT
         hashTable_ = new long[1 << lgArrLongs_];
       }
-      moveDataToTgt(sketchIn.getCache(), curCount_);
+      moveDataToTgt(sketchIn);
     } //end of state 5
 
     //state 7
@@ -434,8 +434,6 @@ long getThetaLong() {
   private void performIntersect(final Sketch sketchIn) {
     // curCount and input data are nonzero, match against HT
     assert curCount_ > 0 && !empty_;
-    final long[] cacheIn = sketchIn.getCache();
-    final int arrLongsIn = cacheIn.length;
     final long[] hashTable;
     if (wmem_ != null) {
       final int htLen = 1 << lgArrLongs_;
@@ -448,27 +446,17 @@ private void performIntersect(final Sketch sketchIn) {
     final long[] matchSet = new long[ min(curCount_, sketchIn.getRetainedEntries(true)) ];
 
     int matchSetCount = 0;
-    if (sketchIn.isOrdered()) {
-      //ordered compact, which enables early stop
-      for (int i = 0; i < arrLongsIn; i++ ) {
-        final long hashIn = cacheIn[i];
-        //if (hashIn <= 0L) continue;  //<= 0 should not happen
-        if (hashIn >= thetaLong_) {
-          break; //early stop assumes that hashes in input sketch are ordered!
-        }
+    final boolean isOrdered = sketchIn.isOrdered();
+    final HashIterator it = sketchIn.iterator();
+    while (it.next()) {
+      final long hashIn = it.get();
+      if (hashIn < thetaLong_) {
         final int foundIdx = hashSearch(hashTable, lgArrLongs_, hashIn);
-        if (foundIdx == -1) { continue; }
-        matchSet[matchSetCount++] = hashIn;
-      }
-    }
-    else {
-      //either unordered compact or hash table
-      for (int i = 0; i < arrLongsIn; i++ ) {
-        final long hashIn = cacheIn[i];
-        if (hashIn <= 0L || hashIn >= thetaLong_) { continue; }
-        final int foundIdx = hashSearch(hashTable, lgArrLongs_, hashIn);
-        if (foundIdx == -1) { continue; }
-        matchSet[matchSetCount++] = hashIn;
+        if (foundIdx != -1) {
+          matchSet[matchSetCount++] = hashIn;
+        }
+      } else {
+        if (isOrdered) { break; } // early stop
       }
     }
     //reduce effective array size to minimum
@@ -515,6 +503,32 @@ private void moveDataToTgt(final long[] arr, final int count) {
     assert tmpCnt == count : "Intersection Count Check: got: " + tmpCnt + ", expected: " + count;
   }
 
+  private void moveDataToTgt(final Sketch sketch) {
+    int count = sketch.getRetainedEntries();
+    int tmpCnt = 0;
+    if (wmem_ != null) { //Off Heap puts directly into mem
+      final int preBytes = CONST_PREAMBLE_LONGS << 3;
+      final int lgArrLongs = lgArrLongs_;
+      final long thetaLong = thetaLong_;
+      HashIterator it = sketch.iterator();
+      while (it.next()) {
+        final long hash = it.get();
+        if (continueCondition(thetaLong, hash)) { continue; }
+        hashInsertOnlyMemory(wmem_, lgArrLongs, hash, preBytes);
+        tmpCnt++;
+      }
+    } else { //On Heap. Assumes HT exists and is large enough
+      HashIterator it = sketch.iterator();
+      while (it.next()) {
+        final long hash = it.get();
+        if (continueCondition(thetaLong_, hash)) { continue; }
+        hashInsertOnly(hashTable_, lgArrLongs_, hash);
+        tmpCnt++;
+      }
+    }
+    assert tmpCnt == count : "Intersection Count Check: got: " + tmpCnt + ", expected: " + count;
+  }
+
   private void hardReset() {
     resetCommon();
     if (wmem_ != null) {