Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Jan 28, 2025
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 27 additions & 7 deletions src/main/java/org/apache/datasketches/theta/AnotBimpl.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand All @@ -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);
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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
18 changes: 7 additions & 11 deletions src/main/java/org/apache/datasketches/theta/CompactSketch.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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)) {
Expand Down Expand Up @@ -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();
Expand All @@ -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);
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
/*
* 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 with the order bit properly set.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Rather than "properly set" maybe just say it must be ordered and with the bit set to true?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this refers to memory byte order. Perhaps can be rephrased for clarity.

*/
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. The order bit must be set properly.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

again

* @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);
}

@Override
public int getRetainedEntries(final boolean valid) { //compact is always valid
final int preLongs = extractPreLongs(mem_);
final int numEntriesBytes = extractNumEntriesBytesV4(mem_);
int offsetBytes = preLongs > 1 ? 16 : 8;
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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;
Expand Down
59 changes: 36 additions & 23 deletions src/main/java/org/apache/datasketches/theta/IntersectionImpl.java
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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_;
Expand All @@ -448,27 +446,16 @@ 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!
}
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 (sketchIn.isOrdered()) { break; } // early stop
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This won't change over the course of the intersection. Should it be a flag pre-computed in advance to avoid a function call every time? Not sure if the JVM can easily determine there's no chance of a side-effect from other code.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Generally I prefer not to introduce aliases, but if you think this can help, I can do that. Perhaps we could try measuring the effect one day.

}
}
//reduce effective array size to minimum
Expand Down Expand Up @@ -515,6 +502,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) {
Expand Down
Loading
Loading