You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datasketches.apache.org by le...@apache.org on 2020/04/25 04:29:25 UTC
[incubator-datasketches-java] 02/05: Fix Druid bug.
This is an automated email from the ASF dual-hosted git repository.
leerho pushed a commit to branch FixDruidIssue9736
in repository https://gitbox.apache.org/repos/asf/incubator-datasketches-java.git
commit 635d31660e9c5140ca7ca3a1153d3fd35e9e9bb0
Author: Lee Rhodes <le...@users.noreply.github.com>
AuthorDate: Fri Apr 24 18:06:15 2020 -0700
Fix Druid bug.
---
.../org/apache/datasketches/hll/HllSketch.java | 54 ++++++++++-----
.../java/org/apache/datasketches/hll/HllUtil.java | 8 +--
.../java/org/apache/datasketches/hll/Union.java | 79 +++++++++-------------
.../org/apache/datasketches/hll/UnionCaseTest.java | 6 +-
.../org/apache/datasketches/hll/UnionTest.java | 75 +++++++++++++++++---
5 files changed, 139 insertions(+), 83 deletions(-)
diff --git a/src/main/java/org/apache/datasketches/hll/HllSketch.java b/src/main/java/org/apache/datasketches/hll/HllSketch.java
index 00b627e..b073953 100644
--- a/src/main/java/org/apache/datasketches/hll/HllSketch.java
+++ b/src/main/java/org/apache/datasketches/hll/HllSketch.java
@@ -158,6 +158,11 @@ public class HllSketch extends BaseHllSketch {
* @return an HllSketch on the java heap.
*/
public static final HllSketch heapify(final Memory srcMem) {
+ return heapify(srcMem, true);
+ }
+
+ //used by union and above
+ static final HllSketch heapify(final Memory srcMem, final boolean checkRebuild) {
final CurMode curMode = checkPreamble(srcMem);
final HllSketch heapSketch;
if (curMode == CurMode.HLL) {
@@ -168,6 +173,9 @@ public class HllSketch extends BaseHllSketch {
heapSketch = new HllSketch(Hll6Array.heapify(srcMem));
} else { //Hll_8
heapSketch = new HllSketch(Hll8Array.heapify(srcMem));
+ if (checkRebuild) {
+ Union.checkRebuildCurMinNumKxQ(heapSketch);
+ }
}
} else if (curMode == CurMode.LIST) {
heapSketch = new HllSketch(CouponList.heapifyList(srcMem));
@@ -185,46 +193,52 @@ public class HllSketch extends BaseHllSketch {
*
* <p>The given <i>dstMem</i> is checked for the required capacity as determined by
* {@link #getMaxUpdatableSerializationBytes(int, TgtHllType)}.
- * @param wmem an writable image of a valid sketch with data.
+ * @param srcWmem an writable image of a valid source sketch with data.
* @return an HllSketch where the sketch data is in the given dstMem.
*/
- public static final HllSketch writableWrap(final WritableMemory wmem) {
- final boolean compact = extractCompactFlag(wmem);
- if (compact) {
+ public static final HllSketch writableWrap(final WritableMemory srcWmem) {
+ if (extractCompactFlag(srcWmem)) {
throw new SketchesArgumentException(
"Cannot perform a writableWrap of a writable sketch image that is in compact form.");
}
- final int lgConfigK = extractLgK(wmem);
- final TgtHllType tgtHllType = extractTgtHllType(wmem);
+ return writableWrap(srcWmem, true);
+ }
+
+ //used by union and above
+ static final HllSketch writableWrap( final WritableMemory srcWmem, final boolean checkRebuild) {
+ final int lgConfigK = extractLgK(srcWmem);
+ final TgtHllType tgtHllType = extractTgtHllType(srcWmem);
final long minBytes = getMaxUpdatableSerializationBytes(lgConfigK, tgtHllType);
- final long capBytes = wmem.getCapacity();
+ final long capBytes = srcWmem.getCapacity();
HllUtil.checkMemSize(minBytes, capBytes);
-
- final CurMode curMode = checkPreamble(wmem);
+ final CurMode curMode = checkPreamble(srcWmem);
final HllSketch directSketch;
if (curMode == CurMode.HLL) {
if (tgtHllType == TgtHllType.HLL_4) {
- directSketch = new HllSketch(new DirectHll4Array(lgConfigK, wmem));
+ directSketch = new HllSketch(new DirectHll4Array(lgConfigK, srcWmem));
} else if (tgtHllType == TgtHllType.HLL_6) {
- directSketch = new HllSketch(new DirectHll6Array(lgConfigK, wmem));
+ directSketch = new HllSketch(new DirectHll6Array(lgConfigK, srcWmem));
} else { //Hll_8
- directSketch = new HllSketch(new DirectHll8Array(lgConfigK, wmem));
+ directSketch = new HllSketch(new DirectHll8Array(lgConfigK, srcWmem));
+ if (checkRebuild) { //union only uses HLL_8, we allow non-finalized from a union call.
+ Union.checkRebuildCurMinNumKxQ(directSketch);
+ }
}
} else if (curMode == CurMode.LIST) {
directSketch =
- new HllSketch(new DirectCouponList(lgConfigK, tgtHllType, curMode, wmem));
- } else {
+ new HllSketch(new DirectCouponList(lgConfigK, tgtHllType, curMode, srcWmem));
+ } else { //SET
directSketch =
- new HllSketch(new DirectCouponHashSet(lgConfigK, tgtHllType, wmem));
+ new HllSketch(new DirectCouponHashSet(lgConfigK, tgtHllType, srcWmem));
}
return directSketch;
}
/**
* Wraps the given read-only Memory that must be a image of a valid sketch,
- * which may be in compact or updatable form, and should have data. Any attempt to update this
- * sketch will throw an exception.
- * @param srcMem a read-only image of a valid sketch.
+ * which may be in compact or updatable form, and should have data. Any attempt to update the
+ * given source Memory will throw an exception.
+ * @param srcMem a read-only image of a valid source sketch.
* @return an HllSketch, where the read-only data of the sketch is in the given srcMem.
*
*/
@@ -241,6 +255,8 @@ public class HllSketch extends BaseHllSketch {
directSketch = new HllSketch(new DirectHll6Array(lgConfigK, srcMem));
} else { //Hll_8
directSketch = new HllSketch(new DirectHll8Array(lgConfigK, srcMem));
+ //rebuild if srcMem came from a union and was not finalized, rather than throw exception.
+ Union.checkRebuildCurMinNumKxQ(directSketch);
}
} else if (curMode == CurMode.LIST) {
directSketch =
@@ -252,6 +268,8 @@ public class HllSketch extends BaseHllSketch {
return directSketch;
}
+
+
/**
* Return a copy of this sketch onto the Java heap.
* @return a copy of this sketch onto the Java heap.
diff --git a/src/main/java/org/apache/datasketches/hll/HllUtil.java b/src/main/java/org/apache/datasketches/hll/HllUtil.java
index 916ad06..976a375 100644
--- a/src/main/java/org/apache/datasketches/hll/HllUtil.java
+++ b/src/main/java/org/apache/datasketches/hll/HllUtil.java
@@ -106,7 +106,8 @@ final class HllUtil {
|| ((curMode == CurMode.SET) && (preInts != HASH_SET_PREINTS))
|| ((curMode == CurMode.HLL) && (preInts != HLL_PREINTS))
) {
- HllUtil.badPreambleState(mem);
+ throw new SketchesArgumentException("Possible Corruption, Invalid Preamble:"
+ + PreambleUtil.toString(mem));
}
return curMode;
}
@@ -117,11 +118,6 @@ final class HllUtil {
"This sketch is compact or does not have write access to the underlying resource.");
}
- static final void badPreambleState(final Memory mem) {
- throw new SketchesArgumentException("Possible Corruption, Invalid Preamble:"
- + PreambleUtil.toString(mem));
- }
-
//Used for thrown exceptions
static String pairString(final int pair) {
return "SlotNo: " + getPairLow26(pair) + ", Value: "
diff --git a/src/main/java/org/apache/datasketches/hll/Union.java b/src/main/java/org/apache/datasketches/hll/Union.java
index 25c6381..f560b11 100644
--- a/src/main/java/org/apache/datasketches/hll/Union.java
+++ b/src/main/java/org/apache/datasketches/hll/Union.java
@@ -22,6 +22,7 @@ package org.apache.datasketches.hll;
import static org.apache.datasketches.Util.invPow2;
import static org.apache.datasketches.hll.HllUtil.EMPTY;
import static org.apache.datasketches.hll.PreambleUtil.HLL_BYTE_ARR_START;
+import static org.apache.datasketches.hll.PreambleUtil.extractTgtHllType;
import static org.apache.datasketches.hll.TgtHllType.HLL_8;
import org.apache.datasketches.SketchesArgumentException;
@@ -82,24 +83,16 @@ public class Union extends BaseHllSketch {
* {@link HllSketch#getMaxUpdatableSerializationBytes(int, TgtHllType)}.
* @param lgMaxK the desired maximum log-base-2 of <i>K</i>. This value must be
* between 4 and 21 inclusively.
- * @param dstMem the destination memory for the sketch.
+ * @param dstWmem the destination writable memory for the sketch.
*/
- public Union(final int lgMaxK, final WritableMemory dstMem) {
+ public Union(final int lgMaxK, final WritableMemory dstWmem) {
this.lgMaxK = HllUtil.checkLgK(lgMaxK);
- gadget = new HllSketch(lgMaxK, HLL_8, dstMem);
+ gadget = new HllSketch(lgMaxK, HLL_8, dstWmem);
}
- Union(final HllSketch sketch) {
+ //used only by writableWrap
+ private Union(final HllSketch sketch) {
lgMaxK = sketch.getLgConfigK();
- final TgtHllType tgtHllType = sketch.getTgtHllType();
- if (tgtHllType != TgtHllType.HLL_8) {
- throw new SketchesArgumentException("Union can only wrap writable HLL_8 sketches.");
- }
- //This should not happen, this is in case it does.
- if (sketch.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- throw new SketchesArgumentException(
- "Incomming sketch is corrupted, Rebuild_CurMin_Num_KxQ flag is set.");
- }
gadget = sketch;
}
@@ -119,7 +112,7 @@ public class Union extends BaseHllSketch {
*/
public static final Union heapify(final Memory mem) {
final int lgK = HllUtil.checkLgK(mem.getByte(PreambleUtil.LG_K_BYTE));
- final HllSketch sk = HllSketch.heapify(mem);
+ final HllSketch sk = HllSketch.heapify(mem, false); //allows non-finalized image
final Union union = new Union(lgK);
union.update(sk);
return union;
@@ -133,18 +126,22 @@ public class Union extends BaseHllSketch {
*
* <p>The given <i>dstMem</i> is checked for the required capacity as determined by
* {@link HllSketch#getMaxUpdatableSerializationBytes(int, TgtHllType)}, and for the correct type.
- * @param wmem an writable image of a valid sketch with data.
+ * @param srcWmem an writable image of a valid sketch with data.
* @return a Union operator where the sketch data is in the given dstMem.
*/
- public static final Union writableWrap(final WritableMemory wmem) {
- return new Union(HllSketch.writableWrap(wmem));
+ public static final Union writableWrap(final WritableMemory srcWmem) {
+ final TgtHllType tgtHllType = extractTgtHllType(srcWmem);
+ if (tgtHllType != TgtHllType.HLL_8) {
+ throw new SketchesArgumentException(
+ "Union can only wrap writable HLL_8 sketches that were the Gadget of a Union.");
+ }
+ //allows writableWrap of non-finalized image
+ return new Union(HllSketch.writableWrap(srcWmem, false));
}
@Override
public double getCompositeEstimate() {
- if (gadget.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- rebuildCurMinNumKxQ((AbstractHllArray)(gadget.hllSketchImpl));
- }
+ checkRebuildCurMinNumKxQ(gadget);
return gadget.hllSketchImpl.getCompositeEstimate();
}
@@ -160,9 +157,7 @@ public class Union extends BaseHllSketch {
@Override
public double getEstimate() {
- if (gadget.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- rebuildCurMinNumKxQ((AbstractHllArray)(gadget.hllSketchImpl));
- }
+ checkRebuildCurMinNumKxQ(gadget);
return gadget.getEstimate();
}
@@ -178,9 +173,7 @@ public class Union extends BaseHllSketch {
@Override
public double getLowerBound(final int numStdDev) {
- if (gadget.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- rebuildCurMinNumKxQ((AbstractHllArray)(gadget.hllSketchImpl));
- }
+ checkRebuildCurMinNumKxQ(gadget);
return gadget.getLowerBound(numStdDev);
}
@@ -209,9 +202,7 @@ public class Union extends BaseHllSketch {
* @return the result of this union operator with the specified TgtHllType
*/
public HllSketch getResult(final TgtHllType tgtHllType) {
- if (gadget.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- rebuildCurMinNumKxQ((AbstractHllArray)(gadget.hllSketchImpl));
- }
+ checkRebuildCurMinNumKxQ(gadget);
return gadget.copyAs(tgtHllType);
}
@@ -227,9 +218,7 @@ public class Union extends BaseHllSketch {
@Override
public double getUpperBound(final int numStdDev) {
- if (gadget.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- rebuildCurMinNumKxQ((AbstractHllArray)(gadget.hllSketchImpl));
- }
+ checkRebuildCurMinNumKxQ(gadget);
return gadget.getUpperBound(numStdDev);
}
@@ -289,26 +278,20 @@ public class Union extends BaseHllSketch {
*/
@Override
public byte[] toCompactByteArray() {
- if (gadget.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- rebuildCurMinNumKxQ((AbstractHllArray)(gadget.hllSketchImpl));
- }
+ checkRebuildCurMinNumKxQ(gadget);
return gadget.toCompactByteArray();
}
@Override
public byte[] toUpdatableByteArray() {
- if (gadget.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- rebuildCurMinNumKxQ((AbstractHllArray)(gadget.hllSketchImpl));
- }
+ checkRebuildCurMinNumKxQ(gadget);
return gadget.toUpdatableByteArray();
}
@Override
public String toString(final boolean summary, final boolean hllDetail,
final boolean auxDetail, final boolean all) {
- if (gadget.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- rebuildCurMinNumKxQ((AbstractHllArray)(gadget.hllSketchImpl));
- }
+ checkRebuildCurMinNumKxQ(gadget);
return gadget.toString(summary, hllDetail, auxDetail, all);
}
@@ -317,11 +300,6 @@ public class Union extends BaseHllSketch {
* @param sketch the given sketch.
*/
public void update(final HllSketch sketch) {
- //This should not happen, this is in case it does.
- if (sketch.hllSketchImpl.isRebuildCurMinNumKxQFlag()) {
- throw new SketchesArgumentException(
- "Incomming sketch is corrupted, Rebuild_CurMin_Num_KxQ flag is set.");
- }
gadget.hllSketchImpl = unionImpl(sketch, gadget, lgMaxK);
}
@@ -654,8 +632,13 @@ public class Union extends BaseHllSketch {
}
//Used to rebuild curMin, numAtCurMin and KxQ registers, due to high performance merge operation
- //performed in 1st switch cases 4, 5, 12, 13, 20, 21
- private static final void rebuildCurMinNumKxQ(final AbstractHllArray absHllArr) {
+ static final void checkRebuildCurMinNumKxQ(final HllSketch sketch) {
+ final HllSketchImpl hllSketchImpl = sketch.hllSketchImpl;
+ final CurMode curMode = sketch.getCurMode();
+ final TgtHllType tgtHllType = sketch.getTgtHllType();
+ final boolean rebuild = hllSketchImpl.isRebuildCurMinNumKxQFlag();
+ if ( !rebuild || (curMode != CurMode.HLL) || (tgtHllType != HLL_8) ) { return; }
+ final AbstractHllArray absHllArr = (AbstractHllArray)(hllSketchImpl);
int curMin = 64;
int numAtCurMin = 0;
double kxq0 = 1 << absHllArr.getLgConfigK();
diff --git a/src/test/java/org/apache/datasketches/hll/UnionCaseTest.java b/src/test/java/org/apache/datasketches/hll/UnionCaseTest.java
index 196c261..33edb58 100644
--- a/src/test/java/org/apache/datasketches/hll/UnionCaseTest.java
+++ b/src/test/java/org/apache/datasketches/hll/UnionCaseTest.java
@@ -359,7 +359,7 @@ public class UnionCaseTest {
union.update(sk); //throws
}
- @Test(expectedExceptions = SketchesArgumentException.class)
+ //@Test(expectedExceptions = SketchesArgumentException.class)
public void checkRebuildCurMinNumKxQFlag2() {
HllSketch sk = buildMemorySketch(4, HLL_8, 16);
DirectHllArray hllArr = (DirectHllArray)(sk.hllSketchImpl);
@@ -437,7 +437,7 @@ public class UnionCaseTest {
* @param o value to print
*/
static void print(Object o) {
- System.out.print(o.toString()); //disable here
+ //System.out.print(o.toString()); //disable here
}
/**
@@ -445,7 +445,7 @@ public class UnionCaseTest {
* @param args arguments
*/
static void printf(String fmt, Object...args) {
- System.out.printf(fmt, args); //disable here
+ //System.out.printf(fmt, args); //disable here
}
}
diff --git a/src/test/java/org/apache/datasketches/hll/UnionTest.java b/src/test/java/org/apache/datasketches/hll/UnionTest.java
index 2d4bf5c..dcba9d7 100644
--- a/src/test/java/org/apache/datasketches/hll/UnionTest.java
+++ b/src/test/java/org/apache/datasketches/hll/UnionTest.java
@@ -19,19 +19,19 @@
package org.apache.datasketches.hll;
+import static java.lang.Math.min;
import static org.apache.datasketches.hll.TgtHllType.HLL_4;
import static org.apache.datasketches.hll.TgtHllType.HLL_6;
import static org.apache.datasketches.hll.TgtHllType.HLL_8;
-import static java.lang.Math.min;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
-import org.testng.annotations.Test;
-
-import org.apache.datasketches.memory.Memory;
import org.apache.datasketches.SketchesArgumentException;
+import org.apache.datasketches.memory.Memory;
+import org.apache.datasketches.memory.WritableMemory;
+import org.testng.annotations.Test;
/**
* @author Lee Rhodes
@@ -434,6 +434,65 @@ public class UnionTest {
assertEquals(est3, est1, 0.0);
}
+ @Test
+ public void checkUnionHeapifyRebuildAfterMerge() {
+ int lgK = 12;
+ //Build 2 sketches in HLL (dense) mode.
+ int u = (lgK < 8) ? 16 : 1 << (lgK - 3);
+ HllSketch sk1 = new HllSketch(lgK);
+ HllSketch sk2 = new HllSketch(lgK);
+ for (int i = 0; i < u; i++) {
+ sk1.update(i);
+ sk2.update(i + u);
+ }
+ final int bytes = Union.getMaxSerializationBytes(lgK);
+ WritableMemory wmem = WritableMemory.allocate(bytes);
+ Union union1 = new Union(lgK, wmem); //Create original union off-heap
+ union1.update(sk1);
+ union1.update(sk2); //oooFlag = Rebuild_KxQ = TRUE
+ boolean rebuild = PreambleUtil.extractRebuildCurMinNumKxQFlag(wmem);
+ double hipAccum = PreambleUtil.extractHipAccum(wmem);
+ assertTrue(rebuild);
+ assertTrue(hipAccum == 0.0);
+ //Heapify byteArr as if it were a sketch, but it is actually a union!
+ HllSketch sk3 = HllSketch.heapify(wmem); //rebuilds sk3
+ rebuild = sk3.hllSketchImpl.isRebuildCurMinNumKxQFlag();
+ assertFalse(rebuild);
+ }
+
+ @SuppressWarnings("unused")
+ @Test //similar to above except uses wrap instead of heapify
+ public void druidUseCase() {
+ final int lgK = 12;
+ final int bytes = Union.getMaxSerializationBytes(lgK);
+ WritableMemory wmem = WritableMemory.allocate(bytes);
+ new Union(lgK, wmem); // result is unused, relying on side effect
+ int trueCount = 0;
+ int delta = (lgK < 8) ? 16 : 1 << (lgK - 3);
+ for (int i = 0; i < 3; i++) {
+ Union.writableWrap(wmem).update(buildSketch(trueCount, delta));
+ trueCount += delta;
+ }
+ boolean rebuild = PreambleUtil.extractRebuildCurMinNumKxQFlag(wmem);
+ assertTrue(rebuild);
+ HllSketch result = Union.writableWrap(wmem).getResult(); //rebuilds result
+ rebuild = result.hllSketchImpl.isRebuildCurMinNumKxQFlag();
+ assertFalse(rebuild);
+ double est = result.getEstimate();
+ double err = (est / trueCount) - 1.0;
+ double rse3 = (3 * 1.04)/Math.sqrt(1 << lgK);
+ println(err + " < " + rse3);
+ assertTrue(err < rse3);
+ }
+
+ private static HllSketch buildSketch(final int start, final int count) {
+ HllSketch sketch = new HllSketch(10);
+ for (int i = start; i < (start + count); i++) {
+ sketch.update(i);
+ }
+ return sketch;
+ }
+
private static Union newUnion(int lgK) {
return new Union(lgK);
}
@@ -451,15 +510,15 @@ public class UnionTest {
/**
* @param s value to print
*/
- static void println(String s) {
- print(s + LS);
+ static void println(Object s) {
+ print(s.toString() + LS);
}
/**
* @param s value to print
*/
- static void print(String s) {
- //System.out.print(s); //disable here
+ static void print(Object s) {
+ System.out.print(s.toString()); //disable here
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org