query post-processing operators

- Allows post-processing of query results
- Example timewarp post-processing operator

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

+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2014  Metamarkets Group Inc.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+
+package io.druid.query;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes(value = {
+    @JsonSubTypes.Type(name = "timewarp", value = TimewarpOperator.class)
+})
+public interface PostProcessingOperator<T>
+{
+  public QueryRunner<T> postProcess(QueryRunner<T> baseQueryRunner);
+}

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

+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2014  Metamarkets Group Inc.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+
+package io.druid.query;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableMap;
+import com.metamx.common.ISE;
+import com.metamx.common.guava.Sequence;
+import com.metamx.common.guava.Sequences;
+import io.druid.data.input.MapBasedRow;
+import io.druid.query.spec.MultipleIntervalSegmentSpec;
+import io.druid.query.timeboundary.TimeBoundaryQuery;
+import io.druid.query.timeboundary.TimeBoundaryResultValue;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.joda.time.Period;
+
+import java.util.Arrays;
+
+public class TimewarpOperator<T> implements PostProcessingOperator<T>
+{
+  private final Interval dataInterval;
+  private final Period period;
+  private final DateTime origin;
+
+  @JsonCreator
+  public TimewarpOperator(
+      @JsonProperty("dataInterval") Interval dataInterval,
+      @JsonProperty("period") Period period,
+      @JsonProperty("origin") DateTime origin
+  )
+  {
+    this.origin = origin;
+    this.dataInterval = dataInterval;
+    this.period = period;
+  }
+
+
+  @Override
+  public QueryRunner<T> postProcess(final QueryRunner<T> baseRunner)
+  {
+    return new QueryRunner<T>()
+    {
+      @Override
+      public Sequence<T> run(Query<T> query)
+      {
+        final long t = DateTime.now().getMillis();
+        final long originMillis = origin.getMillis();
+
+        // this will fail for periods that do not map to millis (e.g. P1M)
+        final long periodMillis = period.toStandardDuration().getMillis();
+
+        // map time t into the last `period` fully contained within dataInterval
+        long startMillis = dataInterval.getEnd().minus(period).getMillis();
+        startMillis -= startMillis % periodMillis - originMillis % periodMillis;
+        final long offset = startMillis + (t % periodMillis) - (originMillis % periodMillis) - t;
+
+        final Interval interval = query.getIntervals().get(0);
+        final Interval modifiedInterval = new Interval(
+            interval.getStartMillis() + offset,
+            Math.min(interval.getEndMillis() + offset, t)
+        );
+        return Sequences.map(
+            baseRunner.run(
+                query.withQuerySegmentSpec(new MultipleIntervalSegmentSpec(Arrays.asList(modifiedInterval)))
+            ),
+            new Function<T, T>()
+            {
+              @Override
+              public T apply(T input)
+              {
+                if (input instanceof Result) {
+                  Result res = (Result) input;
+                  Object value = res.getValue();
+                  if (value instanceof TimeBoundaryResultValue) {
+                    TimeBoundaryResultValue boundary = (TimeBoundaryResultValue) value;
+                    value = new TimeBoundaryResultValue(
+                        ImmutableMap.of(
+                            TimeBoundaryQuery.MIN_TIME, boundary.getMinTime().minus(offset),
+                            TimeBoundaryQuery.MAX_TIME, new DateTime(Math.min(boundary.getMaxTime().getMillis() - offset, t))
+                        )
+                    );
+                  }
+                  return (T) new Result(res.getTimestamp().minus(offset), value);
+                } else if (input instanceof MapBasedRow) {
+                  MapBasedRow row = (MapBasedRow) input;
+                  return (T) new MapBasedRow(row.getTimestamp().minus(offset), row.getEvent());
+                }
+                throw new ISE("Don't know how to timewarp results of type[%s]", input.getClass());
+              }
+            }
+        );
+      }
+    };
+  }
+}

server/src/main/java/io/druid/server/ClientQuerySegmentWalker.java

@@ -19,6 +19,8 @@
 
 package io.druid.server;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Function;
 import com.google.inject.Inject;
 import com.metamx.emitter.service.ServiceEmitter;
@@ -26,6 +28,7 @@ import com.metamx.emitter.service.ServiceMetricEvent;
 import io.druid.client.CachingClusteredClient;
 import io.druid.query.FinalizeResultsQueryRunner;
 import io.druid.query.MetricsEmittingQueryRunner;
+import io.druid.query.PostProcessingOperator;
 import io.druid.query.Query;
 import io.druid.query.QueryRunner;
 import io.druid.query.QuerySegmentWalker;
@@ -44,17 +47,20 @@ public class ClientQuerySegmentWalker implements QuerySegmentWalker
   private final ServiceEmitter emitter;
   private final CachingClusteredClient baseClient;
   private final QueryToolChestWarehouse warehouse;
+  private final ObjectMapper objectMapper;
 
   @Inject
   public ClientQuerySegmentWalker(
       ServiceEmitter emitter,
       CachingClusteredClient baseClient,
-      QueryToolChestWarehouse warehouse
+      QueryToolChestWarehouse warehouse,
+      ObjectMapper objectMapper
   )
   {
     this.emitter = emitter;
     this.baseClient = baseClient;
     this.warehouse = warehouse;
+    this.objectMapper = objectMapper;
   }
 
   @Override
@@ -69,10 +75,10 @@ public class ClientQuerySegmentWalker implements QuerySegmentWalker
     return makeRunner(query);
   }
 
-  private <T> FinalizeResultsQueryRunner<T> makeRunner(final Query<T> query)
+  private <T> QueryRunner<T> makeRunner(final Query<T> query)
   {
     final QueryToolChest<T, Query<T>> toolChest = warehouse.getToolChest(query);
-    return new FinalizeResultsQueryRunner<T>(
+    final FinalizeResultsQueryRunner<T> baseRunner = new FinalizeResultsQueryRunner<T>(
         toolChest.postMergeQueryDecoration(
             toolChest.mergeResults(
                 new UnionQueryRunner<T>(
@@ -94,5 +100,15 @@ public class ClientQuerySegmentWalker implements QuerySegmentWalker
         ),
         toolChest
     );
+
+
+    final PostProcessingOperator<T> postProcessing = objectMapper.convertValue(
+        query.getContext().get("postProcessing"),
+        new TypeReference<PostProcessingOperator<T>>() {
+        }
+    );
+
+    return postProcessing != null ?
+           postProcessing.postProcess(baseRunner) : baseRunner;
   }
 }