diff --git a/processing/src/main/java/org/apache/druid/hll/HyperLogLogCollector.java b/processing/src/main/java/org/apache/druid/hll/HyperLogLogCollector.java index d285f1cd044e..2e0460a56495 100644 --- a/processing/src/main/java/org/apache/druid/hll/HyperLogLogCollector.java +++ b/processing/src/main/java/org/apache/druid/hll/HyperLogLogCollector.java @@ -171,16 +171,21 @@ private static double estimateSparse( final byte minNum, final byte overflowValue, final short overflowPosition, - final boolean isUpperNibble + final boolean isUpperNibble, + final int numHeaderBytes ) { final ByteBuffer copy = buf.asReadOnlyBuffer(); double e = 0.0d; int zeroCount = NUM_BUCKETS - 2 * (buf.remaining() / 3); + boolean overflowRegisterApplied = false; while (copy.hasRemaining()) { short position = copy.getShort(); final int register = (int) copy.get() & 0xff; - if (overflowValue != 0 && position == overflowPosition) { + // `position` is the serialized offset: the payload byte index plus the header size. Normalize it back to the + // payload byte index before comparing with overflowPosition (overflowRegister >>> 1); otherwise the overflow + // bucket's own entry is never matched here and the fallback below double-counts that register. + if (overflowValue != 0 && (position - numHeaderBytes) == overflowPosition) { int upperNibble = ((register & 0xf0) >>> BITS_PER_BUCKET) + minNum; int lowerNibble = (register & 0x0f) + minNum; if (isUpperNibble) { @@ -189,13 +194,25 @@ private static double estimateSparse( lowerNibble = Math.max(lowerNibble, overflowValue); } e += 1.0d / Math.pow(2, upperNibble) + 1.0d / Math.pow(2, lowerNibble); - zeroCount += (((upperNibble & 0xf0) == 0) ? 1 : 0) + (((lowerNibble & 0x0f) == 0) ? 1 : 0); + // upperNibble/lowerNibble hold decoded scalar values here (nibble + minNum, possibly the overflow value), + // so test them against zero directly: nibble masks would wrongly flag decoded values like 16 as empty. + zeroCount += ((upperNibble == 0) ? 1 : 0) + ((lowerNibble == 0) ? 1 : 0); + overflowRegisterApplied = true; } else { e += MIN_NUM_REGISTER_LOOKUP[minNum][register]; zeroCount += NUM_ZERO_LOOKUP[register]; } } + // A sparse buffer only stores registers set through the regular nibble range. When the only information about + // a bucket lives in the overflow register (positionOf1 exceeded registerOffset + RANGE) and that bucket is not + // among the entries above, it was implicitly counted as an empty register. Account for it explicitly so linear + // counting does not treat the collector as empty and estimate a cardinality of zero. + if (overflowValue != 0 && !overflowRegisterApplied) { + zeroCount -= 1; + e += 1.0d / Math.pow(2, overflowValue); + } + e += zeroCount; return applyCorrection(e, zeroCount); } @@ -223,7 +240,9 @@ private static double estimateDense( lowerNibble = Math.max(lowerNibble, overflowValue); } e += 1.0d / Math.pow(2, upperNibble) + 1.0d / Math.pow(2, lowerNibble); - zeroCount += (((upperNibble & 0xf0) == 0) ? 1 : 0) + (((lowerNibble & 0x0f) == 0) ? 1 : 0); + // upperNibble/lowerNibble hold decoded scalar values here (nibble + minNum, possibly the overflow value), + // so test them against zero directly: nibble masks would wrongly flag decoded values like 16 as empty. + zeroCount += ((upperNibble == 0) ? 1 : 0) + ((lowerNibble == 0) ? 1 : 0); } else { e += MIN_NUM_REGISTER_LOOKUP[minNum][register]; zeroCount += NUM_ZERO_LOOKUP[register]; @@ -535,7 +554,8 @@ public double estimateCardinality() registerOffset, overflowValue, overflowPosition, - isUpperNibble + isUpperNibble, + getNumHeaderBytes() ); } else { estimatedCardinality = estimateDense( diff --git a/processing/src/test/java/org/apache/druid/hll/HyperLogLogCollectorTest.java b/processing/src/test/java/org/apache/druid/hll/HyperLogLogCollectorTest.java index 45a0116fa81d..5a2152c3099c 100644 --- a/processing/src/test/java/org/apache/druid/hll/HyperLogLogCollectorTest.java +++ b/processing/src/test/java/org/apache/druid/hll/HyperLogLogCollectorTest.java @@ -609,6 +609,107 @@ public void testEstimation() Assert.assertEquals(expectedVals[valsToCheckIndex], collector.estimateCardinality(), 0.0d); } + @Test + public void testSparseOverflowRegisterEstimation() + { + // Reproduces the case where a single element whose positionOf1 exceeds the 4-bit nibble range + // (registerOffset 0 + RANGE 15) is stored only in the overflow register, leaving the sparse buffer empty. + // estimateSparse() must still account for that one non-empty register instead of reporting a cardinality of 0. + + // Direct form: bucket 5 with positionOf1 = 16 (> RANGE) goes straight to the overflow register. + HyperLogLogCollector direct = HyperLogLogCollector.makeLatestCollector(); + direct.add((short) 5, (byte) 16); + Assert.assertEquals(1L, direct.estimateCardinalityRound()); + Assert.assertEquals(1.0d, direct.estimateCardinality(), 0.05d); + + // Same situation reached through the byte[] hashing path used during ingestion: leading bytes 0x00 0x80 + // give positionOf1 = 16, and the trailing bytes select the bucket. + byte[] hashedValue = new byte[16]; + hashedValue[1] = (byte) 0x80; + hashedValue[15] = 0x05; + HyperLogLogCollector hashed = HyperLogLogCollector.makeLatestCollector(); + hashed.add(hashedValue); + Assert.assertEquals(1L, hashed.estimateCardinalityRound()); + } + + @Test + public void testSparseOverflowRegisterOnPopulatedBucket() + { + // Adding an in-range value converts the collector to dense storage, so estimateSparse's overflow handling is + // only reachable through a sparse-serialized buffer. A single bucket that first receives an in-range value and + // then an overflow value is still one populated register (the overflow value supersedes the nibble). After a + // sparse round-trip that register must be counted exactly once: estimateSparse has to recognize that the + // overflow register already has a sparse entry (comparing positions in the same coordinate system) and skip the + // standalone compensation, otherwise the register is counted twice and a single element estimates as 2. + HyperLogLogCollector source = HyperLogLogCollector.makeLatestCollector(); + source.add((short) 4, (byte) 3); // in-range value -> nibble for bucket 4 + source.add((short) 4, (byte) 16); // overflow on the same bucket -> overflow register + + HyperLogLogCollector sparse = HyperLogLogCollector.makeCollector(ByteBuffer.wrap(source.toByteArray())); + Assert.assertEquals(1L, sparse.estimateCardinalityRound()); + } + + @Test + public void testSparseOverflowRegisterAmongOtherEntries() + { + // A sparse buffer holding an overflow register plus other populated registers: estimateSparse must fold the + // overflow into the overflow bucket's own entry and leave the other entries untouched, which exercises both + // sides of the normalized position comparison. Two distinct populated registers should estimate as 2. + HyperLogLogCollector source = HyperLogLogCollector.makeLatestCollector(); + source.add((short) 4, (byte) 3); // bucket 4 in-range value + source.add((short) 4, (byte) 16); // overflow on bucket 4 -> overflow register + source.add((short) 20, (byte) 5); // an unrelated populated register in a different byte + + HyperLogLogCollector sparse = HyperLogLogCollector.makeCollector(ByteBuffer.wrap(source.toByteArray())); + Assert.assertEquals(2L, sparse.estimateCardinalityRound()); + } + + @Test + public void testSparseOverflowRegisterOnOddBucket() + { + // Odd bucket: the overflow value is folded into the decoded lowerNibble (16), which is a scalar and no longer + // fits the 4-bit nibble mask. zeroCount must treat it as populated, otherwise the single register is counted + // as empty and the estimate collapses to 0. + HyperLogLogCollector source = HyperLogLogCollector.makeLatestCollector(); + source.add((short) 5, (byte) 3); // in-range value -> lower nibble of byte 2 + source.add((short) 5, (byte) 16); // overflow on the same odd bucket + + HyperLogLogCollector sparse = HyperLogLogCollector.makeCollector(ByteBuffer.wrap(source.toByteArray())); + Assert.assertEquals(1L, sparse.estimateCardinalityRound()); + } + + @Test + public void testSparseOverflowRegisterOnOddBucketWithPopulatedNeighbor() + { + // The overflow byte also holds an in-range value for the neighboring even bucket. Both decoded nibbles are + // non-zero scalars (2 and 16), so neither register may be counted as empty: two populated registers should + // estimate as 2. + HyperLogLogCollector source = HyperLogLogCollector.makeLatestCollector(); + source.add((short) 4, (byte) 2); // neighbor bucket in the same byte, upper nibble + source.add((short) 5, (byte) 3); // odd bucket in-range value, lower nibble + source.add((short) 5, (byte) 16); // overflow on the odd bucket + + HyperLogLogCollector sparse = HyperLogLogCollector.makeCollector(ByteBuffer.wrap(source.toByteArray())); + Assert.assertEquals(2L, sparse.estimateCardinalityRound()); + } + + @Test + public void testDenseOverflowRegisterEstimation() + { + // Adding an in-range value converts the collector to dense storage, so estimating without a sparse round-trip + // exercises estimateDense's overflow handling. The same decoded-scalar zero check applies there: one populated + // register must estimate as 1 for both nibble parities. + HyperLogLogCollector even = HyperLogLogCollector.makeLatestCollector(); + even.add((short) 4, (byte) 3); + even.add((short) 4, (byte) 16); + Assert.assertEquals(1L, even.estimateCardinalityRound()); + + HyperLogLogCollector odd = HyperLogLogCollector.makeLatestCollector(); + odd.add((short) 5, (byte) 3); + odd.add((short) 5, (byte) 16); + Assert.assertEquals(1L, odd.estimateCardinalityRound()); + } + @Test public void testEstimationReadOnlyByteBuffers() {