fixed #1879: Add support for range filter

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

@@ -0,0 +1,161 @@
+/*
+* 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.query.filter;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import com.metamx.common.StringUtils;
+import com.sun.org.apache.xpath.internal.operations.Bool;
+import io.druid.segment.ObjectColumnSelector;
+import org.apache.commons.lang.math.NumberUtils;
+
+import javax.annotation.Nonnull;
+import java.nio.ByteBuffer;
+
+/**
+ * Between filter, support "less or equal than upper bound and great or equal than lower bound"
+ * on String values and float values.
+ * String comparisons are implemented through String#compareTo method which is case sensitive.
+ * Float comparisons are implemented through Float#compare method.
+ * Created by zhxiaog on 15/11/12.
+ **/
+public class BetweenDimFilter implements DimFilter
+{
+  private String dimension;
+  private Object lower;
+  private Object upper;
+  private boolean numerically;
+
+  @JsonCreator
+  public BetweenDimFilter(
+      @JsonProperty("dimension") @Nonnull String dimension,
+      @JsonProperty("lower") @Nonnull Object lower,
+      @JsonProperty("upper") @Nonnull Object upper,
+      @JsonProperty("numerically") Boolean numerically
+  )
+  {
+    // Preconditions.checkArgument(dimension != null, "dimension must not be blank.");
+    // Preconditions.checkArgument(lower != null && upper != null, "dimension must not be blank.");
+    this.dimension = dimension;
+    this.numerically = numerically != null ?
+                       numerically : (Number.class.isInstance(lower) && Number.class.isInstance(upper) ? true : false);
+
+    if (this.numerically) {
+      this.lower = Float.parseFloat(lower.toString());
+      this.upper = Float.parseFloat(upper.toString());
+      Preconditions.checkArgument(
+          Float.compare((float) this.lower, (float) this.upper) <= 0,
+          "required: lower <= upper"
+      );
+    } else {
+      this.lower = lower.toString();
+      this.upper = upper.toString();
+      Preconditions.checkArgument(
+          ((String) this.lower).compareTo((String) this.upper) <= 0,
+          "required: lower <= upper"
+      );
+    }
+  }
+
+  @JsonProperty("dimension")
+  public String getDimension()
+  {
+    return dimension;
+  }
+
+  @JsonProperty("upper")
+  public Object getUpper()
+  {
+    return upper;
+  }
+
+  @JsonProperty("lower")
+  public Object getLower()
+  {
+    return lower;
+  }
+
+  @JsonProperty("numerically")
+  public boolean getNumerically()
+  {
+    return numerically;
+  }
+
+  @Override
+  public byte[] getCacheKey()
+  {
+    final byte[] dimensionBytes = StringUtils.toUtf8(dimension);
+    if (numerically) {
+      return ByteBuffer.allocate(1 + dimensionBytes.length + 8)
+                       .put(DimFilterCacheHelper.BETWEEN_CACHE_ID)
+                       .putInt(1)
+                       .put(dimensionBytes)
+                       .putFloat((float) lower)
+                       .putFloat((float) upper)
+                       .array();
+    } else {
+      final byte[] lower_bytes = StringUtils.toUtf8(this.lower.toString());
+      final byte[] upper_bytes = StringUtils.toUtf8(this.upper.toString());
+      return ByteBuffer.allocate(1 + dimensionBytes.length + lower_bytes.length + upper_bytes.length)
+                       .put(DimFilterCacheHelper.BETWEEN_CACHE_ID)
+                       .putInt(0)
+                       .put(dimensionBytes)
+                       .put(lower_bytes)
+                       .put(upper_bytes)
+                       .array();
+    }
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    BetweenDimFilter that = (BetweenDimFilter) o;
+
+    if (numerically != that.numerically) {
+      return false;
+    }
+    if (!dimension.equals(that.dimension)) {
+      return false;
+    }
+    if (!lower.equals(that.lower)) {
+      return false;
+    }
+    return upper.equals(that.upper);
+
+  }
+
+  @Override
+  public int hashCode()
+  {
+    int result = dimension.hashCode();
+    result = 31 * result + lower.hashCode();
+    result = 31 * result + upper.hashCode();
+    result = 31 * result + (numerically ? 1 : 0);
+    return result;
+  }
+}

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

@@ -22,18 +22,19 @@
 
 /**
  */
-@JsonTypeInfo(use=JsonTypeInfo.Id.NAME, property="type")
-@JsonSubTypes(value={
-    @JsonSubTypes.Type(name="and", value=AndDimFilter.class),
-    @JsonSubTypes.Type(name="or", value=OrDimFilter.class),
-    @JsonSubTypes.Type(name="not", value=NotDimFilter.class),
-    @JsonSubTypes.Type(name="selector", value=SelectorDimFilter.class),
-    @JsonSubTypes.Type(name="extraction", value=ExtractionDimFilter.class),
-    @JsonSubTypes.Type(name="regex", value=RegexDimFilter.class),
-    @JsonSubTypes.Type(name="search", value=SearchQueryDimFilter.class),
-    @JsonSubTypes.Type(name="javascript", value=JavaScriptDimFilter.class),
-    @JsonSubTypes.Type(name="spatial", value=SpatialDimFilter.class),
-    @JsonSubTypes.Type(name="in", value=InDimFilter.class)
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes(value = {
+    @JsonSubTypes.Type(name = "and", value = AndDimFilter.class),
+    @JsonSubTypes.Type(name = "or", value = OrDimFilter.class),
+    @JsonSubTypes.Type(name = "not", value = NotDimFilter.class),
+    @JsonSubTypes.Type(name = "selector", value = SelectorDimFilter.class),
+    @JsonSubTypes.Type(name = "extraction", value = ExtractionDimFilter.class),
+    @JsonSubTypes.Type(name = "regex", value = RegexDimFilter.class),
+    @JsonSubTypes.Type(name = "search", value = SearchQueryDimFilter.class),
+    @JsonSubTypes.Type(name = "javascript", value = JavaScriptDimFilter.class),
+    @JsonSubTypes.Type(name = "spatial", value = SpatialDimFilter.class),
+    @JsonSubTypes.Type(name = "in", value = InDimFilter.class),
+    @JsonSubTypes.Type(name = "between", value = BetweenDimFilter.class),
 })
 public interface DimFilter
 {

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

@@ -35,6 +35,7 @@ class DimFilterCacheHelper
   static final byte JAVASCRIPT_CACHE_ID = 0x7;
   static final byte SPATIAL_CACHE_ID = 0x8;
   static final byte IN_CACHE_ID = 0x9;
+  static final byte BETWEEN_CACHE_ID = 0xA;
 
   static byte[] computeCacheKey(byte cacheIdKey, List<DimFilter> filters)
   {

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

@@ -0,0 +1,146 @@
+/*
+* 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.metamx.collections.bitmap.ImmutableBitmap;
+import com.metamx.common.guava.FunctionalIterable;
+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 io.druid.segment.ColumnSelectorFactory;
+import io.druid.segment.data.Indexed;
+import org.apache.commons.lang.math.NumberUtils;
+
+import javax.annotation.Nullable;
+
+/**
+ * Created by zhxiaog on 15/11/12.
+ */
+public class BetweenFilter implements Filter
+{
+
+  private final String dimension;
+  private final Predicate<String> predicate;
+
+  public BetweenFilter(String dimension, boolean numrically, Object lower, Object upper)
+  {
+    this.dimension = dimension;
+    if (numrically) {
+      this.predicate = new FloatBoundPredicate((float) lower, (float) upper);
+    } else {
+      this.predicate = new StringBoundPredicate((String) lower, (String) upper);
+    }
+  }
+
+  @Override
+  public ImmutableBitmap getBitmapIndex(final BitmapIndexSelector selector)
+  {
+    Indexed<String> dimValues = selector.getDimensionValues(this.dimension);
+    ImmutableBitmap result = null;
+    if (dimValues == null) {
+      result = selector.getBitmapFactory().makeEmptyImmutableBitmap();
+    } else {
+      result = selector.getBitmapFactory().union(
+          FunctionalIterable.create(dimValues)
+                            .filter(predicate)
+                            .transform(
+                                new Function<String, ImmutableBitmap>()
+                                {
+                                  @Nullable
+                                  @Override
+                                  public ImmutableBitmap apply(String input)
+                                  {
+                                    return selector.getBitmapIndex(
+                                        BetweenFilter.this.dimension,
+                                        input
+                                    );
+                                  }
+                                }
+                            )
+
+      );
+    }
+    return result;
+  }
+
+  @Override
+  public ValueMatcher makeMatcher(ValueMatcherFactory factory)
+  {
+    return factory.makeValueMatcher(this.dimension, this.predicate);
+  }
+
+  @Override
+  public ValueMatcher makeMatcher(ColumnSelectorFactory columnSelectorFactory)
+  {
+    throw new UnsupportedOperationException();
+  }
+
+  /**
+   * between operator for float values
+   */
+  private static class FloatBoundPredicate implements com.google.common.base.Predicate<String>
+  {
+    private float lower;
+    private float upper;
+
+    public FloatBoundPredicate(float lower, float upper)
+    {
+      this.lower = lower;
+      this.upper = upper;
+    }
+
+    @Override
+    public boolean apply(@Nullable String input)
+    {
+      if (NumberUtils.isNumber(input)) {
+        float num = NumberUtils.toFloat(input);
+        return Float.compare(num, this.upper) <= 0 && Float.compare(num, this.lower) >= 0;
+      } else {
+        return false;
+      }
+    }
+  }
+
+  /**
+   * between operator for string values
+   */
+  private static class StringBoundPredicate implements com.google.common.base.Predicate<String>
+  {
+    private String lower;
+    private String upper;
+
+    public StringBoundPredicate(String lower, String upper)
+    {
+      this.lower = lower;
+      this.upper = upper;
+    }
+
+    @Override
+    public boolean apply(@Nullable String input)
+    {
+      return input != null && input.compareTo(this.upper) <= 0 && input.compareTo(this.lower) >= 0;
+    }
+  }
+
+
+}

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

@@ -20,6 +20,7 @@
 import com.google.common.base.Function;
 import com.google.common.collect.Lists;
 import io.druid.query.filter.AndDimFilter;
+import io.druid.query.filter.BetweenDimFilter;
 import io.druid.query.filter.DimFilter;
 import io.druid.query.filter.ExtractionDimFilter;
 import io.druid.query.filter.Filter;
@@ -39,7 +40,8 @@
  */
 public class Filters
 {
-  public static List<Filter> convertDimensionFilters(List<DimFilter> filters){
+  public static List<Filter> convertDimensionFilters(List<DimFilter> filters)
+  {
     return Lists.transform(
         filters,
         new Function<DimFilter, Filter>()
@@ -109,6 +111,14 @@ public Filter apply(@Nullable String input)
       );
 
       filter = new OrFilter(listFilters);
+    } else if (dimFilter instanceof BetweenDimFilter) {
+      final BetweenDimFilter between = (BetweenDimFilter) dimFilter;
+      filter = new BetweenFilter(
+          between.getDimension(),
+          between.getNumerically(),
+          between.getLower(),
+          between.getUpper()
+      );
     }
 
     return filter;