Support for disabling bitmap indexes.

api/src/main/java/io/druid/data/input/impl/DimensionSchema.java

@@ -120,11 +120,13 @@ public static MultiValueHandling ofDefault()
 
   private final String name;
   private final MultiValueHandling multiValueHandling;
+  private final boolean createBitmapIndex;
 
-  protected DimensionSchema(String name, MultiValueHandling multiValueHandling)
+  protected DimensionSchema(String name, MultiValueHandling multiValueHandling, boolean createBitmapIndex)
   {
     this.name = Preconditions.checkNotNull(name, "Dimension name cannot be null.");
     this.multiValueHandling = multiValueHandling == null ? MultiValueHandling.ofDefault() : multiValueHandling;
+    this.createBitmapIndex = createBitmapIndex;
   }
 
   @JsonProperty
@@ -139,48 +141,50 @@ public MultiValueHandling getMultiValueHandling()
     return multiValueHandling;
   }
 
+  @JsonProperty("createBitmapIndex")
+  public boolean hasBitmapIndex()
+  {
+    return createBitmapIndex;
+  }
+
   @JsonIgnore
   public abstract String getTypeName();
 
   @JsonIgnore
   public abstract ValueType getValueType();
 
   @Override
-  public boolean equals(Object o)
+  public boolean equals(final Object o)
   {
     if (this == o) {
       return true;
     }
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-
-    DimensionSchema that = (DimensionSchema) o;
-
-    if (!name.equals(that.name)) {
-      return false;
-    }
-
-    if (!getValueType().equals(that.getValueType())) {
-      return false;
-    }
-
-    return Objects.equals(multiValueHandling, that.multiValueHandling);
+    final DimensionSchema that = (DimensionSchema) o;
+    return createBitmapIndex == that.createBitmapIndex &&
+           Objects.equals(name, that.name) &&
+           Objects.equals(getTypeName(), that.getTypeName()) &&
+           Objects.equals(getValueType(), that.getValueType()) &&
+           multiValueHandling == that.multiValueHandling;
   }
 
   @Override
   public int hashCode()
   {
-    return Objects.hash(name, getValueType(), multiValueHandling);
+    return Objects.hash(name, multiValueHandling, createBitmapIndex, getTypeName(), getValueType());
   }
 
   @Override
   public String toString()
   {
     return "DimensionSchema{" +
-           "name='" + name + "'" +
-           ", valueType='" + getValueType() + "'" +
-           ", multiValueHandling='" + getMultiValueHandling() + "'" +
-           "}";
+           "name='" + name + '\'' +
+           ", valueType=" + getValueType() +
+           ", typeName=" + getTypeName() +
+           ", multiValueHandling=" + multiValueHandling +
+           ", createBitmapIndex=" + createBitmapIndex +
+           '}';
   }
 }

api/src/main/java/io/druid/data/input/impl/DimensionsSpec.java

@@ -36,6 +36,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 @PublicApi
 public class DimensionsSpec
@@ -56,17 +57,9 @@ public static List<DimensionSchema> getDefaultSchemas(
       final DimensionSchema.MultiValueHandling multiValueHandling
   )
   {
-    return Lists.transform(
-        dimNames,
-        new Function<String, DimensionSchema>()
-        {
-          @Override
-          public DimensionSchema apply(String input)
-          {
-            return new StringDimensionSchema(input, multiValueHandling);
-          }
-        }
-    );
+    return dimNames.stream()
+                   .map(input -> new StringDimensionSchema(input, multiValueHandling, true))
+                   .collect(Collectors.toList());
   }
 
   public static DimensionSchema convertSpatialSchema(SpatialDimensionSchema spatialSchema)

api/src/main/java/io/druid/data/input/impl/DoubleDimensionSchema.java

@@ -19,7 +19,6 @@
 
 package io.druid.data.input.impl;
 
-
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -28,7 +27,7 @@ public class DoubleDimensionSchema extends DimensionSchema
   @JsonCreator
   public DoubleDimensionSchema(@JsonProperty("name") String name)
   {
-    super(name, null);
+    super(name, null, false);
   }
 
   @Override

api/src/main/java/io/druid/data/input/impl/FloatDimensionSchema.java

@@ -30,7 +30,7 @@ public FloatDimensionSchema(
       @JsonProperty("name") String name
   )
   {
-    super(name, null);
+    super(name, null, false);
   }
 
   @Override

api/src/main/java/io/druid/data/input/impl/LongDimensionSchema.java

@@ -30,7 +30,7 @@ public LongDimensionSchema(
       @JsonProperty("name") String name
   )
   {
-    super(name, null);
+    super(name, null, false);
   }
 
   @Override

api/src/main/java/io/druid/data/input/impl/NewSpatialDimensionSchema.java

@@ -43,7 +43,7 @@ public NewSpatialDimensionSchema(
       @JsonProperty("dims") List<String> dims
   )
   {
-    super(name, null);
+    super(name, null, true);
     this.dims = dims;
   }
 

api/src/main/java/io/druid/data/input/impl/StringDimensionSchema.java

@@ -25,6 +25,8 @@
 
 public class StringDimensionSchema extends DimensionSchema
 {
+  private static final boolean DEFAULT_CREATE_BITMAP_INDEX = true;
+
   @JsonCreator
   public static StringDimensionSchema create(String name)
   {
@@ -34,15 +36,16 @@ public static StringDimensionSchema create(String name)
   @JsonCreator
   public StringDimensionSchema(
       @JsonProperty("name") String name,
-      @JsonProperty("multiValueHandling") MultiValueHandling multiValueHandling
+      @JsonProperty("multiValueHandling") MultiValueHandling multiValueHandling,
+      @JsonProperty("createBitmapIndex") Boolean createBitmapIndex
   )
   {
-    super(name, multiValueHandling);
+    super(name, multiValueHandling, createBitmapIndex == null ? DEFAULT_CREATE_BITMAP_INDEX : createBitmapIndex);
   }
 
   public StringDimensionSchema(String name)
   {
-    this(name, null);
+    this(name, null, DEFAULT_CREATE_BITMAP_INDEX);
   }
 
   @Override

api/src/test/java/io/druid/data/input/impl/DimensionSchemaTest.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.data.input.impl;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class DimensionSchemaTest
+{
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+  @Test
+  public void testStringDimensionSchemaSerde() throws Exception
+  {
+    final StringDimensionSchema schema1 = new StringDimensionSchema("foo");
+    Assert.assertEquals(
+        schema1,
+        OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(schema1), DimensionSchema.class)
+    );
+
+    final StringDimensionSchema schema2 = new StringDimensionSchema(
+        "foo",
+        DimensionSchema.MultiValueHandling.ARRAY,
+        false
+    );
+    Assert.assertEquals(
+        schema2,
+        OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(schema2), DimensionSchema.class)
+    );
+  }
+}

api/src/test/java/io/druid/data/input/impl/DimensionsSpecSerdeTest.java

@@ -19,6 +19,7 @@
 
 package io.druid.data.input.impl;
 
+import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import junit.framework.Assert;
 import org.junit.Test;
@@ -30,7 +31,11 @@
  */
 public class DimensionsSpecSerdeTest
 {
-  private final ObjectMapper mapper = new ObjectMapper();
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+  static {
+    OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+  }
 
   @Test
   public void testDimensionsSpecSerde() throws Exception
@@ -59,9 +64,9 @@ public void testDimensionsSpecSerde() throws Exception
                      + "\"spatialDimensions\": [{\"dimName\":\"IMPR\", \"dims\":[\"S\",\"P\",\"Q\",\"R\"]}]"
                      + "}";
 
-    DimensionsSpec actual = mapper.readValue(
-        mapper.writeValueAsString(
-            mapper.readValue(jsonStr, DimensionsSpec.class)
+    DimensionsSpec actual = OBJECT_MAPPER.readValue(
+        OBJECT_MAPPER.writeValueAsString(
+            OBJECT_MAPPER.readValue(jsonStr, DimensionsSpec.class)
         ),
         DimensionsSpec.class
     );

docs/content/ingestion/index.md

@@ -198,7 +198,13 @@ handle all formatting decisions on their own, without using the ParseSpec.
 #### Dimension Schema
 A dimension schema specifies the type and name of a dimension to be ingested.
 
-For example, the following `dimensionsSpec` section from a `dataSchema` ingests one column as Long (`countryNum`), two columns as Float (`userLatitude`, `userLongitude`), and the other columns as Strings:
+For string columns, the dimension schema can also be used to enable or disable bitmap indexing by setting the
+`createBitmapIndex` boolean. By default, bitmap indexes are enabled for all string columns. Only string columns can have
+bitmap indexes; they are not supported for numeric columns.
+
+For example, the following `dimensionsSpec` section from a `dataSchema` ingests one column as Long (`countryNum`), two
+columns as Float (`userLatitude`, `userLongitude`), and the other columns as Strings, with bitmap indexes disabled
+for the `comment` column.
 
 ```json
 "dimensionsSpec" : {
@@ -215,6 +221,11 @@ For example, the following `dimensionsSpec` section from a `dataSchema` ingests
     "country",
     "region",
     "city",
+    {
+      "type": "string",
+      "name": "comment",
+      "createBitmapIndex": false
+    },
     {
       "type": "long",
       "name": "countryNum"
@@ -233,7 +244,6 @@ For example, the following `dimensionsSpec` section from a `dataSchema` ingests
 }
 ```
 
-
 ## GranularitySpec
 
 The default granularity spec is `uniform`, and can be changed by setting the `type` field.

indexing-service/src/main/java/io/druid/indexing/common/task/CompactionTask.java

@@ -354,7 +354,8 @@ private static DimensionsSpec createDimensionsSpec(List<QueryableIndex> queryabl
               createDimensionSchema(
                   column.getCapabilities().getType(),
                   dimension,
-                  dimensionHandler.getMultivalueHandling()
+                  dimensionHandler.getMultivalueHandling(),
+                  column.getCapabilities().hasBitmapIndexes()
               )
           );
         }
@@ -402,7 +403,8 @@ private static List<QueryableIndex> loadSegments(
   private static DimensionSchema createDimensionSchema(
       ValueType type,
       String name,
-      MultiValueHandling multiValueHandling
+      MultiValueHandling multiValueHandling,
+      boolean hasBitmapIndexes
   )
   {
     switch (type) {
@@ -428,7 +430,7 @@ private static DimensionSchema createDimensionSchema(
         );
         return new DoubleDimensionSchema(name);
       case STRING:
-        return new StringDimensionSchema(name, multiValueHandling);
+        return new StringDimensionSchema(name, multiValueHandling, hasBitmapIndexes);
       default:
         throw new ISE("Unsupported value type[%s] for dimension[%s]", type, name);
     }

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

@@ -69,8 +69,8 @@
 import io.druid.segment.column.Column;
 import io.druid.segment.column.ColumnBuilder;
 import io.druid.segment.column.ValueType;
-import io.druid.segment.data.CompressionStrategy;
 import io.druid.segment.data.CompressionFactory.LongEncodingStrategy;
+import io.druid.segment.data.CompressionStrategy;
 import io.druid.segment.data.ListIndexed;
 import io.druid.segment.data.RoaringBitmapSerdeFactory;
 import io.druid.segment.incremental.IncrementalIndex;
@@ -109,13 +109,13 @@ public class CompactionTaskTest
   private static final Interval COMPACTION_INTERVAL = Intervals.of("2017-01-01/2017-06-01");
   private static final Map<Interval, DimensionSchema> MIXED_TYPE_COLUMN_MAP = ImmutableMap.of(
       Intervals.of("2017-01-01/2017-02-01"),
-      new StringDimensionSchema(MIXED_TYPE_COLUMN, null),
+      new StringDimensionSchema(MIXED_TYPE_COLUMN),
       Intervals.of("2017-02-01/2017-03-01"),
-      new StringDimensionSchema(MIXED_TYPE_COLUMN, null),
+      new StringDimensionSchema(MIXED_TYPE_COLUMN),
       Intervals.of("2017-03-01/2017-04-01"),
-      new StringDimensionSchema(MIXED_TYPE_COLUMN, null),
+      new StringDimensionSchema(MIXED_TYPE_COLUMN),
       Intervals.of("2017-04-01/2017-05-01"),
-      new StringDimensionSchema(MIXED_TYPE_COLUMN, null),
+      new StringDimensionSchema(MIXED_TYPE_COLUMN),
       Intervals.of("2017-05-01/2017-06-01"),
       new DoubleDimensionSchema(MIXED_TYPE_COLUMN)
   );
@@ -138,6 +138,7 @@ public static void setup()
     for (int i = 0; i < 5; i++) {
       final StringDimensionSchema schema = new StringDimensionSchema(
           "string_dim_" + i,
+          null,
           null
       );
       DIMENSIONS.put(schema.getName(), schema);

integration-tests/src/test/resources/indexer/wikipedia_index_queries.json

@@ -25,6 +25,11 @@
             "dimensions":[
                 "page"
             ],
+            "filter":{
+                "type":"selector",
+                "dimension":"language",
+                "value":"zh"
+            },
             "aggregations":[
                 {
                     "type":"count",