From 761aa7bd1b758200060a2e298ad03dc67c484d77 Mon Sep 17 00:00:00 2001
From: kangkaisen <kangkaisen@live.com>
Date: Sat, 31 Dec 2016 15:41:07 +0800
Subject: [PATCH] KYLIN-2349 Serialize BitmapCounter with peekLength

---
 .../apache/kylin/measure/bitmap/BitmapCounter.java | 52 +++++++++++++++-------
 1 file changed, 36 insertions(+), 16 deletions(-)

diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java b/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java
index a7f277e..609e55d 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/bitmap/BitmapCounter.java
@@ -34,6 +34,7 @@ import org.roaringbitmap.buffer.MutableRoaringBitmap;
 public class BitmapCounter implements Comparable<BitmapCounter> {
 
     private MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
+    private final int VERSION = 2;
 
     public BitmapCounter() {
     }
@@ -102,15 +103,50 @@ public class BitmapCounter implements Comparable<BitmapCounter> {
         bitmap.serialize(dos);
         dos.close();
         ByteBuffer bb = ByteBuffer.wrap(bos.toByteArray());
+        out.putInt(VERSION);
+        out.putInt(bos.size() + 4 + 4);
         out.put(bb);
     }
 
     public void readRegisters(ByteBuffer in) throws IOException {
+        int mark = in.position();
+        int version = in.getInt();
+
+        // keep forward compatibility
+        if (version == VERSION) {
+            int size = in.getInt();
+        } else {
+            in.position(mark);
+        }
+
         try (DataInputStream is = new DataInputStream(new ByteBufferBackedInputStream(in))) {
             bitmap.deserialize(is);
         }
     }
 
+    public int peekLength(ByteBuffer in) {
+        int mark = in.position();
+        int len;
+        int version = in.getInt();
+
+        // keep forward compatibility
+        if (version == VERSION) {
+            len = in.getInt() ;
+        } else {
+            in.position(mark);
+            try (DataInputStream is = new DataInputStream(new ByteBufferBackedInputStream(in))) {
+                MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
+                bitmap.deserialize(is);
+                len = in.position() - mark;
+            } catch (IOException e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        in.position(mark);
+        return len;
+    }
+
     @Override
     public String toString() {
         long count = getCount();
@@ -169,22 +205,6 @@ public class BitmapCounter implements Comparable<BitmapCounter> {
             return -1;
     }
 
-    public int peekLength(ByteBuffer in) {
-        int mark = in.position();
-        int len;
-
-        MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
-        try (DataInputStream is = new DataInputStream(new ByteBufferBackedInputStream(in))) {
-            bitmap.deserialize(is);
-        } catch (IOException e) {
-            throw new IllegalStateException(e);
-        }
-
-        len = in.position() - mark;
-        in.position(mark);
-        return len;
-    }
-
     private class ByteBufferBackedInputStream extends InputStream {
         private final ByteBuffer buffer;
 
-- 
2.10.1 (Apple Git-78)