Skip to content

Frequent items: empty means zero stream weight, not zero active items #770

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.

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
leerho merged 1 commit into from
Sep 24, 2026
Merged

Frequent items: empty means zero stream weight, not zero active items #770

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
25 changes: 16 additions & 9 deletions src/main/java/org/apache/datasketches/frequencies/FrequentItemsSketch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import static org.apache.datasketches.frequencies.PreambleUtil.insertSerVer;
import static org.apache.datasketches.frequencies.Util.LG_MIN_MAP_SIZE;
import static org.apache.datasketches.frequencies.Util.SAMPLE_SIZE;
import static org.apache.datasketches.frequencies.Util.checkStreamWeight;

import java.lang.foreign.MemorySegment;
import java.lang.reflect.Array;
Expand Down Expand Up @@ -242,7 +243,7 @@ public static <T> FrequentItemsSketch<T> getInstance(final MemorySegment srcSeg,
final int familyID = extractFamilyID(pre0); //Byte 2
final int lgMaxMapSize = extractLgMaxMapSize(pre0); //Byte 3
final int lgCurMapSize = extractLgCurMapSize(pre0); //Byte 4
final boolean empty = (extractFlags(pre0) & EMPTY_FLAG_MASK) != 0; //Byte 5
final boolean emptyFlag = (extractFlags(pre0) & EMPTY_FLAG_MASK) != 0; //Byte 5

// Checks
final boolean preLongsEq1 = (preLongs == 1); //Byte 0
Expand All @@ -260,17 +261,19 @@ public static <T> FrequentItemsSketch<T> getInstance(final MemorySegment srcSeg,
throw new SketchesArgumentException(
"Possible Corruption: FamilyID must be " + actFamID + ": " + familyID);
}
if (empty ^ preLongsEq1) { //Byte 5 and Byte 0
if (emptyFlag ^ preLongsEq1) { //Byte 5 and Byte 0
throw new SketchesArgumentException(
"Possible Corruption: (PreLongs == 1) ^ Empty == True.");
"Possible Corruption: Empty flag does not match PreLongs: flags "
+ extractFlags(pre0) + ", preLongs " + preLongs);
}

if (empty) {
if (preLongsEq1) { //empty is determined by PreLongs
return new FrequentItemsSketch<>(lgMaxMapSize, LG_MIN_MAP_SIZE);
}
//get full preamble
final long[] preArr = new long[preLongs];
MemorySegment.copy(srcSeg, JAVA_LONG_UNALIGNED, 0, preArr, 0, preLongs);
checkStreamWeight(preArr[2]);

final FrequentItemsSketch<T> fis = new FrequentItemsSketch<>(lgMaxMapSize, lgCurMapSize);
fis.streamWeight = 0; //update after
Expand Down Expand Up @@ -448,12 +451,13 @@ public long getUpperBound(final T item) {
}

/**
* Returns true if this sketch is empty
* Returns true if this sketch is empty, that is, it has not been updated with any positive count.
* A sketch that is not empty may retain no items if a purge removed all of them.
*
* @return true if this sketch is empty
*/
public boolean isEmpty() {
return getNumActiveItems() == 0;
return streamWeight == 0;
}

/**
Expand Down Expand Up @@ -506,7 +510,8 @@ public byte[] toByteArray(final ArrayOfItemsSerDe<T> serDe) {
outBytes = 8;
} else {
preLongs = Family.FREQUENCY.getMaxPreLongs();
bytes = serDe.serializeToByteArray(hashMap.getActiveKeys());
//a purge may have removed all items
bytes = (activeItems > 0) ? serDe.serializeToByteArray(hashMap.getActiveKeys()) : new byte[0];
outBytes = ((preLongs + activeItems) << 3) + bytes.length;
}
final byte[] outArr = new byte[outBytes];
Expand All @@ -533,8 +538,10 @@ public byte[] toByteArray(final ArrayOfItemsSerDe<T> serDe) {
MemorySegment.copy(preArr, 0, seg, JAVA_LONG_UNALIGNED, 0, preLongs);

final int preBytes = preLongs << 3;
MemorySegment.copy(hashMap.getActiveValues(), 0, seg, JAVA_LONG_UNALIGNED, preBytes, activeItems);
MemorySegment.copy(bytes, 0, seg, JAVA_BYTE, preBytes + (activeItems << 3), bytes.length);
if (activeItems > 0) {
MemorySegment.copy(hashMap.getActiveValues(), 0, seg, JAVA_LONG_UNALIGNED, preBytes, activeItems);
MemorySegment.copy(bytes, 0, seg, JAVA_BYTE, preBytes + (activeItems << 3), bytes.length);
}
}
return outArr;
}
Expand Down
32 changes: 20 additions & 12 deletions src/main/java/org/apache/datasketches/frequencies/FrequentLongsSketch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import static org.apache.datasketches.frequencies.PreambleUtil.insertSerVer;
import static org.apache.datasketches.frequencies.Util.LG_MIN_MAP_SIZE;
import static org.apache.datasketches.frequencies.Util.SAMPLE_SIZE;
import static org.apache.datasketches.frequencies.Util.checkStreamWeight;

import java.lang.foreign.MemorySegment;
import java.util.ArrayList;
Expand Down Expand Up @@ -235,7 +236,7 @@ public static FrequentLongsSketch getInstance(final MemorySegment srcSeg) {
final int familyID = extractFamilyID(pre0); //Byte 2
final int lgMaxMapSize = extractLgMaxMapSize(pre0); //Byte 3
final int lgCurMapSize = extractLgCurMapSize(pre0); //Byte 4
final boolean empty = (extractFlags(pre0) & EMPTY_FLAG_MASK) != 0; //Byte 5
final boolean emptyFlag = (extractFlags(pre0) & EMPTY_FLAG_MASK) != 0; //Byte 5

// Checks
final boolean preLongsEq1 = (preLongs == 1); //Byte 0
Expand All @@ -253,17 +254,19 @@ public static FrequentLongsSketch getInstance(final MemorySegment srcSeg) {
throw new SketchesArgumentException(
"Possible Corruption: FamilyID must be " + actFamID + ": " + familyID);
}
if (empty ^ preLongsEq1) { //Byte 5 and Byte 0
if (emptyFlag ^ preLongsEq1) { //Byte 5 and Byte 0
throw new SketchesArgumentException(
"Possible Corruption: (PreLongs == 1) ^ Empty == True.");
"Possible Corruption: Empty flag does not match PreLongs: flags "
+ extractFlags(pre0) + ", preLongs " + preLongs);
}

if (empty) {
if (preLongsEq1) { //empty is determined by PreLongs
return new FrequentLongsSketch(lgMaxMapSize, LG_MIN_MAP_SIZE);
}
//get full preamble
final long[] preArr = new long[preLongs];
MemorySegment.copy(srcSeg, JAVA_LONG_UNALIGNED, 0, preArr, 0, preLongs);
checkStreamWeight(preArr[2]);

final FrequentLongsSketch fls = new FrequentLongsSketch(lgMaxMapSize, lgCurMapSize);
fls.streamWeight = 0; //update after
Expand Down Expand Up @@ -319,11 +322,13 @@ public static FrequentLongsSketch getInstance(final String string) {
throw new SketchesArgumentException("Possible Corruption: Bad SerVer: " + serVer);
}
Family.FREQUENCY.checkFamilyID(famID);
final boolean empty = flags > 0;
if (!empty && (numActive == 0)) {
final boolean emptyFlag = (flags & EMPTY_FLAG_MASK) != 0;
if (emptyFlag != (streamWt == 0)) {
throw new SketchesArgumentException(
"Possible Corruption: !Empty && NumActive=0; strLen: " + numActive);
"Possible Corruption: Empty flag does not match stream weight: flags "
+ flags + ", stream weight " + streamWt);
}
if (streamWt != 0) { checkStreamWeight(streamWt); }
final int numTokens = tokens.length;
if ((2 * numActive) != (numTokens - STR_PREAMBLE_TOKENS - 2)) {
throw new SketchesArgumentException(
Expand Down Expand Up @@ -499,12 +504,13 @@ public long getUpperBound(final long item) {
}

/**
* Returns true if this sketch is empty
* Returns true if this sketch is empty, that is, it has not been updated with any positive count.
* A sketch that is not empty may retain no items if a purge removed all of them.
*
* @return true if this sketch is empty
*/
public boolean isEmpty() {
return getNumActiveItems() == 0;
return streamWeight == 0;
}

/**
Expand Down Expand Up @@ -552,7 +558,7 @@ public String serializeToString() {
final int serVer = SER_VER; //0
final int famID = Family.FREQUENCY.getID(); //1
final int lgMaxMapSz = lgMaxMapSize; //2
final int flags = (hashMap.getNumActive() == 0) ? EMPTY_FLAG_MASK : 0; //3
final int flags = isEmpty() ? EMPTY_FLAG_MASK : 0; //3
final String fmt = "%d,%d,%d,%d,%d,%d,";
final String s =
String.format(fmt, serVer, famID, lgMaxMapSz, flags, streamWeight, offset);
Expand Down Expand Up @@ -602,8 +608,10 @@ public byte[] toByteArray() {
MemorySegment.copy(preArr, 0, seg, JAVA_LONG_UNALIGNED, 0, preLongs);

final int preBytes = preLongs << 3;
MemorySegment.copy(hashMap.getActiveValues(), 0, seg, JAVA_LONG_UNALIGNED, preBytes, activeItems);
MemorySegment.copy(hashMap.getActiveKeys(), 0, seg, JAVA_LONG_UNALIGNED, preBytes + (activeItems << 3), activeItems);
if (activeItems > 0) { //a purge may have removed all items
MemorySegment.copy(hashMap.getActiveValues(), 0, seg, JAVA_LONG_UNALIGNED, preBytes, activeItems);
MemorySegment.copy(hashMap.getActiveKeys(), 0, seg, JAVA_LONG_UNALIGNED, preBytes + (activeItems << 3), activeItems);
}
}
return outArr;
}
Expand Down
12 changes: 10 additions & 2 deletions src/main/java/org/apache/datasketches/frequencies/PreambleUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,9 +57,18 @@
* || 31 | 30 | 29 | 28 | 27 | 26 | 25 | 24 |
* 3 ||---------------------------------offset------------------------------------------|
* || 39 | 38 | 37 | 36 | 35 | 34 | 33 | 32 |
* 5 ||----------start of values buffer, followed by keys buffer------------------------|
* 4 ||----------start of values buffer, followed by keys buffer------------------------|
* </pre>
*
* <p>Emptiness is determined by PreambleLongs: 1 if the sketch is empty, 4 otherwise.
* A sketch is empty if its stream length is zero. A non-empty sketch may have zero active items
* if a purge removed all of them; it is serialized with the full preamble and no items.</p>
*
* <p>Flags (byte 5): only the empty flag is defined, as bits 0 and 2 (mask 0x05). Due to a
* historical mistake C++ and Java used different bits, so both are set when writing and either
* is accepted when reading. It is only checked for consistency with PreambleLongs.
* No other flag bits are defined.</p>
*
* @author Lee Rhodes
*/
final class PreambleUtil {
Expand All @@ -74,7 +83,6 @@ private PreambleUtil() {}
static final int LG_MAX_MAP_SIZE_BYTE = 3;
static final int LG_CUR_MAP_SIZE_BYTE = 4;
static final int FLAGS_BYTE = 5;
static final int SER_DE_ID_SHORT = 6; // to 7
static final int ACTIVE_ITEMS_INT = 8; // to 11 : 0 to 4 in pre1
static final int STREAMLENGTH_LONG = 16; // to 23 : pre2
static final int OFFSET_LONG = 24; // to 31 : pre3
Expand Down
14 changes: 14 additions & 0 deletions src/main/java/org/apache/datasketches/frequencies/Util.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@

package org.apache.datasketches.frequencies;

import org.apache.datasketches.common.SketchesArgumentException;

final class Util {

private Util() {}
Expand Down Expand Up @@ -51,4 +53,16 @@ static long hash(long key) {
return key;
}

/**
* Checks the stream weight read from the serialized image of a non-empty sketch.
* @param streamWeight the stream weight
* @throws SketchesArgumentException if the stream weight is not positive
*/
static void checkStreamWeight(final long streamWeight) {
if (streamWeight <= 0) {
throw new SketchesArgumentException(
"Possible Corruption: stream weight of a non-empty sketch must be positive: " + streamWeight);
}
}

}
60 changes: 58 additions & 2 deletions src/test/java/org/apache/datasketches/frequencies/FrequentItemsSketchCrossLanguageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,34 @@ public void generateBinariesForCompatibilityTestingStringsSketch() throws IOExce
}
}

// lgMaxMapSize=8 -> capacity 192; the 193rd distinct item triggers a purge whose
// median (1) removes every counter: not empty, with no retained items
@Test(groups = {GENERATE_JAVA_FILES}, priority = 0)
public void generateBinariesForCompatibilityTestingLongsSketchPurged() throws IOException {
final FrequentLongsSketch sk = new FrequentLongsSketch(1 << 8);
for (int i = 1; i <= 193; i++) {
sk.update(i);
}
assertFalse(sk.isEmpty());
assertEquals(sk.getNumActiveItems(), 0);
assertEquals(sk.getStreamLength(), 193);
assertEquals(sk.getMaximumError(), 1);
putBytesToJavaPath("frequent_long_purged_java.sk", sk.toByteArray());
}

@Test(groups = {GENERATE_JAVA_FILES}, priority = 0)
public void generateBinariesForCompatibilityTestingStringsSketchPurged() throws IOException {
final FrequentItemsSketch<String> sk = new FrequentItemsSketch<>(1 << 8);
for (int i = 1; i <= 193; i++) {
sk.update(Integer.toString(i));
}
assertFalse(sk.isEmpty());
assertEquals(sk.getNumActiveItems(), 0);
assertEquals(sk.getStreamLength(), 193);
assertEquals(sk.getMaximumError(), 1);
putBytesToJavaPath("frequent_string_purged_java.sk", sk.toByteArray(new ArrayOfStringsSerDe()));
}

@Test(groups = {GENERATE_JAVA_FILES}, priority = 0)
public void generateBinariesForCompatibilityTestingStringsSketchAscii() throws IOException {
final FrequentItemsSketch<String> sk = new FrequentItemsSketch<>(64);
Expand Down Expand Up @@ -99,6 +127,8 @@ public void generateBinariesForCompatibilityTestingStringsSketchUtf8() throws IO
public void checkJava() {
longs(GroupLanguage.JAVA);
strings(GroupLanguage.JAVA);
longsPurged(GroupLanguage.JAVA);
stringsPurged(GroupLanguage.JAVA);
stringsAscii(GroupLanguage.JAVA);
stringsUtf8(GroupLanguage.JAVA);
}
Expand All @@ -107,14 +137,18 @@ public void checkJava() {
public void checkCpp() {
longs(GroupLanguage.CPP);
strings(GroupLanguage.CPP);
stringsAscii(GroupLanguage.JAVA);
stringsUtf8(GroupLanguage.JAVA);
longsPurged(GroupLanguage.CPP);
stringsPurged(GroupLanguage.CPP);
stringsAscii(GroupLanguage.CPP);
stringsUtf8(GroupLanguage.CPP);
}

@Test(groups = {CHECK_GO_FILES})
public void checkGo() {
longs(GroupLanguage.GO);
strings(GroupLanguage.GO);
longsPurged(GroupLanguage.GO);
stringsPurged(GroupLanguage.GO);
stringsAscii(GroupLanguage.GO);
stringsUtf8(GroupLanguage.GO);
}
Expand Down Expand Up @@ -155,6 +189,28 @@ private static void strings(final GroupLanguage lang) {
}
}

private static void longsPurged(final GroupLanguage lang) {
final String fileName = "frequent_long_purged" + lang.sfx + ".sk";
final byte[] bytes = getFileBytes(lang.pth, fileName);
if (bytes.length == 0) { return; }
final FrequentLongsSketch sketch = FrequentLongsSketch.getInstance(MemorySegment.ofArray(bytes));
assertFalse(sketch.isEmpty());
assertEquals(sketch.getNumActiveItems(), 0);
assertEquals(sketch.getStreamLength(), 193);
assertEquals(sketch.getMaximumError(), 1);
}

private static void stringsPurged(final GroupLanguage lang) {
final String fileName = "frequent_string_purged" + lang.sfx + ".sk";
final byte[] bytes = getFileBytes(lang.pth, fileName);
if (bytes.length == 0) { return; }
final FrequentItemsSketch<String> sketch = FrequentItemsSketch.getInstance(MemorySegment.ofArray(bytes), new ArrayOfStringsSerDe());
assertFalse(sketch.isEmpty());
assertEquals(sketch.getNumActiveItems(), 0);
assertEquals(sketch.getStreamLength(), 193);
assertEquals(sketch.getMaximumError(), 1);
}

private static void stringsAscii(final GroupLanguage lang) {
final String fileName = "frequent_string_ascii" + lang.sfx + ".sk";
final byte[] bytes = getFileBytes(lang.pth, fileName);
Expand Down
Loading
Loading