Skip to content

Commit 325574e

Browse files
authored
Merge pull request #759 from permutive-engineering/perf/cpc-union-memory-segment
Add CpcUnion.update(MemorySegment)
2 parents 3a1bf7f + 509903e commit 325574e

3 files changed

Lines changed: 247 additions & 7 deletions

File tree

‎src/main/java/org/apache/datasketches/cpc/CpcCompression.java‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -486,19 +486,18 @@ private static void compressTheWindow(final CompressedState target, final CpcSke
486486
target.cwStream = windowBuf; //avoid extra copy
487487
}
488488

489-
private static void uncompressTheWindow(final CpcSketch target, final CompressedState source) {
489+
static byte[] uncompressTheWindow(final CompressedState source) {
490490
final int srcLgK = source.lgK;
491491
final int srcK = 1 << srcLgK;
492492
final byte[] window = new byte[srcK];
493493
// bzero ((void *) window, (size_t) k); // zeroing not needed here (unlike the Hybrid Flavor)
494-
assert (target.slidingWindow == null);
495-
target.slidingWindow = window;
496494
final int pseudoPhase = determinePseudoPhase(srcLgK, source.numCoupons);
497495
assert (source.cwStream != null);
498-
lowLevelUncompressBytes(target.slidingWindow, srcK,
496+
lowLevelUncompressBytes(window, srcK,
499497
decodingTablesForHighEntropyByte[pseudoPhase],
500498
source.cwStream,
501499
source.cwLengthInts);
500+
return window;
502501
}
503502

504503
private static void compressTheSurprisingValues(final CompressedState target, final CpcSketch source,
@@ -521,7 +520,7 @@ private static void compressTheSurprisingValues(final CompressedState target, fi
521520

522521
//allocates and returns an array of uncompressed pairs.
523522
//the length of this array is known to the source sketch.
524-
private static int[] uncompressTheSurprisingValues(final CompressedState source) {
523+
static int[] uncompressTheSurprisingValues(final CompressedState source) {
525524
final int srcK = 1 << source.lgK;
526525
final int numPairs = source.numCsv;
527526
assert numPairs > 0;
@@ -663,7 +662,8 @@ private static void compressPinnedFlavor(final CompressedState target, final Cpc
663662

664663
private static void uncompressPinnedFlavor(final CpcSketch target, final CompressedState source) {
665664
assert (source.cwStream != null);
666-
uncompressTheWindow(target, source);
665+
assert (target.slidingWindow == null);
666+
target.slidingWindow = uncompressTheWindow(source);
667667
final int srcLgK = source.lgK;
668668
final int numPairs = source.numCsv;
669669
if (numPairs == 0) {
@@ -724,7 +724,8 @@ private static void compressSlidingFlavor(final CompressedState target, final Cp
724724

725725
private static void uncompressSlidingFlavor(final CpcSketch target, final CompressedState source) {
726726
assert (source.cwStream != null);
727-
uncompressTheWindow(target, source);
727+
assert (target.slidingWindow == null);
728+
target.slidingWindow = uncompressTheWindow(source);
728729
final int srcLgK = source.lgK;
729730
final int numPairs = source.numCsv;
730731
if (numPairs == 0) {

‎src/main/java/org/apache/datasketches/cpc/CpcUnion.java‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
import static org.apache.datasketches.cpc.Flavor.EMPTY;
2525
import static org.apache.datasketches.cpc.Flavor.SPARSE;
2626

27+
import java.lang.foreign.MemorySegment;
28+
2729
import org.apache.datasketches.common.Family;
2830
import org.apache.datasketches.common.SketchesArgumentException;
2931
import org.apache.datasketches.common.SketchesStateException;
@@ -135,6 +137,20 @@ public void update(final CpcSketch sketch) {
135137
mergeInto(this, sketch);
136138
}
137139

140+
/**
141+
* Update this union with a serialized CpcSketch image.
142+
*
143+
* <p>Where possible the image's coupons are decoded straight into this union's bit matrix,
144+
* which avoids building an intermediate sketch only to walk it once. Images that cannot be
145+
* decoded that way are uncompressed into a sketch and merged as usual, so the result is the
146+
* same either way.</p>
147+
*
148+
* @param seg the given MemorySegment holding a serialized CpcSketch image.
149+
*/
150+
public void update(final MemorySegment seg) {
151+
mergeInto(this, CompressedState.importFromSegment(seg));
152+
}
153+
138154
/**
139155
* Returns the result of union operations as a CPC sketch.
140156
* @return the result of union operations as a CPC sketch.
@@ -201,6 +217,15 @@ private static void walkTableUpdatingSketch(final CpcSketch dest, final PairTabl
201217
}
202218
}
203219

220+
private static void orPairsIntoMatrix(final long[] destMatrix, final int destLgK,
221+
final int[] pairs, final int numPairs, final int colShift) {
222+
final int destMask = (1 << destLgK) - 1; // downsamples when destlgK < srcLgK
223+
for (int i = 0; i < numPairs; i++) {
224+
final int rowCol = pairs[i];
225+
destMatrix[(rowCol >>> 6) & destMask] |= 1L << ((rowCol & 63) + colShift);
226+
}
227+
}
228+
204229
private static void orTableIntoMatrix(final long[] bitMatrix, final int destLgK, final PairTable table) {
205230
final int[] slots = table.getSlotsArr();
206231
final int numSlots = 1 << table.getLgSizeInts();
@@ -275,6 +300,53 @@ private static void reduceUnionK(final CpcUnion union, final int newLgK) {
275300
}
276301
}
277302

303+
private static void mergeInto(final CpcUnion union, final CompressedState source) {
304+
Util.checkSeedHashes(Util.computeSeedHash(union.seed), source.seedHash);
305+
306+
if (source.numCoupons == 0) { return; } //EMPTY
307+
308+
//Accumulator and bitMatrix must be mutually exclusive,
309+
//so bitMatrix != null => accumulator == null and visa versa
310+
//if (Accumulator != null) union must be EMPTY or SPARSE,
311+
checkUnionState(union);
312+
313+
if (source.lgK < union.lgK) { reduceUnionK(union, source.lgK); }
314+
315+
// if source is past SPARSE mode, make sure that union is a bitMatrix.
316+
if ((source.getFlavor().ordinal() > 1) && (union.accumulator != null)) {
317+
union.bitMatrix = CpcUtil.bitMatrixOfSketch(union.accumulator);
318+
union.accumulator = null;
319+
}
320+
321+
//The source's coupons can be decoded straight into the union's bitMatrix, which avoids
322+
//building an intermediate sketch only to walk it once. Flavors that cannot be decoded that
323+
//way leave the switch without having touched the bitMatrix, and go through a sketch below.
324+
if (union.bitMatrix != null) {
325+
final int numPairs = source.numCsv;
326+
switch (source.getFlavor()) {
327+
case SPARSE : //pairs are the whole sketch
328+
case HYBRID : { //the window sits at column offset 0, so the pairs need no shift
329+
orPairsIntoMatrix(union.bitMatrix, union.lgK,
330+
CpcCompression.uncompressTheSurprisingValues(source), numPairs, 0);
331+
return;
332+
}
333+
case PINNED : { //window at column offset 0, pairs shifted up 8 by the compressor
334+
orWindowIntoMatrix(union.bitMatrix, union.lgK,
335+
CpcCompression.uncompressTheWindow(source), 0, source.lgK);
336+
if (numPairs > 0) {
337+
orPairsIntoMatrix(union.bitMatrix, union.lgK,
338+
CpcCompression.uncompressTheSurprisingValues(source), numPairs, 8);
339+
}
340+
return;
341+
}
342+
//Sliding inverts its logic, so a coupon can be signalled by the ABSENCE of a pair.
343+
//EMPTY cannot arrive here, having returned above.
344+
default : break;
345+
}
346+
}
347+
mergeInto(union, CpcSketch.uncompress(source, union.seed));
348+
}
349+
278350
private static void mergeInto(final CpcUnion union, final CpcSketch source) {
279351
if (source == null) { return; }
280352
checkSeeds(union.seed, source.seed);
Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.datasketches.cpc;
21+
22+
import static org.testng.Assert.assertEquals;
23+
import static org.testng.Assert.fail;
24+
25+
import java.lang.foreign.MemorySegment;
26+
import java.util.Random;
27+
28+
import org.apache.datasketches.common.SketchesArgumentException;
29+
import org.apache.datasketches.common.Util;
30+
import org.testng.annotations.Test;
31+
32+
/**
33+
* update(MemorySegment) decodes into the union's bit matrix where it can and uncompresses a
34+
* sketch where it cannot, so it must give the same result as update(heapify(seg)) in every case.
35+
*/
36+
public class CpcUnionSegmentUpdateTest {
37+
38+
private static byte[] sketchBytes(final int lgK, final long from, final long to) {
39+
final CpcSketch sk = new CpcSketch(lgK);
40+
for (long i = from; i < to; i++) { sk.update(i); }
41+
return sk.toByteArray();
42+
}
43+
44+
/** Feeds the same images to both paths and requires byte-identical results. */
45+
private static void assertSamePath(final String what, final int unionLgK, final byte[][] images) {
46+
final CpcUnion viaSegment = new CpcUnion(unionLgK);
47+
final CpcUnion viaSketch = new CpcUnion(unionLgK);
48+
for (final byte[] image : images) {
49+
viaSegment.update(MemorySegment.ofArray(image));
50+
viaSketch.update(CpcSketch.heapify(MemorySegment.ofArray(image)));
51+
}
52+
final CpcSketch a = viaSegment.getResult();
53+
final CpcSketch b = viaSketch.getResult();
54+
assertEquals(a.getEstimate(), b.getEstimate(), 0.0, what + " estimate");
55+
assertEquals(a.getLgK(), b.getLgK(), what + " lgK");
56+
assertEquals(a.toByteArray(), b.toByteArray(), what + " serialized image");
57+
}
58+
59+
/** Every flavor of source, against a union that is still empty. */
60+
@Test
61+
public void checkEachFlavorIntoFreshUnion() {
62+
for (int lgK = 4; lgK <= 14; lgK++) {
63+
final int k = 1 << lgK;
64+
//counts chosen to land in EMPTY, SPARSE, HYBRID, PINNED and SLIDING
65+
final int[] counts = {0, 1, 10, k / 16, k / 4, k, 2 * k, 3 * k, 6 * k, 20 * k};
66+
for (final int n : counts) {
67+
assertSamePath("lgK=" + lgK + " n=" + n, lgK,
68+
new byte[][] {sketchBytes(lgK, 0, n)});
69+
}
70+
}
71+
}
72+
73+
/** Sequences, so the union is exercised in every state a source can arrive into. */
74+
@Test
75+
public void checkSequencesAcrossUnionStates() {
76+
for (final int lgK : new int[] {4, 8, 11, 12}) {
77+
final int k = 1 << lgK;
78+
final int[][] sequences = {
79+
{1, 1, 1}, //stays sparse
80+
{1, 10, k}, //sparse then graduates
81+
{k, 1}, //matrix first, then a sparse source
82+
{k / 16, k, 3 * k, 20 * k}, //ascending through the flavors
83+
{20 * k, 3 * k, k, k / 16}, //descending
84+
{0, k, 0, 3 * k}, //empties interleaved
85+
};
86+
for (final int[] seq : sequences) {
87+
final byte[][] images = new byte[seq.length][];
88+
long from = 0;
89+
for (int i = 0; i < seq.length; i++) {
90+
images[i] = sketchBytes(lgK, from, from + seq[i]);
91+
from += seq[i] / 2; //overlap, so the merges do real work
92+
}
93+
assertSamePath("lgK=" + lgK + " seq=" + java.util.Arrays.toString(seq), lgK, images);
94+
}
95+
}
96+
}
97+
98+
/** A source with a smaller lgK than the union forces the union to be reduced first. */
99+
@Test
100+
public void checkSourceWithSmallerLgK() {
101+
for (final int unionLgK : new int[] {10, 12, 14}) {
102+
for (final int srcLgK : new int[] {4, 8, 9}) {
103+
if (srcLgK >= unionLgK) { continue; }
104+
final int srcK = 1 << srcLgK;
105+
for (final int n : new int[] {1, srcK / 4, srcK, 3 * srcK, 20 * srcK}) {
106+
//seed the union at its own lgK first, then merge the smaller source
107+
assertSamePath("unionLgK=" + unionLgK + " srcLgK=" + srcLgK + " n=" + n, unionLgK,
108+
new byte[][] {sketchBytes(unionLgK, 0, 1 << unionLgK), sketchBytes(srcLgK, 0, n)});
109+
}
110+
}
111+
}
112+
}
113+
114+
/** Randomly fed sketches, which is where the column permutations get exercised hardest. */
115+
@Test
116+
public void checkRandomInput() {
117+
final Random rnd = new Random(9876543L);
118+
for (final int lgK : new int[] {8, 11, 12}) {
119+
for (int trial = 0; trial < 6; trial++) {
120+
final byte[][] images = new byte[5][];
121+
for (int i = 0; i < images.length; i++) {
122+
final CpcSketch sk = new CpcSketch(lgK);
123+
final int n = 1 + rnd.nextInt(40 << lgK);
124+
for (int j = 0; j < n; j++) { sk.update(rnd.nextLong()); }
125+
images[i] = sk.toByteArray();
126+
}
127+
assertSamePath("random lgK=" + lgK + " trial=" + trial, lgK, images);
128+
}
129+
}
130+
}
131+
132+
/** A mismatched seed must be rejected, exactly as the sketch path rejects it. */
133+
@Test
134+
public void checkSeedMismatchIsRejected() {
135+
final byte[] image = sketchBytes(10, 0, 1000);
136+
final CpcUnion u = new CpcUnion(10, Util.DEFAULT_UPDATE_SEED + 1);
137+
try {
138+
u.update(MemorySegment.ofArray(image));
139+
fail("expected a seed hash mismatch");
140+
} catch (final SketchesArgumentException e) {
141+
//expected
142+
}
143+
}
144+
145+
/** Interleaving both entry points on one union must behave like using either alone. */
146+
@Test
147+
public void checkInterleavedEntryPoints() {
148+
final int lgK = 11;
149+
final int k = 1 << lgK;
150+
final byte[][] images = new byte[6][];
151+
for (int i = 0; i < images.length; i++) {
152+
images[i] = sketchBytes(lgK, (long) i * k, ((long) i * k) + (2 * k));
153+
}
154+
155+
final CpcUnion mixed = new CpcUnion(lgK);
156+
final CpcUnion sketchOnly = new CpcUnion(lgK);
157+
for (int i = 0; i < images.length; i++) {
158+
if ((i % 2) == 0) {
159+
mixed.update(MemorySegment.ofArray(images[i]));
160+
} else {
161+
mixed.update(CpcSketch.heapify(MemorySegment.ofArray(images[i])));
162+
}
163+
sketchOnly.update(CpcSketch.heapify(MemorySegment.ofArray(images[i])));
164+
}
165+
assertEquals(mixed.getResult().toByteArray(), sketchOnly.getResult().toByteArray());
166+
}
167+
}

0 commit comments

Comments
 (0)