gt:             if (next_dp.timestamp() >= query_end) {
context: // This file is part of OpenTSDB.
  // Copyright (C) 2014  The OpenTSDB Authors.
  //
  // This program is free software: you can redistribute it and/or modify it
  // under the terms of the GNU Lesser General Public License as published by
  // the Free Software Foundation, either version 2.1 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 Lesser
  // General Public License for more details.  You should have received a copy
  // of the GNU Lesser General Public License along with this program.  If not,
  // see <http://www.gnu.org/licenses/>.
  package net.opentsdb.core;
  
  import java.util.Calendar;
  import java.util.NoSuchElementException;
  
  import net.opentsdb.utils.DateTime;
  
  /**
   * Iterator that downsamples data points using an {@link Aggregator}.
   */
  public class Downsampler implements SeekableView, DataPoint {
    
    /** Matches the weekly downsampler as it requires special handling. */
    protected final static int WEEK_UNIT = DateTime.unitsToCalendarType("w");
    protected final static int DAY_UNIT = DateTime.unitsToCalendarType("d");
    protected final static int WEEK_LENGTH = 7;
    
    /** The downsampling specification when provided */
    protected final DownsamplingSpecification specification;
    
    /** The start timestamp of the actual query for use with "all" */
    protected final long query_start;
    
    /** The end timestamp of the actual query for use with "all" */
    protected final long query_end;
    
    /** The data source */
    protected final SeekableView source;
    
    /** Iterator to iterate the values of the current interval. */
    protected final ValuesInInterval values_in_interval;
    
    /** Last normalized timestamp */ 
    protected long timestamp;
    
    /** Last value as a double */
    protected double value;
    
    /** Whether or not to merge all DPs in the source into one vaalue */
    protected final boolean run_all;
    
    /** The interval to use with a calendar */
    protected final int interval;
    
    /** The unit to use with a calendar as a Calendar integer */
    protected final int unit;
    
    /**
     * Ctor.
     * @param source The iterator to access the underlying data.
     * @param interval_ms The interval in milli seconds wanted between each data
     * point.
     * @param downsampler The downsampling function to use.
     * @deprecated as of 2.3
     */
    Downsampler(final SeekableView source,
                final long interval_ms,
                final Aggregator downsampler) {
      this.source = source;
      if (downsampler == Aggregators.NONE) {
        throw new IllegalArgumentException("cannot use the NONE "
            + "aggregator for downsampling");
      }
      specification = new DownsamplingSpecification(interval_ms, downsampler, 
          DownsamplingSpecification.DEFAULT_FILL_POLICY);
      values_in_interval = new ValuesInInterval();
      query_start = 0;
      query_end = 0;
      interval = unit = 0;
      run_all = false;
    }
    
    /**
     * Ctor.
     * @param source The iterator to access the underlying data.
     * @param specification The downsampling spec to use
     * @param query_start The start timestamp of the actual query for use with "all"
     * @param query_end The end timestamp of the actual query for use with "all"
     * @since 2.3
     */
    Downsampler(final SeekableView source,
                final DownsamplingSpecification specification,
                final long query_start,
                final long query_end
        ) {
      this.source = source;
      this.specification = specification;
      values_in_interval = new ValuesInInterval();
      this.query_start = query_start;
      this.query_end = query_end;
      
      final String s = specification.getStringInterval();
      if (s != null && s.toLowerCase().contains("all")) {
        run_all = true;
        interval = unit = 0;
      } else if (s != null && specification.useCalendar()) {
        if (s.toLowerCase().contains("ms")) {
          interval = Integer.parseInt(s.substring(0, s.length() - 2));
          unit = DateTime.unitsToCalendarType(s.substring(s.length() - 2));
        } else {
          interval = Integer.parseInt(s.substring(0, s.length() - 1));
          unit = DateTime.unitsToCalendarType(s.substring(s.length() - 1));
        }
        run_all = false;
      } else {
        run_all = false;
        interval = unit = 0;
      }
    }
  
    // ------------------ //
    // Iterator interface //
    // ------------------ //
  
    @Override
    public boolean hasNext() {
      return values_in_interval.hasNextValue();
    }
  
    /**
     * @throws NoSuchElementException if no data points remain.
     */
    @Override
    public DataPoint next() {
      if (hasNext()) {
        value = specification.getFunction().runDouble(values_in_interval);
        timestamp = values_in_interval.getIntervalTimestamp();
        values_in_interval.moveToNextInterval();
        return this;
      }
      throw new NoSuchElementException("no more data points in " + this);
    }
  
    @Override
    public void remove() {
      throw new UnsupportedOperationException();
    }
  
    // ---------------------- //
    // SeekableView interface //
    // ---------------------- //
  
    @Override
    public void seek(final long timestamp) {
      values_in_interval.seekInterval(timestamp);
    }
  
    // ------------------- //
    // DataPoint interface //
    // ------------------- //
  
    @Override
    public long timestamp() {
      if (run_all) {
        return query_start;
      }
      return timestamp;
    }
  
    @Override
    public boolean isInteger() {
      return false;
    }
  
    @Override
    public long longValue() {
      throw new ClassCastException("Downsampled values are doubles");
    }
  
    @Override
    public double doubleValue() {
      return value;
    }
  
    @Override
    public double toDouble() {
      return value;
    }
  
    @Override
    public String toString() {
      final StringBuilder buf = new StringBuilder();
      buf.append("Downsampler: ")
         .append(", downsampler=").append(specification)
         .append(", queryStart=").append(query_start)
         .append(", queryEnd=").append(query_end)
         .append(", runAll=").append(run_all)
         .append(", current data=(timestamp=").append(timestamp)
         .append(", value=").append(value)
         .append("), values_in_interval=").append(values_in_interval);
     return buf.toString();
    }
  
    /** Iterates source values for an interval. */
    protected class ValuesInInterval implements Aggregator.Doubles {
  
      /** An optional calendar set to the current timestamp for the data point */
      private Calendar previous_calendar;
      
      /** An optional calendar set to the end of the interval timestamp */
      private Calendar next_calendar;
      
      /** The end of the current interval. */
      private long timestamp_end_interval = Long.MIN_VALUE;
      
      /** True if the last value was successfully extracted from the source. */
      private boolean has_next_value_from_source = false;
      
      /** The last data point extracted from the source. */
      private DataPoint next_dp = null;
  
      /** True if it is initialized for iterating intervals. */
      private boolean initialized = false;
  
      /**
       * Constructor.
       */
      protected ValuesInInterval() {
        if (run_all) {
          timestamp_end_interval = query_end;
        } else if (!specification.useCalendar()) {
          timestamp_end_interval = specification.getInterval();
        }
      }
  
      /** Initializes to iterate intervals. */
      protected void initializeIfNotDone() {
        // NOTE: Delay initialization is required to not access any data point
        // from the source until a user requests it explicitly to avoid the severe
        // performance penalty by accessing the unnecessary first data of a span.
        if (!initialized) {
          initialized = true;
          if (source.hasNext()) {
            moveToNextValue();
            if (!run_all) {
              if (specification.useCalendar()) {
                previous_calendar = DateTime.previousInterval(next_dp.timestamp(), 
                    interval, unit, specification.getTimezone());
                next_calendar = DateTime.previousInterval(next_dp.timestamp(), 
                    interval, unit, specification.getTimezone());
                if (unit == WEEK_UNIT) {
                  next_calendar.add(DAY_UNIT, interval * WEEK_LENGTH);
                } else {
                  next_calendar.add(unit, interval);
                }
                timestamp_end_interval = next_calendar.getTimeInMillis();
              } else {
                timestamp_end_interval = alignTimestamp(next_dp.timestamp()) + 
                    specification.getInterval();
              }
            }
          }
        }
      }
  
      /** Extracts the next value from the source. */
      private void moveToNextValue() {
        if (source.hasNext()) {
          has_next_value_from_source = true;
          // filter out dps that don't match start and end for run_alls
          if (run_all) {
            while (source.hasNext()) {
              next_dp = source.next();
              if (next_dp.timestamp() < query_start) {
                next_dp = null;
                continue;
              }
  