use reverse-iterator if possible

processing/src/main/java/io/druid/segment/BitmapOffset.java

@@ -21,7 +21,10 @@ package io.druid.segment;
 
 import com.metamx.collections.bitmap.BitmapFactory;
 import com.metamx.collections.bitmap.ImmutableBitmap;
+import com.metamx.collections.bitmap.MutableBitmap;
+import com.metamx.collections.bitmap.WrappedImmutableRoaringBitmap;
 import io.druid.segment.data.Offset;
+import io.druid.segment.data.RoaringBitmapSerdeFactory;
 import org.roaringbitmap.IntIterator;
 
 /**
@@ -51,44 +54,17 @@ public class BitmapOffset implements Offset
     if (!descending) {
       return bitmapIndex.iterator();
     }
-    // ImmutableRoaringReverseIntIterator is not cloneable.. looks like a bug
-    // update : it's fixed in 0.5.13
-    int i = bitmapIndex.size();
-    int[] back = new int[bitmapIndex.size()];
-    IntIterator iterator = bitmapIndex.iterator();
+    ImmutableBitmap roaringBitmap = bitmapIndex;
+    if (!(bitmapIndex instanceof WrappedImmutableRoaringBitmap)) {
+      final BitmapFactory factory = RoaringBitmapSerdeFactory.bitmapFactory;
+      final MutableBitmap bitmap = factory.makeEmptyMutableBitmap();
+      final IntIterator iterator = bitmapIndex.iterator();
       while (iterator.hasNext()) {
-      back[--i] = iterator.next();
+        bitmap.add(iterator.next());
       }
-    return new ArrayIntIterator(back, 0);
-  }
-
-  private static class ArrayIntIterator implements IntIterator {
-
-    private final int[] array;
-    private int index;
-
-    private ArrayIntIterator(int[] array, int index) {
-      this.array = array;
-      this.index = index;
-    }
-
-    @Override
-    public boolean hasNext()
-    {
-      return index < array.length;
-    }
-
-    @Override
-    public int next()
-    {
-      return array[index++];
-    }
-
-    @Override
-    public IntIterator clone()
-    {
-      return new ArrayIntIterator(array, index);
+      roaringBitmap = factory.makeImmutableBitmap(bitmap);
     }
+    return ((WrappedImmutableRoaringBitmap) roaringBitmap).getBitmap().getReverseIntIterator();
   }
 
   private BitmapOffset(BitmapOffset otherOffset)

processing/src/main/java/io/druid/segment/data/RoaringBitmapSerdeFactory.java

@@ -32,8 +32,8 @@ import java.nio.ByteBuffer;
  */
 public class RoaringBitmapSerdeFactory implements BitmapSerdeFactory
 {
-  private static final ObjectStrategy<ImmutableBitmap> objectStrategy = new ImmutableRoaringBitmapObjectStrategy();
-  private static final BitmapFactory bitmapFactory = new RoaringBitmapFactory();
+  public static final ObjectStrategy<ImmutableBitmap> objectStrategy = new ImmutableRoaringBitmapObjectStrategy();
+  public static final BitmapFactory bitmapFactory = new RoaringBitmapFactory();
 
   @Override
   public ObjectStrategy<ImmutableBitmap> getObjectStrategy()

processing/src/test/java/io/druid/segment/BitmapOffsetTest.java

@@ -19,39 +19,79 @@
 
 package io.druid.segment;
 
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
+import com.metamx.collections.bitmap.BitSetBitmapFactory;
+import com.metamx.collections.bitmap.BitmapFactory;
 import com.metamx.collections.bitmap.ConciseBitmapFactory;
-import com.metamx.collections.bitmap.WrappedImmutableConciseBitmap;
+import com.metamx.collections.bitmap.MutableBitmap;
+import com.metamx.collections.bitmap.RoaringBitmapFactory;
 import io.druid.segment.data.Offset;
-import it.uniroma3.mat.extendedset.intset.ConciseSet;
-import it.uniroma3.mat.extendedset.intset.ImmutableConciseSet;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.List;
 
 /**
  */
+@RunWith(Parameterized.class)
 public class BitmapOffsetTest
 {
+  private static final int[] TEST_VALS = {1, 2, 4, 291, 27412, 49120, 212312, 2412101};
+  private static final int[] TEST_VALS_FLIP = {2412101, 212312, 49120, 27412, 291, 4, 2, 1};
+
+  @Parameterized.Parameters
+  public static Iterable<Object[]> constructorFeeder() throws IOException
+  {
+    return Iterables.transform(
+        Sets.cartesianProduct(
+            ImmutableSet.of(new ConciseBitmapFactory(), new RoaringBitmapFactory(), new BitSetBitmapFactory()),
+            ImmutableSet.of(false, true)
+        ),
+        new Function<List<?>, Object[]>()
+        {
+          @Override
+          public Object[] apply(List<?> input)
+          {
+            return input.toArray();
+          }
+        }
+    );
+  }
+
+  private final BitmapFactory factory;
+  private final boolean descending;
+
+  public BitmapOffsetTest(BitmapFactory factory, boolean descending)
+  {
+    this.factory = factory;
+    this.descending = descending;
+  }
+
   @Test
   public void testSanity() throws Exception
   {
-    int[] vals = {1, 2, 4, 291, 27412, 49120, 212312, 2412101};
-    ConciseSet mutableSet = new ConciseSet();
-    for (int val : vals) {
-      mutableSet.add(val);
+    MutableBitmap mutable = factory.makeEmptyMutableBitmap();
+    for (int val : TEST_VALS) {
+      mutable.add(val);
     }
 
-    ImmutableConciseSet set = ImmutableConciseSet.newImmutableFromMutable(mutableSet);
-
-    BitmapOffset offset = new BitmapOffset(new ConciseBitmapFactory(), new WrappedImmutableConciseBitmap(set), false);
+    final BitmapOffset offset = new BitmapOffset(factory, factory.makeImmutableBitmap(mutable), descending);
+    final int[] expected = descending ? TEST_VALS_FLIP : TEST_VALS;
 
     int count = 0;
     while (offset.withinBounds()) {
-      Assert.assertEquals(vals[count], offset.getOffset());
+      Assert.assertEquals(expected[count], offset.getOffset());
 
       int cloneCount = count;
       Offset clonedOffset = offset.clone();
       while (clonedOffset.withinBounds()) {
-        Assert.assertEquals(vals[cloneCount], clonedOffset.getOffset());
+        Assert.assertEquals(expected[cloneCount], clonedOffset.getOffset());
 
         ++cloneCount;
         clonedOffset.increment();
@@ -60,6 +100,6 @@ public class BitmapOffsetTest
       ++count;
       offset.increment();
     }
-    Assert.assertEquals(count, vals.length);
+    Assert.assertEquals(count, expected.length);
   }
 }