From 28bb5e25cf8c9ddbbb4330c2c262a1918d65e18f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Marek=20Vavrus=CC=8Ca?= <mvavrusa@cloudflare.com>
Date: Thu, 28 Sep 2017 00:01:54 -0700
Subject: [PATCH] AggregateFunctionTopK: read alphaMap for generic

* the alpha_map vector always (de)serialises
  the actual version (could empty sometimes)
* AggregateFunctionTopK generic variant deserialises
  it as well instead of ignoring it
* AggregateFunctionTopK generic variant clears the
  array before deserialising

refs #1283
---
 .../AggregateFunctions/AggregateFunctionTopK.h   |  4 ++++
 dbms/src/Common/SpaceSaving.h                    | 16 +++++++++++++++-
 2 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/dbms/src/AggregateFunctions/AggregateFunctionTopK.h b/dbms/src/AggregateFunctions/AggregateFunctionTopK.h
index 64938d97f38..814b3564eb8 100644
--- a/dbms/src/AggregateFunctions/AggregateFunctionTopK.h
+++ b/dbms/src/AggregateFunctions/AggregateFunctionTopK.h
@@ -195,7 +195,9 @@ public:
     {
         auto & set = this->data(place).value;
         set.resize(reserved);
+        set.clear();
 
+        // Specialised here because there's no deserialiser for StringRef
         size_t count = 0;
         readVarUInt(count, buf);
         for (size_t i = 0; i < count; ++i) {
@@ -206,6 +208,8 @@ public:
             set.insert(ref, count, error);
             arena->rollback(ref.size);
         }
+
+        set.readAlphaMap(buf);
     }
 
     void addImpl(AggregateDataPtr place, const IColumn & column, size_t row_num, Arena * arena) const
diff --git a/dbms/src/Common/SpaceSaving.h b/dbms/src/Common/SpaceSaving.h
index 6e5f39725aa..ebf8659b32a 100644
--- a/dbms/src/Common/SpaceSaving.h
+++ b/dbms/src/Common/SpaceSaving.h
@@ -130,6 +130,11 @@ public:
         return m_capacity;
     }
 
+    void clear()
+    {
+        return destroyElements();
+    }
+
     void resize(size_t new_capacity)
     {
         counter_list.reserve(new_capacity);
@@ -255,6 +260,8 @@ public:
         writeVarUInt(size(), wb);
         for (auto counter : counter_list)
             counter->write(wb);
+
+        writeVarUInt(alpha_map.size(), wb);
         for (auto alpha : alpha_map)
             writeVarUInt(alpha, wb);
     }
@@ -273,7 +280,14 @@ public:
             push(counter);
         }
 
-        for (size_t i = 0; i < nextAlphaSize(m_capacity); ++i)
+        readAlphaMap(rb);
+    }
+
+    void readAlphaMap(ReadBuffer & rb)
+    {
+        size_t alpha_size = 0;
+        readVarUInt(alpha_size, rb);
+        for (size_t i = 0; i < alpha_size; ++i)
         {
             UInt64 alpha = 0;
             readVarUInt(alpha, rb);