docs: @Override
    public DataPoint next() {
   // Don't proceed if we've already completed iteration.
      if (hasNext()) {
        // Ensure that the timestamp we request is valid.
        values_in_interval.initializeIfNotDone();
  
        // Skip any leading data outside the query bounds.
        long actual = values_in_interval.hasNextValue() ? 
            values_in_interval.getIntervalTimestamp() : Long.MAX_VALUE;
        
        while (!run_all && values_in_interval.hasNextValue() 
            && actual < timestamp) {
          // The actual timestamp precedes our expected, so there's data in the
          // values-in-interval object that we wish to ignore.
          specification.getFunction().runDouble(values_in_interval);
          values_in_interval.moveToNextInterval();
          actual = values_in_interval.getIntervalTimestamp();
        }
        
        // Check whether the timestamp of the calculation interval matches what
        // we expect.
        if (run_all || actual == timestamp) {
          // The calculated interval timestamp matches what we expect, so we can
          // do normal processing.
          if (rollup_query != null && 
              (rollup_query.getRollupAgg() == Aggregators.AVG || 
              rollup_query.getRollupAgg() == Aggregators.DEV)) {
            if (rollup_query.getRollupAgg() == Aggregators.AVG) {
              if (specification.getFunction() == Aggregators.AVG) {
                double sum = 0;
                long count = 0;
                while (values_in_interval.hasNextValue()) {
                  count += values_in_interval.nextValueCount();
                  sum += values_in_interval.nextDoubleValue();
                }
                if (count == 0) { // avoid # / 0
                  value = 0;
                } else {
                  value = sum / (double)count;
                }
              } else {
                class Accumulator implements Doubles {
                  List<Double> values = new ArrayList<Double>();
                  Iterator<Double> iterator;
                  @Override
                  public boolean hasNextValue() {
                    return iterator.hasNext();
                  }
                  @Override
                  public double nextDoubleValue() {
                    return iterator.next();
                  }
                }
                
                final Accumulator accumulator = new Accumulator();
                while (values_in_interval.hasNextValue()) {
                  long count = values_in_interval.nextValueCount();
                  double sum = values_in_interval.nextDoubleValue();
                  if (count == 0) {
                    accumulator.values.add(0D);
                  } else {
                    accumulator.values.add(sum / (double) count);
                  }
                }
                accumulator.iterator = accumulator.values.iterator();
                value = specification.getFunction().runDouble(accumulator);
              }
            } else if (specification.getFunction() == Aggregators.DEV) {
              throw new UnsupportedOperationException("Standard deviation over "
                  + "rolled up data is not supported at this time");
            }
          } else if (rollup_query != null && 
              specification.getFunction() == Aggregators.COUNT) {
            double count = 0;
            while (values_in_interval.hasNextValue()) {
              count += values_in_interval.nextValueCount();
              // WARNING: consume and move next or we'll be stuck in an infinite
              // loop here.
              values_in_interval.nextDoubleValue(); 
            }
            value = count;
          } else {
            value = specification.getFunction().runDouble(values_in_interval);
          }
          values_in_interval.moveToNextInterval();
        } else {
          // Our expected timestamp precedes the actual, so the interval is
          // missing. We will use a special value, based on the fill policy, to
          // represent this case.
          switch (specification.getFillPolicy()) {
          case NOT_A_NUMBER:
          case NULL:
            value = Double.NaN;
            break;
  
          case ZERO:
            value = 0.0;
            break;
            
            // TODO - scalar
  
          default:
            throw new RuntimeException("unhandled fill policy");
          }
        }
  
        // Advance the expected timestamp to the next interval.
        if (!run_all) {
          if (specification.useCalendar()) {
            if (unit == WEEK_UNIT) {
              previous_calendar.add(DAY_UNIT, interval * WEEK_LENGTH);
              next_calendar.add(DAY_UNIT, interval * WEEK_LENGTH);
            } else {
              previous_calendar.add(unit, interval);
              next_calendar.add(unit, interval);
            }
            timestamp = next_calendar.getTimeInMillis();
          } else {
            timestamp += specification.getInterval();
          }
        }
  
        // This object also represents the data.
        return this;
      }
  
      // Ideally, the user will not call this method when no data remains, but
      // we can't enforce that.
      throw new NoSuchElementException("no more data points in " + this);
    }