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.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;

/**
 * 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 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,
      @JsonProperty("period") Period period,
      @JsonProperty("origin") DateTime origin
  )
  {
    this.originMillis = origin.getMillis();
    this.dataInterval = dataInterval;
    // this will fail for periods that do not map to millis (e.g. P1M)
    this.periodMillis = period.toStandardDuration().getMillis();
  }


  @Override
  public QueryRunner<T> postProcess(QueryRunner<T> baseQueryRunner)
  {
    return postProcess(baseQueryRunner, DateTime.now().getMillis());
  }

  public QueryRunner<T> postProcess(final QueryRunner<T> baseRunner, final long now)
  {
    return new QueryRunner<T>()
    {
      @Override
      public Sequence<T> run(Query<T> query)
      {
        final long offset = computeOffset(now);

        final Interval interval = query.getIntervals().get(0);
        final Interval modifiedInterval = new Interval(
            interval.getStartMillis() + offset,
            Math.min(interval.getEndMillis() + offset, now + offset)
        );
        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, now))
                        )
                    );
                  }
                  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());
                }

                // default to noop for unknown result types
                return input;
              }
            }
        );
      }
    };
  }

  /**
   * 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;
  }
}