add tests for timeboundary timewarp

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

@@ -100,7 +100,6 @@ public class TimewarpOperator<T> implements PostProcessingOperator<T>
                 if (input instanceof Result) {
                   Result res = (Result) input;
                   Object value = res.getValue();
-                  final DateTime timestamp = res.getTimestamp().minus(offset);
                   if (value instanceof TimeBoundaryResultValue) {
                     TimeBoundaryResultValue boundary = (TimeBoundaryResultValue) value;
 
@@ -112,12 +111,12 @@ public class TimewarpOperator<T> implements PostProcessingOperator<T>
                     final DateTime maxTime = boundary.getMaxTime();
 
                     return (T) ((TimeBoundaryQuery) query).buildResult(
-                        timestamp,
+                        new DateTime(Math.min(res.getTimestamp().getMillis() - offset, now)),
                         minTime != null ? minTime.minus(offset) : null,
-                        maxTime != null ? maxTime.minus(offset) : null
+                        maxTime != null ? new DateTime(Math.min(maxTime.getMillis() - offset, now)) : null
                     ).iterator().next();
                   }
-                  return (T) new Result(timestamp, value);
+                  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());

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

@@ -26,6 +26,7 @@ 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.timeboundary.TimeBoundaryResultValue;
 import io.druid.query.timeseries.TimeseriesResultValue;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
@@ -122,5 +123,46 @@ public class TimewarpOperatorTest
         Sequences.toList(queryRunner.run(query), Lists.<Result<TimeseriesResultValue>>newArrayList())
     );
 
+
+    TimewarpOperator<Result<TimeBoundaryResultValue>> timeBoundaryOperator = 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
+    );
+
+    QueryRunner<Result<TimeBoundaryResultValue>> timeBoundaryRunner = timeBoundaryOperator.postProcess(
+        new QueryRunner<Result<TimeBoundaryResultValue>>()
+        {
+          @Override
+          public Sequence<Result<TimeBoundaryResultValue>> run(Query<Result<TimeBoundaryResultValue>> query)
+          {
+            return Sequences.simple(
+                ImmutableList.of(
+                    new Result<>(
+                        new DateTime("2014-01-12"),
+                        new TimeBoundaryResultValue(ImmutableMap.<String, Object>of("maxTime", new DateTime("2014-01-12")))
+                    )
+                )
+            );
+          }
+        },
+        new DateTime("2014-08-02").getMillis()
+    );
+
+    final Query<Result<TimeBoundaryResultValue>> timeBoundaryQuery =
+        Druids.newTimeBoundaryQueryBuilder()
+              .dataSource("dummy")
+              .build();
+
+    Assert.assertEquals(
+        Lists.newArrayList(
+            new Result<>(
+                new DateTime("2014-08-02"),
+                new TimeBoundaryResultValue(ImmutableMap.<String, Object>of("maxTime", new DateTime("2014-08-02")))
+            )
+        ),
+        Sequences.toList(timeBoundaryRunner.run(timeBoundaryQuery), Lists.<Result<TimeBoundaryResultValue>>newArrayList())
+    );
+
   }
 }