From 30d1a5db75ed86d7e7173a5831ddabafe296d856 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Xavier=20L=C3=A9aut=C3=A9?= <xavier@metamarkets.com>
Date: Thu, 24 Apr 2014 10:48:35 -0700
Subject: [PATCH] HLL: avoid direct buffers + fix swap target buffer

---
 .../hyperloglog/HyperLogLogCollector.java     | 25 +++++++++----------
 .../HyperUniquesBufferAggregator.java         |  2 +-
 2 files changed, 13 insertions(+), 14 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 bd5cfd84058..dbd766c391a 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
@@ -326,27 +326,26 @@ public abstract class HyperLogLogCollector implements Comparable<HyperLogLogColl
       convertToMutableByteBuffer();
     }
 
+    if (storageBuffer.remaining() != getNumBytesForDenseStorage()) {
+      convertToDenseStorage();
+    }
+
     estimatedCardinality = null;
 
     if (getRegisterOffset() < other.getRegisterOffset()) {
       // "Swap" the buffers so that we are folding into the one with the higher offset
-      ByteBuffer newStorage = ByteBuffer.allocateDirect(other.storageBuffer.remaining());
-      newStorage.put(other.storageBuffer.asReadOnlyBuffer());
-      newStorage.clear();
+      final ByteBuffer tmpBuffer = ByteBuffer.allocate(storageBuffer.remaining());
+      tmpBuffer.put(storageBuffer.asReadOnlyBuffer());
+      tmpBuffer.clear();
 
-      other.storageBuffer = storageBuffer;
-      other.initPosition = initPosition;
-      storageBuffer = newStorage;
-      initPosition = 0;
+      storageBuffer.duplicate().put(other.storageBuffer.asReadOnlyBuffer());
+
+      other = HyperLogLogCollector.makeCollector(tmpBuffer);
     }
 
     final ByteBuffer otherBuffer = other.storageBuffer.asReadOnlyBuffer();
     final byte otherOffset = other.getRegisterOffset();
 
-    if (storageBuffer.remaining() != getNumBytesForDenseStorage()) {
-      convertToDenseStorage();
-    }
-
     byte myOffset = getRegisterOffset();
     short numNonZero = getNumNonZeroRegisters();
 
@@ -540,7 +539,7 @@ public abstract class HyperLogLogCollector implements Comparable<HyperLogLogColl
 
   private void convertToMutableByteBuffer()
   {
-    ByteBuffer tmpBuffer = ByteBuffer.allocateDirect(storageBuffer.remaining());
+    ByteBuffer tmpBuffer = ByteBuffer.allocate(storageBuffer.remaining());
     tmpBuffer.put(storageBuffer.asReadOnlyBuffer());
     tmpBuffer.position(0);
     storageBuffer = tmpBuffer;
@@ -549,7 +548,7 @@ public abstract class HyperLogLogCollector implements Comparable<HyperLogLogColl
 
   private void convertToDenseStorage()
   {
-    ByteBuffer tmpBuffer = ByteBuffer.allocateDirect(getNumBytesForDenseStorage());
+    ByteBuffer tmpBuffer = ByteBuffer.allocate(getNumBytesForDenseStorage());
     // put header
     setVersion(tmpBuffer);
     setRegisterOffset(tmpBuffer, getRegisterOffset());
diff --git a/processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperUniquesBufferAggregator.java b/processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperUniquesBufferAggregator.java
index 3229e813d6d..c088b37beca 100644
--- a/processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperUniquesBufferAggregator.java
+++ b/processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperUniquesBufferAggregator.java
@@ -67,7 +67,7 @@ public class HyperUniquesBufferAggregator implements BufferAggregator
   public Object get(ByteBuffer buf, int position)
   {
     final int size = HyperLogLogCollector.getLatestNumBytesForDenseStorage();
-    ByteBuffer dataCopyBuffer = ByteBuffer.allocateDirect(size);
+    ByteBuffer dataCopyBuffer = ByteBuffer.allocate(size);
     ByteBuffer mutationBuffer = buf.duplicate();
     mutationBuffer.position(position);
     mutationBuffer.limit(position + size);