/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.statistics;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.text.DateFormat;
import java.text.NumberFormat;
import java.text.SimpleDateFormat;
import java.util.AbstractList;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SimpleTimeZone;
import java.util.TimeZone;
import java.util.zip.GZIPInputStream;

import org.apache.geode.GemFireIOException;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.ExitCode;
import org.apache.geode.internal.logging.DateFormatter;

/**
 * StatArchiveReader provides APIs to read statistic snapshots from an archive file.
 */
public class StatArchiveReader implements StatArchiveFormat, AutoCloseable {

  private final StatArchiveFile[] archives;
  private final boolean dump;
  private boolean closed = false;

  /**
   * Creates a StatArchiveReader that will read the named archive file.
   *
   * @param autoClose if its <code>true</code> then the reader will close input files as soon as it
   *        finds their end.
   * @throws IOException if <code>archiveName</code> could not be opened read, or closed.
   */
  public StatArchiveReader(File[] archiveNames, ValueFilter[] filters, boolean autoClose)
      throws IOException {
    archives = new StatArchiveFile[archiveNames.length];
    dump = Boolean.getBoolean("StatArchiveReader.dumpall");
    for (int i = 0; i < archiveNames.length; i++) {
      archives[i] = new StatArchiveFile(this, archiveNames[i], dump, filters);
    }

    update(false, autoClose);

    if (dump || Boolean.getBoolean("StatArchiveReader.dump")) {
      dump(new PrintWriter(System.out));
    }
  }

  /**
   * Creates a StatArchiveReader that will read the named archive file.
   *
   * @throws IOException if <code>archiveName</code> could not be opened read, or closed.
   */
  public StatArchiveReader(String archiveName) throws IOException {
    this(new File[] {new File(archiveName)}, null, false);
  }

  /**
   * Returns an array of stat values that match the specified spec. If nothing matches then an empty
   * array is returned.
   */
  public StatValue[] matchSpec(StatSpec spec) {
    if (spec.getCombineType() == StatSpec.GLOBAL) {
      StatValue[] allValues = matchSpec(new RawStatSpec(spec));
      if (allValues.length == 0) {
        return allValues;
      } else {
        ComboValue cv = new ComboValue(allValues);
        // need to save this in reader's combo value list
        return new StatValue[] {cv};
      }
    } else {
      List l = new ArrayList();
      StatArchiveReader.StatArchiveFile[] archives = getArchives();
      for (StatArchiveFile f : archives) {
        if (spec.archiveMatches(f.getFile())) {
          f.matchSpec(spec, l);
        }
      }
      StatValue[] result = new StatValue[l.size()];
      return (StatValue[]) l.toArray(result);
    }
  }

  /**
   * Checks to see if any archives have changed since the StatArchiverReader instance was created or
   * last updated. If an archive has additional samples then those are read the resource instances
   * maintained by the reader are updated.
   * <p>
   * Once closed a reader can no longer be updated.
   *
   * @return true if update read some new data.
   * @throws IOException if an archive could not be opened read, or closed.
   */
  public boolean update() throws IOException {
    return update(true, false);
  }

  private boolean update(boolean doReset, boolean autoClose) throws IOException {
    if (closed) {
      return false;
    }
    boolean result = false;
    StatArchiveReader.StatArchiveFile[] archives = getArchives();
    for (StatArchiveFile f : archives) {
      if (f.update(doReset)) {
        result = true;
      }
      if (autoClose) {
        f.close();
      }
    }
    return result;
  }

  /**
   * Returns an unmodifiable list of all the {@link ResourceInst} this reader contains.
   */
  public List getResourceInstList() {
    return new ResourceInstList();
  }

  public StatArchiveFile[] getArchives() {
    return archives;
  }

  /**
   * Closes all archives.
   */
  @Override
  public void close() throws IOException {
    if (!closed) {
      StatArchiveReader.StatArchiveFile[] archives = getArchives();
      for (StatArchiveFile f : archives) {
        f.close();
      }
      closed = true;
    }
  }

  private int getMemoryUsed() {
    int result = 0;
    StatArchiveReader.StatArchiveFile[] archives = getArchives();
    for (StatArchiveFile f : archives) {
      result += f.getMemoryUsed();
    }
    return result;
  }

  private void dump(PrintWriter stream) {
    StatArchiveReader.StatArchiveFile[] archives = getArchives();
    for (StatArchiveFile f : archives) {
      f.dump(stream);
    }
  }

  protected static double bitsToDouble(int type, long bits) {
    switch (type) {
      case BOOLEAN_CODE:
      case BYTE_CODE:
      case CHAR_CODE:
      case WCHAR_CODE:
      case SHORT_CODE:
      case INT_CODE:
      case LONG_CODE:
        return bits;
      case FLOAT_CODE:
        return Float.intBitsToFloat((int) bits);
      case DOUBLE_CODE:
        return Double.longBitsToDouble(bits);
      default:
        throw new InternalGemFireException(String.format("Unexpected typecode %s",
            type));
    }
  }

  private static class SingleStatRawStatSpec implements StatSpec {

    private final String archive;
    private final String statType;
    private final String statName;

    SingleStatRawStatSpec(String archive, String typeAndStat) {
      this.archive = archive;
      String[] parts = typeAndStat.split("\\.", 0);
      statType = parts[0];
      statName = parts[1];
    }

    @Override
    public boolean archiveMatches(File archive) {
      return true; // this.archive.equalsIgnoreCase(archive.getName());
    }

    @Override
    public boolean typeMatches(String typeName) {
      return statType.equalsIgnoreCase(typeName);
    }

    @Override
    public boolean statMatches(String statName) {
      return this.statName.equalsIgnoreCase(statName);
    }

    @Override
    public boolean instanceMatches(String textId, long numericId) {
      return true;
    }

    @Override
    public int getCombineType() {
      return StatSpec.NONE;
    }
  }

  private static void printStatValue(StatArchiveReader.StatValue v, long startTime, long endTime,
      boolean nofilter, boolean persec, boolean persample, boolean prunezeros, boolean details) {
    v = v.createTrimmed(startTime, endTime);
    if (nofilter) {
      v.setFilter(StatArchiveReader.StatValue.FILTER_NONE);
    } else if (persec) {
      v.setFilter(StatArchiveReader.StatValue.FILTER_PERSEC);
    } else if (persample) {
      v.setFilter(StatArchiveReader.StatValue.FILTER_PERSAMPLE);
    }
    if (prunezeros) {
      if (v.getSnapshotsMinimum() == 0.0 && v.getSnapshotsMaximum() == 0.0) {
        return;
      }
    }
    System.out.println("  " + v.toString());
    if (details) {
      System.out.print("  values=");
      double[] snapshots = v.getSnapshots();
      for (final double snapshot : snapshots) {
        System.out.print(' ');
        System.out.print(snapshot);
      }
      System.out.println();
      String desc = v.getDescriptor().getDescription();
      if (desc != null && desc.length() > 0) {
        System.out.println("    " + desc);
      }
    }
  }


  /**
   * Simple utility to read and dump statistic archive.
   */
  public static void main(String[] args) throws IOException {
    String archiveName = null;
    final StatArchiveReader reader;
    if (args.length > 1) {
      if (!args[0].equals("stat") || args.length > 3) {
        System.err.println("Usage: stat archiveName statType.statName");
        ExitCode.FATAL.doSystemExit();
      }
      archiveName = args[1];
      String statSpec = args[2];
      if (!statSpec.contains(".")) {
        throw new IllegalArgumentException(
            "stat spec '" + statSpec + "' is malformed - use StatType.statName");
      }
      File archiveFile = new File(archiveName);
      if (!archiveFile.exists()) {
        throw new IllegalArgumentException("archive file does not exist: " + archiveName);
      }
      if (!archiveFile.canRead()) {
        throw new IllegalArgumentException("archive file exists but is unreadable: " + archiveName);
      }
      File[] archives = new File[] {archiveFile};
      SingleStatRawStatSpec[] filters =
          new SingleStatRawStatSpec[] {new SingleStatRawStatSpec(archiveName, args[2])};
      reader = new StatArchiveReader(archives, filters, false);
      final StatValue[] statValues = reader.matchSpec(filters[0]);
      System.out.println(statSpec + " matched " + statValues.length + " stats...");
      for (StatValue value : statValues) {
        printStatValue(value, -1, -1, true, false, false, false, true);
      }
      System.out.println();
      System.out.flush();
    } else {
      if (args.length == 1) {
        archiveName = args[0];
      } else {
        archiveName = "statArchive.gfs";
      }
      reader = new StatArchiveReader(archiveName);
      System.out.println("DEBUG: memory used = " + reader.getMemoryUsed());
    }
    reader.close();
  }

  /**
   * Wraps an instance of StatSpec but alwasy returns a combine type of NONE.
   */
  private static class RawStatSpec implements StatSpec {
    private final StatSpec spec;

    RawStatSpec(StatSpec wrappedSpec) {
      spec = wrappedSpec;
    }

    @Override
    public int getCombineType() {
      return StatSpec.NONE;
    }

    @Override
    public boolean typeMatches(String typeName) {
      return spec.typeMatches(typeName);
    }

    @Override
    public boolean statMatches(String statName) {
      return spec.statMatches(statName);
    }

    @Override
    public boolean instanceMatches(String textId, long numericId) {
      return spec.instanceMatches(textId, numericId);
    }

    @Override
    public boolean archiveMatches(File archive) {
      return spec.archiveMatches(archive);
    }
  }

  private class ResourceInstList extends AbstractList {
    protected ResourceInstList() {
      // nothing needed.
    }

    @Override
    public Object get(int idx) {
      int archiveIdx = 0;
      StatArchiveReader.StatArchiveFile[] archives = getArchives();
      for (StatArchiveFile f : archives) {
        if (idx < (archiveIdx + f.resourceInstSize)) {
          return f.resourceInstTable[idx - archiveIdx];
        }
        archiveIdx += f.resourceInstSize;
      }
      return null;
    }

    @Override
    public int size() {
      int result = 0;
      StatArchiveReader.StatArchiveFile[] archives = getArchives();
      for (final StatArchiveFile archive : archives) {
        result += archive.resourceInstSize;
      }
      return result;
    }
  }

  /**
   * Describes a single statistic.
   */
  public static class StatDescriptor {
    private boolean loaded;
    private String name;
    private final int offset;
    private final boolean isCounter;
    private final boolean largerBetter;
    private final byte typeCode;
    private String units;
    private String desc;

    protected void dump(PrintWriter stream) {
      stream.println(
          "  " + name + ": type=" + typeCode + " offset=" + offset + (isCounter ? " counter" : "")
              + " units=" + units + " largerBetter=" + largerBetter + " desc=" + desc);
    }

    protected StatDescriptor(String name, int offset, boolean isCounter, boolean largerBetter,
        byte typeCode, String units, String desc) {
      loaded = true;
      this.name = name;
      this.offset = offset;
      this.isCounter = isCounter;
      this.largerBetter = largerBetter;
      this.typeCode = typeCode;
      this.units = units;
      this.desc = desc;
    }

    public boolean isLoaded() {
      return loaded;
    }

    void unload() {
      loaded = false;
      name = null;
      units = null;
      desc = null;
    }

    /**
     * Returns the type code of this statistic. It will be one of the following values:
     * <ul>
     * <li>{@link #BOOLEAN_CODE}
     * <li>{@link #WCHAR_CODE}
     * <li>{@link #CHAR_CODE}
     * <li>{@link #BYTE_CODE}
     * <li>{@link #SHORT_CODE}
     * <li>{@link #INT_CODE}
     * <li>{@link #LONG_CODE}
     * <li>{@link #FLOAT_CODE}
     * <li>{@link #DOUBLE_CODE}
     * </ul>
     */
    public byte getTypeCode() {
      return typeCode;
    }

    /**
     * Returns the name of this statistic.
     */
    public String getName() {
      return name;
    }

    /**
     * Returns true if this statistic's value will always increase.
     */
    public boolean isCounter() {
      return isCounter;
    }

    /**
     * Returns true if larger values indicate better performance.
     */
    public boolean isLargerBetter() {
      return largerBetter;
    }

    /**
     * Returns a string that describes the units this statistic measures.
     */
    public String getUnits() {
      return units;
    }

    /**
     * Returns a textual description of this statistic.
     */
    public String getDescription() {
      return desc;
    }

    /**
     * Returns the offset of this stat in its type.
     */
    public int getOffset() {
      return offset;
    }
  }

  public interface StatValue {
    /**
     * {@link StatArchiveReader.StatValue} filter that causes the statistic values to be unfiltered.
     * This causes the raw values written to the archive to be used.
     * <p>
     * This is the default filter for non-counter statistics. To determine if a statistic is not a
     * counter use {@link StatArchiveReader.StatDescriptor#isCounter}.
     */
    int FILTER_NONE = 0;
    /**
     * {@link StatArchiveReader.StatValue} filter that causes the statistic values to be filtered to
     * reflect how often they change per second. Since the difference between two samples is used to
     * calculate the value this causes the {@link StatArchiveReader.StatValue} to have one less
     * sample than {@link #FILTER_NONE}. The instance time stamp that does not have a per second
     * value is the instance's first time stamp
     * {@link StatArchiveReader.ResourceInst#getFirstTimeMillis}.
     * <p>
     * This is the default filter for counter statistics. To determine if a statistic is a counter
     * use {@link StatArchiveReader.StatDescriptor#isCounter}.
     */
    int FILTER_PERSEC = 1;
    /**
     * {@link StatArchiveReader.StatValue} filter that causes the statistic values to be filtered to
     * reflect how much they changed between sample periods. Since the difference between two
     * samples is used to calculate the value this causes the {@link StatArchiveReader.StatValue} to
     * have one less sample than {@link #FILTER_NONE}. The instance time stamp that does not have a
     * per second value is the instance's first time stamp
     * {@link StatArchiveReader.ResourceInst#getFirstTimeMillis}.
     */
    int FILTER_PERSAMPLE = 2;

    /**
     * Creates and returns a trimmed version of this stat value. Any samples taken before
     * <code>startTime</code> and after <code>endTime</code> are discarded from the resulting value.
     * Set a time parameter to <code>-1</code> to not trim that side.
     */
    StatValue createTrimmed(long startTime, long endTime);

    /**
     * Returns true if value has data that has been trimmed off it by a start timestamp.
     */
    boolean isTrimmedLeft();

    /**
     * Gets the {@link StatArchiveReader.ResourceType type} of the resources that this value belongs
     * to.
     */
    ResourceType getType();

    /**
     * Gets the {@link StatArchiveReader.ResourceInst resources} that this value belongs to.
     */
    ResourceInst[] getResources();

    /**
     * Returns an array of timestamps for each unfiltered snapshot in this value. Each returned time
     * stamp is the number of millis since midnight, Jan 1, 1970 UTC.
     */
    long[] getRawAbsoluteTimeStamps();

    /**
     * Returns an array of timestamps for each unfiltered snapshot in this value. Each returned time
     * stamp is the number of millis since midnight, Jan 1, 1970 UTC. The resolution is seconds.
     */
    long[] getRawAbsoluteTimeStampsWithSecondRes();

    /**
     * Returns an array of doubles containing the unfiltered value of this statistic for each point
     * in time that it was sampled.
     */
    double[] getRawSnapshots();

    /**
     * Returns an array of doubles containing the filtered value of this statistic for each point in
     * time that it was sampled.
     */
    double[] getSnapshots();

    /**
     * Returns the number of samples taken of this statistic's value.
     */
    int getSnapshotsSize();

    /**
     * Returns the smallest of all the samples taken of this statistic's value.
     */
    double getSnapshotsMinimum();

    /**
     * Returns the largest of all the samples taken of this statistic's value.
     */
    double getSnapshotsMaximum();

    /**
     * Returns the average of all the samples taken of this statistic's value.
     */
    double getSnapshotsAverage();

    /**
     * Returns the standard deviation of all the samples taken of this statistic's value.
     */
    double getSnapshotsStandardDeviation();

    /**
     * Returns the most recent value of all the samples taken of this statistic's value.
     */
    double getSnapshotsMostRecent();

    /**
     * Returns true if sample whose value was different from previous values has been added to this
     * StatValue since the last time this method was called.
     */
    boolean hasValueChanged();

    /**
     * Returns the current filter used to calculate this statistic's values. It will be one of these
     * values:
     * <ul>
     * <li>{@link #FILTER_NONE}
     * <li>{@link #FILTER_PERSAMPLE}
     * <li>{@link #FILTER_PERSEC}
     * </ul>
     */
    int getFilter();

    /**
     * Sets the current filter used to calculate this statistic's values. The default filter is
     * {@link #FILTER_NONE} unless the statistic is a counter,
     * {@link StatArchiveReader.StatDescriptor#isCounter}, in which case its {@link #FILTER_PERSEC}.
     *
     * @param filter It must be one of these values:
     *        <ul>
     *        <li>{@link #FILTER_NONE}
     *        <li>{@link #FILTER_PERSAMPLE}
     *        <li>{@link #FILTER_PERSEC}
     *        </ul>
     * @throws IllegalArgumentException if <code>filter</code> is not a valid filter constant.
     */
    void setFilter(int filter);

    /**
     * Returns a description of this statistic.
     */
    StatDescriptor getDescriptor();
  }

  protected abstract static class AbstractValue implements StatValue {
    protected StatDescriptor descriptor;
    protected int filter;

    protected long startTime = -1;
    protected long endTime = -1;

    protected boolean statsValid = false;
    protected int size;
    protected double min;
    protected double max;
    protected double avg;
    protected double stddev;
    protected double mostRecent;

    public void calcStats() {
      if (!statsValid) {
        getSnapshots();
      }
    }

    @Override
    public int getSnapshotsSize() {
      calcStats();
      return size;
    }

    @Override
    public double getSnapshotsMinimum() {
      calcStats();
      return min;
    }

    @Override
    public double getSnapshotsMaximum() {
      calcStats();
      return max;
    }

    @Override
    public double getSnapshotsAverage() {
      calcStats();
      return avg;
    }

    @Override
    public double getSnapshotsStandardDeviation() {
      calcStats();
      return stddev;
    }

    @Override
    public double getSnapshotsMostRecent() {
      calcStats();
      return mostRecent;
    }

    @Override
    public StatDescriptor getDescriptor() {
      return descriptor;
    }

    @Override
    public int getFilter() {
      return filter;
    }

    @Override
    public void setFilter(int filter) {
      if (filter != this.filter) {
        if (filter != FILTER_NONE && filter != FILTER_PERSEC && filter != FILTER_PERSAMPLE) {
          throw new IllegalArgumentException(
              String.format("Filter value %s must be %s, %s, or %s.",
                  filter, FILTER_NONE,
                  FILTER_PERSEC, FILTER_PERSAMPLE));
        }
        this.filter = filter;
        statsValid = false;
      }
    }

    /**
     * Calculates each stat given the result of calling getSnapshots
     */
    protected void calcStats(double[] values) {
      if (statsValid) {
        return;
      }
      size = values.length;
      if (size == 0) {
        min = 0.0;
        max = 0.0;
        avg = 0.0;
        stddev = 0.0;
        mostRecent = 0.0;
      } else {
        min = values[0];
        max = values[0];
        mostRecent = values[values.length - 1];
        double total = values[0];
        for (int i = 1; i < size; i++) {
          total += values[i];
          if (values[i] < min) {
            min = values[i];
          } else if (values[i] > max) {
            max = values[i];
          }
        }
        avg = total / size;
        stddev = 0.0;
        if (size > 1) {
          for (int i = 0; i < size; i++) {
            double dv = values[i] - avg;
            stddev += (dv * dv);
          }
          stddev /= (size - 1);
          stddev = Math.sqrt(stddev);
        }
      }
      statsValid = true;
    }

    /**
     * Returns a string representation of this object.
     */
    @Override
    public String toString() {
      calcStats();
      StringBuilder result = new StringBuilder();
      result.append(getDescriptor().getName());
      String units = getDescriptor().getUnits();
      if (units != null && units.length() > 0) {
        result.append(' ').append(units);
      }
      if (filter == FILTER_PERSEC) {
        result.append("/sec");
      } else if (filter == FILTER_PERSAMPLE) {
        result.append("/sample");
      }
      result.append(": samples=").append(getSnapshotsSize());
      if (startTime != -1) {
        result.append(" startTime=\"").append(new Date(startTime)).append("\"");
      }
      if (endTime != -1) {
        result.append(" endTime=\"").append(new Date(endTime)).append("\"");
      }

      NumberFormat nf = getNumberFormat();
      result.append(" min=").append(nf.format(min));
      result.append(" max=").append(nf.format(max));
      result.append(" average=").append(nf.format(avg));
      result.append(" stddev=").append(nf.format(stddev));
      result.append(" last=") // for bug 42532
          .append(nf.format(mostRecent));
      return result.toString();
    }
  }

  /**
   * A ComboValue is a value that is the logical combination of a set of other stat values.
   * <p>
   * For now ComboValue has a simple implementation that does not suppport updates.
   */
  private static class ComboValue extends AbstractValue {
    private final ResourceType type;
    private final StatValue[] values;

    /**
     * Creates a ComboValue by adding all the specified values together.
     */
    ComboValue(List valueList) {
      this((StatValue[]) valueList.toArray(new StatValue[0]));
    }

    /**
     * Creates a ComboValue by adding all the specified values together.
     */
    ComboValue(StatValue[] values) {
      this.values = values;
      filter = this.values[0].getFilter();
      String typeName = this.values[0].getType().getName();
      String statName = this.values[0].getDescriptor().getName();
      int bestTypeIdx = 0;
      for (int i = 1; i < this.values.length; i++) {
        if (filter != this.values[i].getFilter()) {
          /*
           * I'm not sure why this would happen. If it really can happen then this code should
           * change the filter since a client has no way to select values based on the filter.
           */
          throw new IllegalArgumentException(
              "Cannot combine values with different filters.");
        }
        if (!typeName.equals(this.values[i].getType().getName())) {
          throw new IllegalArgumentException(
              "Cannot combine values with different types.");
        }
        if (!statName.equals(this.values[i].getDescriptor().getName())) {
          throw new IllegalArgumentException(
              "Cannot combine different stats.");
        }
        if (this.values[i].getDescriptor().isCounter()) {
          // it is a counter which is not the default
          if (!this.values[i].getDescriptor().isLargerBetter()) {
            // this value has non-defaults for both, use it
            bestTypeIdx = i;
          } else if (this.values[bestTypeIdx].getDescriptor()
              .isCounter() == this.values[bestTypeIdx].getDescriptor().isLargerBetter()) {
            // as long as we haven't already found a value with defaults
            // make this value the best type
            bestTypeIdx = i;
          }
        } else {
          // its a gauge, see if it has a non-default largerBetter
          if (this.values[i].getDescriptor().isLargerBetter()) {
            // as long as we haven't already found a value with defaults
            if (this.values[bestTypeIdx].getDescriptor().isCounter() == this.values[bestTypeIdx]
                .getDescriptor().isLargerBetter()) {
              // make this value the best type
              bestTypeIdx = i;
            }
          }
        }
      }
      type = this.values[bestTypeIdx].getType();
      descriptor = this.values[bestTypeIdx].getDescriptor();
    }

    private ComboValue(ComboValue original, long startTime, long endTime) {
      this.startTime = startTime;
      this.endTime = endTime;
      type = original.getType();
      descriptor = original.getDescriptor();
      filter = original.getFilter();
      values = new StatValue[original.values.length];
      for (int i = 0; i < values.length; i++) {
        values[i] = original.values[i].createTrimmed(startTime, endTime);
      }
    }

    @Override
    public StatValue createTrimmed(long startTime, long endTime) {
      if (startTime == this.startTime && endTime == this.endTime) {
        return this;
      } else {
        return new ComboValue(this, startTime, endTime);
      }
    }

    @Override
    public ResourceType getType() {
      return type;
    }

    @Override
    public ResourceInst[] getResources() {
      Set set = new HashSet();
      for (final StatValue value : values) {
        set.addAll(Arrays.asList(value.getResources()));
      }
      ResourceInst[] result = new ResourceInst[set.size()];
      return (ResourceInst[]) set.toArray(result);
    }

    @Override
    public boolean hasValueChanged() {
      return true;
    }

    public static boolean closeEnough(long v1, long v2, long delta) {
      return (v1 == v2) || ((Math.abs(v1 - v2) / 2) <= delta);
    }

    /**
     * Return true if v is closer to prev. Return false if v is closer to next. Return true if v is
     * the same distance from both.
     */
    public static boolean closer(long v, long prev, long next) {
      return Math.abs(v - prev) <= Math.abs(v - next);
    }


    /**
     * Return true if the current ts must be inserted instead of being mapped to the tsAtInsertPoint
     */
    private static boolean mustInsert(int nextIdx, long[] valueTimeStamps, long tsAtInsertPoint) {
      return (nextIdx < valueTimeStamps.length) && (valueTimeStamps[nextIdx] <= tsAtInsertPoint);
    }

    @Override
    public long[] getRawAbsoluteTimeStampsWithSecondRes() {
      return getRawAbsoluteTimeStamps();
    }

    @Override
    public long[] getRawAbsoluteTimeStamps() {
      if (values.length == 0) {
        return new long[0];
      }
      long[] valueTimeStamps = values[0].getRawAbsoluteTimeStamps();
      int tsCount = valueTimeStamps.length + 1;
      long[] ourTimeStamps = new long[(tsCount * 2) + 1];
      System.arraycopy(valueTimeStamps, 0, ourTimeStamps, 0, valueTimeStamps.length);
      // Note we add a MAX sample to make the insert logic simple
      ourTimeStamps[valueTimeStamps.length] = Long.MAX_VALUE;
      for (int i = 1; i < values.length; i++) {
        valueTimeStamps = values[i].getRawAbsoluteTimeStamps();
        if (valueTimeStamps.length == 0) {
          continue;
        }
        int ourIdx = 0;
        int j = 0;
        long tsToInsert = valueTimeStamps[0] - 1000; // default to 1 second
        if (valueTimeStamps.length > 1) {
          tsToInsert = valueTimeStamps[0] - (valueTimeStamps[1] - valueTimeStamps[0]);
        }
        // tsToInsert is now initialized to a value that we can pretend
        // was the previous timestamp inserted.
        while (j < valueTimeStamps.length) {
          long timeDelta = (valueTimeStamps[j] - tsToInsert) / 2;
          tsToInsert = valueTimeStamps[j];
          long tsAtInsertPoint = ourTimeStamps[ourIdx];
          while (tsToInsert > tsAtInsertPoint
              && !closeEnough(tsToInsert, tsAtInsertPoint, timeDelta)) {
            // System.out.println("DEBUG: skipping " + ourIdx + " because it was not closeEnough");
            ourIdx++;
            tsAtInsertPoint = ourTimeStamps[ourIdx];
          }
          if (closeEnough(tsToInsert, tsAtInsertPoint, timeDelta)
              && !mustInsert(j + 1, valueTimeStamps, tsAtInsertPoint)) {
            // It was already in our list so just go to the next one
            j++;
            ourIdx++; // never put the next timestamp at this index
            while (!closer(tsToInsert, ourTimeStamps[ourIdx - 1], ourTimeStamps[ourIdx])
                && !mustInsert(j, valueTimeStamps, ourTimeStamps[ourIdx])) {
              ourIdx++; // it is closer to the next one so skip forward on more
            }
          } else {
            // its not in our list so add it
            int endRunIdx = j + 1;
            while (endRunIdx < valueTimeStamps.length
                && valueTimeStamps[endRunIdx] < tsAtInsertPoint
                && !closeEnough(valueTimeStamps[endRunIdx], tsAtInsertPoint, timeDelta)) {
              endRunIdx++;
            }
            int numToCopy = endRunIdx - j;
            if (tsCount + numToCopy > ourTimeStamps.length) {
              // grow our timestamp array
              long[] tmp = new long[(tsCount + numToCopy) * 2];
              System.arraycopy(ourTimeStamps, 0, tmp, 0, tsCount);
              ourTimeStamps = tmp;
            }
            // make room for insert
            System.arraycopy(ourTimeStamps, ourIdx, ourTimeStamps, ourIdx + numToCopy,
                tsCount - ourIdx);
            // insert the elements
            if (numToCopy == 1) {
              ourTimeStamps[ourIdx] = valueTimeStamps[j];
            } else {
              System.arraycopy(valueTimeStamps, j, ourTimeStamps, ourIdx, numToCopy);
            }
            ourIdx += numToCopy;
            tsCount += numToCopy;
            // skip over all inserted elements
            j += numToCopy;
          }
        }
      }
      tsCount--;
      {
        int startIdx = 0;
        int endIdx = tsCount - 1;
        if (startTime != -1) {
          Assert.assertTrue(ourTimeStamps[startIdx] >= startTime);
        }
        if (endTime != -1) {
          Assert.assertTrue(endIdx == startIdx - 1 || ourTimeStamps[endIdx] < endTime);
        }
        tsCount = (endIdx - startIdx) + 1;

        // shrink and trim our timestamp array
        long[] tmp = new long[tsCount];
        System.arraycopy(ourTimeStamps, startIdx, tmp, 0, tsCount);
        ourTimeStamps = tmp;
      }
      return ourTimeStamps;
    }

    @Override
    public double[] getRawSnapshots() {
