From 72eb64a40c7a7f337bcf36928b24e8052d87ceb3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Xavier=20L=C3=A9aut=C3=A9?= <xavier@metamarkets.com>
Date: Fri, 22 Aug 2014 13:42:43 -0700
Subject: [PATCH] fix HLL equals when comparing sparse and dense

---
 .../hyperloglog/HyperLogLogCollector.java     | 32 +++++++++++++++++--
 1 file changed, 29 insertions(+), 3 deletions(-)

diff --git a/processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperLogLogCollector.java b/processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperLogLogCollector.java
index 586dc86d3c7..8e84e690130 100644
--- a/processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperLogLogCollector.java
+++ b/processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperLogLogCollector.java
@@ -195,6 +195,13 @@ public abstract class HyperLogLogCollector implements Comparable<HyperLogLogColl
     return applyCorrection(e, zeroCount);
   }
 
+  /**
+   * Checks if the payload for the given ByteBuffer is sparse or not.
+   * The given buffer must be positioned at getPayloadBytePosition() prior to calling isSparse
+   *
+   * @param buffer
+   * @return
+   */
   private static boolean isSparse(ByteBuffer buffer)
   {
     return buffer.remaining() != NUM_BYTES_FOR_BUCKETS;
@@ -495,13 +502,32 @@ public abstract class HyperLogLogCollector implements Comparable<HyperLogLogColl
       return false;
     }
 
-    HyperLogLogCollector collector = (HyperLogLogCollector) o;
+    ByteBuffer otherBuffer = ((HyperLogLogCollector) o).storageBuffer;
 
-    if (storageBuffer != null ? !storageBuffer.equals(collector.storageBuffer) : collector.storageBuffer != null) {
+    if (storageBuffer != null ? false : otherBuffer != null) {
       return false;
     }
 
-    return true;
+    if(storageBuffer == null && otherBuffer == null) {
+      return true;
+    }
+
+    final ByteBuffer denseStorageBuffer;
+    if(storageBuffer.remaining() != getNumBytesForDenseStorage()) {
+      HyperLogLogCollector denseCollector = HyperLogLogCollector.makeCollector(storageBuffer);
+      denseCollector.convertToDenseStorage();
+      denseStorageBuffer = denseCollector.storageBuffer;
+    } else {
+      denseStorageBuffer = storageBuffer;
+    }
+
+    if(otherBuffer.remaining() != getNumBytesForDenseStorage()) {
+      HyperLogLogCollector otherCollector = HyperLogLogCollector.makeCollector(otherBuffer);
+      otherCollector.convertToDenseStorage();
+      otherBuffer = otherCollector.storageBuffer;
+    }
+
+    return denseStorageBuffer.equals(otherBuffer);
   }
 
   @Override