/*
 * 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.druid.query.aggregation.histogram;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonValue;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.druid.common.config.NullHandling;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;

import javax.annotation.Nullable;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Objects;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class FixedBucketsHistogram
{
  public static final byte SERIALIZATION_VERSION = 0x01;

  /**
   * Serialization header format:
   *
   * byte: serialization version, must be 0x01
   * byte: encoding mode, 0x01 for full, 0x02 for sparse
   */
  public static final int SERDE_HEADER_SIZE = Byte.BYTES +
                                              Byte.BYTES;

  /**
   * Common serialization fields format:
   *
   * double: lowerLimit
   * double: upperLimit
   * int: numBuckets
   * byte: outlier handling mode (0x00 for `ignore`, 0x01 for `overflow`, and 0x02 for `clip`)
   * long: count, total number of values contained in the histogram, excluding outliers
   * long: lowerOutlierCount
   * long: upperOutlierCount
   * long: missingValueCount
   * double: max
   * double: min
   */
  public static final int COMMON_FIELDS_SIZE = Double.BYTES +
                                                Double.BYTES +
                                                Integer.BYTES +
                                                Byte.BYTES +
                                                Long.BYTES * 4 +
                                                Double.BYTES +
                                                Double.BYTES;

  /**
   * Full serialization format:
   *
   * serialization header
   * common fields
   * array of longs: bucket counts for the histogram
   */
  public static final byte FULL_ENCODING_MODE = 0x01;

  /**
   * Sparse serialization format:
   *
   * serialization header
   * common fields
   * int: number of following (bucketNum, count) pairs
   * sequence of (int, long) pairs:
   *  int: bucket number
   *  count: bucket count
   */
  public static final byte SPARSE_ENCODING_MODE = 0x02;

  /**
   * Determines how the histogram handles outliers.
   *
   * Ignore:   do not track outliers at all
   * Overflow: track outlier counts in upperOutlierCount and lowerOutlierCount.
   *           The min and max do not take outlier values into account.
   * Clip:     Clip outlier values to either the start or end of the histogram's range, adding them to the first or
   *           last bucket, respectively. The min and max are affected by such clipped outlier values.
   */
  public enum OutlierHandlingMode
  {
    IGNORE,
    OVERFLOW,
    CLIP;

    @JsonValue
    @Override
    public String toString()
    {
      return StringUtils.toLowerCase(this.name());
    }

    @JsonCreator
    public static OutlierHandlingMode fromString(String name)
    {
      return valueOf(StringUtils.toUpperCase(name));
    }

    public byte[] getCacheKey()
    {
      return new byte[] {(byte) this.ordinal()};
    }
  }

  @JsonIgnore
  private final OutlierHandler outlierHandler;

  /**
   * Locking is needed when the non-buffer aggregator is used within a realtime ingestion task.
   *
   * The following areas are locked:
   * - Add value
   * - Merge histograms
   * - Compute percentiles
   * - Serialization
   */
  @JsonIgnore
  private final ReadWriteLock readWriteLock;

  private double lowerLimit;
  private double upperLimit;
  private int numBuckets;

  private long upperOutlierCount = 0;
  private long lowerOutlierCount = 0;
  private long missingValueCount = 0;
  private long[] histogram;
  private double bucketSize;
  private OutlierHandlingMode outlierHandlingMode;

  private long count = 0;
  private double max = Double.NEGATIVE_INFINITY;
  private double min = Double.POSITIVE_INFINITY;

  public FixedBucketsHistogram(
      double lowerLimit,
      double upperLimit,
      int numBuckets,
      OutlierHandlingMode outlierHandlingMode
  )
  {
    Preconditions.checkArgument(
        upperLimit > lowerLimit,
        "Upper limit [%s] must be greater than lower limit [%s].",
        upperLimit,
        lowerLimit
    );
    Preconditions.checkArgument(
        numBuckets > 0,
        "numBuckets must be > 0, got [%s] instead.",
        numBuckets
    );
    this.lowerLimit = lowerLimit;
    this.upperLimit = upperLimit;
    this.numBuckets = numBuckets;
    this.outlierHandlingMode = outlierHandlingMode;

    this.histogram = new long[numBuckets];
    this.bucketSize = (upperLimit - lowerLimit) / numBuckets;
    this.readWriteLock = new ReentrantReadWriteLock(true);

    this.outlierHandler = makeOutlierHandler();
  }

  @VisibleForTesting
  protected FixedBucketsHistogram(
      double lowerLimit,
      double upperLimit,
      int numBuckets,
      OutlierHandlingMode outlierHandlingMode,
      long[] histogram,
      long count,
      double max,
      double min,
      long lowerOutlierCount,
      long upperOutlierCount,
      long missingValueCount
  )
  {
    this.lowerLimit = lowerLimit;
    this.upperLimit = upperLimit;
    this.numBuckets = numBuckets;
    this.outlierHandlingMode = outlierHandlingMode;
    this.histogram = histogram;
    this.count = count;
    this.max = max;
    this.min = min;
    this.upperOutlierCount = upperOutlierCount;
    this.lowerOutlierCount = lowerOutlierCount;
    this.missingValueCount = missingValueCount;

    this.bucketSize = (upperLimit - lowerLimit) / numBuckets;
    this.readWriteLock = new ReentrantReadWriteLock(true);
    this.outlierHandler = makeOutlierHandler();
  }

  @VisibleForTesting
  public FixedBucketsHistogram getCopy()
  {
    // only used for tests/benchmarks
    return new FixedBucketsHistogram(
        lowerLimit,
        upperLimit,
        numBuckets,
        outlierHandlingMode,
        Arrays.copyOf(histogram, histogram.length),
        count,
        max,
        min,
        lowerOutlierCount,
        upperOutlierCount,
        missingValueCount
    );
  }

  @JsonProperty
  public double getLowerLimit()
  {
    return lowerLimit;
  }

  @JsonProperty
  public double getUpperLimit()
  {
    return upperLimit;
  }

  @JsonProperty
  public int getNumBuckets()
  {
    return numBuckets;
  }

  @JsonProperty
  public long getUpperOutlierCount()
  {
    return upperOutlierCount;
  }

  @JsonProperty
  public long getLowerOutlierCount()
  {
    return lowerOutlierCount;
  }

  @JsonProperty
  public long getMissingValueCount()
  {
    return missingValueCount;
  }

  @JsonProperty
  public long[] getHistogram()
  {
    return histogram;
  }

  @JsonProperty
  public double getBucketSize()
  {
    return bucketSize;
  }

  @JsonProperty
  public long getCount()
  {
    return count;
  }

  @JsonProperty
  public double getMax()
  {
    return max;
  }

  @JsonProperty
  public double getMin()
  {
    return min;
  }

  @JsonProperty
  public OutlierHandlingMode getOutlierHandlingMode()
  {
    return outlierHandlingMode;
  }

  @Override
  public String toString()
  {
    return "{" +
           "lowerLimit=" + lowerLimit +
           ", upperLimit=" + upperLimit +
           ", numBuckets=" + numBuckets +
           ", upperOutlierCount=" + upperOutlierCount +
           ", lowerOutlierCount=" + lowerOutlierCount +
           ", missingValueCount=" + missingValueCount +
           ", histogram=" + Arrays.toString(histogram) +
           ", outlierHandlingMode=" + outlierHandlingMode +
           ", count=" + count +
           ", max=" + max +
           ", min=" + min +
           '}';
  }

  @Override
  public boolean equals(Object o)
  {
    if (this == o) {
      return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }
    FixedBucketsHistogram that = (FixedBucketsHistogram) o;
    return Double.compare(that.getLowerLimit(), getLowerLimit()) == 0 &&
           Double.compare(that.getUpperLimit(), getUpperLimit()) == 0 &&
           getNumBuckets() == that.getNumBuckets() &&
           getUpperOutlierCount() == that.getUpperOutlierCount() &&
           getLowerOutlierCount() == that.getLowerOutlierCount() &&
           getMissingValueCount() == that.getMissingValueCount() &&
           Double.compare(that.getBucketSize(), getBucketSize()) == 0 &&
           getCount() == that.getCount() &&
           Double.compare(that.getMax(), getMax()) == 0 &&
           Double.compare(that.getMin(), getMin()) == 0 &&
           Arrays.equals(getHistogram(), that.getHistogram()) &&
           getOutlierHandlingMode() == that.getOutlierHandlingMode();
  }

  @Override
  public int hashCode()
  {
    return Objects.hash(
        getLowerLimit(),
        getUpperLimit(),
        getNumBuckets(),
        getUpperOutlierCount(),
        getLowerOutlierCount(),
        getMissingValueCount(),
        Arrays.hashCode(getHistogram()),
        getBucketSize(),
        getOutlierHandlingMode(),
        getCount(),
        getMax(),
        getMin()
    );
  }

  public ReadWriteLock getReadWriteLock()
  {
    return readWriteLock;
  }

  /**
   * Add a value to the histogram, using the outlierHandler to account for outliers.
   *
   * @param value value to be added.
   */
  public void add(
      double value
  )
  {
    readWriteLock.writeLock().lock();

    try {
      if (value < lowerLimit) {
        outlierHandler.handleOutlierAdd(false);
        return;
      } else if (value >= upperLimit) {
        outlierHandler.handleOutlierAdd(true);
        return;
      }

      count += 1;
      if (value > max) {
        max = value;
      }
      if (value < min) {
        min = value;
      }

      double valueRelativeToRange = value - lowerLimit;
      int targetBucket = (int) (valueRelativeToRange / bucketSize);

      if (targetBucket >= histogram.length) {
        targetBucket = histogram.length - 1;
      }

      histogram[targetBucket] += 1;
    }
    finally {
      readWriteLock.writeLock().unlock();
    }
  }

  /**
   * Called when the histogram encounters a null value.
   */
  public void incrementMissing()
  {
    readWriteLock.writeLock().lock();

    try {
      missingValueCount++;
    }
    finally {
      readWriteLock.writeLock().unlock();
    }
  }

  /**
   * Merge another datapoint into this one. The other datapoint could be
   *  - base64 encoded string of {@code FixedBucketsHistogram}
   *  - {@code FixedBucketsHistogram} object
   *  - Numeric value
   *
   * @param val
   */
  @VisibleForTesting
  public void combine(@Nullable Object val)
  {
    if (val == null) {
      if (NullHandling.replaceWithDefault()) {
        add(NullHandling.defaultDoubleValue());
      } else {
        incrementMissing();
      }
    } else if (val instanceof String) {
      combineHistogram(fromBase64((String) val));
    } else if (val instanceof FixedBucketsHistogram) {
      combineHistogram((FixedBucketsHistogram) val);
    } else if (val instanceof Number) {
      add(((Number) val).doubleValue());
    } else {
      throw new ISE("Unknown class for object: " + val.getClass());
    }
  }

  /**
   * Merge another histogram into this one. Only the state of this histogram is updated.
   *
   * If the two histograms have identical buckets, a simpler algorithm is used.
   *
   * @param otherHistogram
   */
  public void combineHistogram(FixedBucketsHistogram otherHistogram)
  {
    if (otherHistogram == null) {
      return;
    }

    readWriteLock.writeLock().lock();
    otherHistogram.getReadWriteLock().readLock().lock();

    try {
      missingValueCount += otherHistogram.getMissingValueCount();

      if (bucketSize == otherHistogram.getBucketSize() &&
          lowerLimit == otherHistogram.getLowerLimit() &&
          upperLimit == otherHistogram.getUpperLimit()) {
        combineHistogramSameBuckets(otherHistogram);
      } else {
        combineHistogramDifferentBuckets(otherHistogram);
      }
    }
    finally {
      readWriteLock.writeLock().unlock();
      otherHistogram.getReadWriteLock().readLock().unlock();
    }
  }

  /**
   * Merge another histogram that has the same range and same buckets.
   *
   * @param otherHistogram
   */
  private void combineHistogramSameBuckets(FixedBucketsHistogram otherHistogram)
  {
    long[] otherHistogramArray = otherHistogram.getHistogram();
    for (int i = 0; i < numBuckets; i++) {
      histogram[i] += otherHistogramArray[i];
    }

    count += otherHistogram.getCount();
    max = Math.max(max, otherHistogram.getMax());
    min = Math.min(min, otherHistogram.getMin());

    outlierHandler.handleOutliersForCombineSameBuckets(otherHistogram);
  }

  /**
   * Merge another histogram that has different buckets from mine.
   *
   * First, check if the other histogram falls entirely outside of my defined range. If so, call the appropriate
   * function for optimized handling.
   *
   * Otherwise, this merges the histograms with a more general function.
   *
   * @param otherHistogram
   */
  private void combineHistogramDifferentBuckets(FixedBucketsHistogram otherHistogram)
  {
    if (otherHistogram.getLowerLimit() >= upperLimit) {
      outlierHandler.handleOutliersCombineDifferentBucketsAllUpper(otherHistogram);
    } else if (otherHistogram.getUpperLimit() <= lowerLimit) {
      outlierHandler.handleOutliersCombineDifferentBucketsAllLower(otherHistogram);
    } else {
      simpleInterpolateMerge(otherHistogram);
    }
  }

  /**
   * Get a sum of bucket counts from either the start of a histogram's range or end, up to a specified cutoff value.
   *
   * For example, if I have the following histogram with a range of 0-40, with 4 buckets and
   * per-bucket counts of 5, 2, 10, and 7:
   *
   * |   5   |   2   |   24   |   7   |
   * 0       10      20       30      40
   *
   * Calling this function with a cutoff of 25 and fromStart = true would:
   * - Sum the first two bucket counts 5 + 2
   * - Since the cutoff falls in the third bucket, multiply the third bucket's count by the fraction of the bucket range
   *   covered by the cutoff, in this case the fraction is ((25 - 20) / 10) = 0.5
   * - The total count returned is 5 + 2 + 12
   *
   * @param cutoff Cutoff point within the histogram's range
   * @param fromStart If true, sum the bucket counts starting from the beginning of the histogram range.
   *                  If false, sum from the other direction, starting from the end of the histogram range.
   * @return Sum of bucket counts up to the cutoff point
   */
  private double getCumulativeCount(double cutoff, boolean fromStart)
  {
    int cutoffBucket = (int) ((cutoff - lowerLimit) / bucketSize);
    double count = 0;

    if (fromStart) {
      for (int i = 0; i <= cutoffBucket; i++) {
        if (i == cutoffBucket) {
          double bucketStart = i * bucketSize + lowerLimit;
          double partialCount = ((cutoff - bucketStart) / bucketSize) * histogram[i];
          count += partialCount;
        } else {
          count += histogram[i];
        }
      }
    } else {
      for (int i = cutoffBucket; i < histogram.length; i++) {
        if (i == cutoffBucket) {
          double bucketEnd = ((i + 1) * bucketSize) + lowerLimit;
          double partialCount = ((bucketEnd - cutoff) / bucketSize) * histogram[i];
          count += partialCount;
        } else {
          count += histogram[i];
        }
      }
    }
    return count;
  }

  /**
   * Combines this histogram with another histogram by "rebucketing" the other histogram to match our bucketing scheme,
   * assuming that the original input values are uniformly distributed within each bucket in the other histogram.
   *
   * Suppose we have the following histograms:
   *
   * |   0   |   0   |   0   |   0   |   0   |   0   |
   * 0       1       2       3       4       5       6
   *
   * |   4   |   4   |   0   |   0   |
   * 0       3       6       9       12
   *
   * We will preserve the bucketing scheme of our histogram, and determine what cut-off points within the other
   * histogram align with our bucket boundaries.
   *
   * Using this example, we would effectively rebucket the second histogram to the following:
   *
   * | 1.333 | 1.333 | 1.333 | 1.333 | 1.333 | 1.333 |   0   |   0   |   0   |   0   |   0   |   0   |
   * 0       1       2       3       4       5       6       7       8       9       10      11      12
   *
   * The 0-3 bucket in the second histogram is rebucketed across new buckets with size 1, with the original frequency 4
   * multiplied by the fraction (new bucket size / original bucket size), in this case 1/3.
   *
   * These new rebucketed counts are then added to the base histogram's buckets, with rounding, resulting in:
   *
   * |   1   |   1   |   1   |   1   |   1   |   1   |
   * 0       1       2       3       4       5       6
   *
   * @param otherHistogram other histogram to be merged
   */
  private void simpleInterpolateMerge(FixedBucketsHistogram otherHistogram)
  {
    double rangeStart = Math.max(lowerLimit, otherHistogram.getLowerLimit());
    double rangeEnd = Math.min(upperLimit, otherHistogram.getUpperLimit());

    int myCurBucket = (int) ((rangeStart - lowerLimit) / bucketSize);
    double myNextCursorBoundary = (myCurBucket + 1) * bucketSize + lowerLimit;
    double myCursor = rangeStart;

    int theirCurBucket = (int) ((rangeStart - otherHistogram.getLowerLimit()) / otherHistogram.getBucketSize());
    double theirNextCursorBoundary = (theirCurBucket + 1) * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit();
    double theirCursor = rangeStart;

    double intraBucketStride = otherHistogram.getBucketSize() / otherHistogram.getHistogram()[theirCurBucket];

    myNextCursorBoundary = Math.min(myNextCursorBoundary, rangeEnd);
    theirNextCursorBoundary = Math.min(theirNextCursorBoundary, rangeEnd);

    outlierHandler.simpleInterpolateMergeHandleOutliers(otherHistogram, rangeStart, rangeEnd);

    double theirCurrentLowerBucketBoundary = theirCurBucket * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit();

    while (myCursor < rangeEnd) {
      double needToConsume = myNextCursorBoundary - myCursor;
      double canConsumeFromOtherBucket = theirNextCursorBoundary - theirCursor;
      double toConsume = Math.min(needToConsume, canConsumeFromOtherBucket);

      // consume one of our bucket's worth of data from the other histogram
      while (needToConsume > 0) {
        double fractional = toConsume / otherHistogram.getBucketSize();
        double fractionalCount = otherHistogram.getHistogram()[theirCurBucket] * fractional;

        if (Math.round(fractionalCount) > 0) {
          // in this chunk from the other histogram bucket, calculate min and max interpolated values that would be added
          double minStride = Math.ceil(
              (theirCursor - (theirCurBucket * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit()))
              / intraBucketStride
          );

          minStride *= intraBucketStride;
          minStride += theirCurrentLowerBucketBoundary;
          if (minStride >= theirCursor) {
            min = Math.min(minStride, min);
            max = Math.max(minStride, max);
          }

          double maxStride = Math.floor(
              (theirCursor + toConsume - (theirCurBucket * otherHistogram.getBucketSize()
                                          + otherHistogram.getLowerLimit())) / intraBucketStride
          );
          maxStride *= intraBucketStride;
          maxStride += theirCurrentLowerBucketBoundary;
          if (maxStride < theirCursor + toConsume) {
            max = Math.max(maxStride, max);
            min = Math.min(maxStride, min);
          }
        }

        histogram[myCurBucket] += Math.round(fractionalCount);
        count += Math.round(fractionalCount);

        needToConsume -= toConsume;
        theirCursor += toConsume;
        myCursor += toConsume;


        if (theirCursor >= rangeEnd) {
          break;
        }

        // we've crossed buckets in the other histogram
        if (theirCursor >= theirNextCursorBoundary) {
          theirCurBucket += 1;
          intraBucketStride = otherHistogram.getBucketSize() / otherHistogram.getHistogram()[theirCurBucket];
          theirCurrentLowerBucketBoundary = theirCurBucket * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit();
          theirNextCursorBoundary = (theirCurBucket + 1) * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit();
          theirNextCursorBoundary = Math.min(theirNextCursorBoundary, rangeEnd);
        }

        canConsumeFromOtherBucket = theirNextCursorBoundary - theirCursor;
        toConsume = Math.min(needToConsume, canConsumeFromOtherBucket);
      }

      if (theirCursor >= rangeEnd) {
        break;
      }

      myCurBucket += 1;
      myNextCursorBoundary = (myCurBucket + 1) * bucketSize + lowerLimit;
      myNextCursorBoundary = Math.min(myNextCursorBoundary, rangeEnd);
    }
  }


  /**
   * Estimate percentiles from the histogram bucket counts, using the following process:
   *
   * Suppose we have the following histogram with range 0-10, with bucket size = 2, and the following counts:
   *
   * |   0  |   1  |   2  |   4  |   0  |
   * 0      2      4      6      8      10
   *
   * If we wanted to estimate the 75th percentile:
   * 1. Find the frequency cut-off for the 75th percentile: 0.75 * 7 (the total count for the entire histogram) = 5.25
   * 2. Find the bucket for the cut-off point: in this case, the 6-8 bucket, since the cumulative frequency
   *    from 0-6 is 3 and cumulative frequency from 0-8 is 7
   * 3. The percentile estimate is L + F * W, where:
   *
   *   L = lower boundary of target bucket
   *   F = (frequency cut-off - cumulative frequency up to target bucket) / frequency of target bucket
   *   W = bucket size
   *
   *   In this case:
   *   75th percentile estimate = 6 + ((5.25 - 3) / 4) * 2 = 7.125
   *
   * Based off PercentileBuckets code from Netflix Spectator:
   * - https://github.com/Netflix/spectator/blob/0d083da3a60221de9cc710e314b0749e47a40e67/spectator-api/src/main/java/com/netflix/spectator/api/histogram/PercentileBuckets.java#L105
   *
   * @param pcts Array of percentiles to be estimated. Must be sorted in ascending order.
   * @return Estimates of percentile values, in the same order as the provided input.
   */
  public float[] percentilesFloat(double[] pcts)
  {
    readWriteLock.readLock().lock();

    try {
      float[] results = new float[pcts.length];
      long total = count;

      int pctIdx = 0;

      long prev = 0;
      double prevP = 0.0;
      double prevB = lowerLimit;
      for (int i = 0; i < numBuckets; ++i) {
        long next = prev + histogram[i];
        double nextP = 100.0 * next / total;
        double nextB = (i + 1) * bucketSize + lowerLimit;
        while (pctIdx < pcts.length && nextP >= pcts[pctIdx]) {
          double f = (pcts[pctIdx] - prevP) / (nextP - prevP);
          results[pctIdx] = (float) (f * (nextB - prevB) + prevB);
          ++pctIdx;
        }
        if (pctIdx >= pcts.length) {
          break;
        }
        prev = next;
        prevP = nextP;
        prevB = nextB;
      }

      return results;
    }
    finally {
      readWriteLock.readLock().unlock();
    }
  }

  /**
   * Encode the serialized form generated by toBytes() in Base64, used as the JSON serialization format.
   *
   * @return Base64 serialization
   */
  @JsonValue
  public String toBase64()
  {
    byte[] asBytes = toBytes();
    return StringUtils.fromUtf8(StringUtils.encodeBase64(asBytes));
  }

  /**
   * Serialize the histogram, with two possible encodings chosen based on the number of filled buckets:
   *
   * Full: Store the histogram buckets as an array of bucket counts, with size numBuckets
   * Sparse: Store the histogram buckets as a list of (bucketNum, count) pairs.
   */
  public byte[] toBytes()
  {
    readWriteLock.readLock().lock();

    try {
      int nonEmptyBuckets = getNonEmptyBucketCount();
      if (nonEmptyBuckets < (numBuckets / 2)) {
        return toBytesSparse(nonEmptyBuckets);
      } else {
        return toBytesFull(true);
      }
    }
    finally {
      readWriteLock.readLock().unlock();
    }
  }

  /**
   * Write a serialization header containing the serde version byte and full/sparse encoding mode byte.
   *
   * This header is not needed when serializing the histogram for localized internal use within the
   * buffer aggregator implementation.
   *
   * @param buf Destination buffer
   * @param mode Full or sparse mode
   */
  private void writeByteBufferSerdeHeader(ByteBuffer buf, byte mode)
  {
    buf.put(SERIALIZATION_VERSION);
    buf.put(mode);
  }

  /**
   * Serializes histogram fields that are common to both the full and sparse encoding modes.
   *
   * @param buf Destination buffer
   */
  private void writeByteBufferCommonFields(ByteBuffer buf)
  {
    buf.putDouble(lowerLimit);
    buf.putDouble(upperLimit);
    buf.putInt(numBuckets);
    buf.put((byte) outlierHandlingMode.ordinal());

    buf.putLong(count);
    buf.putLong(lowerOutlierCount);
    buf.putLong(upperOutlierCount);
    buf.putLong(missingValueCount);

    buf.putDouble(max);
    buf.putDouble(min);
  }

  /**
   * Serialize the histogram in full encoding mode.
   *
   * For efficiency, the header is not written when the histogram
   * is serialized for localized internal use within the buffer aggregator implementation.
   *
   * @param withHeader If true, include the serialization header
   * @return Serialized histogram with full encoding
   */
  public byte[] toBytesFull(boolean withHeader)
  {
    int size = getFullStorageSize(numBuckets);
    if (withHeader) {
      size += SERDE_HEADER_SIZE;
    }
    ByteBuffer buf = ByteBuffer.allocate(size);
    writeByteBufferFull(buf, withHeader);
    return buf.array();
  }

  /**
   * Helper method for toBytesFull
   *
   * @param buf Destination buffer
   * @param withHeader If true, include the serialization header
   */
  private void writeByteBufferFull(ByteBuffer buf, boolean withHeader)
  {
    if (withHeader) {
      writeByteBufferSerdeHeader(buf, FULL_ENCODING_MODE);
    }

    writeByteBufferCommonFields(buf);

    buf.asLongBuffer().put(histogram);
    buf.position(buf.position() + Long.BYTES * histogram.length);
  }

  /**
   * Serialize the histogram in sparse encoding mode.
   *
   * The serialization header is always written, since the sparse encoding is only used in situations where the
   * header is required.
   *
   * @param nonEmptyBuckets Number of non-empty buckets in the histogram
   * @return Serialized histogram with sparse encoding
   */
  public byte[] toBytesSparse(int nonEmptyBuckets)
  {
    int size = SERDE_HEADER_SIZE + getSparseStorageSize(nonEmptyBuckets);
    ByteBuffer buf = ByteBuffer.allocate(size);
    writeByteBufferSparse(buf, nonEmptyBuckets);
    return buf.array();
  }

  /**
   * Helper method for toBytesSparse
   *
   * @param buf Destination buffer
   * @param nonEmptyBuckets Number of non-empty buckets in the histogram
   */
  public void writeByteBufferSparse(ByteBuffer buf, int nonEmptyBuckets)
  {
    writeByteBufferSerdeHeader(buf, SPARSE_ENCODING_MODE);
    writeByteBufferCommonFields(buf);

    buf.putInt(nonEmptyBuckets);
    int bucketsWritten = 0;
    for (int i = 0; i < numBuckets; i++) {
      if (histogram[i] > 0) {
        buf.putInt(i);
        buf.putLong(histogram[i]);
        bucketsWritten += 1;
      }
      if (bucketsWritten == nonEmptyBuckets) {
        break;
      }
    }
  }

  /**
   * Deserialize a Base64-encoded histogram, used when deserializing from JSON (such as when transferring query results)
   * or when ingesting a pre-computed histogram.
   *
   * @param encodedHistogram Base64-encoded histogram
   * @return Deserialized object
   */
  public static FixedBucketsHistogram fromBase64(String encodedHistogram)
  {
    byte[] asBytes = StringUtils.decodeBase64(encodedHistogram.getBytes(StandardCharsets.UTF_8));
    return fromBytes(asBytes);
  }

  /**
   * General deserialization method for FixedBucketsHistogram.
   *
   * @param bytes
   * @return
   */
  public static FixedBucketsHistogram fromBytes(byte[] bytes)
  {
    ByteBuffer buf = ByteBuffer.wrap(bytes);
    return fromByteBuffer(buf);
  }

  /**
   * Deserialization helper method
   *
   * @param buf Source buffer containing serialized histogram
   * @return Deserialized object
   */
  public static FixedBucketsHistogram fromByteBuffer(ByteBuffer buf)
  {
    byte serializationVersion = buf.get();
    Preconditions.checkArgument(
        serializationVersion == SERIALIZATION_VERSION,
        StringUtils.format("Only serialization version %s is supported.", SERIALIZATION_VERSION)
    );
    byte mode = buf.get();
    if (mode == FULL_ENCODING_MODE) {
      return fromByteBufferFullNoSerdeHeader(buf);
    } else if (mode == SPARSE_ENCODING_MODE) {
      return fromBytesSparse(buf);
    } else {
      throw new ISE("Invalid histogram serde mode: %s", mode);
    }
  }

  /**
   * Helper method for deserializing histograms with full encoding mode. Assumes that the serialization header is not
   * present or has already been read.
   *
   * @param buf Source buffer containing serialized full-encoding histogram.
   * @return Deserialized object
   */
  protected static FixedBucketsHistogram fromByteBufferFullNoSerdeHeader(ByteBuffer buf)
  {
    double lowerLimit = buf.getDouble();
    double upperLimit = buf.getDouble();
    int numBuckets = buf.getInt();
    OutlierHandlingMode outlierHandlingMode = OutlierHandlingMode.values()[buf.get()];

    long count = buf.getLong();
    long lowerOutlierCount = buf.getLong();
    long upperOutlierCount = buf.getLong();
    long missingValueCount = buf.getLong();

    double max = buf.getDouble();
    double min = buf.getDouble();

    long[] histogram = new long[numBuckets];
    buf.asLongBuffer().get(histogram);
    buf.position(buf.position() + Long.BYTES * histogram.length);

    return new FixedBucketsHistogram(
        lowerLimit,
        upperLimit,
        numBuckets,
        outlierHandlingMode,
        histogram,
        count,
        max,
        min,
        lowerOutlierCount,
        upperOutlierCount,
        missingValueCount
    );
  }

  /**
   * Helper method for deserializing histograms with sparse encoding mode. Assumes that the serialization header is not
   * present or has already been read.
   *
   * @param buf Source buffer containing serialized sparse-encoding histogram.
   * @return Deserialized object
   */
  private static FixedBucketsHistogram fromBytesSparse(ByteBuffer buf)
  {
    double lowerLimit = buf.getDouble();
    double upperLimit = buf.getDouble();
    int numBuckets = buf.getInt();
    OutlierHandlingMode outlierHandlingMode = OutlierHandlingMode.values()[buf.get()];

    long count = buf.getLong();
    long lowerOutlierCount = buf.getLong();
    long upperOutlierCount = buf.getLong();
    long missingValueCount = buf.getLong();

    double max = buf.getDouble();
    double min = buf.getDouble();

    int nonEmptyBuckets = buf.getInt();
    long[] histogram = new long[numBuckets];
    for (int i = 0; i < nonEmptyBuckets; i++) {
      int bucket = buf.getInt();
      long bucketCount = buf.getLong();
      histogram[bucket] = bucketCount;
    }

    return new FixedBucketsHistogram(
        lowerLimit,
        upperLimit,
        numBuckets,
        outlierHandlingMode,
        histogram,
        count,
        max,
        min,
        lowerOutlierCount,
        upperOutlierCount,
        missingValueCount
    );
  }

  /**
   * Compute the size in bytes of a full-encoding serialized histogram, without the serialization header
   *
   * @param numBuckets number of buckets
   * @return full serialized size in bytes
   */
  public static int getFullStorageSize(int numBuckets)
  {
    return COMMON_FIELDS_SIZE +
           Long.BYTES * numBuckets;
  }

  /**
   * Compute the size in bytes of a sparse-encoding serialized histogram, without the serialization header
   *
   * @param nonEmptyBuckets number of non-empty buckets
   * @return sparse serialized size in bytes
   */
  public static int getSparseStorageSize(int nonEmptyBuckets)
  {
    return COMMON_FIELDS_SIZE +
           Integer.BYTES +
           (Integer.BYTES + Long.BYTES) * nonEmptyBuckets;
  }

  @VisibleForTesting
  public int getNonEmptyBucketCount()
  {
    int count = 0;
    for (int i = 0; i < numBuckets; i++) {
      if (histogram[i] != 0) {
        count++;
      }
    }
    return count;
  }

  /**
   * Updates the histogram state when an outlier value is encountered. An implementation is provided for each
   * supported outlierHandlingMode.
   */
  private interface OutlierHandler
  {
    /**
     * Handle an outlier when a single numeric value is added to the histogram.
     *
     * @param exceededMax If true, the value was an upper outlier. If false, the value was a lower outlier.
     */
    void handleOutlierAdd(boolean exceededMax);

    /**
     * Merge outlier information from another histogram.
     *
     * Called when merging two histograms that have overlapping ranges and potentially different bucket sizes.
     *
     * @param otherHistogram
     * @param rangeStart
     * @param rangeEnd
     */
    void simpleInterpolateMergeHandleOutliers(
        FixedBucketsHistogram otherHistogram,
        double rangeStart,
        double rangeEnd
    );

    /**
     * Merge outlier information from another histogram.
     *
     * Called when merging two histograms with identical buckets (same range and number of buckets)
     *
     * @param otherHistogram other histogram being merged
     */
    void handleOutliersForCombineSameBuckets(FixedBucketsHistogram otherHistogram);

    /**
     * Merge outlier information from another histogram.
     *
     * Called when merging two histograms, where the histogram to be merged has lowerLimit greater than my upperLimit
     * @param otherHistogram other histogram being merged
     */
    void handleOutliersCombineDifferentBucketsAllUpper(FixedBucketsHistogram otherHistogram);

    /**
     * Merge outlier information from another histogram.
     *
     * Called when merging two histograms, where the histogram to be merged has upperLimit less than my lowerLimit
     * @param otherHistogram other histogram being merged
     */

    void handleOutliersCombineDifferentBucketsAllLower(FixedBucketsHistogram otherHistogram);
  }

  private OutlierHandler makeOutlierHandler()
  {
    switch (outlierHandlingMode) {
      case IGNORE:
        return new IgnoreOutlierHandler();
      case OVERFLOW:
        return new OverflowOutlierHandler();
      case CLIP:
        return new ClipOutlierHandler();
      default:
        throw new ISE("Unknown outlier handling mode: %s", outlierHandlingMode);
    }
  }

  private static class IgnoreOutlierHandler implements OutlierHandler
  {
    @Override
    public void handleOutlierAdd(boolean exceededMax)
    {
    }

    @Override
    public void simpleInterpolateMergeHandleOutliers(
        FixedBucketsHistogram otherHistogram,
        double rangeStart,
        double rangeEnd
    )
    {
    }

    @Override
    public void handleOutliersForCombineSameBuckets(FixedBucketsHistogram otherHistogram)
    {
    }

    @Override
    public void handleOutliersCombineDifferentBucketsAllUpper(FixedBucketsHistogram otherHistogram)
    {
    }

    @Override
    public void handleOutliersCombineDifferentBucketsAllLower(FixedBucketsHistogram otherHistogram)
    {
    }
  }

  private class OverflowOutlierHandler implements OutlierHandler
  {
    @Override
    public void handleOutlierAdd(boolean exceededMax)
    {
      if (exceededMax) {
        upperOutlierCount += 1;
      } else {
        lowerOutlierCount += 1;
      }
    }

    @Override
    public void simpleInterpolateMergeHandleOutliers(
        FixedBucketsHistogram otherHistogram,
        double rangeStart,
        double rangeEnd
    )
    {
      // They contain me
      if (lowerLimit == rangeStart && upperLimit == rangeEnd) {
        long lowerCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeStart, true));
        long upperCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeEnd, false));
        upperOutlierCount += otherHistogram.getUpperOutlierCount() + upperCountFromOther;
        lowerOutlierCount += otherHistogram.getLowerOutlierCount() + lowerCountFromOther;
      } else if (rangeStart == lowerLimit) {
        // assume that none of the other histogram's outliers fall within our range
        // how many values were there in the portion of the other histogram that falls under my lower limit?
        long lowerCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeStart, true));
        lowerOutlierCount += lowerCountFromOther;
        lowerOutlierCount += otherHistogram.getLowerOutlierCount();
        upperOutlierCount += otherHistogram.getUpperOutlierCount();
      } else if (rangeEnd == upperLimit) {
        // how many values were there in the portion of the other histogram that is greater than my upper limit?
        long upperCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeEnd, false));
        upperOutlierCount += upperCountFromOther;
        upperOutlierCount += otherHistogram.getUpperOutlierCount();
        lowerOutlierCount += otherHistogram.getLowerOutlierCount();
      } else if (rangeStart > lowerLimit && rangeEnd < upperLimit) {
        upperOutlierCount += otherHistogram.getUpperOutlierCount();
        lowerOutlierCount += otherHistogram.getLowerOutlierCount();
      }
    }

    @Override
    public void handleOutliersForCombineSameBuckets(FixedBucketsHistogram otherHistogram)
    {
      lowerOutlierCount += otherHistogram.getLowerOutlierCount();
      upperOutlierCount += otherHistogram.getUpperOutlierCount();
    }

    @Override
    public void handleOutliersCombineDifferentBucketsAllUpper(FixedBucketsHistogram otherHistogram)
    {
      // ignore any lower outliers in the other histogram, we're not sure where those outliers would fall
      // within our range.
      upperOutlierCount += otherHistogram.getCount() + otherHistogram.getUpperOutlierCount();
    }

    @Override
    public void handleOutliersCombineDifferentBucketsAllLower(FixedBucketsHistogram otherHistogram)
    {
      // ignore any upper outliers in the other histogram, we're not sure where those outliers would fall
      // within our range.
      lowerOutlierCount += otherHistogram.getCount() + otherHistogram.getLowerOutlierCount();
    }
  }

  private class ClipOutlierHandler implements OutlierHandler
  {
    @Override
    public void handleOutlierAdd(boolean exceededMax)
    {
      double clippedValue;
      count += 1;
      if (exceededMax) {
        clippedValue = upperLimit;
        histogram[histogram.length - 1] += 1;
      } else {
        clippedValue = lowerLimit;
        histogram[0] += 1;
      }
      if (clippedValue > max) {
        max = clippedValue;
      }
      if (clippedValue < min) {
        min = clippedValue;
      }
    }

    @Override
    public void simpleInterpolateMergeHandleOutliers(
        FixedBucketsHistogram otherHistogram,
        double rangeStart,
        double rangeEnd
    )
    {
      // They contain me
      if (lowerLimit == rangeStart && upperLimit == rangeEnd) {
        long lowerCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeStart, true));
        long upperCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeEnd, false));
        histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount() + upperCountFromOther;
        histogram[0] += otherHistogram.getLowerOutlierCount() + lowerCountFromOther;
        count += otherHistogram.getUpperOutlierCount() + otherHistogram.getLowerOutlierCount();
        count += upperCountFromOther + lowerCountFromOther;
        if (otherHistogram.getUpperOutlierCount() + upperCountFromOther > 0) {
          max = Math.max(max, upperLimit);
        }
        if (otherHistogram.getLowerOutlierCount() + lowerCountFromOther > 0) {
          min = Math.min(min, lowerLimit);
        }
      } else if (rangeStart == lowerLimit) {
        // assume that none of the other histogram's outliers fall within our range
        // how many values were there in the portion of the other histogram that falls under my lower limit?
        long lowerCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeStart, true));

        histogram[0] += lowerCountFromOther;
        histogram[0] += otherHistogram.getLowerOutlierCount();
        histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount();
        count += lowerCountFromOther + otherHistogram.getLowerOutlierCount() + otherHistogram.getUpperOutlierCount();
        if (lowerCountFromOther + otherHistogram.getLowerOutlierCount() > 0) {
          min = lowerLimit;
        }
        if (otherHistogram.getUpperOutlierCount() > 0) {
          max = upperLimit;
        }
      } else if (rangeEnd == upperLimit) {
        // how many values were there in the portion of the other histogram that is greater than my upper limit?
        long upperCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeEnd, false));

        histogram[histogram.length - 1] += upperCountFromOther;
        histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount();
        histogram[0] += otherHistogram.getLowerOutlierCount();
        count += upperCountFromOther + otherHistogram.getLowerOutlierCount() + otherHistogram.getUpperOutlierCount();
        if (upperCountFromOther + otherHistogram.getUpperOutlierCount() > 0) {
          max = upperLimit;
        }
        if (otherHistogram.getLowerOutlierCount() > 0) {
          min = lowerLimit;
        }
      } else if (rangeStart > lowerLimit && rangeEnd < upperLimit) {
        // I contain them

        histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount();
        histogram[0] += otherHistogram.getLowerOutlierCount();
        count += otherHistogram.getUpperOutlierCount() + otherHistogram.getLowerOutlierCount();
        if (otherHistogram.getUpperOutlierCount() > 0) {
          max = Math.max(max, upperLimit);
        } else if (otherHistogram.getCount() > 0) {
          max = Math.max(max, otherHistogram.getMax());
        }
        if (otherHistogram.getLowerOutlierCount() > 0) {
          min = Math.min(min, lowerLimit);
        } else if (otherHistogram.getCount() > 0) {
          min = Math.min(min, otherHistogram.getMin());
        }
      }
    }

    @Override
    public void handleOutliersForCombineSameBuckets(FixedBucketsHistogram otherHistogram)
    {
      if (otherHistogram.getLowerOutlierCount() > 0) {
        histogram[0] += otherHistogram.getLowerOutlierCount();
        count += otherHistogram.getLowerOutlierCount();
        min = Math.min(lowerLimit, min);
      }
      if (otherHistogram.getUpperOutlierCount() > 0) {
        histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount();
        count += otherHistogram.getUpperOutlierCount();
        max = Math.max(upperLimit, max);
      }
    }

    @Override
    public void handleOutliersCombineDifferentBucketsAllUpper(FixedBucketsHistogram otherHistogram)
    {
      long otherCount = otherHistogram.getCount() + otherHistogram.getUpperOutlierCount();
      histogram[histogram.length - 1] += otherCount;
      count += otherCount;
      if (otherCount > 0) {
        max = Math.max(max, upperLimit);
      }
    }

    @Override
    public void handleOutliersCombineDifferentBucketsAllLower(FixedBucketsHistogram otherHistogram)
    {
      long otherCount = otherHistogram.getCount() + otherHistogram.getLowerOutlierCount();
      histogram[0] += otherCount;
      count += otherCount;
      if (otherCount > 0) {
        min = Math.min(min, lowerLimit);
      }
    }
  }
}