Improve memory efficiency of WrappedRoaringBitmap. (#13889)

* Improve memory efficiency of WrappedRoaringBitmap.

Two changes:

1) Use an int[] for sizes 4 or below.
2) Remove the boolean compressRunOnSerialization. Doesn't save much
   space, but it does save a little, and it isn't adding a ton of value
   to have it be configurable. It was originally configurable in case
   anything broke when enabling it, but it's been a while and nothing
   has broken.

* Slight adjustment.

* Adjust for inspection.

* Updates.

* Update snaps.

* Update test.

* Adjust test.

* Fix snaps.

benchmarks/src/test/java/org/apache/druid/benchmark/BoundFilterBenchmark.java

@@ -159,7 +159,7 @@ public class BoundFilterBenchmark
   {
     step = (END_INT - START_INT) / cardinality;
     final BitmapFactory bitmapFactory = new RoaringBitmapFactory();
-    final BitmapSerdeFactory serdeFactory = new RoaringBitmapSerdeFactory(null);
+    final BitmapSerdeFactory serdeFactory = RoaringBitmapSerdeFactory.getInstance();
     final List<Integer> ints = generateInts();
     final GenericIndexed<String> dictionary = GenericIndexed.fromIterable(
         FluentIterable.from(ints).transform(Object::toString),

benchmarks/src/test/java/org/apache/druid/benchmark/DictionaryEncodedStringIndexSupplierBenchmark.java

@@ -91,7 +91,7 @@ public class DictionaryEncodedStringIndexSupplierBenchmark
     public void setup()
     {
       final BitmapFactory bitmapFactory = new RoaringBitmapFactory();
-      final BitmapSerdeFactory serdeFactory = new RoaringBitmapSerdeFactory(null);
+      final BitmapSerdeFactory serdeFactory = RoaringBitmapSerdeFactory.getInstance();
       final Iterable<Integer> ints = intGenerator();
       final GenericIndexed<String> dictionary = GenericIndexed.fromIterable(
           FluentIterable.from(ints)

benchmarks/src/test/java/org/apache/druid/benchmark/DimensionPredicateFilterBenchmark.java

@@ -120,7 +120,7 @@ public class DimensionPredicateFilterBenchmark
   public void setup()
   {
     final BitmapFactory bitmapFactory = new RoaringBitmapFactory();
-    final BitmapSerdeFactory serdeFactory = new RoaringBitmapSerdeFactory(null);
+    final BitmapSerdeFactory serdeFactory = RoaringBitmapSerdeFactory.getInstance();
     final List<Integer> ints = generateInts();
     final GenericIndexed<String> dictionary = GenericIndexed.fromIterable(
         FluentIterable.from(ints)

benchmarks/src/test/java/org/apache/druid/benchmark/InFilterBenchmark.java

@@ -83,7 +83,7 @@ public class InFilterBenchmark
   public void setup()
   {
     final BitmapFactory bitmapFactory = new RoaringBitmapFactory();
-    final BitmapSerdeFactory serdeFactory = new RoaringBitmapSerdeFactory(null);
+    final BitmapSerdeFactory serdeFactory = RoaringBitmapSerdeFactory.getInstance();
     final Iterable<Integer> ints = intGenerator();
     final GenericIndexed<String> dictionary = GenericIndexed.fromIterable(
         FluentIterable.from(ints)

benchmarks/src/test/java/org/apache/druid/benchmark/LikeFilterBenchmark.java

@@ -120,7 +120,7 @@ public class LikeFilterBenchmark
   {
     step = (END_INT - START_INT) / cardinality;
     final BitmapFactory bitmapFactory = new RoaringBitmapFactory();
-    final BitmapSerdeFactory serdeFactory = new RoaringBitmapSerdeFactory(null);
+    final BitmapSerdeFactory serdeFactory = RoaringBitmapSerdeFactory.getInstance();
     final List<Integer> ints = generateInts();
     final GenericIndexed<String> dictionary = GenericIndexed.fromIterable(
         FluentIterable.from(ints)

docs/development/extensions-core/kafka-supervisor-reference.md

@@ -232,7 +232,6 @@ For Roaring bitmaps:
 |Field|Type|Description|Required|
 |-----|----|-----------|--------|
 |`type`|String|Must be `roaring`.|yes|
-|`compressRunOnSerialization`|Boolean|Use a run-length encoding where it is estimated as more space efficient.|no (default == `true`)|
 
 For Concise bitmaps:
 

docs/development/extensions-core/kinesis-ingestion.md

@@ -322,7 +322,6 @@ For Roaring bitmaps:
 |Field|Type|Description|Required|
 |-----|----|-----------|--------|
 |`type`|String|Must be `roaring`.|yes|
-|`compressRunOnSerialization`|Boolean|Use a run-length encoding where it is estimated as more space efficient.|no (default == `true`)|
 
 For Concise bitmaps:
 

docs/ingestion/ingestion-spec.md

@@ -475,7 +475,7 @@ The `indexSpec` object can include the following properties:
 
 |Field|Description|Default|
 |-----|-----------|-------|
-|bitmap|Compression format for bitmap indexes. Should be a JSON object with `type` set to `roaring` or `concise`. For type `roaring`, the boolean property `compressRunOnSerialization` (defaults to true) controls whether or not run-length encoding will be used when it is determined to be more space-efficient.|`{"type": "roaring"}`|
+|bitmap|Compression format for bitmap indexes. Should be a JSON object with `type` set to `roaring` or `concise`.|`{"type": "roaring"}`|
 |dimensionCompression|Compression format for dimension columns. Options are `lz4`, `lzf`, `zstd`, or `uncompressed`.|`lz4`|
 |stringDictionaryEncoding|Encoding format for STRING value dictionaries used by STRING and COMPLEX&lt;json&gt; columns. <br /><br />Example to enable front coding: `{"type":"frontCoded", "bucketSize": 4}`<br />`bucketSize` is the number of values to place in a bucket to perform delta encoding. Must be a power of 2, maximum is 128. Defaults to 4.<br /><br />See [Front coding](#front-coding) for more information.|`{"type":"utf8"}`|
 |metricCompression|Compression format for primitive type metric columns. Options are `lz4`, `lzf`, `zstd`, `uncompressed`, or `none` (which is more efficient than `uncompressed`, but not supported by older versions of Druid).|`lz4`|

docs/operations/dump-segment.md

@@ -92,8 +92,7 @@ Sample output:
 ```
 {
   "bitmapSerdeFactory": {
-    "type": "roaring",
+    "type": "roaring"
-    "compressRunOnSerialization": true
   },
   "bitmaps": {
     "isRobot": {
@@ -195,8 +194,7 @@ Sample output:
 ```json
 {
   "bitmapSerdeFactory": {
-    "type": "roaring",
+    "type": "roaring"
-    "compressRunOnSerialization": true
   },
   "nest": {
     "$.x": {

docs/querying/sql-metadata-tables.md

@@ -190,7 +190,7 @@ SELECT * FROM sys.segments WHERE is_active = 1 AND last_compaction_state IS NOT
 or if you want to retrieve segment that was compacted only by a particular compaction spec (such as that of the auto compaction):
 
 ```sql
-SELECT * FROM sys.segments WHERE is_active = 1 AND last_compaction_state = 'CompactionState{partitionsSpec=DynamicPartitionsSpec{maxRowsPerSegment=5000000, maxTotalRows=9223372036854775807}, indexSpec={bitmap={type=roaring, compressRunOnSerialization=true}, dimensionCompression=lz4, metricCompression=lz4, longEncoding=longs, segmentLoader=null}}'
+SELECT * FROM sys.segments WHERE is_active = 1 AND last_compaction_state = 'CompactionState{partitionsSpec=DynamicPartitionsSpec{maxRowsPerSegment=5000000, maxTotalRows=9223372036854775807}, indexSpec={bitmap={type=roaring}, dimensionCompression=lz4, metricCompression=lz4, longEncoding=longs, segmentLoader=null}}'
 ```
 
 ### SERVERS table

indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java

@@ -327,7 +327,7 @@ public class CompactionTaskTest
         null,
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -627,7 +627,7 @@ public class CompactionTaskTest
             null,
             null,
             new IndexSpec(
-                new RoaringBitmapSerdeFactory(true),
+                RoaringBitmapSerdeFactory.getInstance(),
                 CompressionStrategy.LZ4,
                 CompressionStrategy.LZF,
                 LongEncodingStrategy.LONGS
@@ -690,7 +690,7 @@ public class CompactionTaskTest
         null,
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -721,7 +721,7 @@ public class CompactionTaskTest
         null,
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -766,7 +766,7 @@ public class CompactionTaskTest
         null,
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -805,7 +805,7 @@ public class CompactionTaskTest
         null,
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -954,7 +954,7 @@ public class CompactionTaskTest
         null,
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -1029,7 +1029,7 @@ public class CompactionTaskTest
         null,
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -1104,7 +1104,7 @@ public class CompactionTaskTest
         null,
         new HashedPartitionsSpec(null, 3, null),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -1852,7 +1852,7 @@ public class CompactionTaskTest
             null,
             new HashedPartitionsSpec(5000000, null, null), // automatically computed targetPartitionSize
             new IndexSpec(
-                new RoaringBitmapSerdeFactory(true),
+                RoaringBitmapSerdeFactory.getInstance(),
                 CompressionStrategy.LZ4,
                 CompressionStrategy.LZF,
                 LongEncodingStrategy.LONGS

indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTuningConfigTest.java

@@ -79,7 +79,7 @@ public class CompactionTuningConfigTest
         null,
         new DynamicPartitionsSpec(100, 100L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -121,7 +121,7 @@ public class CompactionTuningConfigTest
         null,
         new DynamicPartitionsSpec(100, 100L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -164,7 +164,7 @@ public class CompactionTuningConfigTest
         null,
         new DynamicPartitionsSpec(100, 100L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS

indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskSerdeTest.java

@@ -69,7 +69,7 @@ public class IndexTaskSerdeTest
         null,
         new DynamicPartitionsSpec(1000, 2000L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(false),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -106,7 +106,7 @@ public class IndexTaskSerdeTest
         null,
         new HashedPartitionsSpec(null, 10, ImmutableList.of("dim1", "dim2")),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(false),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -143,7 +143,7 @@ public class IndexTaskSerdeTest
         null,
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(false),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -180,7 +180,7 @@ public class IndexTaskSerdeTest
         ImmutableList.of("dim1", "dim2"),
         null,
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(false),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -219,7 +219,7 @@ public class IndexTaskSerdeTest
         null,
         new DynamicPartitionsSpec(1000, 2000L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(false),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -257,7 +257,7 @@ public class IndexTaskSerdeTest
         null,
         new HashedPartitionsSpec(null, 10, ImmutableList.of("dim1", "dim2")),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(false),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.LZ4,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS

indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskTest.java

@@ -247,7 +247,7 @@ public class ParallelIndexSupervisorTaskTest
           null,
           new HashedPartitionsSpec(null, 10, null),
           new IndexSpec(
-              new RoaringBitmapSerdeFactory(true),
+              RoaringBitmapSerdeFactory.getInstance(),
               CompressionStrategy.UNCOMPRESSED,
               CompressionStrategy.LZF,
               LongEncodingStrategy.LONGS
@@ -319,7 +319,7 @@ public class ParallelIndexSupervisorTaskTest
           null,
           new HashedPartitionsSpec(null, 10, null),
           new IndexSpec(
-              new RoaringBitmapSerdeFactory(true),
+              RoaringBitmapSerdeFactory.getInstance(),
               CompressionStrategy.UNCOMPRESSED,
               CompressionStrategy.LZF,
               LongEncodingStrategy.LONGS

indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexTuningConfigTest.java

@@ -79,7 +79,7 @@ public class ParallelIndexTuningConfigTest
         null,
         new DynamicPartitionsSpec(100, 100L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -126,7 +126,7 @@ public class ParallelIndexTuningConfigTest
         null,
         new DynamicPartitionsSpec(100, 100L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -173,7 +173,7 @@ public class ParallelIndexTuningConfigTest
         null,
         new DynamicPartitionsSpec(100, 100L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -222,7 +222,7 @@ public class ParallelIndexTuningConfigTest
         null,
         new DynamicPartitionsSpec(100, 100L),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -268,7 +268,7 @@ public class ParallelIndexTuningConfigTest
         null,
         new HashedPartitionsSpec(null, 10, null),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -314,7 +314,7 @@ public class ParallelIndexTuningConfigTest
         null,
         new SingleDimensionPartitionsSpec(100, null, "dim", false),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS
@@ -360,7 +360,7 @@ public class ParallelIndexTuningConfigTest
         null,
         new DynamicPartitionsSpec(100, null),
         new IndexSpec(
-            new RoaringBitmapSerdeFactory(true),
+            RoaringBitmapSerdeFactory.getInstance(),
             CompressionStrategy.UNCOMPRESSED,
             CompressionStrategy.LZF,
             LongEncodingStrategy.LONGS

integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java

@@ -458,8 +458,8 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
           fullDatasourceName,
           AutoCompactionSnapshot.AutoCompactionScheduleStatus.RUNNING,
           0,
-          14586,
+          13702,
-          14585,
+          13701,
           0,
           2,
           2,
@@ -476,7 +476,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
           fullDatasourceName,
           AutoCompactionSnapshot.AutoCompactionScheduleStatus.RUNNING,
           0,
-          22892,
+          21566,
           0,
           0,
           3,
@@ -592,8 +592,8 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
       getAndAssertCompactionStatus(
           fullDatasourceName,
           AutoCompactionSnapshot.AutoCompactionScheduleStatus.RUNNING,
-          14586,
+          13702,
-          14585,
+          13701,
           0,
           2,
           2,
@@ -601,7 +601,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
           1,
           1,
           0);
-      Assert.assertEquals(compactionResource.getCompactionProgress(fullDatasourceName).get("remainingSegmentSize"), "14586");
+      Assert.assertEquals(compactionResource.getCompactionProgress(fullDatasourceName).get("remainingSegmentSize"), "13702");
       // Run compaction again to compact the remaining day
       // Remaining day compacted (1 new segment). Now both days compacted (2 total)
       forceTriggerAutoCompaction(2);
@@ -612,7 +612,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
           fullDatasourceName,
           AutoCompactionSnapshot.AutoCompactionScheduleStatus.RUNNING,
           0,
-          22892,
+          21566,
           0,
           0,
           3,

processing/src/main/java/org/apache/druid/collections/bitmap/RoaringBitmapFactory.java

@@ -35,7 +35,7 @@ import java.util.Iterator;
  */
 public class RoaringBitmapFactory implements BitmapFactory
 {
-  static final boolean DEFAULT_COMPRESS_RUN_ON_SERIALIZATION = false;
+  public static final RoaringBitmapFactory INSTANCE = new RoaringBitmapFactory();
   private static final ImmutableRoaringBitmap EMPTY_IMMUTABLE_BITMAP;
 
   static {
@@ -54,18 +54,6 @@ public class RoaringBitmapFactory implements BitmapFactory
   private static final WrappedImmutableRoaringBitmap WRAPPED_IMMUTABLE_ROARING_BITMAP =
       new WrappedImmutableRoaringBitmap(EMPTY_IMMUTABLE_BITMAP);
 
-  private final boolean compressRunOnSerialization;
-
-  public RoaringBitmapFactory()
-  {
-    this(DEFAULT_COMPRESS_RUN_ON_SERIALIZATION);
-  }
-
-  public RoaringBitmapFactory(boolean compressRunOnSerialization)
-  {
-    this.compressRunOnSerialization = compressRunOnSerialization;
-  }
-
   private static Iterable<ImmutableRoaringBitmap> unwrap(
       final Iterable<ImmutableBitmap> b
   )
@@ -109,7 +97,7 @@ public class RoaringBitmapFactory implements BitmapFactory
   @Override
   public MutableBitmap makeEmptyMutableBitmap()
   {
-    return new WrappedRoaringBitmap(compressRunOnSerialization);
+    return new WrappedRoaringBitmap();
   }
 
   @Override

processing/src/main/java/org/apache/druid/collections/bitmap/WrappedRoaringBitmap.java

@@ -20,62 +20,59 @@
 package org.apache.druid.collections.bitmap;
 
 import com.google.common.annotations.VisibleForTesting;
+import it.unimi.dsi.fastutil.ints.IntIterators;
+import it.unimi.dsi.fastutil.ints.IntListIterator;
+import org.apache.druid.extendedset.intset.EmptyIntIterator;
 import org.roaringbitmap.IntIterator;
 import org.roaringbitmap.PeekableIntIterator;
-import org.roaringbitmap.RoaringBitmap;
 import org.roaringbitmap.RoaringBitmapWriter;
 import org.roaringbitmap.buffer.MutableRoaringBitmap;
 
+import javax.annotation.Nullable;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 
 public class WrappedRoaringBitmap implements MutableBitmap
 {
-  // attempt to compress long runs prior to serialization (requires RoaringBitmap version 0.5 or better)
+  private static final int ARRAY_SIZE = 4;
-  // this may improve compression greatly in some cases at the expense of slower serialization
+  private static final int NOT_SET = -1;
-  // in the worst case.
+
-  private final boolean compressRunOnSerialization;
   /**
    * Underlying bitmap.
    */
+  @Nullable
   private RoaringBitmapWriter<MutableRoaringBitmap> writer;
 
   /**
-   * Creates a new WrappedRoaringBitmap wrapping an empty MutableRoaringBitmap
+   * Array used instead of {@link #writer} when the number of distinct values is less than {@link #ARRAY_SIZE}.
+   * Saves memory vs. a full {@link RoaringBitmapWriter}.
+   */
+  @Nullable
+  private int[] smallArray;
+
+  /**
+   * Creates a new, empty bitmap.
    */
   public WrappedRoaringBitmap()
   {
-    this(RoaringBitmapFactory.DEFAULT_COMPRESS_RUN_ON_SERIALIZATION);
-  }
-
-  /**
-   * Creates a new WrappedRoaringBitmap wrapping an empty MutableRoaringBitmap
-   *
-   * @param compressRunOnSerialization indicates whether to call {@link RoaringBitmap#runOptimize()} before serializing
-   */
-  public WrappedRoaringBitmap(boolean compressRunOnSerialization)
-  {
-    this.writer = RoaringBitmapWriter.bufferWriter().get();
-    this.compressRunOnSerialization = compressRunOnSerialization;
   }
 
   @VisibleForTesting
   public ImmutableBitmap toImmutableBitmap()
   {
+    initializeWriterIfNeeded();
     MutableRoaringBitmap bitmap = writer.get().clone();
-    if (compressRunOnSerialization) {
+    bitmap.runOptimize();
-      bitmap.runOptimize();
-    }
     return new WrappedImmutableRoaringBitmap(bitmap.toImmutableRoaringBitmap());
   }
 
   @Override
   public byte[] toBytes()
   {
+    initializeWriterIfNeeded();
     try {
       MutableRoaringBitmap bitmap = writer.get();
-      if (compressRunOnSerialization) {
+      bitmap.runOptimize();
-        bitmap.runOptimize();
-      }
       ByteBuffer buffer = ByteBuffer.allocate(bitmap.serializedSizeInBytes());
       bitmap.serialize(buffer);
       return buffer.array();
@@ -88,83 +85,173 @@ public class WrappedRoaringBitmap implements MutableBitmap
   @Override
   public void clear()
   {
-    this.writer.reset();
+    this.writer = null;
+    this.smallArray = null;
   }
 
   @Override
   public void or(MutableBitmap mutableBitmap)
   {
+    initializeWriterIfNeeded();
     WrappedRoaringBitmap other = (WrappedRoaringBitmap) mutableBitmap;
+    other.initializeWriterIfNeeded();
     MutableRoaringBitmap unwrappedOtherBitmap = other.writer.get();
     writer.get().or(unwrappedOtherBitmap);
   }
 
-
   @Override
   public int getSizeInBytes()
   {
+    initializeWriterIfNeeded();
     MutableRoaringBitmap bitmap = writer.get();
-    if (compressRunOnSerialization) {
+    bitmap.runOptimize();
-      bitmap.runOptimize();
-    }
     return bitmap.serializedSizeInBytes();
   }
 
   @Override
   public void add(int entry)
   {
-    writer.add(entry);
+    if (entry < 0) {
+      throw new IllegalArgumentException("Cannot add negative ints");
+    } else if (writer != null) {
+      writer.add(entry);
+    } else {
+      if (smallArray == null) {
+        smallArray = new int[ARRAY_SIZE];
+        Arrays.fill(smallArray, NOT_SET);
+      }
+
+      for (int i = 0; i < smallArray.length; i++) {
+        if (smallArray[i] == NOT_SET) {
+          if (i > 0 && entry <= smallArray[i - 1]) {
+            // Can't handle nonascending order with smallArray
+            break;
+          }
+
+          smallArray[i] = entry;
+          return;
+        }
+      }
+
+      initializeWriterIfNeeded();
+      smallArray = null;
+      writer.add(entry);
+    }
   }
 
   @Override
   public int size()
   {
-    return writer.get().getCardinality();
+    if (writer != null) {
+      return writer.get().getCardinality();
+    } else if (smallArray != null) {
+      for (int i = 0; i < smallArray.length; i++) {
+        if (smallArray[i] == NOT_SET) {
+          return i;
+        }
+      }
+
+      return ARRAY_SIZE;
+    } else {
+      return 0;
+    }
   }
 
   public void serialize(ByteBuffer buffer)
   {
+    initializeWriterIfNeeded();
     MutableRoaringBitmap bitmap = writer.get();
-    if (compressRunOnSerialization) {
+    bitmap.runOptimize();
-      bitmap.runOptimize();
-    }
     bitmap.serialize(buffer);
   }
 
   @Override
   public String toString()
   {
-    return getClass().getSimpleName() + writer.getUnderlying();
+    if (writer != null) {
+      return getClass().getSimpleName() + writer.getUnderlying();
+    } else if (smallArray != null) {
+      return getClass().getSimpleName() + Arrays.toString(smallArray);
+    } else {
+      return getClass().getSimpleName() + "[]";
+    }
   }
 
   @Override
   public void remove(int entry)
   {
+    initializeWriterIfNeeded();
     writer.get().remove(entry);
   }
 
   @Override
   public IntIterator iterator()
   {
-    return writer.get().getIntIterator();
+    if (writer != null) {
+      return writer.get().getIntIterator();
+    } else if (smallArray != null) {
+      final int sz = size();
+
+      class SmallArrayIterator implements IntIterator
+      {
+        private final IntListIterator iterator;
+
+        public SmallArrayIterator(IntListIterator iterator)
+        {
+          this.iterator = iterator;
+        }
+
+        @Override
+        public IntIterator clone()
+        {
+          return new SmallArrayIterator(IntIterators.wrap(smallArray, 0, sz));
+        }
+
+        @Override
+        public boolean hasNext()
+        {
+          return iterator.hasNext();
+        }
+
+        @Override
+        public int next()
+        {
+          return iterator.nextInt();
+        }
+      }
+
+      return new SmallArrayIterator(IntIterators.wrap(smallArray, 0, sz));
+    } else {
+      return EmptyIntIterator.instance();
+    }
   }
 
   @Override
   public PeekableIntIterator peekableIterator()
   {
-    return writer.get().getIntIterator();
+    if (writer != null) {
+      return writer.get().getIntIterator();
+    } else {
+      return new PeekableIteratorAdapter<>(iterator());
+    }
   }
 
   @Override
   public boolean isEmpty()
   {
-    return writer.get().isEmpty();
+    if (writer != null) {
+      return writer.get().isEmpty();
+    } else {
+      return smallArray == null;
+    }
   }
 
   @Override
   public ImmutableBitmap intersection(ImmutableBitmap otherBitmap)
   {
+    initializeWriterIfNeeded();
     WrappedRoaringBitmap other = (WrappedRoaringBitmap) otherBitmap;
+    other.initializeWriterIfNeeded();
     MutableRoaringBitmap unwrappedOtherBitmap = other.writer.get();
     return new WrappedImmutableRoaringBitmap(MutableRoaringBitmap.and(writer.get(), unwrappedOtherBitmap));
   }
@@ -172,6 +259,37 @@ public class WrappedRoaringBitmap implements MutableBitmap
   @Override
   public boolean get(int value)
   {
-    return writer.get().contains(value);
+    if (value < 0) {
+      return false;
+    } else if (writer != null) {
+      return writer.get().contains(value);
+    } else if (smallArray != null) {
+      for (int i : smallArray) {
+        if (i == value) {
+          return true;
+        }
+      }
+
+      return false;
+    } else {
+      return false;
+    }
+  }
+
+  private void initializeWriterIfNeeded()
+  {
+    if (writer == null) {
+      writer = RoaringBitmapWriter.bufferWriter().get();
+
+      if (smallArray != null) {
+        for (int i : smallArray) {
+          if (i != NOT_SET) {
+            writer.add(i);
+          }
+        }
+
+        smallArray = null;
+      }
+    }
   }
 }

processing/src/main/java/org/apache/druid/segment/BitmapOffset.java

@@ -19,16 +19,15 @@
 
 package org.apache.druid.segment;
 
-import org.apache.druid.collections.bitmap.BitmapFactory;
 import org.apache.druid.collections.bitmap.ImmutableBitmap;
 import org.apache.druid.collections.bitmap.MutableBitmap;
+import org.apache.druid.collections.bitmap.RoaringBitmapFactory;
 import org.apache.druid.collections.bitmap.WrappedImmutableRoaringBitmap;
 import org.apache.druid.extendedset.intset.EmptyIntIterator;
 import org.apache.druid.java.util.common.RE;
 import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
 import org.apache.druid.segment.data.Offset;
 import org.apache.druid.segment.data.ReadableOffset;
-import org.apache.druid.segment.data.RoaringBitmapSerdeFactory;
 import org.roaringbitmap.IntIterator;
 
 import java.util.Arrays;
@@ -39,7 +38,6 @@ import java.util.HashSet;
 public class BitmapOffset extends Offset
 {
   private static final int INVALID_VALUE = -1;
-  private static final BitmapFactory ROARING_BITMAP_FACTORY = new RoaringBitmapSerdeFactory(false).getBitmapFactory();
 
   /**
    * Currently the default stops are not consciously optimized for the goals described in {@link #factorizeFullness}.
@@ -133,12 +131,12 @@ public class BitmapOffset extends Offset
   {
     ImmutableBitmap roaringBitmap = bitmapIndex;
     if (!(bitmapIndex instanceof WrappedImmutableRoaringBitmap)) {
-      final MutableBitmap bitmap = ROARING_BITMAP_FACTORY.makeEmptyMutableBitmap();
+      final MutableBitmap bitmap = RoaringBitmapFactory.INSTANCE.makeEmptyMutableBitmap();
       final IntIterator iterator = bitmapIndex.iterator();
       while (iterator.hasNext()) {
         bitmap.add(iterator.next());
       }
-      roaringBitmap = ROARING_BITMAP_FACTORY.makeImmutableBitmap(bitmap);
+      roaringBitmap = RoaringBitmapFactory.INSTANCE.makeImmutableBitmap(bitmap);
     }
     return ((WrappedImmutableRoaringBitmap) roaringBitmap).getBitmap().getReverseIntIterator();
   }

processing/src/main/java/org/apache/druid/segment/data/BitmapSerde.java

@@ -20,6 +20,7 @@
 package org.apache.druid.segment.data;
 
 import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.druid.collections.bitmap.RoaringBitmapFactory;
 
 public class BitmapSerde
 {
@@ -32,7 +33,7 @@ public class BitmapSerde
   {
     public DefaultBitmapSerdeFactory()
     {
-      super(RoaringBitmapSerdeFactory.DEFAULT_COMPRESS_RUN_ON_SERIALIZATION);
+      super(RoaringBitmapFactory.INSTANCE);
     }
   }
 

processing/src/main/java/org/apache/druid/segment/data/RoaringBitmapSerdeFactory.java

@@ -20,7 +20,6 @@
 package org.apache.druid.segment.data;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.druid.collections.bitmap.BitmapFactory;
 import org.apache.druid.collections.bitmap.ImmutableBitmap;
 import org.apache.druid.collections.bitmap.RoaringBitmapFactory;
@@ -34,27 +33,20 @@ import java.nio.ByteBuffer;
  */
 public class RoaringBitmapSerdeFactory implements BitmapSerdeFactory
 {
-  public static final boolean DEFAULT_COMPRESS_RUN_ON_SERIALIZATION = true;
+  public static final RoaringBitmapSerdeFactory INSTANCE = new RoaringBitmapSerdeFactory(RoaringBitmapFactory.INSTANCE);
   private static final ObjectStrategy<ImmutableBitmap> OBJECT_STRATEGY = new ImmutableRoaringBitmapObjectStrategy();
 
-  private final boolean compressRunOnSerialization;
   private final BitmapFactory bitmapFactory;
 
-  @JsonCreator
+  protected RoaringBitmapSerdeFactory(BitmapFactory bitmapFactory)
-  public RoaringBitmapSerdeFactory(
-      @JsonProperty("compressRunOnSerialization") @Nullable Boolean compressRunOnSerialization
-  )
   {
-    this.compressRunOnSerialization = compressRunOnSerialization == null
+    this.bitmapFactory = bitmapFactory;
-                                      ? DEFAULT_COMPRESS_RUN_ON_SERIALIZATION
-                                      : compressRunOnSerialization;
-    this.bitmapFactory = new RoaringBitmapFactory(this.compressRunOnSerialization);
   }
 
-  @JsonProperty
+  @JsonCreator
-  public boolean getCompressRunOnSerialization()
+  public static RoaringBitmapSerdeFactory getInstance()
   {
-    return compressRunOnSerialization;
+    return INSTANCE;
   }
 
   @Override

processing/src/main/java/org/apache/druid/segment/nested/GlobalDictionaryEncodedFieldColumnWriter.java

@@ -220,10 +220,12 @@ public abstract class GlobalDictionaryEncodedFieldColumnWriter<T>
       bitmaps[sortedLocalId].add(rowCount++);
     }
 
-    for (MutableBitmap bitmap : bitmaps) {
+    for (int i = 0; i < bitmaps.length; i++) {
+      final MutableBitmap bitmap = bitmaps[i];
       bitmapIndexWriter.write(
           indexSpec.getBitmapSerdeFactory().getBitmapFactory().makeImmutableBitmap(bitmap)
       );
+      bitmaps[i] = null; // Reclaim memory
     }
 
     final Serializer fieldSerializer = new Serializer()

processing/src/test/java/org/apache/druid/collections/bitmap/WrappedRoaringBitmapTest.java

@@ -19,68 +19,114 @@
 
 package org.apache.druid.collections.bitmap;
 
-import junit.framework.Assert;
+import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
+import org.roaringbitmap.IntIterator;
 
 import java.nio.ByteBuffer;
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.List;
 
 @RunWith(Parameterized.class)
 public class WrappedRoaringBitmapTest
 {
-  private final RoaringBitmapFactory factory;
+  private static final int[] DATA = new int[]{1, 3, 5, 7, 9, 10, 11, 100, 122};
 
-  public WrappedRoaringBitmapTest(RoaringBitmapFactory factory)
+  private final int cardinality;
+  private WrappedRoaringBitmap bitmap;
+
+  public WrappedRoaringBitmapTest(int cardinality)
   {
-    this.factory = factory;
+    this.cardinality = cardinality;
   }
 
   @Parameterized.Parameters
-  public static List<RoaringBitmapFactory[]> factoryClasses()
+  public static List<Object[]> constructorFeeder()
   {
-    return Arrays.asList(
+    final List<Object[]> constructors = new ArrayList<>();
-        new RoaringBitmapFactory[] {
+    for (int i = 0; i < DATA.length; i++) {
-            new RoaringBitmapFactory(false)
+      constructors.add(new Object[]{i});
-        },
+    }
-        new RoaringBitmapFactory[] {
+    return constructors;
-            new RoaringBitmapFactory(true)
-        }
-    );
   }
 
-  private WrappedRoaringBitmap createWrappedRoaringBitmap()
+  @Before
+  public void setUp()
   {
-    WrappedRoaringBitmap set = (WrappedRoaringBitmap) factory.makeEmptyMutableBitmap();
+    bitmap = (WrappedRoaringBitmap) RoaringBitmapFactory.INSTANCE.makeEmptyMutableBitmap();
-    set.add(1);
+    for (int i = 0; i < cardinality; i++) {
-    set.add(3);
+      bitmap.add(DATA[i]);
-    set.add(5);
+    }
-    set.add(7);
+  }
-    set.add(9);
+
-    return set;
+  @Test
+  public void testGet()
+  {
+    for (int i = 0; i < DATA.length; i++) {
+      Assert.assertEquals(String.valueOf(i), i < cardinality, bitmap.get(DATA[i]));
+    }
+
+    Assert.assertFalse(bitmap.get(-1));
+    Assert.assertFalse(bitmap.get(Integer.MAX_VALUE));
+  }
+
+  @Test
+  public void testSize()
+  {
+    Assert.assertEquals(cardinality, bitmap.size());
+  }
+
+  @Test
+  public void testRemove()
+  {
+    bitmap.remove(Integer.MAX_VALUE);
+    Assert.assertEquals(cardinality, bitmap.size());
+
+    if (cardinality > 0) {
+      bitmap.remove(DATA[0]);
+      Assert.assertEquals(cardinality - 1, bitmap.size());
+    }
+  }
+
+  @Test
+  public void testClear()
+  {
+    bitmap.clear();
+    Assert.assertEquals(0, bitmap.size());
+  }
+
+  @Test
+  public void testIterator()
+  {
+    final IntIterator iterator = bitmap.iterator();
+
+    int i = 0;
+    while (iterator.hasNext()) {
+      final int n = iterator.next();
+      Assert.assertEquals(String.valueOf(i), DATA[i], n);
+      i++;
+    }
+    Assert.assertEquals("number of elements", i, cardinality);
   }
 
   @Test
   public void testSerialize()
   {
-    WrappedRoaringBitmap set = createWrappedRoaringBitmap();
+    byte[] buffer = new byte[bitmap.getSizeInBytes()];
-
-    byte[] buffer = new byte[set.getSizeInBytes()];
     ByteBuffer byteBuffer = ByteBuffer.wrap(buffer);
-    set.serialize(byteBuffer);
+    bitmap.serialize(byteBuffer);
     byteBuffer.flip();
     ImmutableBitmap immutableBitmap = new RoaringBitmapFactory().mapImmutableBitmap(byteBuffer);
-    Assert.assertEquals(5, immutableBitmap.size());
+    Assert.assertEquals(cardinality, immutableBitmap.size());
   }
 
   @Test
   public void testToByteArray()
   {
-    WrappedRoaringBitmap set = createWrappedRoaringBitmap();
+    ImmutableBitmap immutableBitmap = new RoaringBitmapFactory().mapImmutableBitmap(ByteBuffer.wrap(bitmap.toBytes()));
-    ImmutableBitmap immutableBitmap = new RoaringBitmapFactory().mapImmutableBitmap(ByteBuffer.wrap(set.toBytes()));
+    Assert.assertEquals(cardinality, immutableBitmap.size());
-    Assert.assertEquals(5, immutableBitmap.size());
   }
-
 }

processing/src/test/java/org/apache/druid/segment/IndexSpecTest.java

@@ -43,7 +43,7 @@ public class IndexSpecTest
         + ", \"longEncoding\" : \"auto\", \"stringDictionaryEncoding\":{\"type\":\"frontCoded\", \"bucketSize\":16}}";
 
     final IndexSpec spec = objectMapper.readValue(json, IndexSpec.class);
-    Assert.assertEquals(new RoaringBitmapSerdeFactory(null), spec.getBitmapSerdeFactory());
+    Assert.assertEquals(RoaringBitmapSerdeFactory.getInstance(), spec.getBitmapSerdeFactory());
     Assert.assertEquals(CompressionStrategy.LZ4, spec.getDimensionCompression());
     Assert.assertEquals(CompressionStrategy.LZF, spec.getMetricCompression());
     Assert.assertEquals(CompressionFactory.LongEncodingStrategy.AUTO, spec.getLongEncoding());

processing/src/test/java/org/apache/druid/segment/RoaringBitmapIndexMergerV9Test.java

@@ -37,7 +37,7 @@ public class RoaringBitmapIndexMergerV9Test extends IndexMergerTestBase
   )
   {
     super(
-        new RoaringBitmapSerdeFactory(null),
+        RoaringBitmapSerdeFactory.getInstance(),
         compressionStrategy,
         dimCompressionStrategy,
         longEncodingStrategy

processing/src/test/java/org/apache/druid/segment/data/BitmapSerdeFactoryTest.java

@@ -30,12 +30,10 @@ public class BitmapSerdeFactoryTest
   public void testSerialization() throws Exception
   {
     ObjectMapper mapper = new DefaultObjectMapper();
-    Assert.assertEquals("{\"type\":\"roaring\",\"compressRunOnSerialization\":true}", mapper.writeValueAsString(new RoaringBitmapSerdeFactory(null)));
+    Assert.assertEquals("{\"type\":\"roaring\"}", mapper.writeValueAsString(RoaringBitmapSerdeFactory.getInstance()));
-    Assert.assertEquals("{\"type\":\"roaring\",\"compressRunOnSerialization\":false}", mapper.writeValueAsString(new RoaringBitmapSerdeFactory(false)));
-    Assert.assertEquals("{\"type\":\"roaring\",\"compressRunOnSerialization\":true}", mapper.writeValueAsString(new RoaringBitmapSerdeFactory(true)));
     Assert.assertEquals("{\"type\":\"concise\"}", mapper.writeValueAsString(new ConciseBitmapSerdeFactory()));
     Assert.assertEquals("{\"type\":\"concise\"}", mapper.writeValueAsString(BitmapSerde.createLegacyFactory()));
-    Assert.assertEquals("{\"type\":\"roaring\",\"compressRunOnSerialization\":true}", mapper.writeValueAsString(new BitmapSerde.DefaultBitmapSerdeFactory()));
+    Assert.assertEquals("{\"type\":\"roaring\"}", mapper.writeValueAsString(new BitmapSerde.DefaultBitmapSerdeFactory()));
     Assert.assertEquals("{\"type\":\"concise\"}", mapper.writeValueAsString(new BitmapSerde.LegacyBitmapSerdeFactory()));
   }
 
@@ -45,15 +43,6 @@ public class BitmapSerdeFactoryTest
     ObjectMapper mapper = new DefaultObjectMapper();
     final BitmapSerdeFactory roaringFactory = mapper.readValue("{\"type\":\"roaring\"}", BitmapSerdeFactory.class);
     Assert.assertTrue(roaringFactory instanceof RoaringBitmapSerdeFactory);
-    Assert.assertTrue(((RoaringBitmapSerdeFactory) roaringFactory).getCompressRunOnSerialization());
-
-    final BitmapSerdeFactory compressingRoaringFactory = mapper.readValue(
-        "{\"type\":\"roaring\", \"compressRunOnSerialization\":false}",
-        BitmapSerdeFactory.class
-    );
-    Assert.assertTrue(compressingRoaringFactory instanceof RoaringBitmapSerdeFactory);
-    Assert.assertFalse(((RoaringBitmapSerdeFactory) compressingRoaringFactory).getCompressRunOnSerialization());
-
     Assert.assertTrue(mapper.readValue("{\"type\":\"concise\"}", BitmapSerdeFactory.class) instanceof ConciseBitmapSerdeFactory);
     Assert.assertTrue(mapper.readValue("{\"type\":\"BitmapSerde$SomeRandomClass\"}", BitmapSerdeFactory.class) instanceof RoaringBitmapSerdeFactory);
   }

processing/src/test/java/org/apache/druid/segment/filter/BaseFilterTest.java

@@ -299,7 +299,7 @@ public abstract class BaseFilterTest extends InitializedNullHandlingTest
 
     final Map<String, BitmapSerdeFactory> bitmapSerdeFactories = ImmutableMap.of(
         "concise", new ConciseBitmapSerdeFactory(),
-        "roaring", new RoaringBitmapSerdeFactory(true)
+        "roaring", RoaringBitmapSerdeFactory.getInstance()
     );
 
     final Map<String, SegmentWriteOutMediumFactory> segmentWriteOutMediumFactories = ImmutableMap.of(

processing/src/test/java/org/apache/druid/segment/filter/ExtractionDimFilterTest.java

@@ -69,7 +69,7 @@ public class ExtractionDimFilterTest extends InitializedNullHandlingTest
   {
     return ImmutableList.of(
         new Object[]{new ConciseBitmapFactory(), new ConciseBitmapSerdeFactory()},
-        new Object[]{new RoaringBitmapFactory(), new RoaringBitmapSerdeFactory(null)}
+        new Object[]{new RoaringBitmapFactory(), RoaringBitmapSerdeFactory.getInstance()}
     );
   }
 

processing/src/test/java/org/apache/druid/segment/nested/NestedFieldLiteralColumnIndexSupplierTest.java

@@ -62,7 +62,7 @@ import java.util.TreeSet;
 
 public class NestedFieldLiteralColumnIndexSupplierTest extends InitializedNullHandlingTest
 {
-  BitmapSerdeFactory roaringFactory = new RoaringBitmapSerdeFactory(null);
+  BitmapSerdeFactory roaringFactory = RoaringBitmapSerdeFactory.getInstance();
   BitmapResultFactory<ImmutableBitmap> bitmapResultFactory = new DefaultBitmapResultFactory(
       roaringFactory.getBitmapFactory()
   );

processing/src/test/java/org/apache/druid/segment/serde/DictionaryEncodedStringIndexSupplierTest.java

@@ -44,7 +44,7 @@ import java.util.TreeSet;
 
 public class DictionaryEncodedStringIndexSupplierTest extends InitializedNullHandlingTest
 {
-  BitmapSerdeFactory roaringFactory = new RoaringBitmapSerdeFactory(null);
+  BitmapSerdeFactory roaringFactory = RoaringBitmapSerdeFactory.getInstance();
   BitmapResultFactory<ImmutableBitmap> bitmapResultFactory = new DefaultBitmapResultFactory(
       roaringFactory.getBitmapFactory()
   );

processing/src/test/java/org/apache/druid/segment/serde/NullColumnPartSerdeTest.java

@@ -57,7 +57,7 @@ public class NullColumnPartSerdeTest extends InitializedNullHandlingTest
   {
     final ObjectMapper mapper = new DefaultObjectMapper();
 
-    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, new RoaringBitmapSerdeFactory(null));
+    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, RoaringBitmapSerdeFactory.getInstance());
     final String json = mapper.writeValueAsString(partSerde);
     Assert.assertEquals(partSerde, mapper.readValue(json, ColumnPartSerde.class));
   }
@@ -65,7 +65,7 @@ public class NullColumnPartSerdeTest extends InitializedNullHandlingTest
   @Test
   public void testDeserializer()
   {
-    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, new RoaringBitmapSerdeFactory(null));
+    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, RoaringBitmapSerdeFactory.getInstance());
     final ColumnBuilder builder = new ColumnBuilder().setType(ValueType.DOUBLE);
     partSerde.getDeserializer().read(EMPTY_BUFFER, builder, A_CONFIG);
     final ColumnCapabilities columnCapabilities = builder.build().getCapabilities();
@@ -82,7 +82,7 @@ public class NullColumnPartSerdeTest extends InitializedNullHandlingTest
   @Test
   public void testDimensionSelector()
   {
-    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, new RoaringBitmapSerdeFactory(null));
+    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, RoaringBitmapSerdeFactory.getInstance());
     final ColumnBuilder builder = new ColumnBuilder().setType(ValueType.STRING);
     partSerde.getDeserializer().read(EMPTY_BUFFER, builder, A_CONFIG);
     ColumnHolder holder = builder.build();
@@ -104,7 +104,7 @@ public class NullColumnPartSerdeTest extends InitializedNullHandlingTest
   @Test
   public void testDimensionVectorSelector()
   {
-    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, new RoaringBitmapSerdeFactory(null));
+    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, RoaringBitmapSerdeFactory.getInstance());
     final ColumnBuilder builder = new ColumnBuilder().setType(ValueType.STRING);
     partSerde.getDeserializer().read(EMPTY_BUFFER, builder, A_CONFIG);
     ColumnHolder holder = builder.build();
@@ -132,7 +132,7 @@ public class NullColumnPartSerdeTest extends InitializedNullHandlingTest
   @Test
   public void testVectorObjectSelector()
   {
-    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, new RoaringBitmapSerdeFactory(null));
+    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, RoaringBitmapSerdeFactory.getInstance());
     final ColumnBuilder builder = new ColumnBuilder().setType(ValueType.STRING);
     partSerde.getDeserializer().read(EMPTY_BUFFER, builder, A_CONFIG);
     ColumnHolder holder = builder.build();
@@ -151,7 +151,7 @@ public class NullColumnPartSerdeTest extends InitializedNullHandlingTest
   @Test
   public void testColumnValueSelector()
   {
-    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, new RoaringBitmapSerdeFactory(null));
+    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, RoaringBitmapSerdeFactory.getInstance());
     final ColumnBuilder builder = new ColumnBuilder().setType(ValueType.DOUBLE);
     partSerde.getDeserializer().read(EMPTY_BUFFER, builder, A_CONFIG);
     ColumnHolder holder = builder.build();
@@ -172,7 +172,7 @@ public class NullColumnPartSerdeTest extends InitializedNullHandlingTest
   @Test
   public void testVectorValueSelector()
   {
-    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, new RoaringBitmapSerdeFactory(null));
+    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, RoaringBitmapSerdeFactory.getInstance());
     final ColumnBuilder builder = new ColumnBuilder().setType(ValueType.DOUBLE);
     partSerde.getDeserializer().read(EMPTY_BUFFER, builder, A_CONFIG);
     ColumnHolder holder = builder.build();
@@ -196,7 +196,7 @@ public class NullColumnPartSerdeTest extends InitializedNullHandlingTest
   @Test
   public void testIndexSupplier()
   {
-    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, new RoaringBitmapSerdeFactory(null));
+    final NullColumnPartSerde partSerde = new NullColumnPartSerde(10, RoaringBitmapSerdeFactory.getInstance());
     final ColumnBuilder builder = new ColumnBuilder().setType(ValueType.DOUBLE);
     partSerde.getDeserializer().read(EMPTY_BUFFER, builder, A_CONFIG);
     ColumnHolder holder = builder.build();

server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java

@@ -2373,7 +2373,7 @@ public class CompactSegmentsTest
                 transformSpec,
                 ImmutableMap.of(
                     "bitmap",
-                    ImmutableMap.of("type", "roaring", "compressRunOnSerialization", true),
+                    ImmutableMap.of("type", "roaring"),
                     "dimensionCompression",
                     "lz4",
                     "metricCompression",

services/src/main/java/org/apache/druid/cli/DumpSegment.java

@@ -364,7 +364,7 @@ public class DumpSegment extends GuiceRunnable
     if (bitmapFactory instanceof ConciseBitmapFactory) {
       bitmapSerdeFactory = new ConciseBitmapSerdeFactory();
     } else if (bitmapFactory instanceof RoaringBitmapFactory) {
-      bitmapSerdeFactory = new RoaringBitmapSerdeFactory(null);
+      bitmapSerdeFactory = RoaringBitmapSerdeFactory.getInstance();
     } else {
       throw new ISE(
           "Don't know which BitmapSerdeFactory to use for BitmapFactory[%s]!",
@@ -565,7 +565,7 @@ public class DumpSegment extends GuiceRunnable
     if (bitmapFactory instanceof ConciseBitmapFactory) {
       bitmapSerdeFactory = new ConciseBitmapSerdeFactory();
     } else if (bitmapFactory instanceof RoaringBitmapFactory) {
-      bitmapSerdeFactory = new RoaringBitmapSerdeFactory(null);
+      bitmapSerdeFactory = RoaringBitmapSerdeFactory.getInstance();
     } else {
       throw new ISE(
           "Don't know which BitmapSerdeFactory to use for BitmapFactory[%s]!",

services/src/test/java/org/apache/druid/cli/DumpSegmentTest.java

@@ -191,7 +191,7 @@ public class DumpSegmentTest extends InitializedNullHandlingTest
     );
     final byte[] fileBytes = Files.readAllBytes(outputFile.toPath());
     final String output = StringUtils.fromUtf8(fileBytes);
-    Assert.assertEquals("{\"bitmapSerdeFactory\":{\"type\":\"roaring\",\"compressRunOnSerialization\":true},\"nest\":{\"$.x\":{\"types\":[\"LONG\"],\"dictionary\":[{\"localId\":0,\"globalId\":0,\"value\":null,\"rows\":[4]},{\"localId\":1,\"globalId\":3,\"value\":\"100\",\"rows\":[3]},{\"localId\":2,\"globalId\":4,\"value\":\"200\",\"rows\":[0,2]},{\"localId\":3,\"globalId\":5,\"value\":\"400\",\"rows\":[1]}],\"column\":[{\"row\":0,\"raw\":{\"x\":200,\"y\":2.2},\"fieldId\":2,\"fieldValue\":\"200\"},{\"row\":1,\"raw\":{\"x\":400,\"y\":1.1,\"z\":\"a\"},\"fieldId\":3,\"fieldValue\":\"400\"},{\"row\":2,\"raw\":{\"x\":200,\"z\":\"b\"},\"fieldId\":2,\"fieldValue\":\"200\"},{\"row\":3,\"raw\":{\"x\":100,\"y\":1.1,\"z\":\"a\"},\"fieldId\":1,\"fieldValue\":\"100\"},{\"row\":4,\"raw\":{\"y\":3.3,\"z\":\"b\"},\"fieldId\":0,\"fieldValue\":null}]}}}", output);
+    Assert.assertEquals("{\"bitmapSerdeFactory\":{\"type\":\"roaring\"},\"nest\":{\"$.x\":{\"types\":[\"LONG\"],\"dictionary\":[{\"localId\":0,\"globalId\":0,\"value\":null,\"rows\":[4]},{\"localId\":1,\"globalId\":3,\"value\":\"100\",\"rows\":[3]},{\"localId\":2,\"globalId\":4,\"value\":\"200\",\"rows\":[0,2]},{\"localId\":3,\"globalId\":5,\"value\":\"400\",\"rows\":[1]}],\"column\":[{\"row\":0,\"raw\":{\"x\":200,\"y\":2.2},\"fieldId\":2,\"fieldValue\":\"200\"},{\"row\":1,\"raw\":{\"x\":400,\"y\":1.1,\"z\":\"a\"},\"fieldId\":3,\"fieldValue\":\"400\"},{\"row\":2,\"raw\":{\"x\":200,\"z\":\"b\"},\"fieldId\":2,\"fieldValue\":\"200\"},{\"row\":3,\"raw\":{\"x\":100,\"y\":1.1,\"z\":\"a\"},\"fieldId\":1,\"fieldValue\":\"100\"},{\"row\":4,\"raw\":{\"y\":3.3,\"z\":\"b\"},\"fieldId\":0,\"fieldValue\":null}]}}}", output);
   }
 
   public static List<Segment> createSegments(

web-console/src/dialogs/index-spec-dialog/__snapshots__/index-spec-dialog.spec.tsx.snap

@@ -56,16 +56,6 @@ exports[`IndexSpecDialog matches snapshot with indexSpec 1`] = `
             ],
             "type": "string",
           },
-          Object {
-            "defaultValue": true,
-            "defined": [Function],
-            "info": <React.Fragment>
-              Controls whether or not run-length encoding will be used when it is determined to be more space-efficient.
-            </React.Fragment>,
-            "label": "Bitmap compress run on serialization",
-            "name": "bitmap.compressRunOnSerialization",
-            "type": "boolean",
-          },
           Object {
             "defaultValue": "lz4",
             "info": <React.Fragment>
@@ -216,16 +206,6 @@ exports[`IndexSpecDialog matches snapshot without compactionConfig 1`] = `
             ],
             "type": "string",
           },
-          Object {
-            "defaultValue": true,
-            "defined": [Function],
-            "info": <React.Fragment>
-              Controls whether or not run-length encoding will be used when it is determined to be more space-efficient.
-            </React.Fragment>,
-            "label": "Bitmap compress run on serialization",
-            "name": "bitmap.compressRunOnSerialization",
-            "type": "boolean",
-          },
           Object {
             "defaultValue": "lz4",
             "info": <React.Fragment>

web-console/src/druid-models/index-spec/index-spec.tsx

@@ -33,7 +33,6 @@ export interface IndexSpec {
 
 export interface Bitmap {
   type: string;
-  compressRunOnSerialization?: boolean;
 }
 
 export function summarizeIndexSpec(indexSpec: IndexSpec | undefined): string {
@@ -103,19 +102,6 @@ export const INDEX_SPEC_FIELDS: Field<IndexSpec>[] = [
     suggestions: ['roaring', 'concise'],
     info: <>Compression format for bitmap indexes.</>,
   },
-  {
-    name: 'bitmap.compressRunOnSerialization',
-    label: 'Bitmap compress run on serialization',
-    type: 'boolean',
-    defaultValue: true,
-    defined: spec => (deepGet(spec, 'bitmap.type') || 'roaring') === 'roaring',
-    info: (
-      <>
-        Controls whether or not run-length encoding will be used when it is determined to be more
-        space-efficient.
-      </>
-    ),
-  },
 
   {
     name: 'dimensionCompression',

web-console/src/druid-models/stages/stages.mock.ts

@@ -671,14 +671,14 @@ export const STAGES = new Stages(
               maxTotalRows: null,
             },
             indexSpec: {
-              bitmap: { type: 'roaring', compressRunOnSerialization: true },
+              bitmap: { type: 'roaring' },
               dimensionCompression: 'lz4',
               metricCompression: 'lz4',
               longEncoding: 'longs',
               segmentLoader: null,
             },
             indexSpecForIntermediatePersists: {
-              bitmap: { type: 'roaring', compressRunOnSerialization: true },
+              bitmap: { type: 'roaring' },
               dimensionCompression: 'lz4',
               metricCompression: 'lz4',
               longEncoding: 'longs',

website/.spelling

@@ -1099,7 +1099,7 @@ KinesisSupervisorIOConfig
 KinesisSupervisorTuningConfig
 Resharding
 resharding
-LZ4LZFuncompressedLZ4LZ4LZFuncompressednoneLZ4autolongsautolongslongstypeconcisetyperoaringcompressRunOnSerializationtruetypestreamendpointreplicastaskCounttaskCount
+LZ4LZFuncompressedLZ4LZ4LZFuncompressednoneLZ4autolongsautolongslongstypeconcisetyperoaringtypestreamendpointreplicastaskCounttaskCount
 deaggregate
 druid-kinesis-indexing-service
 maxRecordsPerPoll