Add virtual column types, holder serde, and safety features. (#3823)

* Add virtual column types, holder serde, and safety features.

Virtual columns:
- add long, float, dimension selectors
- put cache IDs in VirtualColumnCacheHelper
- adjust serde so VirtualColumns can be the holder object for Jackson
- add fail-fast validation for cycle detection and duplicates
- add expression virtual column in core

Storage adapters:
- move virtual column hooks before checking base columns, to prevent surprises
  when a new base column is added that happens to have the same name as a
  virtual column.

* Fix ExtractionDimensionSpecs with virtual dimensions.

* Fix unused imports.

* CR comments

* Merge one more time, with feeling.

common/src/main/java/io/druid/math/expr/Evals.java

@@ -20,7 +20,6 @@
 package io.druid.math.expr;
 
 import com.google.common.base.Strings;
-import io.druid.common.guava.GuavaUtils;
 import io.druid.java.util.common.logger.Logger;
 
 import java.util.Arrays;
@@ -32,27 +31,6 @@ public class Evals
 {
   private static final Logger log = new Logger(Evals.class);
 
-  public static Number toNumber(Object value)
-  {
-    if (value == null) {
-      return 0L;
-    }
-    if (value instanceof Number) {
-      return (Number) value;
-    }
-    String stringValue = String.valueOf(value);
-    Long longValue = GuavaUtils.tryParseLong(stringValue);
-    if (longValue == null) {
-      return Double.valueOf(stringValue);
-    }
-    return longValue;
-  }
-
-  public static boolean isConstant(Expr expr)
-  {
-    return expr instanceof ConstantExpr;
-  }
-
   public static boolean isAllConstants(Expr... exprs)
   {
     return isAllConstants(Arrays.asList(exprs));

extensions-contrib/virtual-columns/src/main/java/io/druid/segment/MapVirtualColumn.java

@@ -22,13 +22,20 @@ package io.druid.segment;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
 import com.metamx.common.StringUtils;
 import io.druid.query.dimension.DefaultDimensionSpec;
+import io.druid.query.dimension.DimensionSpec;
 import io.druid.query.filter.DimFilterUtils;
+import io.druid.segment.column.ColumnCapabilities;
+import io.druid.segment.column.ColumnCapabilitiesImpl;
+import io.druid.segment.column.ValueType;
 import io.druid.segment.data.IndexedInts;
+import io.druid.segment.virtual.VirtualColumnCacheHelper;
 
 import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
@@ -36,8 +43,6 @@ import java.util.Objects;
  */
 public class MapVirtualColumn implements VirtualColumn
 {
-  private static final byte VC_TYPE_ID = 0x00;
-
   private final String outputName;
   private final String keyDimension;
   private final String valueDimension;
@@ -59,13 +64,14 @@ public class MapVirtualColumn implements VirtualColumn
   }
 
   @Override
-  public ObjectColumnSelector init(String dimension, ColumnSelectorFactory factory)
+  public ObjectColumnSelector makeObjectColumnSelector(String dimension, ColumnSelectorFactory factory)
   {
     final DimensionSelector keySelector = factory.makeDimensionSelector(DefaultDimensionSpec.of(keyDimension));
     final DimensionSelector valueSelector = factory.makeDimensionSelector(DefaultDimensionSpec.of(valueDimension));
 
-    int index = dimension.indexOf('.');
+    final String subColumnName = VirtualColumns.splitColumnName(dimension).rhs;
-    if (index < 0) {
+
+    if (subColumnName == null) {
       return new ObjectColumnSelector<Map>()
       {
         @Override
@@ -97,7 +103,7 @@ public class MapVirtualColumn implements VirtualColumn
 
     IdLookup keyIdLookup = keySelector.idLookup();
     if (keyIdLookup != null) {
-      final int keyId = keyIdLookup.lookupId(dimension.substring(index + 1));
+      final int keyId = keyIdLookup.lookupId(subColumnName);
       if (keyId < 0) {
         return NullStringObjectColumnSelector.instance();
       }
@@ -127,7 +133,6 @@ public class MapVirtualColumn implements VirtualColumn
         }
       };
     } else {
-      final String key = dimension.substring(index + 1);
       return new ObjectColumnSelector<String>()
       {
         @Override
@@ -146,7 +151,7 @@ public class MapVirtualColumn implements VirtualColumn
           }
           final int limit = Math.min(keyIndices.size(), valueIndices.size());
           for (int i = 0; i < limit; i++) {
-            if (Objects.equals(keySelector.lookupName(keyIndices.get(i)), key)) {
+            if (Objects.equals(keySelector.lookupName(keyIndices.get(i)), subColumnName)) {
               return valueSelector.lookupName(valueIndices.get(i));
             }
           }
@@ -156,6 +161,38 @@ public class MapVirtualColumn implements VirtualColumn
     }
   }
 
+  @Override
+  public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec, ColumnSelectorFactory factory)
+  {
+    // Could probably do something useful here if the column name is dot-style. But for now just return nothing.
+    return null;
+  }
+
+  @Override
+  public FloatColumnSelector makeFloatColumnSelector(String columnName, ColumnSelectorFactory factory)
+  {
+    return null;
+  }
+
+  @Override
+  public LongColumnSelector makeLongColumnSelector(String columnName, ColumnSelectorFactory factory)
+  {
+    return null;
+  }
+
+  @Override
+  public ColumnCapabilities capabilities(String columnName)
+  {
+    final ValueType valueType = columnName.indexOf('.') < 0 ? ValueType.COMPLEX : ValueType.STRING;
+    return new ColumnCapabilitiesImpl().setType(valueType);
+  }
+
+  @Override
+  public List<String> requiredColumns()
+  {
+    return ImmutableList.of(keyDimension, valueDimension);
+  }
+
   @Override
   public boolean usesDotNotation()
   {
@@ -170,7 +207,7 @@ public class MapVirtualColumn implements VirtualColumn
     byte[] output = StringUtils.toUtf8(outputName);
 
     return ByteBuffer.allocate(3 + key.length + value.length + output.length)
-                     .put(VC_TYPE_ID)
+                     .put(VirtualColumnCacheHelper.CACHE_TYPE_ID_MAP)
                      .put(key).put(DimFilterUtils.STRING_SEPARATOR)
                      .put(value).put(DimFilterUtils.STRING_SEPARATOR)
                      .put(output)

indexing-hadoop/src/main/java/io/druid/indexer/InputRowSerde.java

@@ -36,6 +36,7 @@ import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.segment.incremental.IncrementalIndex;
 import io.druid.segment.serde.ComplexMetricSerde;
 import io.druid.segment.serde.ComplexMetrics;
+import io.druid.segment.VirtualColumns;
 import org.apache.hadoop.io.ArrayWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.WritableUtils;
@@ -87,6 +88,7 @@ public class InputRowSerde
 
         Aggregator agg = aggFactory.factorize(
             IncrementalIndex.makeColumnSelectorFactory(
+                VirtualColumns.EMPTY,
                 aggFactory,
                 supplier,
                 true

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

@@ -55,6 +55,7 @@ import io.druid.query.timeboundary.TimeBoundaryQuery;
 import io.druid.query.timeboundary.TimeBoundaryResultValue;
 import io.druid.query.timeseries.TimeseriesQuery;
 import io.druid.segment.VirtualColumn;
+import io.druid.segment.VirtualColumns;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
@@ -1104,7 +1105,7 @@ public class Druids
     private QueryGranularity granularity;
     private List<DimensionSpec> dimensions;
     private List<String> metrics;
-    private List<VirtualColumn> virtualColumns;
+    private VirtualColumns virtualColumns;
     private PagingSpec pagingSpec;
 
     public SelectQueryBuilder()
@@ -1233,12 +1234,22 @@ public class Druids
       return this;
     }
 
-    public SelectQueryBuilder virtualColumns(List<VirtualColumn> vcs)
+    public SelectQueryBuilder virtualColumns(VirtualColumns vcs)
     {
       virtualColumns = vcs;
       return this;
     }
 
+    public SelectQueryBuilder virtualColumns(List<VirtualColumn> vcs)
+    {
+      return virtualColumns(VirtualColumns.create(vcs));
+    }
+
+    public SelectQueryBuilder virtualColumns(VirtualColumn... vcs)
+    {
+      return virtualColumns(VirtualColumns.create(Arrays.asList(vcs)));
+    }
+
     public SelectQueryBuilder pagingSpec(PagingSpec p)
     {
       pagingSpec = p;

processing/src/main/java/io/druid/query/extraction/BucketExtractionFn.java

@@ -58,7 +58,7 @@ public class BucketExtractionFn implements ExtractionFn
   public String apply(Object value)
   {
     if (value instanceof Number) {
-      return bucket((Double) value);
+      return bucket(((Number) value).doubleValue());
     } else if (value instanceof String) {
       return apply(value);
     }

processing/src/main/java/io/druid/query/select/SelectQuery.java

@@ -31,7 +31,7 @@ import io.druid.query.Result;
 import io.druid.query.dimension.DimensionSpec;
 import io.druid.query.filter.DimFilter;
 import io.druid.query.spec.QuerySegmentSpec;
-import io.druid.segment.VirtualColumn;
+import io.druid.segment.VirtualColumns;
 
 import java.util.List;
 import java.util.Map;
@@ -46,7 +46,7 @@ public class SelectQuery extends BaseQuery<Result<SelectResultValue>>
   private final QueryGranularity granularity;
   private final List<DimensionSpec> dimensions;
   private final List<String> metrics;
-  private final List<VirtualColumn> virtualColumns;
+  private final VirtualColumns virtualColumns;
   private final PagingSpec pagingSpec;
 
   @JsonCreator
@@ -58,7 +58,7 @@ public class SelectQuery extends BaseQuery<Result<SelectResultValue>>
       @JsonProperty("granularity") QueryGranularity granularity,
       @JsonProperty("dimensions") List<DimensionSpec> dimensions,
       @JsonProperty("metrics") List<String> metrics,
-      @JsonProperty("virtualColumns") List<VirtualColumn> virtualColumns,
+      @JsonProperty("virtualColumns") VirtualColumns virtualColumns,
       @JsonProperty("pagingSpec") PagingSpec pagingSpec,
       @JsonProperty("context") Map<String, Object> context
   )
@@ -67,7 +67,7 @@ public class SelectQuery extends BaseQuery<Result<SelectResultValue>>
     this.dimFilter = dimFilter;
     this.granularity = granularity;
     this.dimensions = dimensions;
-    this.virtualColumns = virtualColumns;
+    this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY : virtualColumns;
     this.metrics = metrics;
     this.pagingSpec = pagingSpec;
 
@@ -134,7 +134,7 @@ public class SelectQuery extends BaseQuery<Result<SelectResultValue>>
   }
 
   @JsonProperty
-  public List<VirtualColumn> getVirtualColumns()
+  public VirtualColumns getVirtualColumns()
   {
     return virtualColumns;
   }

processing/src/main/java/io/druid/query/select/SelectQueryEngine.java

@@ -43,7 +43,6 @@ import io.druid.segment.LongColumnSelector;
 import io.druid.segment.ObjectColumnSelector;
 import io.druid.segment.Segment;
 import io.druid.segment.StorageAdapter;
-import io.druid.segment.VirtualColumns;
 import io.druid.segment.column.Column;
 import io.druid.segment.column.ColumnCapabilities;
 import io.druid.segment.column.ValueType;
@@ -161,7 +160,7 @@ public class SelectQueryEngine
         adapter,
         query.getQuerySegmentSpec().getIntervals(),
         filter,
-        VirtualColumns.valueOf(query.getVirtualColumns()),
+        query.getVirtualColumns(),
         query.isDescending(),
         query.getGranularity(),
         new Function<Cursor, Result<SelectResultValue>>()

processing/src/main/java/io/druid/query/select/SelectQueryQueryToolChest.java

@@ -49,7 +49,6 @@ import io.druid.query.aggregation.MetricManipulationFn;
 import io.druid.query.dimension.DimensionSpec;
 import io.druid.query.filter.DimFilter;
 import io.druid.timeline.DataSegmentUtils;
-import io.druid.segment.VirtualColumn;
 import io.druid.timeline.LogicalSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -192,20 +191,7 @@ public class SelectQueryQueryToolChest extends QueryToolChest<Result<SelectResul
           ++index;
         }
 
-        List<VirtualColumn> virtualColumns = query.getVirtualColumns();
+        final byte[] virtualColumnsCacheKey = query.getVirtualColumns().getCacheKey();
-        if (virtualColumns == null) {
-          virtualColumns = Collections.emptyList();
-        }
-
-        final byte[][] virtualColumnsBytes = new byte[virtualColumns.size()][];
-        int virtualColumnsBytesSize = 0;
-        index = 0;
-        for (VirtualColumn vc : virtualColumns) {
-          virtualColumnsBytes[index] = vc.getCacheKey();
-          virtualColumnsBytesSize += virtualColumnsBytes[index].length;
-          ++index;
-        }
-
         final ByteBuffer queryCacheKey = ByteBuffer
             .allocate(
                 1
@@ -214,7 +200,7 @@ public class SelectQueryQueryToolChest extends QueryToolChest<Result<SelectResul
                 + query.getPagingSpec().getCacheKey().length
                 + dimensionsBytesSize
                 + metricBytesSize
-                + virtualColumnsBytesSize
+                + virtualColumnsCacheKey.length
             )
             .put(SELECT_QUERY)
             .put(granularityBytes)
@@ -229,9 +215,7 @@ public class SelectQueryQueryToolChest extends QueryToolChest<Result<SelectResul
           queryCacheKey.put(metricByte);
         }
 
-        for (byte[] vcByte : virtualColumnsBytes) {
+        queryCacheKey.put(virtualColumnsCacheKey);
-          queryCacheKey.put(vcByte);
-        }
 
         return queryCacheKey.array();
       }

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

@@ -22,14 +22,29 @@ package io.druid.segment;
 import io.druid.query.dimension.DimensionSpec;
 import io.druid.segment.column.ColumnCapabilities;
 
+import javax.annotation.Nullable;
+
 /**
  * Factory class for MetricSelectors
  */
 public interface ColumnSelectorFactory
 {
-  public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec);
+  DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec);
-  public FloatColumnSelector makeFloatColumnSelector(String columnName);
+  FloatColumnSelector makeFloatColumnSelector(String columnName);
-  public LongColumnSelector makeLongColumnSelector(String columnName);
+  LongColumnSelector makeLongColumnSelector(String columnName);
-  public ObjectColumnSelector makeObjectColumnSelector(String columnName);
+
-  public ColumnCapabilities getColumnCapabilities(String columnName);
+  @Nullable
+  ObjectColumnSelector makeObjectColumnSelector(String columnName);
+
+  /**
+   * Returns capabilities of a particular column, if known. May be null if the column doesn't exist, or if
+   * the column does exist but the capabilities are unknown. The latter is possible with dynamically discovered
+   * columns.
+   *
+   * @param column column name
+   *
+   * @return capabilities, or null
+   */
+  @Nullable
+  ColumnCapabilities getColumnCapabilities(String column);
 }

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

@@ -30,6 +30,18 @@ import javax.annotation.Nullable;
 
 public class NullDimensionSelector implements DimensionSelector, IdLookup
 {
+  private static final NullDimensionSelector INSTANCE = new NullDimensionSelector();
+
+  private NullDimensionSelector()
+  {
+    // Singleton.
+  }
+
+  public static NullDimensionSelector instance()
+  {
+    return INSTANCE;
+  }
+
   @Override
   public IndexedInts getRow()
   {

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

@@ -41,7 +41,6 @@ import io.druid.query.filter.ValueMatcher;
 import io.druid.segment.column.BitmapIndex;
 import io.druid.segment.column.Column;
 import io.druid.segment.column.ColumnCapabilities;
-import io.druid.segment.column.ColumnCapabilitiesImpl;
 import io.druid.segment.column.ComplexColumn;
 import io.druid.segment.column.DictionaryEncodedColumn;
 import io.druid.segment.column.GenericColumn;
@@ -68,8 +67,6 @@ import java.util.Map;
  */
 public class QueryableIndexStorageAdapter implements StorageAdapter
 {
-  private static final NullDimensionSelector NULL_DIMENSION_SELECTOR = new NullDimensionSelector();
-
   private final QueryableIndex index;
 
   public QueryableIndexStorageAdapter(
@@ -430,6 +427,10 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                         DimensionSpec dimensionSpec
                     )
                     {
+                      if (virtualColumns.exists(dimensionSpec.getDimension())) {
+                        return virtualColumns.makeDimensionSelector(dimensionSpec, this);
+                      }
+
                       return dimensionSpec.decorate(makeDimensionSelectorUndecorated(dimensionSpec));
                     }
 
@@ -442,7 +443,7 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
 
                       final Column columnDesc = index.getColumn(dimension);
                       if (columnDesc == null) {
-                        return NULL_DIMENSION_SELECTOR;
+                        return NullDimensionSelector.instance();
                       }
 
                       if (dimension.equals(Column.TIME_COLUMN_NAME)) {
@@ -463,7 +464,7 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                       final DictionaryEncodedColumn<String> column = cachedColumn;
 
                       if (column == null) {
-                        return NULL_DIMENSION_SELECTOR;
+                        return NullDimensionSelector.instance();
                       } else if (columnDesc.getCapabilities().hasMultipleValues()) {
                         class MultiValueDimensionSelector implements DimensionSelector, IdLookup
                         {
@@ -652,6 +653,10 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                     @Override
                     public FloatColumnSelector makeFloatColumnSelector(String columnName)
                     {
+                      if (virtualColumns.exists(columnName)) {
+                        return virtualColumns.makeFloatColumnSelector(columnName, this);
+                      }
+
                       GenericColumn cachedMetricVals = genericColumnCache.get(columnName);
 
                       if (cachedMetricVals == null) {
@@ -665,14 +670,7 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                       }
 
                       if (cachedMetricVals == null) {
-                        return new FloatColumnSelector()
+                        return ZeroFloatColumnSelector.instance();
-                        {
-                          @Override
-                          public float get()
-                          {
-                            return 0.0f;
-                          }
-                        };
                       }
 
                       final GenericColumn metricVals = cachedMetricVals;
@@ -689,6 +687,10 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                     @Override
                     public LongColumnSelector makeLongColumnSelector(String columnName)
                     {
+                      if (virtualColumns.exists(columnName)) {
+                        return virtualColumns.makeLongColumnSelector(columnName, this);
+                      }
+
                       GenericColumn cachedMetricVals = genericColumnCache.get(columnName);
 
                       if (cachedMetricVals == null) {
@@ -702,14 +704,7 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                       }
 
                       if (cachedMetricVals == null) {
-                        return new LongColumnSelector()
+                        return ZeroLongColumnSelector.instance();
-                        {
-                          @Override
-                          public long get()
-                          {
-                            return 0L;
-                          }
-                        };
                       }
 
                       final GenericColumn metricVals = cachedMetricVals;
@@ -727,6 +722,10 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                     @Override
                     public ObjectColumnSelector makeObjectColumnSelector(String column)
                     {
+                      if (virtualColumns.exists(column)) {
+                        return virtualColumns.makeObjectColumnSelector(column, this);
+                      }
+
                       Object cachedColumnVals = objectColumnCache.get(column);
 
                       if (cachedColumnVals == null) {
@@ -751,10 +750,6 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                       }
 
                       if (cachedColumnVals == null) {
-                        VirtualColumn vc = virtualColumns.getVirtualColumn(column);
-                        if (vc != null) {
-                          return vc.init(column, this);
-                        }
                         return null;
                       }
 
@@ -881,19 +876,14 @@ public class QueryableIndexStorageAdapter implements StorageAdapter
                       };
                     }
 
-                    @Nullable
                     @Override
                     public ColumnCapabilities getColumnCapabilities(String columnName)
                     {
-                      ColumnCapabilities capabilities = getColumnCapabilites(index, columnName);
+                      if (virtualColumns.exists(columnName)) {
-                      if (capabilities == null && !virtualColumns.isEmpty()) {
+                        return virtualColumns.getColumnCapabilities(columnName);
-                        VirtualColumn virtualColumn = virtualColumns.getVirtualColumn(columnName);
-                        if (virtualColumn != null) {
-                          Class clazz = virtualColumn.init(columnName, this).classOfObject();
-                          capabilities = new ColumnCapabilitiesImpl().setType(ValueType.typeFor(clazz));
-                        }
                       }
-                      return capabilities;
+
+                      return getColumnCapabilites(index, columnName);
                     }
                   }
 

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

@@ -19,15 +19,26 @@
 
 package io.druid.segment;
 
+import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.druid.query.dimension.DimensionSpec;
+import io.druid.segment.column.ColumnCapabilities;
+import io.druid.segment.virtual.ExpressionVirtualColumn;
+
+import javax.annotation.Nullable;
+import java.util.List;
 
-/**
- */
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 /**
  * Virtual columns are "views" created over a ColumnSelectorFactory. They can potentially draw from multiple
  * underlying columns, although they always present themselves as if they were a single column.
+ *
+ * A virtual column object will be shared amongst threads and must be thread safe. The selectors returned
+ * from the various makeXXXSelector methods need not be thread safe.
  */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes(value = {
+    @JsonSubTypes.Type(name = "expression", value = ExpressionVirtualColumn.class)
+})
 public interface VirtualColumn
 {
   /**
@@ -42,10 +53,73 @@ public interface VirtualColumn
    * virtual column was referenced with, which is useful if this column uses dot notation.
    *
    * @param columnName the name this virtual column was referenced with
-   * @param factory column selector factory
+   * @param factory    column selector factory
-   * @return the selector
+   *
+   * @return the selector, must not be null
    */
-  ObjectColumnSelector init(String columnName, ColumnSelectorFactory factory);
+  ObjectColumnSelector makeObjectColumnSelector(String columnName, ColumnSelectorFactory factory);
+
+  /**
+   * Build a selector corresponding to this virtual column. Also provides the name that the
+   * virtual column was referenced with (through {@link DimensionSpec#getDimension()}, which
+   * is useful if this column uses dot notation. The virtual column is expected to apply any
+   * necessary decoration from the dimensionSpec.
+   *
+   * @param dimensionSpec the dimensionSpec this column was referenced with
+   * @param factory       column selector factory
+   *
+   * @return the selector, or null if we can't make a selector
+   */
+  @Nullable
+  DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec, ColumnSelectorFactory factory);
+
+  /**
+   * Build a selector corresponding to this virtual column. Also provides the name that the
+   * virtual column was referenced with, which is useful if this column uses dot notation.
+   *
+   * @param columnName the name this virtual column was referenced with
+   * @param factory    column selector factory
+   *
+   * @return the selector, or null if we can't make a selector
+   */
+  @Nullable
+  FloatColumnSelector makeFloatColumnSelector(String columnName, ColumnSelectorFactory factory);
+
+  /**
+   * Build a selector corresponding to this virtual column. Also provides the name that the
+   * virtual column was referenced with, which is useful if this column uses dot notation.
+   *
+   * @param columnName the name this virtual column was referenced with
+   * @param factory    column selector factory
+   *
+   * @return the selector, or null if we can't make a selector
+   */
+  @Nullable
+  LongColumnSelector makeLongColumnSelector(String columnName, ColumnSelectorFactory factory);
+
+  /**
+   * Returns the capabilities of this virtual column, which includes a type that should match
+   * the type returned by "makeObjectColumnSelector" and should correspond to the best
+   * performing selector. May vary based on columnName if this column uses dot notation.
+   *
+   * @param columnName the name this virtual column was referenced with
+   *
+   * @return capabilities, must not be null
+   */
+  ColumnCapabilities capabilities(String columnName);
+
+  /**
+   * Returns a list of columns that this virtual column will access. This may include the
+   * names of other virtual columns. May be empty if a virtual column doesn't access any
+   * underlying columns.
+   *
+   * Does not pass columnName because there is an assumption that the list of columns
+   * needed by a dot-notation supporting virtual column will not vary based on the
+   * columnName.
+   *
+   * @return column names
+   */
+  List<String> requiredColumns();
 
   /**
    * Indicates that this virtual column can be referenced with dot notation. For example,

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

@@ -19,62 +19,256 @@
 
 package io.druid.segment;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.primitives.Ints;
+import io.druid.java.util.common.IAE;
+import io.druid.java.util.common.Pair;
+import io.druid.query.dimension.DimensionSpec;
+import io.druid.segment.column.Column;
+import io.druid.segment.column.ColumnCapabilities;
+import io.druid.segment.virtual.VirtualizedColumnSelectorFactory;
 
+import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
+ * Class allowing lookup and usage of virtual columns.
  */
 public class VirtualColumns
 {
   public static final VirtualColumns EMPTY = new VirtualColumns(
-      ImmutableMap.<String, VirtualColumn>of(), ImmutableMap.<String, VirtualColumn>of()
+      ImmutableList.<VirtualColumn>of(),
+      ImmutableMap.<String, VirtualColumn>of(),
+      ImmutableMap.<String, VirtualColumn>of()
   );
 
-  public static VirtualColumns valueOf(List<VirtualColumn> virtualColumns) {
+  /**
+   * Split a dot-style columnName into the "main" columnName and the subColumn name after the dot. Useful for
+   * columns that support dot notation.
+   *
+   * @param columnName columnName like "foo" or "foo.bar"
+   *
+   * @return pair of main column name (will not be null) and subColumn name (may be null)
+   */
+  public static Pair<String, String> splitColumnName(String columnName)
+  {
+    final int i = columnName.indexOf('.');
+    if (i < 0) {
+      return Pair.of(columnName, null);
+    } else {
+      return Pair.of(columnName.substring(0, i), columnName.substring(i + 1));
+    }
+  }
+
+  @JsonCreator
+  public static VirtualColumns create(List<VirtualColumn> virtualColumns)
+  {
     if (virtualColumns == null || virtualColumns.isEmpty()) {
       return EMPTY;
     }
     Map<String, VirtualColumn> withDotSupport = Maps.newHashMap();
     Map<String, VirtualColumn> withoutDotSupport = Maps.newHashMap();
     for (VirtualColumn vc : virtualColumns) {
+      if (vc.getOutputName().equals(Column.TIME_COLUMN_NAME)) {
+        throw new IAE("virtualColumn name[%s] not allowed", vc.getOutputName());
+      }
+
+      if (withDotSupport.containsKey(vc.getOutputName()) || withoutDotSupport.containsKey(vc.getOutputName())) {
+        throw new IAE("Duplicate virtualColumn name[%s]", vc.getOutputName());
+      }
+
       if (vc.usesDotNotation()) {
         withDotSupport.put(vc.getOutputName(), vc);
       } else {
         withoutDotSupport.put(vc.getOutputName(), vc);
       }
     }
-    return new VirtualColumns(withDotSupport, withoutDotSupport);
+    return new VirtualColumns(ImmutableList.copyOf(virtualColumns), withDotSupport, withoutDotSupport);
   }
 
-  public VirtualColumns(Map<String, VirtualColumn> withDotSupport, Map<String, VirtualColumn> withoutDotSupport)
+  private VirtualColumns(
+      List<VirtualColumn> virtualColumns,
+      Map<String, VirtualColumn> withDotSupport,
+      Map<String, VirtualColumn> withoutDotSupport
+  )
   {
+    this.virtualColumns = virtualColumns;
     this.withDotSupport = withDotSupport;
     this.withoutDotSupport = withoutDotSupport;
+
+    for (VirtualColumn virtualColumn : virtualColumns) {
+      detectCycles(virtualColumn, null);
+    }
   }
 
+  // For equals, hashCode, toString, and serialization:
+  private final List<VirtualColumn> virtualColumns;
+
+  // For getVirtualColumn:
   private final Map<String, VirtualColumn> withDotSupport;
   private final Map<String, VirtualColumn> withoutDotSupport;
 
-  public VirtualColumn getVirtualColumn(String dimension)
+  public boolean exists(String columnName)
   {
-    VirtualColumn vc = withoutDotSupport.get(dimension);
+    return getVirtualColumn(columnName) != null;
+  }
+
+  public VirtualColumn getVirtualColumn(String columnName)
+  {
+    final VirtualColumn vc = withoutDotSupport.get(columnName);
     if (vc != null) {
       return vc;
     }
-    for (int index = dimension.indexOf('.'); index >= 0; index = dimension.indexOf('.', index + 1)) {
+    final String baseColumnName = splitColumnName(columnName).lhs;
-      vc = withDotSupport.get(dimension.substring(0, index));
+    return withDotSupport.get(baseColumnName);
-      if (vc != null) {
+  }
-        return vc;
+
-      }
+  public ObjectColumnSelector makeObjectColumnSelector(String columnName, ColumnSelectorFactory factory)
+  {
+    final VirtualColumn virtualColumn = getVirtualColumn(columnName);
+    if (virtualColumn == null) {
+      return null;
+    } else {
+      return Preconditions.checkNotNull(
+          virtualColumn.makeObjectColumnSelector(columnName, factory),
+          "VirtualColumn[%s] returned a null ObjectColumnSelector for columnName[%s]",
+          virtualColumn.getOutputName(),
+          columnName
+      );
+    }
+  }
+
+  public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec, ColumnSelectorFactory factory)
+  {
+    final VirtualColumn virtualColumn = getVirtualColumn(dimensionSpec.getDimension());
+    if (virtualColumn == null) {
+      return dimensionSpec.decorate(NullDimensionSelector.instance());
+    } else {
+      final DimensionSelector selector = virtualColumn.makeDimensionSelector(dimensionSpec, factory);
+      return selector == null ? dimensionSpec.decorate(NullDimensionSelector.instance()) : selector;
+    }
+  }
+
+  public FloatColumnSelector makeFloatColumnSelector(String columnName, ColumnSelectorFactory factory)
+  {
+    final VirtualColumn virtualColumn = getVirtualColumn(columnName);
+    if (virtualColumn == null) {
+      return ZeroFloatColumnSelector.instance();
+    } else {
+      final FloatColumnSelector selector = virtualColumn.makeFloatColumnSelector(columnName, factory);
+      return selector == null ? ZeroFloatColumnSelector.instance() : selector;
+    }
+  }
+
+  public LongColumnSelector makeLongColumnSelector(String columnName, ColumnSelectorFactory factory)
+  {
+    final VirtualColumn virtualColumn = getVirtualColumn(columnName);
+    if (virtualColumn == null) {
+      return ZeroLongColumnSelector.instance();
+    } else {
+      final LongColumnSelector selector = virtualColumn.makeLongColumnSelector(columnName, factory);
+      return selector == null ? ZeroLongColumnSelector.instance() : selector;
+    }
+  }
+
+  public ColumnCapabilities getColumnCapabilities(String columnName)
+  {
+    final VirtualColumn virtualColumn = getVirtualColumn(columnName);
+    if (virtualColumn != null) {
+      return Preconditions.checkNotNull(
+          virtualColumn.capabilities(columnName),
+          "capabilities for column[%s]",
+          columnName
+      );
+    } else {
+      return null;
     }
-    return withDotSupport.get(dimension);
   }
 
   public boolean isEmpty()
   {
     return withDotSupport.isEmpty() && withoutDotSupport.isEmpty();
   }
+
+  @JsonValue
+  public VirtualColumn[] getVirtualColumns()
+  {
+    // VirtualColumn[] instead of List<VirtualColumn> to aid Jackson serialization.
+    return virtualColumns.toArray(new VirtualColumn[]{});
+  }
+
+  public ColumnSelectorFactory wrap(final ColumnSelectorFactory baseFactory)
+  {
+    return new VirtualizedColumnSelectorFactory(baseFactory, this);
+  }
+
+  public byte[] getCacheKey()
+  {
+    final byte[][] cacheKeys = new byte[virtualColumns.size()][];
+    int len = Ints.BYTES;
+    for (int i = 0; i < virtualColumns.size(); i++) {
+      cacheKeys[i] = virtualColumns.get(i).getCacheKey();
+      len += Ints.BYTES + cacheKeys[i].length;
+    }
+    final ByteBuffer buf = ByteBuffer.allocate(len).putInt(virtualColumns.size());
+    for (byte[] cacheKey : cacheKeys) {
+      buf.putInt(cacheKey.length);
+      buf.put(cacheKey);
+    }
+    return buf.array();
+  }
+
+  private void detectCycles(VirtualColumn virtualColumn, Set<String> columnNames)
+  {
+    // Copy columnNames to avoid modifying it
+    final Set<String> nextSet = columnNames == null
+                                ? Sets.newHashSet(virtualColumn.getOutputName())
+                                : Sets.newHashSet(columnNames);
+
+    for (String columnName : virtualColumn.requiredColumns()) {
+      if (!nextSet.add(columnName)) {
+        throw new IAE("Self-referential column[%s]", columnName);
+      }
+
+      final VirtualColumn dependency = getVirtualColumn(columnName);
+      if (dependency != null) {
+        detectCycles(dependency, nextSet);
+      }
+    }
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    VirtualColumns that = (VirtualColumns) o;
+
+    return virtualColumns.equals(that.virtualColumns);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return virtualColumns.hashCode();
+  }
+
+  @Override
+  public String toString()
+  {
+    return virtualColumns.toString();
+  }
 }

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

@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+public final class ZeroFloatColumnSelector implements FloatColumnSelector
+{
+  private static final ZeroFloatColumnSelector INSTANCE = new ZeroFloatColumnSelector();
+
+  private ZeroFloatColumnSelector()
+  {
+    // No instantiation.
+  }
+
+  public static ZeroFloatColumnSelector instance()
+  {
+    return INSTANCE;
+  }
+
+  @Override
+  public float get()
+  {
+    return 0.0f;
+  }
+}

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

@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+public final class ZeroLongColumnSelector implements LongColumnSelector
+{
+  private static final ZeroLongColumnSelector INSTANCE = new ZeroLongColumnSelector();
+
+  private ZeroLongColumnSelector()
+  {
+    // No instantiation.
+  }
+
+  public static ZeroLongColumnSelector instance()
+  {
+    return INSTANCE;
+  }
+
+  @Override
+  public long get()
+  {
+    return 0;
+  }
+}

processing/src/main/java/io/druid/segment/column/ValueType.java

@@ -26,17 +26,5 @@ public enum ValueType
   FLOAT,
   LONG,
   STRING,
-  COMPLEX;
+  COMPLEX
-
-  public static ValueType typeFor(Class clazz)
-  {
-    if (clazz == String.class) {
-      return STRING;
-    } else if (clazz == float.class || clazz == Float.TYPE) {
-      return FLOAT;
-    } else if (clazz == long.class || clazz == Long.TYPE) {
-      return LONG;
-    }
-    return COMPLEX;
-  }
 }

processing/src/main/java/io/druid/segment/incremental/IncrementalIndex.java

@@ -53,6 +53,7 @@ import io.druid.segment.FloatColumnSelector;
 import io.druid.segment.LongColumnSelector;
 import io.druid.segment.Metadata;
 import io.druid.segment.ObjectColumnSelector;
+import io.druid.segment.VirtualColumns;
 import io.druid.segment.column.Column;
 import io.druid.segment.column.ColumnCapabilities;
 import io.druid.segment.column.ColumnCapabilitiesImpl;
@@ -100,14 +101,25 @@ public abstract class IncrementalIndex<AggregatorType> implements Iterable<Row>,
       .put(DimensionSchema.ValueType.STRING, ValueType.STRING)
       .build();
 
+  /**
+   * Column selector used at ingestion time for inputs to aggregators.
+   *
+   * @param agg                       the aggregator
+   * @param in                        ingestion-time input row supplier
+   * @param deserializeComplexMetrics whether complex objects should be deserialized by a {@link ComplexMetricExtractor}
+   *
+   * @return column selector factory
+   */
   public static ColumnSelectorFactory makeColumnSelectorFactory(
+      final VirtualColumns virtualColumns,
       final AggregatorFactory agg,
       final Supplier<InputRow> in,
       final boolean deserializeComplexMetrics
   )
   {
     final RowBasedColumnSelectorFactory baseSelectorFactory = RowBasedColumnSelectorFactory.create(in, null);
-    return new ColumnSelectorFactory()
+
+    class IncrementalIndexInputRowColumnSelectorFactory implements ColumnSelectorFactory
     {
       @Override
       public LongColumnSelector makeLongColumnSelector(final String columnName)
@@ -167,13 +179,16 @@ public abstract class IncrementalIndex<AggregatorType> implements Iterable<Row>,
       {
         return baseSelectorFactory.getColumnCapabilities(columnName);
       }
-    };
+    }
+
+    return virtualColumns.wrap(new IncrementalIndexInputRowColumnSelectorFactory());
   }
 
   private final long minTimestamp;
   private final QueryGranularity gran;
   private final boolean rollup;
   private final List<Function<InputRow, InputRow>> rowTransformers;
+  private final VirtualColumns virtualColumns;
   private final AggregatorFactory[] metrics;
   private final AggregatorType[] aggs;
   private final boolean deserializeComplexMetrics;
@@ -217,6 +232,7 @@ public abstract class IncrementalIndex<AggregatorType> implements Iterable<Row>,
     this.minTimestamp = incrementalIndexSchema.getMinTimestamp();
     this.gran = incrementalIndexSchema.getGran();
     this.rollup = incrementalIndexSchema.isRollup();
+    this.virtualColumns = incrementalIndexSchema.getVirtualColumns();
     this.metrics = incrementalIndexSchema.getMetrics();
     this.rowTransformers = new CopyOnWriteArrayList<>();
     this.deserializeComplexMetrics = deserializeComplexMetrics;
@@ -894,6 +910,15 @@ public abstract class IncrementalIndex<AggregatorType> implements Iterable<Row>,
     }
   }
 
+  protected ColumnSelectorFactory makeColumnSelectorFactory(
+      final AggregatorFactory agg,
+      final Supplier<InputRow> in,
+      final boolean deserializeComplexMetrics
+  )
+  {
+    return makeColumnSelectorFactory(virtualColumns, agg, in, deserializeComplexMetrics);
+  }
+
   protected final Comparator<TimeAndDims> dimsComparator()
   {
     return new TimeAndDimsComp(dimensionDescsList);

processing/src/main/java/io/druid/segment/incremental/IncrementalIndexSchema.java

@@ -25,6 +25,7 @@ import io.druid.data.input.impl.TimestampSpec;
 import io.druid.granularity.QueryGranularities;
 import io.druid.granularity.QueryGranularity;
 import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.segment.VirtualColumns;
 
 /**
  */
@@ -34,6 +35,7 @@ public class IncrementalIndexSchema
   private final long minTimestamp;
   private final TimestampSpec timestampSpec;
   private final QueryGranularity gran;
+  private final VirtualColumns virtualColumns;
   private final DimensionsSpec dimensionsSpec;
   private final AggregatorFactory[] metrics;
   private final boolean rollup;
@@ -42,6 +44,7 @@ public class IncrementalIndexSchema
       long minTimestamp,
       TimestampSpec timestampSpec,
       QueryGranularity gran,
+      VirtualColumns virtualColumns,
       DimensionsSpec dimensionsSpec,
       AggregatorFactory[] metrics,
       boolean rollup
@@ -50,6 +53,7 @@ public class IncrementalIndexSchema
     this.minTimestamp = minTimestamp;
     this.timestampSpec = timestampSpec;
     this.gran = gran;
+    this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY : virtualColumns;
     this.dimensionsSpec = dimensionsSpec;
     this.metrics = metrics;
     this.rollup = rollup;
@@ -70,6 +74,11 @@ public class IncrementalIndexSchema
     return gran;
   }
 
+  public VirtualColumns getVirtualColumns()
+  {
+    return virtualColumns;
+  }
+
   public DimensionsSpec getDimensionsSpec()
   {
     return dimensionsSpec;
@@ -90,6 +99,7 @@ public class IncrementalIndexSchema
     private long minTimestamp;
     private TimestampSpec timestampSpec;
     private QueryGranularity gran;
+    private VirtualColumns virtualColumns;
     private DimensionsSpec dimensionsSpec;
     private AggregatorFactory[] metrics;
     private boolean rollup;
@@ -98,6 +108,7 @@ public class IncrementalIndexSchema
     {
       this.minTimestamp = 0L;
       this.gran = QueryGranularities.NONE;
+      this.virtualColumns = VirtualColumns.EMPTY;
       this.dimensionsSpec = new DimensionsSpec(null, null, null);
       this.metrics = new AggregatorFactory[]{};
       this.rollup = true;
@@ -133,6 +144,12 @@ public class IncrementalIndexSchema
       return this;
     }
 
+    public Builder withVirtualColumns(VirtualColumns virtualColumns)
+    {
+      this.virtualColumns = virtualColumns;
+      return this;
+    }
+
     public Builder withDimensionsSpec(DimensionsSpec dimensionsSpec)
     {
       this.dimensionsSpec = dimensionsSpec == null ? DimensionsSpec.ofEmpty() : dimensionsSpec;
@@ -167,7 +184,7 @@ public class IncrementalIndexSchema
     public IncrementalIndexSchema build()
     {
       return new IncrementalIndexSchema(
-          minTimestamp, timestampSpec, gran, dimensionsSpec, metrics, rollup
+          minTimestamp, timestampSpec, gran, virtualColumns, dimensionsSpec, metrics, rollup
       );
     }
   }

processing/src/main/java/io/druid/segment/incremental/IncrementalIndexStorageAdapter.java

@@ -44,12 +44,11 @@ import io.druid.segment.NullDimensionSelector;
 import io.druid.segment.ObjectColumnSelector;
 import io.druid.segment.SingleScanTimeDimSelector;
 import io.druid.segment.StorageAdapter;
-import io.druid.segment.VirtualColumn;
 import io.druid.segment.VirtualColumns;
+import io.druid.segment.ZeroFloatColumnSelector;
+import io.druid.segment.ZeroLongColumnSelector;
 import io.druid.segment.column.Column;
 import io.druid.segment.column.ColumnCapabilities;
-import io.druid.segment.column.ColumnCapabilitiesImpl;
-import io.druid.segment.column.ValueType;
 import io.druid.segment.data.Indexed;
 import io.druid.segment.data.ListIndexed;
 import io.druid.segment.filter.BooleanValueMatcher;
@@ -64,8 +63,6 @@ import java.util.Map;
  */
 public class IncrementalIndexStorageAdapter implements StorageAdapter
 {
-  private static final NullDimensionSelector NULL_DIMENSION_SELECTOR = new NullDimensionSelector();
-
   private final IncrementalIndex<?> index;
 
   public IncrementalIndexStorageAdapter(
@@ -340,6 +337,10 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
                   DimensionSpec dimensionSpec
               )
               {
+                if (virtualColumns.exists(dimensionSpec.getDimension())) {
+                  return virtualColumns.makeDimensionSelector(dimensionSpec, this);
+                }
+
                 final String dimension = dimensionSpec.getDimension();
                 final ExtractionFn extractionFn = dimensionSpec.getExtractionFn();
 
@@ -354,7 +355,7 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
 
                 final IncrementalIndex.DimensionDesc dimensionDesc = index.getDimension(dimensionSpec.getDimension());
                 if (dimensionDesc == null) {
-                  return dimensionSpec.decorate(NULL_DIMENSION_SELECTOR);
+                  return dimensionSpec.decorate(NullDimensionSelector.instance());
                 }
 
                 final DimensionIndexer indexer = dimensionDesc.getIndexer();
@@ -364,6 +365,10 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
               @Override
               public FloatColumnSelector makeFloatColumnSelector(String columnName)
               {
+                if (virtualColumns.exists(columnName)) {
+                  return virtualColumns.makeFloatColumnSelector(columnName, this);
+                }
+
                 final Integer dimIndex = index.getDimensionIndex(columnName);
                 if (dimIndex != null) {
                   final IncrementalIndex.DimensionDesc dimensionDesc = index.getDimension(columnName);
@@ -377,14 +382,7 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
 
                 final Integer metricIndexInt = index.getMetricIndex(columnName);
                 if (metricIndexInt == null) {
-                  return new FloatColumnSelector()
+                  return ZeroFloatColumnSelector.instance();
-                  {
-                    @Override
-                    public float get()
-                    {
-                      return 0.0f;
-                    }
-                  };
                 }
 
                 final int metricIndex = metricIndexInt;
@@ -401,6 +399,10 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
               @Override
               public LongColumnSelector makeLongColumnSelector(String columnName)
               {
+                if (virtualColumns.exists(columnName)) {
+                  return virtualColumns.makeLongColumnSelector(columnName, this);
+                }
+
                 if (columnName.equals(Column.TIME_COLUMN_NAME)) {
                   return new LongColumnSelector()
                   {
@@ -425,14 +427,7 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
 
                 final Integer metricIndexInt = index.getMetricIndex(columnName);
                 if (metricIndexInt == null) {
-                  return new LongColumnSelector()
+                  return ZeroLongColumnSelector.instance();
-                  {
-                    @Override
-                    public long get()
-                    {
-                      return 0L;
-                    }
-                  };
                 }
 
                 final int metricIndex = metricIndexInt;
@@ -453,6 +448,10 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
               @Override
               public ObjectColumnSelector makeObjectColumnSelector(String column)
               {
+                if (virtualColumns.exists(column)) {
+                  return virtualColumns.makeObjectColumnSelector(column, this);
+                }
+
                 if (column.equals(Column.TIME_COLUMN_NAME)) {
                   return new ObjectColumnSelector<Long>()
                   {
@@ -496,10 +495,6 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
                 IncrementalIndex.DimensionDesc dimensionDesc = index.getDimension(column);
 
                 if (dimensionDesc == null) {
-                  VirtualColumn virtualColumn = virtualColumns.getVirtualColumn(column);
-                  if (virtualColumn != null) {
-                    return virtualColumn.init(column, this);
-                  }
                   return null;
                 } else {
 
@@ -539,15 +534,11 @@ public class IncrementalIndexStorageAdapter implements StorageAdapter
               @Override
               public ColumnCapabilities getColumnCapabilities(String columnName)
               {
-                ColumnCapabilities capabilities = index.getCapabilities(columnName);
+                if (virtualColumns.exists(columnName)) {
-                if (capabilities == null && !virtualColumns.isEmpty()) {
+                  return virtualColumns.getColumnCapabilities(columnName);
-                  VirtualColumn virtualColumn = virtualColumns.getVirtualColumn(columnName);
-                  if (virtualColumn != null) {
-                    Class clazz = virtualColumn.init(columnName, this).classOfObject();
-                    capabilities = new ColumnCapabilitiesImpl().setType(ValueType.typeFor(clazz));
-                  }
                 }
-                return capabilities;
+
+                return index.getCapabilities(columnName);
               }
             };
           }

processing/src/main/java/io/druid/segment/incremental/OffheapIncrementalIndex.java

@@ -91,31 +91,6 @@ public class OffheapIncrementalIndex extends IncrementalIndex<BufferAggregator>
     aggBuffers.add(bb);
   }
 
-  public OffheapIncrementalIndex(
-      long minTimestamp,
-      QueryGranularity gran,
-      final AggregatorFactory[] metrics,
-      boolean deserializeComplexMetrics,
-      boolean reportParseExceptions,
-      boolean sortFacts,
-      int maxRowCount,
-      StupidPool<ByteBuffer> bufferPool
-  )
-  {
-    this(
-        new IncrementalIndexSchema.Builder().withMinTimestamp(minTimestamp)
-                                            .withQueryGranularity(gran)
-                                            .withMetrics(metrics)
-                                            .withRollup(IncrementalIndexSchema.DEFAULT_ROLLUP)
-                                            .build(),
-        deserializeComplexMetrics,
-        reportParseExceptions,
-        sortFacts,
-        maxRowCount,
-        bufferPool
-    );
-  }
-
   public OffheapIncrementalIndex(
       long minTimestamp,
       QueryGranularity gran,

processing/src/main/java/io/druid/segment/virtual/BaseSingleValueDimensionSelector.java

@@ -0,0 +1,92 @@
+/*
+ * 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.virtual;
+
+import com.google.common.base.Predicate;
+import io.druid.query.filter.ValueMatcher;
+import io.druid.segment.DimensionSelector;
+import io.druid.segment.IdLookup;
+import io.druid.segment.data.IndexedInts;
+import io.druid.segment.data.ZeroIndexedInts;
+
+import javax.annotation.Nullable;
+import java.util.Objects;
+
+public abstract class BaseSingleValueDimensionSelector implements DimensionSelector
+{
+  protected abstract String getValue();
+
+  @Override
+  public IndexedInts getRow()
+  {
+    return ZeroIndexedInts.instance();
+  }
+
+  @Override
+  public int getValueCardinality()
+  {
+    return DimensionSelector.CARDINALITY_UNKNOWN;
+  }
+
+  @Override
+  public String lookupName(int id)
+  {
+    return getValue();
+  }
+
+  @Override
+  public ValueMatcher makeValueMatcher(final String value)
+  {
+    return new ValueMatcher()
+    {
+      @Override
+      public boolean matches()
+      {
+        return Objects.equals(getValue(), value);
+      }
+    };
+  }
+
+  @Override
+  public ValueMatcher makeValueMatcher(final Predicate<String> predicate)
+  {
+    return new ValueMatcher()
+    {
+      @Override
+      public boolean matches()
+      {
+        return predicate.apply(getValue());
+      }
+    };
+  }
+
+  @Override
+  public boolean nameLookupPossibleInAdvance()
+  {
+    return false;
+  }
+
+  @Nullable
+  @Override
+  public IdLookup idLookup()
+  {
+    return null;
+  }
+}

processing/src/main/java/io/druid/segment/virtual/ExpressionSelectors.java

@@ -20,7 +20,9 @@
 package io.druid.segment.virtual;
 
 import io.druid.math.expr.Expr;
+import io.druid.query.extraction.ExtractionFn;
 import io.druid.segment.ColumnSelectorFactory;
+import io.druid.segment.DimensionSelector;
 import io.druid.segment.FloatColumnSelector;
 import io.druid.segment.LongColumnSelector;
 
@@ -46,7 +48,7 @@ public class ExpressionSelectors
   )
   {
     final ExpressionObjectSelector baseSelector = ExpressionObjectSelector.from(columnSelectorFactory, expression);
-    return new LongColumnSelector()
+    class ExpressionLongColumnSelector implements LongColumnSelector
     {
       @Override
       public long get()
@@ -54,7 +56,8 @@ public class ExpressionSelectors
         final Number number = baseSelector.get();
         return number != null ? number.longValue() : nullValue;
       }
-    };
+    }
+    return new ExpressionLongColumnSelector();
   }
 
   public static FloatColumnSelector makeFloatColumnSelector(
@@ -64,7 +67,7 @@ public class ExpressionSelectors
   )
   {
     final ExpressionObjectSelector baseSelector = ExpressionObjectSelector.from(columnSelectorFactory, expression);
-    return new FloatColumnSelector()
+    class ExpressionFloatColumnSelector implements FloatColumnSelector
     {
       @Override
       public float get()
@@ -72,6 +75,39 @@ public class ExpressionSelectors
         final Number number = baseSelector.get();
         return number != null ? number.floatValue() : nullValue;
       }
-    };
+    }
+    return new ExpressionFloatColumnSelector();
+  }
+
+  public static DimensionSelector makeDimensionSelector(
+      final ColumnSelectorFactory columnSelectorFactory,
+      final Expr expression,
+      final ExtractionFn extractionFn
+  )
+  {
+    final ExpressionObjectSelector baseSelector = ExpressionObjectSelector.from(columnSelectorFactory, expression);
+
+    if (extractionFn == null) {
+      class DefaultExpressionDimensionSelector extends BaseSingleValueDimensionSelector
+      {
+        @Override
+        protected String getValue()
+        {
+          final Number number = baseSelector.get();
+          return number == null ? null : String.valueOf(number);
+        }
+      }
+      return new DefaultExpressionDimensionSelector();
+    } else {
+      class ExtractionExpressionDimensionSelector extends BaseSingleValueDimensionSelector
+      {
+        @Override
+        protected String getValue()
+        {
+          return extractionFn.apply(baseSelector.get());
+        }
+      }
+      return new ExtractionExpressionDimensionSelector();
+    }
   }
 }

processing/src/main/java/io/druid/segment/virtual/ExpressionVirtualColumn.java

@@ -0,0 +1,185 @@
+/*
+ * 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.virtual;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import com.google.common.primitives.Ints;
+import io.druid.math.expr.Expr;
+import io.druid.math.expr.Parser;
+import io.druid.query.dimension.DimensionSpec;
+import io.druid.segment.ColumnSelectorFactory;
+import io.druid.segment.DimensionSelector;
+import io.druid.segment.FloatColumnSelector;
+import io.druid.segment.LongColumnSelector;
+import io.druid.segment.ObjectColumnSelector;
+import io.druid.segment.VirtualColumn;
+import io.druid.segment.column.ColumnCapabilities;
+import io.druid.segment.column.ColumnCapabilitiesImpl;
+import io.druid.segment.column.ValueType;
+import org.apache.commons.codec.Charsets;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+public class ExpressionVirtualColumn implements VirtualColumn
+{
+  private static final ColumnCapabilities CAPABILITIES = new ColumnCapabilitiesImpl().setType(ValueType.FLOAT);
+
+  private final String name;
+  private final String expression;
+  private final Expr parsedExpression;
+
+  @JsonCreator
+  public ExpressionVirtualColumn(
+      @JsonProperty("name") String name,
+      @JsonProperty("expression") String expression
+  )
+  {
+    this.name = Preconditions.checkNotNull(name, "name");
+    this.expression = Preconditions.checkNotNull(expression, "expression");
+    this.parsedExpression = Parser.parse(expression);
+  }
+
+  @JsonProperty("name")
+  @Override
+  public String getOutputName()
+  {
+    return name;
+  }
+
+  @JsonProperty
+  public String getExpression()
+  {
+    return expression;
+  }
+
+  @Override
+  public ObjectColumnSelector makeObjectColumnSelector(
+      final String columnName,
+      final ColumnSelectorFactory columnSelectorFactory
+  )
+  {
+    return ExpressionSelectors.makeObjectColumnSelector(columnSelectorFactory, parsedExpression);
+  }
+
+  @Override
+  public DimensionSelector makeDimensionSelector(
+      final DimensionSpec dimensionSpec,
+      final ColumnSelectorFactory columnSelectorFactory
+  )
+  {
+    return dimensionSpec.decorate(
+        ExpressionSelectors.makeDimensionSelector(
+            columnSelectorFactory,
+            parsedExpression,
+            dimensionSpec.getExtractionFn()
+        )
+    );
+  }
+
+  @Override
+  public FloatColumnSelector makeFloatColumnSelector(
+      final String columnName,
+      final ColumnSelectorFactory columnSelectorFactory
+  )
+  {
+    return ExpressionSelectors.makeFloatColumnSelector(columnSelectorFactory, parsedExpression, 0.0f);
+  }
+
+  @Override
+  public LongColumnSelector makeLongColumnSelector(
+      final String columnName,
+      final ColumnSelectorFactory columnSelectorFactory
+  )
+  {
+    return ExpressionSelectors.makeLongColumnSelector(columnSelectorFactory, parsedExpression, 0L);
+  }
+
+  @Override
+  public ColumnCapabilities capabilities(String columnName)
+  {
+    return CAPABILITIES;
+  }
+
+  @Override
+  public List<String> requiredColumns()
+  {
+    return Parser.findRequiredBindings(expression);
+  }
+
+  @Override
+  public boolean usesDotNotation()
+  {
+    return false;
+  }
+
+  @Override
+  public byte[] getCacheKey()
+  {
+    final byte[] nameBytes = name.getBytes(Charsets.UTF_8);
+    final byte[] expressionBytes = expression.getBytes(Charsets.UTF_8);
+
+    return ByteBuffer
+        .allocate(1 + Ints.BYTES * 2 + nameBytes.length + expressionBytes.length)
+        .put(VirtualColumnCacheHelper.CACHE_TYPE_ID_EXPRESSION)
+        .putInt(nameBytes.length)
+        .put(nameBytes)
+        .putInt(expressionBytes.length)
+        .put(expressionBytes)
+        .array();
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    ExpressionVirtualColumn that = (ExpressionVirtualColumn) o;
+
+    if (!name.equals(that.name)) {
+      return false;
+    }
+    return expression.equals(that.expression);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    int result = name.hashCode();
+    result = 31 * result + expression.hashCode();
+    return result;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ExpressionVirtualColumn{" +
+           "name='" + name + '\'' +
+           ", expression='" + expression + '\'' +
+           '}';
+  }
+}

processing/src/main/java/io/druid/segment/virtual/VirtualColumnCacheHelper.java

@@ -0,0 +1,34 @@
+/*
+ * 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.virtual;
+
+public class VirtualColumnCacheHelper
+{
+  public static final byte CACHE_TYPE_ID_MAP = 0x00;
+  public static final byte CACHE_TYPE_ID_EXPRESSION = 0x01;
+
+  // Starting byte 0xFF is reserved for site-specific virtual columns.
+  public static final byte CACHE_TYPE_ID_USER_DEFINED = (byte) 0xFF;
+
+  private VirtualColumnCacheHelper()
+  {
+    // No instantiation.
+  }
+}

processing/src/main/java/io/druid/segment/virtual/VirtualizedColumnSelectorFactory.java

@@ -0,0 +1,99 @@
+/*
+ * 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.virtual;
+
+import com.google.common.base.Preconditions;
+import io.druid.query.dimension.DimensionSpec;
+import io.druid.segment.ColumnSelectorFactory;
+import io.druid.segment.DimensionSelector;
+import io.druid.segment.FloatColumnSelector;
+import io.druid.segment.LongColumnSelector;
+import io.druid.segment.ObjectColumnSelector;
+import io.druid.segment.VirtualColumns;
+import io.druid.segment.column.ColumnCapabilities;
+
+import javax.annotation.Nullable;
+
+public class VirtualizedColumnSelectorFactory implements ColumnSelectorFactory
+{
+  private final ColumnSelectorFactory baseFactory;
+  private final VirtualColumns virtualColumns;
+
+  public VirtualizedColumnSelectorFactory(
+      ColumnSelectorFactory baseFactory,
+      VirtualColumns virtualColumns
+  )
+  {
+    this.baseFactory = Preconditions.checkNotNull(baseFactory, "baseFactory");
+    this.virtualColumns = Preconditions.checkNotNull(virtualColumns, "virtualColumns");
+  }
+
+  @Override
+  public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec)
+  {
+    if (virtualColumns.exists(dimensionSpec.getDimension())) {
+      return virtualColumns.makeDimensionSelector(dimensionSpec, baseFactory);
+    } else {
+      return baseFactory.makeDimensionSelector(dimensionSpec);
+    }
+  }
+
+  @Override
+  public FloatColumnSelector makeFloatColumnSelector(String columnName)
+  {
+    if (virtualColumns.exists(columnName)) {
+      return virtualColumns.makeFloatColumnSelector(columnName, baseFactory);
+    } else {
+      return baseFactory.makeFloatColumnSelector(columnName);
+    }
+  }
+
+  @Override
+  public LongColumnSelector makeLongColumnSelector(String columnName)
+  {
+    if (virtualColumns.exists(columnName)) {
+      return virtualColumns.makeLongColumnSelector(columnName, baseFactory);
+    } else {
+      return baseFactory.makeLongColumnSelector(columnName);
+    }
+  }
+
+  @Nullable
+  @Override
+  public ObjectColumnSelector makeObjectColumnSelector(String columnName)
+  {
+    if (virtualColumns.exists(columnName)) {
+      return virtualColumns.makeObjectColumnSelector(columnName, baseFactory);
+    } else {
+      return baseFactory.makeObjectColumnSelector(columnName);
+    }
+  }
+
+  @Nullable
+  @Override
+  public ColumnCapabilities getColumnCapabilities(String columnName)
+  {
+    if (virtualColumns.exists(columnName)) {
+      return virtualColumns.getColumnCapabilities(columnName);
+    } else {
+      return baseFactory.getColumnCapabilities(columnName);
+    }
+  }
+}

processing/src/test/java/io/druid/query/select/SelectQuerySpecTest.java

@@ -60,7 +60,7 @@ public class SelectQuerySpecTest
         + "\"granularity\":{\"type\":\"all\"},"
         + "\"dimensions\":[{\"type\":\"default\",\"dimension\":\"market\",\"outputName\":\"market\"},{\"type\":\"default\",\"dimension\":\"quality\",\"outputName\":\"quality\"}],"
         + "\"metrics\":[\"index\"],"
-        + "\"virtualColumns\":null,"
+        + "\"virtualColumns\":[],"
         + "\"pagingSpec\":{\"pagingIdentifiers\":{},\"threshold\":3,\"fromNext\":false},"
         + "\"context\":null}";
 

processing/src/test/java/io/druid/segment/NullDimensionSelectorTest.java

@@ -27,7 +27,7 @@ import java.util.Iterator;
 
 public class NullDimensionSelectorTest {
 
-  private final NullDimensionSelector selector = new NullDimensionSelector();
+  private final NullDimensionSelector selector = NullDimensionSelector.instance();
 
   @Test
   public void testGetRow() throws Exception {

processing/src/test/java/io/druid/segment/TestIndex.java

@@ -41,6 +41,7 @@ import io.druid.segment.incremental.IncrementalIndex;
 import io.druid.segment.incremental.IncrementalIndexSchema;
 import io.druid.segment.incremental.OnheapIncrementalIndex;
 import io.druid.segment.serde.ComplexMetrics;
+import io.druid.segment.virtual.ExpressionVirtualColumn;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
@@ -78,10 +79,15 @@ public class TestIndex
   public static final String[] METRICS = new String[]{"index", "indexMin", "indexMaxPlusTen"};
   private static final Logger log = new Logger(TestIndex.class);
   private static final Interval DATA_INTERVAL = new Interval("2011-01-12T00:00:00.000Z/2011-05-01T00:00:00.000Z");
+  private static final VirtualColumns VIRTUAL_COLUMNS = VirtualColumns.create(
+      Arrays.<VirtualColumn>asList(
+          new ExpressionVirtualColumn("expr", "index + 10")
+      )
+  );
   public static final AggregatorFactory[] METRIC_AGGS = new AggregatorFactory[]{
       new DoubleSumAggregatorFactory(METRICS[0], METRICS[0]),
       new DoubleMinAggregatorFactory(METRICS[1], METRICS[0]),
-      new DoubleMaxAggregatorFactory(METRICS[2], null, "index + 10"),
+      new DoubleMaxAggregatorFactory(METRICS[2], VIRTUAL_COLUMNS.getVirtualColumns()[0].getOutputName()),
       new HyperUniquesAggregatorFactory("quality_uniques", "quality")
   };
   private static final IndexSpec indexSpec = new IndexSpec();
@@ -224,6 +230,7 @@ public class TestIndex
         .withMinTimestamp(new DateTime("2011-01-12T00:00:00.000Z").getMillis())
         .withTimestampSpec(new TimestampSpec("ds", "auto", null))
         .withQueryGranularity(QueryGranularities.NONE)
+        .withVirtualColumns(VIRTUAL_COLUMNS)
         .withMetrics(METRIC_AGGS)
         .withRollup(rollup)
         .build();

processing/src/test/java/io/druid/segment/incremental/IncrementalIndexMultiValueSpecTest.java

@@ -28,6 +28,7 @@ import io.druid.data.input.impl.StringDimensionSchema;
 import io.druid.data.input.impl.TimestampSpec;
 import io.druid.granularity.QueryGranularities;
 import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.segment.VirtualColumns;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -54,6 +55,7 @@ public class IncrementalIndexMultiValueSpecTest
         0,
         new TimestampSpec("ds", "auto", null),
         QueryGranularities.ALL,
+        VirtualColumns.EMPTY,
         dimensionsSpec,
         new AggregatorFactory[0],
         false

processing/src/test/java/io/druid/segment/incremental/IncrementalIndexStorageAdapterTest.java

@@ -49,9 +49,9 @@ import io.druid.query.topn.TopNResultValue;
 import io.druid.segment.Cursor;
 import io.druid.segment.DimensionSelector;
 import io.druid.segment.StorageAdapter;
-import io.druid.segment.VirtualColumns;
 import io.druid.segment.data.IndexedInts;
 import io.druid.segment.filter.SelectorFilter;
+import io.druid.segment.VirtualColumns;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 import org.junit.Assert;
@@ -265,7 +265,7 @@ public class IncrementalIndexStorageAdapterTest
       Sequence<Cursor> cursorSequence = adapter.makeCursors(
           new SelectorFilter("sally", "bo"),
           interval,
-          null,
+          VirtualColumns.EMPTY,
           QueryGranularities.NONE,
           descending
       );

processing/src/test/java/io/druid/segment/virtual/ExpressionVirtualColumnTest.java

@@ -0,0 +1,175 @@
+/*
+ * 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.virtual;
+
+import com.google.common.base.Predicates;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.druid.data.input.InputRow;
+import io.druid.data.input.MapBasedInputRow;
+import io.druid.query.dimension.DefaultDimensionSpec;
+import io.druid.query.dimension.ExtractionDimensionSpec;
+import io.druid.query.extraction.BucketExtractionFn;
+import io.druid.query.filter.ValueMatcher;
+import io.druid.query.groupby.epinephelinae.TestColumnSelectorFactory;
+import io.druid.segment.DimensionSelector;
+import io.druid.segment.FloatColumnSelector;
+import io.druid.segment.LongColumnSelector;
+import io.druid.segment.ObjectColumnSelector;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ExpressionVirtualColumnTest
+{
+  private static final InputRow ROW0 = new MapBasedInputRow(
+      0,
+      ImmutableList.<String>of(),
+      ImmutableMap.<String, Object>of()
+  );
+
+  private static final InputRow ROW1 = new MapBasedInputRow(
+      0,
+      ImmutableList.<String>of(),
+      ImmutableMap.<String, Object>of("x", 4)
+  );
+
+  private static final InputRow ROW2 = new MapBasedInputRow(
+      0,
+      ImmutableList.<String>of(),
+      ImmutableMap.<String, Object>of("x", 2.1, "y", 3L)
+  );
+
+  private static final ExpressionVirtualColumn XPLUSY = new ExpressionVirtualColumn("expr", "x + y");
+  private static final TestColumnSelectorFactory COLUMN_SELECTOR_FACTORY = new TestColumnSelectorFactory();
+
+  @Test
+  public void testObjectSelector()
+  {
+    final ObjectColumnSelector selector = XPLUSY.makeObjectColumnSelector("expr", COLUMN_SELECTOR_FACTORY);
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW0);
+    Assert.assertEquals(null, selector.get());
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW1);
+    Assert.assertEquals(null, selector.get());
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW2);
+    Assert.assertEquals(5.1d, selector.get());
+  }
+
+  @Test
+  public void testLongSelector()
+  {
+    final LongColumnSelector selector = XPLUSY.makeLongColumnSelector("expr", COLUMN_SELECTOR_FACTORY);
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW0);
+    Assert.assertEquals(0L, selector.get());
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW1);
+    Assert.assertEquals(0L, selector.get());
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW2);
+    Assert.assertEquals(5L, selector.get());
+  }
+
+  @Test
+  public void testFloatSelector()
+  {
+    final FloatColumnSelector selector = XPLUSY.makeFloatColumnSelector("expr", COLUMN_SELECTOR_FACTORY);
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW0);
+    Assert.assertEquals(0.0f, selector.get(), 0.0f);
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW1);
+    Assert.assertEquals(0.0f, selector.get(), 0.0f);
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW2);
+    Assert.assertEquals(5.1f, selector.get(), 0.0f);
+  }
+
+  @Test
+  public void testDimensionSelector()
+  {
+    final DimensionSelector selector = XPLUSY.makeDimensionSelector(
+        new DefaultDimensionSpec("expr", "x"),
+        COLUMN_SELECTOR_FACTORY
+    );
+
+    final ValueMatcher nullMatcher = selector.makeValueMatcher((String) null);
+    final ValueMatcher fiveMatcher = selector.makeValueMatcher("5");
+    final ValueMatcher nonNullMatcher = selector.makeValueMatcher(Predicates.<String>notNull());
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW0);
+    Assert.assertEquals(true, nullMatcher.matches());
+    Assert.assertEquals(false, fiveMatcher.matches());
+    Assert.assertEquals(false, nonNullMatcher.matches());
+    Assert.assertEquals(null, selector.lookupName(selector.getRow().get(0)));
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW1);
+    Assert.assertEquals(true, nullMatcher.matches());
+    Assert.assertEquals(false, fiveMatcher.matches());
+    Assert.assertEquals(false, nonNullMatcher.matches());
+    Assert.assertEquals(null, selector.lookupName(selector.getRow().get(0)));
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW2);
+    Assert.assertEquals(false, nullMatcher.matches());
+    Assert.assertEquals(false, fiveMatcher.matches());
+    Assert.assertEquals(true, nonNullMatcher.matches());
+    Assert.assertEquals("5.1", selector.lookupName(selector.getRow().get(0)));
+  }
+
+  @Test
+  public void testDimensionSelectorWithExtraction()
+  {
+    final DimensionSelector selector = XPLUSY.makeDimensionSelector(
+        new ExtractionDimensionSpec("expr", "x", new BucketExtractionFn(1.0, 0.0)),
+        COLUMN_SELECTOR_FACTORY
+    );
+
+    final ValueMatcher nullMatcher = selector.makeValueMatcher((String) null);
+    final ValueMatcher fiveMatcher = selector.makeValueMatcher("5");
+    final ValueMatcher nonNullMatcher = selector.makeValueMatcher(Predicates.<String>notNull());
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW0);
+    Assert.assertEquals(true, nullMatcher.matches());
+    Assert.assertEquals(false, fiveMatcher.matches());
+    Assert.assertEquals(false, nonNullMatcher.matches());
+    Assert.assertEquals(null, selector.lookupName(selector.getRow().get(0)));
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW1);
+    Assert.assertEquals(true, nullMatcher.matches());
+    Assert.assertEquals(false, fiveMatcher.matches());
+    Assert.assertEquals(false, nonNullMatcher.matches());
+    Assert.assertEquals(null, selector.lookupName(selector.getRow().get(0)));
+
+    COLUMN_SELECTOR_FACTORY.setRow(ROW2);
+    Assert.assertEquals(false, nullMatcher.matches());
+    Assert.assertEquals(true, fiveMatcher.matches());
+    Assert.assertEquals(true, nonNullMatcher.matches());
+    Assert.assertEquals("5", selector.lookupName(selector.getRow().get(0)));
+  }
+
+  @Test
+  public void testRequiredColumns()
+  {
+    final ExpressionVirtualColumn virtualColumn = new ExpressionVirtualColumn("expr", "x + y");
+    Assert.assertEquals(ImmutableList.of("x", "y"), virtualColumn.requiredColumns());
+  }
+}

processing/src/test/java/io/druid/segment/virtual/VirtualColumnsTest.java

@@ -0,0 +1,417 @@
+/*
+ * 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.virtual;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableList;
+import com.google.common.primitives.Longs;
+import io.druid.jackson.DefaultObjectMapper;
+import io.druid.query.dimension.DefaultDimensionSpec;
+import io.druid.query.dimension.DimensionSpec;
+import io.druid.query.dimension.ExtractionDimensionSpec;
+import io.druid.query.extraction.BucketExtractionFn;
+import io.druid.query.extraction.ExtractionFn;
+import io.druid.query.filter.ValueMatcher;
+import io.druid.segment.ColumnSelectorFactory;
+import io.druid.segment.DimensionSelector;
+import io.druid.segment.DimensionSelectorUtils;
+import io.druid.segment.FloatColumnSelector;
+import io.druid.segment.IdLookup;
+import io.druid.segment.LongColumnSelector;
+import io.druid.segment.ObjectColumnSelector;
+import io.druid.segment.VirtualColumn;
+import io.druid.segment.VirtualColumns;
+import io.druid.segment.column.ColumnCapabilities;
+import io.druid.segment.column.ColumnCapabilitiesImpl;
+import io.druid.segment.column.ValueType;
+import io.druid.segment.data.IndexedInts;
+import it.unimi.dsi.fastutil.ints.IntIterator;
+import it.unimi.dsi.fastutil.ints.IntIterators;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+public class VirtualColumnsTest
+{
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
+  @Test
+  public void testMakeSelectors()
+  {
+    final VirtualColumns virtualColumns = makeVirtualColumns();
+    final ObjectColumnSelector objectSelector = virtualColumns.makeObjectColumnSelector("expr", null);
+    final DimensionSelector dimensionSelector = virtualColumns.makeDimensionSelector(
+        new DefaultDimensionSpec("expr", "x"),
+        null
+    );
+    final DimensionSelector extractionDimensionSelector = virtualColumns.makeDimensionSelector(
+        new ExtractionDimensionSpec("expr", "x", new BucketExtractionFn(1.0, 0.5)),
+        null
+    );
+    final FloatColumnSelector floatSelector = virtualColumns.makeFloatColumnSelector("expr", null);
+    final LongColumnSelector longSelector = virtualColumns.makeLongColumnSelector("expr", null);
+
+    Assert.assertEquals(1L, objectSelector.get());
+    Assert.assertEquals("1", dimensionSelector.lookupName(dimensionSelector.getRow().get(0)));
+    Assert.assertEquals("0.5", extractionDimensionSelector.lookupName(extractionDimensionSelector.getRow().get(0)));
+    Assert.assertEquals(1.0f, floatSelector.get(), 0.0f);
+    Assert.assertEquals(1L, longSelector.get());
+  }
+
+  @Test
+  public void testMakeSelectorsWithDotSupport()
+  {
+    final VirtualColumns virtualColumns = makeVirtualColumns();
+    final ObjectColumnSelector objectSelector = virtualColumns.makeObjectColumnSelector("foo.5", null);
+    final DimensionSelector dimensionSelector = virtualColumns.makeDimensionSelector(
+        new DefaultDimensionSpec("foo.5", "x"),
+        null
+    );
+    final FloatColumnSelector floatSelector = virtualColumns.makeFloatColumnSelector("foo.5", null);
+    final LongColumnSelector longSelector = virtualColumns.makeLongColumnSelector("foo.5", null);
+
+    Assert.assertEquals(5L, objectSelector.get());
+    Assert.assertEquals("5", dimensionSelector.lookupName(dimensionSelector.getRow().get(0)));
+    Assert.assertEquals(5.0f, floatSelector.get(), 0.0f);
+    Assert.assertEquals(5L, longSelector.get());
+  }
+
+  @Test
+  public void testMakeSelectorsWithDotSupportBaseNameOnly()
+  {
+    final VirtualColumns virtualColumns = makeVirtualColumns();
+    final ObjectColumnSelector objectSelector = virtualColumns.makeObjectColumnSelector("foo", null);
+    final DimensionSelector dimensionSelector = virtualColumns.makeDimensionSelector(
+        new DefaultDimensionSpec("foo", "x"),
+        null
+    );
+    final FloatColumnSelector floatSelector = virtualColumns.makeFloatColumnSelector("foo", null);
+    final LongColumnSelector longSelector = virtualColumns.makeLongColumnSelector("foo", null);
+
+    Assert.assertEquals(-1L, objectSelector.get());
+    Assert.assertEquals("-1", dimensionSelector.lookupName(dimensionSelector.getRow().get(0)));
+    Assert.assertEquals(-1.0f, floatSelector.get(), 0.0f);
+    Assert.assertEquals(-1L, longSelector.get());
+  }
+
+  @Test
+  public void testTimeNotAllowed()
+  {
+    final ExpressionVirtualColumn expr = new ExpressionVirtualColumn("__time", "x + y");
+
+    expectedException.expect(IllegalArgumentException.class);
+    expectedException.expectMessage("virtualColumn name[__time] not allowed");
+
+    VirtualColumns.create(ImmutableList.<VirtualColumn>of(expr));
+  }
+
+  @Test
+  public void testDuplicateNameDetection()
+  {
+    final ExpressionVirtualColumn expr = new ExpressionVirtualColumn("expr", "x + y");
+    final ExpressionVirtualColumn expr2 = new ExpressionVirtualColumn("expr", "x * 2");
+
+    expectedException.expect(IllegalArgumentException.class);
+    expectedException.expectMessage("Duplicate virtualColumn name[expr]");
+
+    VirtualColumns.create(ImmutableList.<VirtualColumn>of(expr, expr2));
+  }
+
+  @Test
+  public void testCycleDetection()
+  {
+    final ExpressionVirtualColumn expr = new ExpressionVirtualColumn("expr", "x + expr2");
+    final ExpressionVirtualColumn expr2 = new ExpressionVirtualColumn("expr2", "expr * 2");
+
+    expectedException.expect(IllegalArgumentException.class);
+    expectedException.expectMessage("Self-referential column[expr]");
+
+    VirtualColumns.create(ImmutableList.<VirtualColumn>of(expr, expr2));
+  }
+
+  @Test
+  public void testGetCacheKey() throws Exception
+  {
+    final VirtualColumns virtualColumns = VirtualColumns.create(
+        ImmutableList.<VirtualColumn>of(
+            new ExpressionVirtualColumn("expr", "x + y")
+        )
+    );
+
+    final VirtualColumns virtualColumns2 = VirtualColumns.create(
+        ImmutableList.<VirtualColumn>of(
+            new ExpressionVirtualColumn("expr", "x + y")
+        )
+    );
+
+    Assert.assertArrayEquals(virtualColumns.getCacheKey(), virtualColumns2.getCacheKey());
+    Assert.assertFalse(Arrays.equals(virtualColumns.getCacheKey(), VirtualColumns.EMPTY.getCacheKey()));
+  }
+
+  @Test
+  public void testEqualsAndHashCode() throws Exception
+  {
+    final VirtualColumns virtualColumns = VirtualColumns.create(
+        ImmutableList.<VirtualColumn>of(
+            new ExpressionVirtualColumn("expr", "x + y")
+        )
+    );
+
+    final VirtualColumns virtualColumns2 = VirtualColumns.create(
+        ImmutableList.<VirtualColumn>of(
+            new ExpressionVirtualColumn("expr", "x + y")
+        )
+    );
+
+    Assert.assertEquals(virtualColumns, virtualColumns);
+    Assert.assertEquals(virtualColumns, virtualColumns2);
+    Assert.assertNotEquals(VirtualColumns.EMPTY, virtualColumns);
+    Assert.assertNotEquals(VirtualColumns.EMPTY, null);
+
+    Assert.assertEquals(virtualColumns.hashCode(), virtualColumns.hashCode());
+    Assert.assertEquals(virtualColumns.hashCode(), virtualColumns2.hashCode());
+    Assert.assertNotEquals(VirtualColumns.EMPTY.hashCode(), virtualColumns.hashCode());
+  }
+
+  @Test
+  public void testSerde() throws Exception
+  {
+    final ObjectMapper mapper = new DefaultObjectMapper();
+    final ImmutableList<VirtualColumn> theColumns = ImmutableList.<VirtualColumn>of(
+        new ExpressionVirtualColumn("expr", "x + y"),
+        new ExpressionVirtualColumn("expr2", "x + z")
+    );
+    final VirtualColumns virtualColumns = VirtualColumns.create(theColumns);
+
+    Assert.assertEquals(
+        virtualColumns,
+        mapper.readValue(
+            mapper.writeValueAsString(virtualColumns),
+            VirtualColumns.class
+        )
+    );
+
+    Assert.assertEquals(
+        theColumns,
+        mapper.readValue(
+            mapper.writeValueAsString(virtualColumns),
+            mapper.getTypeFactory().constructParametricType(List.class, VirtualColumn.class)
+        )
+    );
+  }
+
+  private VirtualColumns makeVirtualColumns()
+  {
+    final ExpressionVirtualColumn expr = new ExpressionVirtualColumn("expr", "1");
+    final DottyVirtualColumn dotty = new DottyVirtualColumn("foo");
+    return VirtualColumns.create(ImmutableList.of(expr, dotty));
+  }
+
+  static class DottyVirtualColumn implements VirtualColumn
+  {
+    private final String name;
+
+    public DottyVirtualColumn(String name)
+    {
+      this.name = name;
+    }
+
+    @Override
+    public String getOutputName()
+    {
+      return name;
+    }
+
+    @Override
+    public ObjectColumnSelector makeObjectColumnSelector(String columnName, ColumnSelectorFactory factory)
+    {
+      final LongColumnSelector selector = makeLongColumnSelector(columnName, factory);
+      return new ObjectColumnSelector()
+      {
+        @Override
+        public Class classOfObject()
+        {
+          return Long.class;
+        }
+
+        @Override
+        public Object get()
+        {
+          return selector.get();
+        }
+      };
+    }
+
+    @Override
+    public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec, ColumnSelectorFactory factory)
+    {
+      final LongColumnSelector selector = makeLongColumnSelector(dimensionSpec.getDimension(), factory);
+      final ExtractionFn extractionFn = dimensionSpec.getExtractionFn();
+      final DimensionSelector dimensionSelector = new DimensionSelector()
+      {
+        @Override
+        public IndexedInts getRow()
+        {
+          return new IndexedInts()
+          {
+            @Override
+            public int size()
+            {
+              return 1;
+            }
+
+            @Override
+            public int get(int index)
+            {
+              return 0;
+            }
+
+            @Override
+            public IntIterator iterator()
+            {
+              return IntIterators.singleton(0);
+            }
+
+            @Override
+            public void fill(int index, int[] toFill)
+            {
+              throw new UnsupportedOperationException("fill not supported");
+            }
+
+            @Override
+            public void close() throws IOException
+            {
+
+            }
+          };
+        }
+
+        @Override
+        public int getValueCardinality()
+        {
+          return DimensionSelector.CARDINALITY_UNKNOWN;
+        }
+
+        @Override
+        public String lookupName(int id)
+        {
+          final String stringValue = String.valueOf(selector.get());
+          return extractionFn == null ? stringValue : extractionFn.apply(stringValue);
+        }
+
+        @Override
+        public ValueMatcher makeValueMatcher(final String value)
+        {
+          return DimensionSelectorUtils.makeValueMatcherGeneric(this, value);
+        }
+
+        @Override
+        public ValueMatcher makeValueMatcher(final Predicate<String> predicate)
+        {
+          return DimensionSelectorUtils.makeValueMatcherGeneric(this, predicate);
+        }
+
+        @Override
+        public boolean nameLookupPossibleInAdvance()
+        {
+          return false;
+        }
+
+        @Nullable
+        @Override
+        public IdLookup idLookup()
+        {
+          return new IdLookup()
+          {
+            @Override
+            public int lookupId(final String name)
+            {
+              return 0;
+            }
+          };
+        }
+      };
+
+      return dimensionSpec.decorate(dimensionSelector);
+    }
+
+    @Override
+    public FloatColumnSelector makeFloatColumnSelector(String columnName, ColumnSelectorFactory factory)
+    {
+      final LongColumnSelector selector = makeLongColumnSelector(columnName, factory);
+      return new FloatColumnSelector()
+      {
+        @Override
+        public float get()
+        {
+          return selector.get();
+        }
+      };
+    }
+
+    @Override
+    public LongColumnSelector makeLongColumnSelector(String columnName, ColumnSelectorFactory factory)
+    {
+      final String subColumn = VirtualColumns.splitColumnName(columnName).rhs;
+      final Long boxed = subColumn == null ? null : Longs.tryParse(subColumn);
+      final long theLong = boxed == null ? -1 : boxed;
+      return new LongColumnSelector()
+      {
+        @Override
+        public long get()
+        {
+          return theLong;
+        }
+      };
+    }
+
+    @Override
+    public ColumnCapabilities capabilities(String columnName)
+    {
+      return new ColumnCapabilitiesImpl().setType(ValueType.LONG);
+    }
+
+    @Override
+    public List<String> requiredColumns()
+    {
+      return ImmutableList.of();
+    }
+
+    @Override
+    public boolean usesDotNotation()
+    {
+      return true;
+    }
+
+    @Override
+    public byte[] getCacheKey()
+    {
+      throw new UnsupportedOperationException();
+    }
+  }
+}