add documentation and tests

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

@@ -35,12 +35,26 @@ import org.joda.time.Period;
 
 import java.util.Arrays;
 
+/**
+ * TimewarpOperator is an example post-processing operator that maps current time
+ * to the latest period ending withing the specified data interval and truncates
+ * the query interval to discard data that would be mapped to the future.
+ *
+ */
 public class TimewarpOperator<T> implements PostProcessingOperator<T>
 {
   private final Interval dataInterval;
-  private final Period period;
-  private final DateTime origin;
+  private final long periodMillis;
+  private final long originMillis;
 
+  /**
+   *
+   * @param dataInterval interval containing the actual data
+   * @param period time will be offset by a multiple of the given period
+   *               until there is at least a full period ending within the data interval
+   * @param origin origin to be used to align time periods
+   *               (e.g. to determine on what day of the week a weekly period starts)
+   */
   @JsonCreator
   public TimewarpOperator(
       @JsonProperty("dataInterval") Interval dataInterval,
@@ -48,35 +62,32 @@ public class TimewarpOperator<T> implements PostProcessingOperator<T>
       @JsonProperty("origin") DateTime origin
   )
   {
-    this.origin = origin;
+    this.originMillis = origin.getMillis();
     this.dataInterval = dataInterval;
-    this.period = period;
+    // this will fail for periods that do not map to millis (e.g. P1M)
+    this.periodMillis = period.toStandardDuration().getMillis();
   }
 
 
   @Override
-  public QueryRunner<T> postProcess(final QueryRunner<T> baseRunner)
+  public QueryRunner<T> postProcess(QueryRunner<T> baseQueryRunner)
+  {
+    return postProcess(baseQueryRunner, DateTime.now().getMillis());
+  }
+
+  public QueryRunner<T> postProcess(final QueryRunner<T> baseRunner, final long t)
   {
     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 long offset = computeOffset(t);
 
         final Interval interval = query.getIntervals().get(0);
         final Interval modifiedInterval = new Interval(
             interval.getStartMillis() + offset,
-            Math.min(interval.getEndMillis() + offset, t)
+            Math.min(interval.getEndMillis() + offset, t + offset)
         );
         return Sequences.map(
             baseRunner.run(
@@ -113,4 +124,29 @@ public class TimewarpOperator<T> implements PostProcessingOperator<T>
       }
     };
   }
+
+  /**
+   * Map time t into the last `period` ending within `dataInterval`
+   *
+   * @param t
+   * @return the offset between the mapped time and time t
+   */
+  protected long computeOffset(final long t)
+  {
+    // start is the beginning of the last period ending within dataInterval
+    long start = dataInterval.getEndMillis() - periodMillis;
+    long startOffset = start % periodMillis - originMillis % periodMillis;
+    if(startOffset < 0) {
+      startOffset += periodMillis;
+    };
+    start -= startOffset;
+
+    // tOffset is the offset time t within the last period
+    long tOffset = t % periodMillis - originMillis % periodMillis;
+    if(tOffset < 0) {
+      tOffset += periodMillis;
+    }
+    tOffset += start;
+    return tOffset - t;
+  }
 }

processing/src/test/java/io/druid/query/TimewarpOperatorTest.java

+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2012, 2013, 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.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.metamx.common.guava.Sequence;
+import com.metamx.common.guava.Sequences;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.CountAggregatorFactory;
+import io.druid.query.timeseries.TimeseriesResultValue;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.joda.time.Period;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+
+public class TimewarpOperatorTest
+{
+  TimewarpOperator<Result<TimeseriesResultValue>> testOperator = new TimewarpOperator<>(
+      new Interval(new DateTime("2014-01-01"), new DateTime("2014-01-15")),
+      new Period("P1W"),
+      new DateTime("2014-01-06") // align on Monday
+  );
+
+  @Test
+  public void testComputeOffset() throws Exception
+  {
+    {
+      final DateTime t = new DateTime("2014-01-23");
+      final DateTime tOffset = new DateTime("2014-01-09");
+
+      Assert.assertEquals(
+          new DateTime(tOffset),
+          t.plus(testOperator.computeOffset(t.getMillis()))
+      );
+    }
+
+    {
+      final DateTime t = new DateTime("2014-08-02");
+      final DateTime tOffset = new DateTime("2014-01-11");
+
+      Assert.assertEquals(
+          new DateTime(tOffset),
+          t.plus(testOperator.computeOffset(t.getMillis()))
+      );
+    }
+  }
+
+  @Test
+  public void testPostProcess() throws Exception
+  {
+    QueryRunner<Result<TimeseriesResultValue>> queryRunner = testOperator.postProcess(
+        new QueryRunner<Result<TimeseriesResultValue>>()
+        {
+          @Override
+          public Sequence<Result<TimeseriesResultValue>> run(Query<Result<TimeseriesResultValue>> query)
+          {
+            return Sequences.simple(
+                ImmutableList.of(
+                    new Result<>(
+                        new DateTime(new DateTime("2014-01-09")),
+                        new TimeseriesResultValue(ImmutableMap.<String, Object>of("metric", 2))
+                    ),
+                    new Result<>(
+                        new DateTime(new DateTime("2014-01-11")),
+                        new TimeseriesResultValue(ImmutableMap.<String, Object>of("metric", 3))
+                    ),
+                    new Result<>(
+                        query.getIntervals().get(0).getEnd(),
+                        new TimeseriesResultValue(ImmutableMap.<String, Object>of("metric", 5))
+                    )
+                )
+            );
+          }
+        },
+        new DateTime("2014-08-02").getMillis()
+    );
+
+    final Query<Result<TimeseriesResultValue>> query =
+        Druids.newTimeseriesQueryBuilder()
+              .dataSource("dummy")
+              .intervals("2014-07-31/2014-08-05")
+              .aggregators(Arrays.<AggregatorFactory>asList(new CountAggregatorFactory("count")))
+              .build();
+
+    Assert.assertEquals(
+        Lists.newArrayList(
+            new Result<>(
+                new DateTime("2014-07-31"),
+                new TimeseriesResultValue(ImmutableMap.<String, Object>of("metric", 2))
+            ),
+            new Result<>(
+                new DateTime("2014-08-02"),
+                new TimeseriesResultValue(ImmutableMap.<String, Object>of("metric", 3))
+            ),
+            new Result<>(
+                new DateTime("2014-08-02"),
+                new TimeseriesResultValue(ImmutableMap.<String, Object>of("metric", 5))
+            )
+        ),
+        Sequences.toList(queryRunner.run(query), Lists.<Result<TimeseriesResultValue>>newArrayList())
+    );
+
+  }
+}