New sketches are written in serialization version 2, the cross-language format defined in + * + * the DataSketches TCK. Version 2 hashes items with XxHash64, so its images give the same results + * in every language. A sketch read from a version 1 image keeps the version 1 hashing, which used + * MurmurHash3 and seeds from {@link Random}, and is written back as version 1. + * Version 1 and version 2 sketches can't be merged.
*/ public class CountMinSketch { + static final byte SERIAL_VERSION_1 = 1; + static final byte SERIAL_VERSION_2 = 2; + + private final byte serialVersion_; private final byte numHashes_; private final int numBuckets_; private final long seed_; @@ -64,17 +77,52 @@ int mask() { } /** - * Creates a CountMin sketch with given number of hash functions and buckets, - * and a user-specified seed. + * Creates an empty CountMin sketch with the given number of hash functions and buckets, + * and the default seed, {@link Util#DEFAULT_UPDATE_SEED}. + * The sketch uses serialization version 2. + * + *{@link #suggestNumHashes(double)} and {@link #suggestNumBuckets(double)} give suitable values + * for a desired confidence and relative error.
+ * + * @param numHashes The number of hash functions (rows), from 1 to 127 + * @param numBuckets The number of buckets per hash function, at least 3. + * The product of numHashes and numBuckets must be less than 230. + * @throws SketchesArgumentException if a parameter is out of range + */ + public CountMinSketch(final int numHashes, final int numBuckets) { + this(numHashes, numBuckets, Util.DEFAULT_UPDATE_SEED); + } + + /** + * Creates an empty CountMin sketch with the given number of hash functions and buckets, + * and the given seed. The sketch uses serialization version 2. + * + *{@link #suggestNumHashes(double)} and {@link #suggestNumBuckets(double)} give suitable values + * for a desired confidence and relative error. Sketches can be merged only if they use the same + * seed, and an image must be read back with the seed it was created with.
* - * @param numHashes The number of hash functions to apply to items + * @param numHashes The number of hash functions (rows), from 1 to 127 + * @param numBuckets The number of buckets per hash function, at least 3. + * The product of numHashes and numBuckets must be less than 230. + * @param seed The base hash seed + * @throws SketchesArgumentException if a parameter is out of range + */ + public CountMinSketch(final int numHashes, final int numBuckets, final long seed) { + this(numHashes, numBuckets, seed, SERIAL_VERSION_2); + } + + /** + * Creates a CountMin sketch that hashes items as the given serialization version does. + * + * @param numHashes The number of hash functions to apply to items, from 1 to 127 * @param numBuckets Array size for each of the hashing function * @param seed The base hash seed + * @param serialVersion 1 for the legacy hashing, 2 for the cross-language hashing */ - CountMinSketch(final byte numHashes, final int numBuckets, final long seed) { + private CountMinSketch(final int numHashes, final int numBuckets, final long seed, final byte serialVersion) { // Validate numHashes - if (numHashes <= 0) { - throw new SketchesArgumentException("Number of hash functions must be positive, got: " + numHashes); + if ((numHashes < 1) || (numHashes > Byte.MAX_VALUE)) { + throw new SketchesArgumentException("Number of hash functions must be from 1 to 127, got: " + numHashes); } // Validate numBuckets with clear mathematical justification @@ -102,16 +150,24 @@ int mask() { + "Consider reducing numHashes or numBuckets."); } - numHashes_ = numHashes; + serialVersion_ = serialVersion; + numHashes_ = (byte) numHashes; //stored as one byte in the image numBuckets_ = numBuckets; seed_ = seed; hashSeeds_ = new long[numHashes]; sketchArray_ = new long[(int) totalSize]; totalWeight_ = 0; - final Random rand = new Random(seed); - for (int i = 0; i < numHashes; i++) { - hashSeeds_[i] = rand.nextLong(); + if (serialVersion == SERIAL_VERSION_1) { + final Random rand = new Random(seed); + for (int i = 0; i < numHashes; i++) { + hashSeeds_[i] = rand.nextLong(); + } + } else { + // rowSeed[i] = XXH64(LE64(i), seed): XxHash64.hash(long, seed) hashes the 8 little-endian bytes of i + for (int i = 0; i < numHashes; i++) { + hashSeeds_[i] = hash(i, seed); + } } } @@ -128,13 +184,42 @@ private long[] getHashes(final byte[] item) { final long[] updateLocations = new long[numHashes_]; for (int i = 0; i < numHashes_; i++) { - final long[] index = MurmurHash3.hash(item, hashSeeds_[i]); - updateLocations[i] = (i * (long)numBuckets_) + Math.floorMod(index[0], numBuckets_); + updateLocations[i] = (i * (long) numBuckets_) + bucket(item, hashSeeds_[i]); } return updateLocations; } + private long[] getHashes(final long item) { + if (serialVersion_ == SERIAL_VERSION_1) { + return getHashes(longToBytes(item)); + } + final long[] updateLocations = new long[numHashes_]; + for (int i = 0; i < numHashes_; i++) { + // XxHash64.hash(long, seed) hashes the 8 little-endian bytes of the item, as version 2 requires + final long h = hash(item, hashSeeds_[i]); + updateLocations[i] = (i * (long) numBuckets_) + ((h >>> 1) % numBuckets_); + } + return updateLocations; + } + + private long bucket(final byte[] item, final long rowSeed) { + if (serialVersion_ == SERIAL_VERSION_1) { + return Math.floorMod(MurmurHash3.hash(item, rowSeed)[0], numBuckets_); + } + final long h = hashBytes(item, 0, item.length, rowSeed); + return (h >>> 1) % numBuckets_; + } + + /** + * Returns the serialization version this sketch hashes and serializes with: + * 2 for new sketches, or 1 for a sketch read from a version 1 image. + * @return the serialization version + */ + byte getSerialVersion() { + return serialVersion_; + } + /** * Checks if the CountMinSketch has processed any items. * @return True if the sketch is empty, otherwise false. @@ -214,7 +299,7 @@ public static int suggestNumBuckets(final double relativeError) { * @param weight The weight of the item. */ public void update(final long item, final long weight) { - update(longToBytes(item), weight); + updateLocations(getHashes(item), weight); } /** @@ -236,12 +321,14 @@ public void update(final String item, final long weight) { * @param weight The weight of the item. */ public void update(final byte[] item, final long weight) { - if (item.length == 0) { + if ((item == null) || (item.length == 0)) { return; } + updateLocations(getHashes(item), weight); + } + private void updateLocations(final long[] hashLocations, final long weight) { totalWeight_ += weight > 0 ? weight : -weight; - final long[] hashLocations = getHashes(item); for (final long h : hashLocations) { sketchArray_[(int) h] += weight; } @@ -253,7 +340,7 @@ public void update(final byte[] item, final long weight) { * @return Estimated frequency. */ public long getEstimate(final long item) { - return getEstimate(longToBytes(item)); + return estimateAt(getHashes(item)); } /** @@ -276,11 +363,13 @@ public long getEstimate(final String item) { * @return Estimated frequency. */ public long getEstimate(final byte[] item) { - if (item.length == 0) { + if ((item == null) || (item.length == 0)) { return 0; } + return estimateAt(getHashes(item)); + } - final long[] hashLocations = getHashes(item); + private long estimateAt(final long[] hashLocations) { long res = sketchArray_[(int) hashLocations[0]]; // Start from index 1 to avoid processing first element twice for (int i = 1; i < hashLocations.length; i++) { @@ -296,7 +385,7 @@ public long getEstimate(final byte[] item) { * @return Upper bound of estimated frequency. */ public long getUpperBound(final long item) { - return getUpperBound(longToBytes(item)); + return getEstimate(item) + (long) (getRelativeError() * getTotalWeight_()); } /** @@ -319,7 +408,7 @@ public long getUpperBound(final String item) { * @return Upper bound of estimated frequency. */ public long getUpperBound(final byte[] item) { - if (item.length == 0) { + if ((item == null) || (item.length == 0)) { return 0; } @@ -332,7 +421,7 @@ public long getUpperBound(final byte[] item) { * @return Lower bound of estimated frequency. */ public long getLowerBound(final long item) { - return getLowerBound(longToBytes(item)); + return getEstimate(item); } /** @@ -373,6 +462,10 @@ public void merge(final CountMinSketch other) { if (!acceptableConfig) { throw new SketchesException("Incompatible sketch configuration."); } + if (serialVersion_ != other.serialVersion_) { + throw new SketchesException("Cannot merge sketches with different serialization versions: " + + serialVersion_ + " and " + other.serialVersion_); + } for (int i = 0; i < sketchArray_.length; i++) { sketchArray_[i] += other.sketchArray_[i]; @@ -404,8 +497,7 @@ public byte[] toByteArray() { // Long 0 final int preambleLongs = Family.COUNTMIN.getMinPreLongs(); posSeg.setByte((byte) preambleLongs); - final int serialVersion = 1; - posSeg.setByte((byte) serialVersion); + posSeg.setByte(serialVersion_); final int familyId = Family.COUNTMIN.getID(); posSeg.setByte((byte) familyId); final int flagsByte = isEmpty() ? Flag.IS_EMPTY.mask() : 0; @@ -457,10 +549,9 @@ public static CountMinSketch deserialize(final byte[] b, final long seed) { throw new SketchesArgumentException("Preamble longs mismatch: expected " + expectedPreambleLongs + ", actual " + preambleLongs); } - final int expectedSerialVersion = 1; - if (serialVersion != expectedSerialVersion) { - throw new SketchesArgumentException("Serial version mismatch: expected " + expectedSerialVersion - + ", actual " + serialVersion); + if ((serialVersion != SERIAL_VERSION_1) && (serialVersion != SERIAL_VERSION_2)) { + throw new SketchesArgumentException("Serial version mismatch: expected " + SERIAL_VERSION_1 + + " or " + SERIAL_VERSION_2 + ", actual " + serialVersion); } final int expectedFamilyId = Family.COUNTMIN.getID(); if (familyId != expectedFamilyId) { @@ -483,7 +574,7 @@ public static CountMinSketch deserialize(final byte[] b, final long seed) { final long arrayBytes = (long) numHashes * numBuckets * Long.BYTES; Util.checkBounds(preambleBytes, Long.BYTES + Math.max(arrayBytes, 0), b.length); } - final CountMinSketch cms = new CountMinSketch(numHashes, numBuckets, seed); + final CountMinSketch cms = new CountMinSketch(numHashes, numBuckets, seed, serialVersion); if (empty) { return cms; } diff --git a/src/test/java/org/apache/datasketches/count/CountMinSketchCrossLanguageTest.java b/src/test/java/org/apache/datasketches/count/CountMinSketchCrossLanguageTest.java new file mode 100644 index 000000000..14ce808e8 --- /dev/null +++ b/src/test/java/org/apache/datasketches/count/CountMinSketchCrossLanguageTest.java @@ -0,0 +1,133 @@ +/* + * 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.count; + +import static org.apache.datasketches.common.UtilityIO.CHECK_CPP_FILES; +import static org.apache.datasketches.common.UtilityIO.CHECK_GO_FILES; +import static org.apache.datasketches.common.UtilityIO.CHECK_JAVA_FILES; +import static org.apache.datasketches.common.UtilityIO.CHECK_RUST_FILES; +import static org.apache.datasketches.common.UtilityIO.GENERATE_JAVA_FILES; +import static org.apache.datasketches.common.UtilityIO.getFileBytes; +import static org.apache.datasketches.common.UtilityIO.putBytesToJavaPath; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +import org.apache.datasketches.common.UtilityIO.GroupLanguage; +import org.testng.annotations.Test; + +/** + * Generates and checks the Count-Min version 2 cross-language files defined in section 8 of the + * TCK spec docs/count-min-format.md. + */ +public class CountMinSketchCrossLanguageTest { + private static final byte NUM_HASHES = 3; + private static final int NUM_BUCKETS = 1000; + + /** One case from the table in section 8.1 of the spec. */ + private record Case(String name, long seed, int n) { } + + private static final Case[] CASES = { + new Case("n0", 9001L, 0), + new Case("n1", 9001L, 1), + new Case("n10", 9001L, 10), + new Case("n100", 9001L, 100), + new Case("n1000", 9001L, 1000), + new Case("n10000", 9001L, 10000), + new Case("seed42_n1000", 42L, 1000) }; + + private static String fileName(final Case testCase, final GroupLanguage lang) { + return "count_min_v2_" + testCase.name() + lang.sfx + ".sk"; + } + + static CountMinSketch build(final Case testCase) { + final int n = testCase.n(); + final CountMinSketch cms = new CountMinSketch(NUM_HASHES, NUM_BUCKETS, testCase.seed()); + for (int i = 0; i < n; i++) { + cms.update(i, i + 1L); + cms.update(Integer.toString(i), 1); + } + return cms; + } + + @Test(groups = {GENERATE_JAVA_FILES}, priority = 0) + public void generateCountMinBinaries() { + for (final Case testCase : CASES) { + putBytesToJavaPath(fileName(testCase, GroupLanguage.JAVA), build(testCase).toByteArray()); + } + } + + @Test(groups = {CHECK_JAVA_FILES}, priority = 1) + public void checkJava() { + checkCountMinBinaries(GroupLanguage.JAVA); + } + + @Test(groups = {CHECK_CPP_FILES}) + public void checkCpp() { + checkCountMinBinaries(GroupLanguage.CPP); + checkCppVersion1Binaries(); + } + + @Test(groups = {CHECK_GO_FILES}) + public void checkGo() { + checkCountMinBinaries(GroupLanguage.GO); + } + + @Test(groups = {CHECK_RUST_FILES}) + public void checkRust() { + checkCountMinBinaries(GroupLanguage.RUST); + } + + private static void checkCountMinBinaries(final GroupLanguage lang) { + for (final Case testCase : CASES) { + final byte[] bytes = getFileBytes(lang.pth, fileName(testCase, lang)); + if (bytes.length == 0) { continue; } + final long seed = testCase.seed(); + final int n = testCase.n(); + final CountMinSketch cms = CountMinSketch.deserialize(bytes, seed); + + assertEquals(cms.getSerialVersion(), CountMinSketch.SERIAL_VERSION_2, testCase.name()); + assertEquals(cms.getNumHashes_(), NUM_HASHES, testCase.name()); + assertEquals(cms.getNumBuckets_(), NUM_BUCKETS, testCase.name()); + assertEquals(cms.isEmpty(), n == 0, testCase.name()); + assertEquals(cms.getTotalWeight_(), ((long) n * (n + 1)) / 2 + n, testCase.name()); + for (int i = 0; i < n; i++) { + final String s = Integer.toString(i); + assertTrue(cms.getEstimate(i) >= (i + 1L), testCase.name() + " integer " + i); + assertTrue(cms.getEstimate(i) <= cms.getUpperBound(i), testCase.name() + " integer " + i); + assertTrue(cms.getEstimate(s) >= 1, testCase.name() + " string " + s); + assertTrue(cms.getEstimate(s) <= cms.getUpperBound(s), testCase.name() + " string " + s); + } + // Version 2 is deterministic, so the image must equal the one Java builds for the same case + assertEquals(bytes, build(testCase).toByteArray(), testCase.name()); + } + } + + // Section 8.3: the C++ version 1 files deserialize as version 1. Their estimates aren't checked, + // because version 1 hashing differs between languages. + private static void checkCppVersion1Binaries() { + for (final String name : new String[] { "count_min_empty_cpp.sk", "count_min_non_empty_cpp.sk" }) { + final byte[] bytes = getFileBytes(GroupLanguage.CPP.pth, name); + if (bytes.length == 0) { continue; } + final CountMinSketch cms = CountMinSketch.deserialize(bytes, org.apache.datasketches.common.Util.DEFAULT_UPDATE_SEED); + assertEquals(cms.getSerialVersion(), CountMinSketch.SERIAL_VERSION_1, name); + assertEquals(cms.isEmpty(), name.startsWith("count_min_empty"), name); + } + } +} diff --git a/src/test/java/org/apache/datasketches/count/CountMinSketchTest.java b/src/test/java/org/apache/datasketches/count/CountMinSketchTest.java index 59590d9a6..45f99b34f 100644 --- a/src/test/java/org/apache/datasketches/count/CountMinSketchTest.java +++ b/src/test/java/org/apache/datasketches/count/CountMinSketchTest.java @@ -24,18 +24,24 @@ import static org.testng.Assert.assertThrows; import static org.testng.Assert.assertTrue; +import java.nio.charset.StandardCharsets; +import java.util.HexFormat; +import java.util.Random; + import org.apache.datasketches.common.AllocationTracker; import org.apache.datasketches.common.ByteArrayUtil; import org.apache.datasketches.common.SketchesArgumentException; import org.apache.datasketches.common.SketchesException; +import org.apache.datasketches.hash.MurmurHash3; +import org.apache.datasketches.hash.XxHash64; import org.testng.annotations.Test; public class CountMinSketchTest { @Test public void createNewCountMinSketchTest() throws Exception { - assertThrows(SketchesArgumentException.class, () -> new CountMinSketch((byte) 5, 1, 123)); - assertThrows(SketchesArgumentException.class, () -> new CountMinSketch((byte) 4, (1 << 28), 123)); + assertThrows(SketchesArgumentException.class, () -> new CountMinSketch(5, 1, 123)); + assertThrows(SketchesArgumentException.class, () -> new CountMinSketch(4, (1 << 28), 123)); final byte numHashes = 3; final int numBuckets = 5; final long seed = 1234567; @@ -104,7 +110,7 @@ public void countMinSketchOneUpdateTest() { @Test public void frequencyCancellationTest() { - final CountMinSketch c = new CountMinSketch((byte) 1, 5, 123456); + final CountMinSketch c = new CountMinSketch(1, 5, 123456); c.update("x", 1); c.update("y", -1); assertEquals(c.getTotalWeight_(), 2); @@ -155,7 +161,7 @@ public void mergeFailTest() { assertThrows("Cannot merge a sketch with itself.", SketchesException.class, () -> s.merge(s)); - final CountMinSketch s1 = new CountMinSketch((byte) (numHashes + 1), numBuckets, seed); + final CountMinSketch s1 = new CountMinSketch(numHashes + 1, numBuckets, seed); final CountMinSketch s2 = new CountMinSketch(numHashes, numBuckets + 1, seed); final CountMinSketch s3 = new CountMinSketch(numHashes, numBuckets, seed + 1); @@ -244,8 +250,8 @@ public void serializeDeserializeTest() { @Test public void emptyImageIsUsable() { final long seed = 123L; - final CountMinSketch cms = CountMinSketch.deserialize(new CountMinSketch((byte) 3, 5, seed).toByteArray(), seed); - final CountMinSketch other = new CountMinSketch((byte) 3, 5, seed); + final CountMinSketch cms = CountMinSketch.deserialize(new CountMinSketch(3, 5, seed).toByteArray(), seed); + final CountMinSketch other = new CountMinSketch(3, 5, seed); cms.merge(other); //both empty assertTrue(cms.isEmpty()); other.update(1L, 2); @@ -260,7 +266,7 @@ public void emptyImageIsUsable() { @Test public void truncatedImageThrows() { final long seed = 123L; - final CountMinSketch cms = new CountMinSketch((byte) 3, 5, seed); + final CountMinSketch cms = new CountMinSketch(3, 5, seed); cms.update(1L, 1); final byte[] bytes = cms.toByteArray(); ByteArrayUtil.putIntLE(bytes, 8, 1 << 28); //numBuckets larger than the image holds @@ -271,4 +277,213 @@ public void truncatedImageThrows() { assertThrows(SketchesArgumentException.class, () -> CountMinSketch.deserialize(shortBytes, seed)); } + // ---- Serialization version 2: test vectors from the TCK spec docs/count-min-format.md, section 7 ---- + + private static final long SPEC_SEED = 9001L; + + // 7.1 XXH64 and 7.2 row seeds + @Test + public void v2XxHashAndRowSeedVectors() { + final byte[] empty = new byte[0]; + assertEquals(XxHash64.hashBytes(empty, 0, 0, 0), 0xEF46DB3751D8E999L); + assertEquals(XxHash64.hashBytes(utf8("a"), 0, 1, 0), 0xD24EC4F1A98C6E5BL); + assertEquals(XxHash64.hashBytes(utf8("abc"), 0, 3, 0), 0x44BC2CF5AD770999L); + + final long[] seeds9001 = { 0xD71D64972584A288L, 0xEFE8454A6499F311L, 0x23819BE21B1B1D9EL }; + final long[] seeds42 = { 0xB71B47EBDA15746CL, 0x9ED50FD59358D232L, 0xD19FE30002B9A3CDL }; + for (int i = 0; i < 3; i++) { + assertEquals(XxHash64.hash(i, 9001L), seeds9001[i]); + assertEquals(XxHash64.hash(i, 42L), seeds42[i]); + } + } + + // 7.3 item buckets, observed through the counters a single update sets + @Test + public void v2BucketVectors() { + assertBuckets(SPEC_SEED, 1L, 185, 898, 4); + assertBuckets(SPEC_SEED, 9L, 490, 217, 440); + assertBuckets(SPEC_SEED, -1L, 256, 641, 308); + assertBuckets(SPEC_SEED, Long.MIN_VALUE, 951, 0, 298); + assertBuckets(SPEC_SEED, "apple", 130, 747, 144); + assertBuckets(SPEC_SEED, "\u00e9t\u00e9", 806, 491, 163); + assertBuckets(42L, 1L, 468, 459, 696); + assertBuckets(42L, 9L, 841, 428, 322); + } + + // 7.4 images + @Test + public void v2ImageVectors() { + final CountMinSketch empty = new CountMinSketch(2, 9, SPEC_SEED); + assertEquals(empty.getSerialVersion(), CountMinSketch.SERIAL_VERSION_2); + assertEquals(empty.toByteArray(), hex("02021201000000000900000002cc9300")); + + final CountMinSketch cms = new CountMinSketch(2, 9, SPEC_SEED); + cms.update(1L, 10); + cms.update(9L, 810); + cms.update("apple", 5); + final byte[] expected = new byte[168]; + System.arraycopy(hex("02021200000000000900000002cc9300"), 0, expected, 0, 16); + ByteArrayUtil.putLongLE(expected, 16, 825); + ByteArrayUtil.putLongLE(expected, 24 + (8 * 0), 810); // row 0, bucket 0 + ByteArrayUtil.putLongLE(expected, 24 + (8 * 4), 5); // row 0, bucket 4 + ByteArrayUtil.putLongLE(expected, 24 + (8 * 6), 10); // row 0, bucket 6 + ByteArrayUtil.putLongLE(expected, 24 + (8 * 10), 5); // row 1, bucket 1 + ByteArrayUtil.putLongLE(expected, 24 + (8 * 11), 810); // row 1, bucket 2 + ByteArrayUtil.putLongLE(expected, 24 + (8 * 16), 10); // row 1, bucket 7 + assertEquals(cms.toByteArray(), expected); + + final CountMinSketch read = CountMinSketch.deserialize(expected, SPEC_SEED); + assertEquals(read.getSerialVersion(), CountMinSketch.SERIAL_VERSION_2); + assertEquals(read.getEstimate(1L), 10); + assertEquals(read.getEstimate(9L), 810); + assertEquals(read.getEstimate("apple"), 5); + assertEquals(read.toByteArray(), expected); + } + + // The long overloads hash the 8 little-endian bytes of the item, like the byte[] overloads + @Test + public void v2LongMatchesLittleEndianBytes() { + final CountMinSketch a = new CountMinSketch(4, 101, SPEC_SEED); + final CountMinSketch b = new CountMinSketch(4, 101, SPEC_SEED); + final long[] items = { 0L, 1L, -1L, Long.MIN_VALUE, Long.MAX_VALUE, 0x0123456789ABCDEFL }; + for (final long item : items) { + final byte[] le = new byte[Long.BYTES]; + ByteArrayUtil.putLongLE(le, 0, item); + a.update(item, 3); + b.update(le, 3); + } + assertEquals(a.toByteArray(), b.toByteArray()); + } + + // Version 1 images keep the legacy hashing and are written back as version 1 + @Test + public void v1ImageKeepsLegacyHashing() { + final long seed = 1234L; + final byte numHashes = 3; + final int numBuckets = 17; + final byte[] emptyV1 = new CountMinSketch(numHashes, numBuckets, seed).toByteArray(); + emptyV1[1] = CountMinSketch.SERIAL_VERSION_1; + + final CountMinSketch v1 = CountMinSketch.deserialize(emptyV1, seed); + assertEquals(v1.getSerialVersion(), CountMinSketch.SERIAL_VERSION_1); + v1.update(7L, 11); + v1.update("legacy", 13); + + final byte[] image = v1.toByteArray(); + assertEquals(image[1], CountMinSketch.SERIAL_VERSION_1); + + // Independent version 1 computation: seeds from Random(seed), MurmurHash3, floorMod, native byte order + final Random rand = new Random(seed); + final long[] rowSeeds = new long[numHashes]; + for (int i = 0; i < numHashes; i++) { rowSeeds[i] = rand.nextLong(); } + final byte[] item7 = java.nio.ByteBuffer.allocate(8).order(java.nio.ByteOrder.nativeOrder()).putLong(7L).array(); + final long[] expected = new long[numHashes * numBuckets]; + for (int i = 0; i < numHashes; i++) { + expected[(i * numBuckets) + Math.floorMod(MurmurHash3.hash(item7, rowSeeds[i])[0], numBuckets)] += 11; + expected[(i * numBuckets) + Math.floorMod(MurmurHash3.hash(utf8("legacy"), rowSeeds[i])[0], numBuckets)] += 13; + } + for (int k = 0; k < expected.length; k++) { + assertEquals(ByteArrayUtil.getLongLE(image, 24 + (8 * k)), expected[k], "counter " + k); + } + + final CountMinSketch reread = CountMinSketch.deserialize(image, seed); + assertEquals(reread.getSerialVersion(), CountMinSketch.SERIAL_VERSION_1); + assertEquals(reread.getEstimate(7L), v1.getEstimate(7L)); + assertEquals(reread.getEstimate("legacy"), v1.getEstimate("legacy")); + } + + @Test + public void mergeAcrossVersionsThrows() { + final long seed = 1234L; + final byte[] emptyV1 = new CountMinSketch(3, 17, seed).toByteArray(); + emptyV1[1] = CountMinSketch.SERIAL_VERSION_1; + final CountMinSketch v1 = CountMinSketch.deserialize(emptyV1, seed); + final CountMinSketch v2 = new CountMinSketch(3, 17, seed); + assertThrows(SketchesException.class, () -> v1.merge(v2)); + assertThrows(SketchesException.class, () -> v2.merge(v1)); + } + + @Test + public void unknownSerialVersionThrows() { + final long seed = 1234L; + final byte[] image = new CountMinSketch(3, 17, seed).toByteArray(); + image[1] = 3; + assertThrows(SketchesArgumentException.class, () -> CountMinSketch.deserialize(image, seed)); + image[1] = 0; + assertThrows(SketchesArgumentException.class, () -> CountMinSketch.deserialize(image, seed)); + } + + @Test + public void nullAndEmptyItemsAreIgnored() { + final CountMinSketch cms = new CountMinSketch(3, 17, SPEC_SEED); + cms.update((byte[]) null, 5); + cms.update(new byte[0], 5); + cms.update((String) null, 5); + cms.update("", 5); + assertTrue(cms.isEmpty()); + assertEquals(cms.getEstimate((byte[]) null), 0); + assertEquals(cms.getEstimate(new byte[0]), 0); + assertEquals(cms.getEstimate((String) null), 0); + assertEquals(cms.getEstimate(""), 0); + assertEquals(cms.getUpperBound((byte[]) null), 0); + } + + @Test + public void publicConstructorsUseDefaultSeedAndVersion2() { + final CountMinSketch dflt = new CountMinSketch(2, 9); + assertEquals(dflt.getSeed_(), org.apache.datasketches.common.Util.DEFAULT_UPDATE_SEED); + assertEquals(dflt.getSerialVersion(), CountMinSketch.SERIAL_VERSION_2); + assertEquals(dflt.toByteArray(), new CountMinSketch(2, 9, SPEC_SEED).toByteArray()); + + final byte numHashes = CountMinSketch.suggestNumHashes(0.95); + final int numBuckets = CountMinSketch.suggestNumBuckets(0.01); + final CountMinSketch cms = new CountMinSketch(numHashes, numBuckets, 42L); + cms.update("x", 3); + assertEquals(cms.getEstimate("x"), 3); + assertThrows(SketchesArgumentException.class, () -> new CountMinSketch(0, 9)); + assertThrows(SketchesArgumentException.class, () -> new CountMinSketch(2, 2)); + } + + @Test + public void numHashesRange() { + // numHashes is an int in the constructors but is stored as one byte, so it must be from 1 to 127 + for (final int bad : new int[] {0, -1, 128, 256 + 3}) { + assertThrows(SketchesArgumentException.class, () -> new CountMinSketch(bad, 9)); + } + final CountMinSketch max = new CountMinSketch(127, 3); + assertEquals(max.getNumHashes_(), (byte) 127); + final CountMinSketch copy = CountMinSketch.deserialize(max.toByteArray(), max.getSeed_()); + assertEquals(copy.getNumHashes_(), (byte) 127); + } + + private static void assertBuckets(final long seed, final long item, final int... buckets) { + final CountMinSketch cms = new CountMinSketch(3, 1000, seed); + cms.update(item, 1); + assertCounters(cms, buckets); + } + + private static void assertBuckets(final long seed, final String item, final int... buckets) { + final CountMinSketch cms = new CountMinSketch(3, 1000, seed); + cms.update(item, 1); + assertCounters(cms, buckets); + } + + private static void assertCounters(final CountMinSketch cms, final int... buckets) { + final byte[] image = cms.toByteArray(); + for (int row = 0; row < buckets.length; row++) { + for (int b = 0; b < 1000; b++) { + final long counter = ByteArrayUtil.getLongLE(image, 24 + (8 * ((row * 1000) + b))); + assertEquals(counter, b == buckets[row] ? 1 : 0, "row " + row + ", bucket " + b); + } + } + } + + private static byte[] utf8(final String s) { + return s.getBytes(StandardCharsets.UTF_8); + } + + private static byte[] hex(final String s) { + return HexFormat.of().parseHex(s); + } + }