Implement native in filter

processing/src/main/java/io/druid/query/Druids.java

@@ -32,6 +32,7 @@
 import io.druid.query.dimension.DimensionSpec;
 import io.druid.query.filter.AndDimFilter;
 import io.druid.query.filter.DimFilter;
+import io.druid.query.filter.InDimFilter;
 import io.druid.query.filter.NoopDimFilter;
 import io.druid.query.filter.NotDimFilter;
 import io.druid.query.filter.OrDimFilter;
@@ -464,11 +465,7 @@ public TimeseriesQueryBuilder filters(String dimensionName, String value)
 
     public TimeseriesQueryBuilder filters(String dimensionName, String value, String... values)
     {
-      List<DimFilter> fields = Lists.<DimFilter>newArrayList(new SelectorDimFilter(dimensionName, value));
-      for (String val : values) {
-        fields.add(new SelectorDimFilter(dimensionName, val));
-      }
-      dimFilter = new OrDimFilter(fields);
+      dimFilter = new InDimFilter(dimensionName, Lists.asList(value, values));
       return this;
     }
 
@@ -624,11 +621,7 @@ public SearchQueryBuilder filters(String dimensionName, String value)
 
     public SearchQueryBuilder filters(String dimensionName, String value, String... values)
     {
-      List<DimFilter> fields = Lists.<DimFilter>newArrayList(new SelectorDimFilter(dimensionName, value));
-      for (String val : values) {
-        fields.add(new SelectorDimFilter(dimensionName, val));
-      }
-      dimFilter = new OrDimFilter(fields);
+      dimFilter = new InDimFilter(dimensionName, Lists.asList(value, values));
       return this;
     }
 
@@ -1172,11 +1165,7 @@ public SelectQueryBuilder filters(String dimensionName, String value)
 
     public SelectQueryBuilder filters(String dimensionName, String value, String... values)
     {
-      List<DimFilter> fields = Lists.<DimFilter>newArrayList(new SelectorDimFilter(dimensionName, value));
-      for (String val : values) {
-        fields.add(new SelectorDimFilter(dimensionName, val));
-      }
-      dimFilter = new OrDimFilter(fields);
+      dimFilter = new InDimFilter(dimensionName, Lists.asList(value, values));
       return this;
     }
 

processing/src/main/java/io/druid/query/filter/ExtractionDimFilter.java

@@ -21,9 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
 import com.metamx.common.StringUtils;
 import io.druid.query.extraction.ExtractionFn;
 import io.druid.query.lookup.LookupExtractionFn;
@@ -104,14 +102,7 @@ public DimFilter optimize()
       final List<String> keys = lookup.unapply(this.getValue());
       final String dimensionName = this.getDimension();
       if (!keys.isEmpty()) {
-        return new OrDimFilter(Lists.transform(keys, new Function<String, DimFilter>()
-        {
-          @Override
-          public DimFilter apply(String input)
-          {
-            return new SelectorDimFilter(dimensionName, input);
-          }
-        }));
+        return new InDimFilter(dimensionName, keys);
       }
     }
     return this;

processing/src/main/java/io/druid/query/filter/InDimFilter.java

@@ -24,14 +24,14 @@
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import com.metamx.common.StringUtils;
-import io.druid.segment.filter.OrFilter;
-import io.druid.segment.filter.SelectorFilter;
+import io.druid.segment.filter.InFilter;
 
 import java.nio.ByteBuffer;
-import java.util.Collections;
 import java.util.List;
 
 public class InDimFilter implements DimFilter
@@ -43,7 +43,20 @@ public class InDimFilter implements DimFilter
   public InDimFilter(@JsonProperty("dimension") String dimension, @JsonProperty("values") List<String> values)
   {
     Preconditions.checkNotNull(dimension, "dimension can not be null");
-    this.values = (values == null) ? Collections.<String>emptyList() : values;
+    Preconditions.checkArgument(values != null && !values.isEmpty(), "values can not be null or empty");
+    this.values = Lists.newArrayList(
+        Iterables.transform(
+            values, new Function<String, String>()
+            {
+              @Override
+              public String apply(String input)
+              {
+                return Strings.nullToEmpty(input);
+              }
+
+            }
+        )
+    );
     this.dimension = dimension;
   }
 
@@ -92,21 +105,7 @@ public DimFilter optimize()
   @Override
   public Filter toFilter()
   {
-    final List<Filter> selectorFilters = ImmutableList.copyOf(
-        Iterables.transform(
-            values,
-            new Function<String, Filter>()
-            {
-              @Override
-              public Filter apply(String input)
-              {
-                return new SelectorFilter(dimension, input);
-              }
-            }
-        )
-    );
-
-    return new OrFilter(selectorFilters);
+    return new InFilter(dimension, ImmutableSet.copyOf(values));
   }
 
   @Override

processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java

@@ -28,7 +28,7 @@
 import io.druid.query.dimension.DefaultDimensionSpec;
 import io.druid.query.dimension.DimensionSpec;
 import io.druid.query.filter.DimFilter;
-import io.druid.query.filter.OrDimFilter;
+import io.druid.query.filter.InDimFilter;
 import io.druid.query.filter.SelectorDimFilter;
 import io.druid.query.spec.LegacySegmentSpec;
 import io.druid.query.spec.QuerySegmentSpec;
@@ -252,11 +252,7 @@ public TopNQueryBuilder filters(String dimensionName, String value)
 
   public TopNQueryBuilder filters(String dimensionName, String value, String... values)
   {
-    List<DimFilter> fields = Lists.<DimFilter>newArrayList(new SelectorDimFilter(dimensionName, value));
-    for (String val : values) {
-      fields.add(new SelectorDimFilter(dimensionName, val));
-    }
-    dimFilter = new OrDimFilter(fields);
+    dimFilter = new InDimFilter(dimensionName, Lists.asList(value, values));
     return this;
   }
 

processing/src/main/java/io/druid/segment/filter/InFilter.java

@@ -0,0 +1,80 @@
+/*
+ * 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.segment.filter;
+
+import com.google.common.base.Function;
+import com.google.common.base.Predicate;
+import com.google.common.base.Strings;
+import com.google.common.collect.Iterables;
+import com.metamx.collections.bitmap.BitmapFactory;
+import com.metamx.collections.bitmap.ImmutableBitmap;
+import io.druid.query.filter.BitmapIndexSelector;
+import io.druid.query.filter.Filter;
+import io.druid.query.filter.ValueMatcher;
+import io.druid.query.filter.ValueMatcherFactory;
+
+import javax.annotation.Nullable;
+import java.util.Set;
+
+/**
+ */
+public class InFilter implements Filter
+{
+  private final String dimension;
+  private final Set<String> values;
+
+  public InFilter(String dimension, Set<String> values)
+  {
+    this.dimension = dimension;
+    this.values = values;
+  }
+
+  @Override
+  public ImmutableBitmap getBitmapIndex(final BitmapIndexSelector selector)
+  {
+    return selector.getBitmapFactory().union(
+        Iterables.transform(
+            values, new Function<String, ImmutableBitmap>()
+            {
+              @Override
+              public ImmutableBitmap apply(String value)
+              {
+                return selector.getBitmapIndex(dimension, value);
+              }
+            }
+        )
+    );
+  }
+
+  @Override
+  public ValueMatcher makeMatcher(ValueMatcherFactory factory)
+  {
+    return factory.makeValueMatcher(
+        dimension, new Predicate<String>()
+        {
+          @Override
+          public boolean apply(String input)
+          {
+            return values.contains(Strings.nullToEmpty(input));
+          }
+        }
+    );
+  }
+}