Add ExpressionFilter.

benchmarks/src/main/java/io/druid/benchmark/BoundFilterBenchmark.java

@@ -200,6 +200,12 @@ public Indexed<String> getDimensionValues(String dimension)
         return dictionary;
       }
 
+      @Override
+      public boolean hasMultipleValues(final String dimension)
+      {
+        throw new UnsupportedOperationException();
+      }
+
       @Override
       public int getNumRows()
       {

benchmarks/src/main/java/io/druid/benchmark/DimensionPredicateFilterBenchmark.java

@@ -154,6 +154,12 @@ public Indexed<String> getDimensionValues(String dimension)
         return dictionary;
       }
 
+      @Override
+      public boolean hasMultipleValues(final String dimension)
+      {
+        throw new UnsupportedOperationException();
+      }
+
       @Override
       public int getNumRows()
       {

benchmarks/src/main/java/io/druid/benchmark/LikeFilterBenchmark.java

@@ -161,6 +161,12 @@ public Indexed<String> getDimensionValues(String dimension)
         return dictionary;
       }
 
+      @Override
+      public boolean hasMultipleValues(final String dimension)
+      {
+        throw new UnsupportedOperationException();
+      }
+
       @Override
       public int getNumRows()
       {

processing/src/main/java/io/druid/query/expression/ExprUtils.java

@@ -0,0 +1,32 @@
+/*
+ * 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.expression;
+
+import io.druid.math.expr.Expr;
+
+public class ExprUtils
+{
+  private static final Expr.ObjectBinding NIL_BINDINGS = name -> null;
+
+  public static Expr.ObjectBinding nilBindings()
+  {
+    return NIL_BINDINGS;
+  }
+}

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

@@ -30,6 +30,7 @@
 public interface BitmapIndexSelector
 {
   public Indexed<String> getDimensionValues(String dimension);
+  public boolean hasMultipleValues(String dimension);
   public int getNumRows();
   public BitmapFactory getBitmapFactory();
   public BitmapIndex getBitmapIndex(String dimension);

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

@@ -41,7 +41,8 @@
     @JsonSubTypes.Type(name="in", value=InDimFilter.class),
     @JsonSubTypes.Type(name="bound", value=BoundDimFilter.class),
     @JsonSubTypes.Type(name="interval", value=IntervalDimFilter.class),
-    @JsonSubTypes.Type(name="like", value=LikeDimFilter.class)
+    @JsonSubTypes.Type(name="like", value=LikeDimFilter.class),
+    @JsonSubTypes.Type(name="expression", value=ExpressionDimFilter.class)
 })
 public interface DimFilter extends Cacheable
 {

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

@@ -51,6 +51,7 @@ public class DimFilterUtils
   static final byte INTERVAL_CACHE_ID = 0xB;
   static final byte LIKE_CACHE_ID = 0xC;
   static final byte COLUMN_COMPARISON_CACHE_ID = 0xD;
+  static final byte EXPRESSION_CACHE_ID = 0xE;
   public static final byte STRING_SEPARATOR = (byte) 0xFF;
 
   static byte[] computeCacheKey(byte cacheIdKey, List<DimFilter> filters)

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

@@ -0,0 +1,107 @@
+/*
+ * 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.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.RangeSet;
+import io.druid.math.expr.Expr;
+import io.druid.math.expr.ExprMacroTable;
+import io.druid.math.expr.Parser;
+import io.druid.query.cache.CacheKeyBuilder;
+import io.druid.segment.filter.ExpressionFilter;
+
+import java.util.Objects;
+
+public class ExpressionDimFilter implements DimFilter
+{
+  private final String expression;
+  private final Expr parsed;
+
+  @JsonCreator
+  public ExpressionDimFilter(
+      @JsonProperty("expression") final String expression,
+      @JacksonInject ExprMacroTable macroTable
+  )
+  {
+    this.expression = expression;
+    this.parsed = Parser.parse(expression, macroTable);
+  }
+
+  @JsonProperty
+  public String getExpression()
+  {
+    return expression;
+  }
+
+  @Override
+  public DimFilter optimize()
+  {
+    return this;
+  }
+
+  @Override
+  public Filter toFilter()
+  {
+    return new ExpressionFilter(parsed);
+  }
+
+  @Override
+  public RangeSet<String> getDimensionRangeSet(final String dimension)
+  {
+    return null;
+  }
+
+  @Override
+  public byte[] getCacheKey()
+  {
+    return new CacheKeyBuilder(DimFilterUtils.EXPRESSION_CACHE_ID)
+        .appendString(expression)
+        .build();
+  }
+
+  @Override
+  public boolean equals(final Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final ExpressionDimFilter that = (ExpressionDimFilter) o;
+    return Objects.equals(expression, that.expression);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(expression);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ExpressionDimFilter{" +
+           "expression='" + expression + '\'' +
+           '}';
+  }
+}

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

@@ -34,7 +34,9 @@
 public interface Filter
 {
   /**
-   * Get a bitmap index, indicating rows that match this filter.
+   * Get a bitmap index, indicating rows that match this filter. Do not call this method unless
+   * {@link #supportsBitmapIndex(BitmapIndexSelector)} returns true. Behavior in the case that
+   * {@link #supportsBitmapIndex(BitmapIndexSelector)} returns false is undefined.
    *
    * @param selector Object used to retrieve bitmap indexes
    *
@@ -47,6 +49,17 @@ default ImmutableBitmap getBitmapIndex(BitmapIndexSelector selector)
     return getBitmapResult(selector, new DefaultBitmapResultFactory(selector.getBitmapFactory()));
   }
 
+  /**
+   * Get a (possibly wrapped) bitmap index, indicating rows that match this filter. Do not call this method unless
+   * {@link #supportsBitmapIndex(BitmapIndexSelector)} returns true. Behavior in the case that
+   * {@link #supportsBitmapIndex(BitmapIndexSelector)} returns false is undefined.
+   *
+   * @param selector Object used to retrieve bitmap indexes
+   *
+   * @return A bitmap indicating rows that match this filter.
+   *
+   * @see Filter#estimateSelectivity(BitmapIndexSelector)
+   */
   default <T> T getBitmapResult(BitmapIndexSelector selector, BitmapResultFactory<T> bitmapResultFactory)
   {
     return bitmapResultFactory.wrapUnknown(getBitmapIndex(selector));

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

@@ -27,7 +27,6 @@
 import io.druid.query.monomorphicprocessing.RuntimeShapeInspector;
 import io.druid.segment.column.BitmapIndex;
 import io.druid.segment.column.Column;
-import io.druid.segment.column.ColumnCapabilities;
 import io.druid.segment.column.DictionaryEncodedColumn;
 import io.druid.segment.column.GenericColumn;
 import io.druid.segment.column.ValueType;
@@ -59,7 +58,7 @@ public ColumnSelectorBitmapIndexSelector(
   @Override
   public Indexed<String> getDimensionValues(String dimension)
   {
-    if (isFilterableVirtualColumn(dimension)) {
+    if (isVirtualColumn(dimension)) {
       // Virtual columns don't have dictionaries or indexes.
       return null;
     }
@@ -109,6 +108,17 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
     };
   }
 
+  @Override
+  public boolean hasMultipleValues(final String dimension)
+  {
+    if (isVirtualColumn(dimension)) {
+      return virtualColumns.getVirtualColumn(dimension).capabilities(dimension).hasMultipleValues();
+    }
+
+    final Column column = index.getColumn(dimension);
+    return column != null && column.getCapabilities().hasMultipleValues();
+  }
+
   @Override
   public int getNumRows()
   {
@@ -126,7 +136,7 @@ public BitmapFactory getBitmapFactory()
   @Override
   public BitmapIndex getBitmapIndex(String dimension)
   {
-    if (isFilterableVirtualColumn(dimension)) {
+    if (isVirtualColumn(dimension)) {
       // Virtual columns don't have dictionaries or indexes.
       return null;
     }
@@ -193,7 +203,7 @@ public ImmutableBitmap getBitmap(int idx)
   @Override
   public ImmutableBitmap getBitmapIndex(String dimension, String value)
   {
-    if (isFilterableVirtualColumn(dimension)) {
+    if (isVirtualColumn(dimension)) {
       // Virtual columns don't have dictionaries or indexes.
       return null;
     }
@@ -218,7 +228,7 @@ public ImmutableBitmap getBitmapIndex(String dimension, String value)
   @Override
   public ImmutableRTree getSpatialIndex(String dimension)
   {
-    if (isFilterableVirtualColumn(dimension)) {
+    if (isVirtualColumn(dimension)) {
       return new ImmutableRTree();
     }
 
@@ -230,14 +240,9 @@ public ImmutableRTree getSpatialIndex(String dimension)
     return column.getSpatialIndex().getRTree();
   }
 
-  private boolean isFilterableVirtualColumn(final String columnName)
+  private boolean isVirtualColumn(final String columnName)
   {
-    final ColumnCapabilities columnCapabilities = virtualColumns.getColumnCapabilities(columnName);
-    if (columnCapabilities == null) {
-      return false;
-    } else {
-      return Filters.FILTERABLE_TYPES.contains(columnCapabilities.getType());
-    }
+    return virtualColumns.getVirtualColumn(columnName) != null;
   }
 
   private static boolean columnSupportsFiltering(Column column)