content: /*
   * 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.calcite.profile;
  
  import org.apache.calcite.config.CalciteSystemProperty;
  import org.apache.calcite.linq4j.Ord;
  import org.apache.calcite.linq4j.tree.Primitive;
  import org.apache.calcite.materialize.Lattice;
  import org.apache.calcite.rel.metadata.NullSentinel;
  import org.apache.calcite.runtime.FlatLists;
  import org.apache.calcite.util.ImmutableBitSet;
  import org.apache.calcite.util.Pair;
  import org.apache.calcite.util.PartiallyOrderedSet;
  import org.apache.calcite.util.Util;
  
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.ImmutableSortedSet;
  import com.google.common.collect.Iterables;
  import com.google.common.collect.Ordering;
  import com.yahoo.sketches.hll.HllSketch;
  
  import org.checkerframework.checker.nullness.qual.Nullable;
  
  import java.nio.ByteBuffer;
  import java.nio.charset.StandardCharsets;
  import java.util.ArrayDeque;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.BitSet;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.Deque;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.NavigableSet;
  import java.util.PriorityQueue;
  import java.util.Queue;
  import java.util.Set;
  import java.util.SortedSet;
  import java.util.TreeSet;
  import java.util.function.Predicate;
  
  import static com.google.common.base.Preconditions.checkArgument;
  
  import static org.apache.calcite.linq4j.Nullness.castNonNull;
  import static org.apache.calcite.profile.ProfilerImpl.CompositeCollector.OF;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * Implementation of {@link Profiler} that only investigates "interesting"
   * combinations of columns.
   */
  public class ProfilerImpl implements Profiler {
    /** The number of combinations to consider per pass.
     * The number is determined by memory, but a value of 1,000 is typical.
     * You need 2KB memory per sketch, and one sketch for each combination. */
    private final int combinationsPerPass;
  
    /** The minimum number of combinations considered "interesting". After that,
     * a combination is only considered "interesting" if its surprise is greater
     * than the median surprise. */
    private final int interestingCount;
  
    /** Whether a successor is considered interesting enough to analyze. */
    private final Predicate<Pair<Space, Column>> predicate;
  
    public static Builder builder() {
      return new Builder();
    }
  
    /**
     * Creates a {@code ProfilerImpl}.
     *
     * @param combinationsPerPass Maximum number of columns (or combinations of
     *   columns) to compute each pass
     * @param interestingCount Minimum number of combinations considered
     *   interesting
     * @param predicate Whether a successor is considered interesting enough to
     *   analyze
     */
    ProfilerImpl(int combinationsPerPass,
        int interestingCount, Predicate<Pair<Space, Column>> predicate) {
      checkArgument(combinationsPerPass > 2);
      checkArgument(interestingCount > 2);
      this.combinationsPerPass = combinationsPerPass;
      this.interestingCount = interestingCount;
      this.predicate = predicate;
    }
  
    @Override public Profile profile(Iterable<List<Comparable>> rows,
        final List<Column> columns, Collection<ImmutableBitSet> initialGroups) {
      return new Run(columns, initialGroups).profile(rows);
    }
  
    /** A run of the profiler. */
    class Run {
      private final List<Column> columns;
      final PartiallyOrderedSet<ImmutableBitSet> keyPoset =
          new PartiallyOrderedSet<>(
              PartiallyOrderedSet.BIT_SET_INCLUSION_ORDERING);
      final Map<ImmutableBitSet, Distribution> distributions = new HashMap<>();
      /** List of spaces that have one column. */
      final List<@Nullable Space> singletonSpaces;
      /** Combinations of columns that we have computed but whose successors have
       * not yet been computed. We may add some of those successors to
       * {@link #spaceQueue}. */
      final Queue<Space> doneQueue =
          new PriorityQueue<>(100, (s0, s1) -> {
            // The space with 0 columns is more interesting than
            // any space with 1 column, and so forth.
            // For spaces with 2 or more columns we compare "surprise":
            // how many fewer values did it have than expected?
            int c = Integer.compare(s0.columns.size(), s1.columns.size());
            if (c == 0) {
              c = Double.compare(s0.surprise(), s1.surprise());
            }
            return c;
          });
      final SurpriseQueue surprises;
  
      /** Combinations of columns that we will compute next pass. */
      final Deque<ImmutableBitSet> spaceQueue = new ArrayDeque<>();
      final List<Unique> uniques = new ArrayList<>();
      final List<FunctionalDependency> functionalDependencies = new ArrayList<>();
      /** Column ordinals that have ever been placed on {@link #spaceQueue}.
       * Ensures that we do not calculate the same combination more than once,
       * even though we generate a column set from multiple parents. */
      final Set<ImmutableBitSet> resultSet = new HashSet<>();
      final PartiallyOrderedSet<Space> results =
          new PartiallyOrderedSet<>((e1, e2) ->
              e2.columnOrdinals.contains(e1.columnOrdinals));
      private final List<ImmutableBitSet> keyOrdinalLists =
          new ArrayList<>();
      private int rowCount;
  
      /**
       * Creates a Run.
       *
       * @param columns List of columns
       *
       * @param initialGroups List of combinations of columns that should be
       *                     profiled early, because they may be interesting
       */
      Run(final List<Column> columns, Collection<ImmutableBitSet> initialGroups) {
        this.columns = ImmutableList.copyOf(columns);
        for (Ord<Column> column : Ord.zip(columns)) {
          if (column.e.ordinal != column.i) {
            throw new IllegalArgumentException();
          }
        }
        this.singletonSpaces =
            new ArrayList<>(Collections.nCopies(columns.size(), (Space) null));
        if (combinationsPerPass > Math.pow(2D, columns.size())) {
          // There are not many columns. We can compute all combinations in the
          // first pass.
          for (ImmutableBitSet ordinals
              : ImmutableBitSet.range(columns.size()).powerSet()) {
            spaceQueue.add(ordinals);
          }
        } else {
          // We will need to take multiple passes.
          // Pass 0, just put the empty combination on the queue.
          // Next pass, we will do its successors, the singleton combinations.
          spaceQueue.add(ImmutableBitSet.of());
          spaceQueue.addAll(initialGroups);
          if (columns.size() < combinationsPerPass) {
            // There are not very many columns. Compute the singleton
            // groups in pass 0.
            for (Column column : columns) {
              spaceQueue.add(ImmutableBitSet.of(column.ordinal));
            }
          }
        }
        // The surprise queue must have enough room for all singleton groups
        // plus all initial groups.
        surprises =
            new SurpriseQueue(1 + columns.size() + initialGroups.size(),
                interestingCount);
      }
  
      Profile profile(Iterable<List<Comparable>> rows) {
        int pass = 0;
        for (;;) {
          final List<Space> spaces = nextBatch(pass);
          if (spaces.isEmpty()) {
            break;
          }
          pass(pass++, spaces, rows);
        }
  
        for (Space s : singletonSpaces) {
          for (ImmutableBitSet dependent : requireNonNull(s, "s").dependents) {
            functionalDependencies.add(
                new FunctionalDependency(toColumns(dependent),
                    Iterables.getOnlyElement(s.columns)));
          }
        }
        return new Profile(columns, new RowCount(rowCount),
            functionalDependencies, distributions.values(), uniques);
      }
  
      /** Populates {@code spaces} with the next batch.
       * Returns an empty list if done. */
      List<Space> nextBatch(int pass) {
        final List<Space> spaces = new ArrayList<>();
      loop:
        for (;;) {
          if (spaces.size() >= combinationsPerPass) {
            // We have enough for the next pass.
            return spaces;
          }
          // First, see if there is a space we did have room for last pass.
          final ImmutableBitSet ordinals = spaceQueue.poll();
          if (ordinals != null) {
            final Space space = new Space(this, ordinals, toColumns(ordinals));
            spaces.add(space);
            if (ordinals.cardinality() == 1) {
              singletonSpaces.set(ordinals.nth(0), space);
            }
          } else {
            // Next, take a space that was done last time, generate its
            // successors, and add the interesting ones to the space queue.
            for (;;) {
              final Space doneSpace = doneQueue.poll();
              if (doneSpace == null) {
                // There are no more done spaces. We're done.
                return spaces;
              }
              if (doneSpace.columnOrdinals.cardinality() > 4) {
                // Do not generate successors for groups with lots of columns,
                // probably initial groups
                continue;
              }
              for (Column column : columns) {
                if (!doneSpace.columnOrdinals.get(column.ordinal)) {
                  if (pass == 0
                      || doneSpace.columnOrdinals.cardinality() == 0
                      || !containsKey(
                          doneSpace.columnOrdinals.set(column.ordinal))
                      && predicate.test(Pair.of(doneSpace, column))) {
                    final ImmutableBitSet nextOrdinals =
                        doneSpace.columnOrdinals.set(column.ordinal);
                    if (resultSet.add(nextOrdinals)) {
                      spaceQueue.add(nextOrdinals);
                    }
                  }
                }
              }
              // We've converted at a space into at least one interesting
              // successor.
              if (!spaceQueue.isEmpty()) {
                continue loop;
              }
            }
          }
        }
      }
  
      private boolean containsKey(ImmutableBitSet ordinals) {
        for (ImmutableBitSet keyOrdinals : keyOrdinalLists) {
          if (ordinals.contains(keyOrdinals)) {
            return true;
          }
        }
        return false;
      }
  
      void pass(int pass, List<Space> spaces, Iterable<List<Comparable>> rows) {
        if (CalciteSystemProperty.DEBUG.value()) {
          System.out.println("pass: " + pass
              + ", spaces.size: " + spaces.size()
              + ", distributions.size: " + distributions.size());
        }
  
        for (Space space : spaces) {
          space.collector = Collector.create(space, 1000);
        }
  
        int rowCount = 0;
        for (final List<Comparable> row : rows) {
          ++rowCount;
          for (Space space : spaces) {
            castNonNull(space.collector).add(row);
          }
        }
  
        // Populate unique keys.
        // If [x, y] is a key,
        // then [x, y, z] is a non-minimal key (therefore not interesting),
        // and [x, y] => [a] is a functional dependency but not interesting,
        // and [x, y, z] is not an interesting distribution.
        for (Space space : spaces) {
          Collector collector = space.collector;
          if (collector != null) {
            collector.finish();
          }
          space.collector = null;
  //        results.add(space);
  
          int nonMinimal = 0;
        dependents:
          for (Space s : results.getDescendants(space)) {
            if (s.cardinality == space.cardinality) {
              // We have discovered a sub-set that has the same cardinality.
              // The column(s) that are not in common are functionally
              // dependent.
              final ImmutableBitSet dependents =
                  space.columnOrdinals.except(s.columnOrdinals);
              for (int i : s.columnOrdinals) {
                final Space s1 = singletonSpaces.get(i);
                final ImmutableBitSet rest = s.columnOrdinals.clear(i);
                for (ImmutableBitSet dependent : requireNonNull(s1, "s1").dependents) {
                  if (rest.contains(dependent)) {
                    // The "key" of this functional dependency is not minimal.
                    // For instance, if we know that
                    //   (a) -> x
                    // then
                    //   (a, b, x) -> y
                    // is not minimal; we could say the same with a smaller key:
                    //   (a, b) -> y
                    ++nonMinimal;
                    continue dependents;
                  }
                }
              }
              for (int dependent : dependents) {
                final Space s1 = singletonSpaces.get(dependent);
                for (ImmutableBitSet d : requireNonNull(s1, "s1").dependents) {
                  if (s.columnOrdinals.contains(d)) {
                    ++nonMinimal;
                    continue dependents;
                  }
                }
              }
              space.dependencies.or(dependents.toBitSet());
              for (int d : dependents) {
                Space spaceD = requireNonNull(singletonSpaces.get(d), "singletonSpaces.get(d)");
                spaceD.dependents.add(s.columnOrdinals);
              }
            }
          }
          if (nonMinimal > 0) {
            continue;
          }
          final String s = space.columns.toString(); // for debug
          Util.discard(s);
          double expectedCardinality =
              expectedCardinality(rowCount, space.columnOrdinals);
  
          final boolean minimal = nonMinimal == 0
              && !space.unique
              && !containsKey(space.columnOrdinals);
          space.expectedCardinality = expectedCardinality;
          if (minimal) {
            final Distribution distribution =
                new Distribution(space.columns, space.valueSet, space.cardinality,
                    space.nullCount, expectedCardinality, minimal);
            final double surprise = distribution.surprise();
            if (CalciteSystemProperty.DEBUG.value() && surprise > 0.1d) {
              System.out.println(distribution.columnOrdinals()
                  + " " + distribution.columns
                  + ", cardinality: " + distribution.cardinality
                  + ", expected: " + distribution.expectedCardinality
                  + ", surprise: " + distribution.surprise());
            }
            if (surprises.offer(surprise)) {
              distributions.put(space.columnOrdinals, distribution);
              keyPoset.add(space.columnOrdinals);
              doneQueue.add(space);
            }
          }
  
          if (space.cardinality == rowCount) {
            // We have discovered a new key. It is not a super-set of a key.
            uniques.add(new Unique(space.columns));
            keyOrdinalLists.add(space.columnOrdinals);
            space.unique = true;
          }
        }
  
        if (pass == 0) {
          this.rowCount = rowCount;
        }
      }
  
      /** Estimates the cardinality of a collection of columns represented by
       * {@code columnOrdinals}, drawing on existing distributions. */
      private double cardinality(double rowCount, ImmutableBitSet columns) {
        final Distribution distribution = distributions.get(columns);
        if (distribution != null) {
          return distribution.cardinality;
        } else {
          return expectedCardinality(rowCount, columns);
        }
      }
  
      /** Estimates the cardinality of a collection of columns represented by
       * {@code columnOrdinals}, drawing on existing distributions. Does not
       * look in the distribution map for this column set. */
      private double expectedCardinality(double rowCount,
          ImmutableBitSet columns) {
        switch (columns.cardinality()) {
        case 0:
          return 1d;
        case 1:
          return rowCount;
        default:
          double c = rowCount;
          List<ImmutableBitSet> parents =
              requireNonNull(keyPoset.getParents(columns, true),
                  () -> "keyPoset.getParents(columns, true) is null for "
                      + columns);
          for (ImmutableBitSet bitSet : parents) {
            if (bitSet.isEmpty()) {
              // If the parent is the empty group (i.e. "GROUP BY ()", the grand
              // total) we cannot improve on the estimate.
              continue;
            }
            final Distribution d1 = distributions.get(bitSet);
            final double c2 = cardinality(rowCount, columns.except(bitSet));
            final double d =
                Lattice.getRowCount(rowCount,
                    requireNonNull(d1, "d1").cardinality, c2);
            c = Math.min(c, d);
          }
          List<ImmutableBitSet> children =
              requireNonNull(keyPoset.getChildren(columns, true),
                  () -> "keyPoset.getChildren(columns, true) is null for "
                      + columns);
          for (ImmutableBitSet bitSet : children) {
            final Distribution d1 = distributions.get(bitSet);
            c = Math.min(c, requireNonNull(d1, "d1").cardinality);
          }
          return c;
        }
      }
  
  
      private ImmutableSortedSet<Column> toColumns(Iterable<Integer> ordinals) {
        //noinspection Convert2MethodRef
        return ImmutableSortedSet.copyOf(
            Util.transform(ordinals, idx -> columns.get(idx)));
      }
    }
  
    /** Work space for a particular combination of columns. */
    static class Space {
      private final Run run;
      final ImmutableBitSet columnOrdinals;
      final ImmutableSortedSet<Column> columns;
      boolean unique;
      final BitSet dependencies = new BitSet();
      final Set<ImmutableBitSet> dependents = new HashSet<>();
      double expectedCardinality;
      @Nullable Collector collector;
      /** Assigned by {@link Collector#finish()}. */
      int nullCount;
      /** Number of distinct values. Null is counted as a value, if present.
       * Assigned by {@link Collector#finish()}. */
      int cardinality;
      /** Assigned by {@link Collector#finish()}. */
      @Nullable SortedSet<Comparable> valueSet;
  
      Space(Run run, ImmutableBitSet columnOrdinals, Iterable<Column> columns) {
        this.run = run;
        this.columnOrdinals = columnOrdinals;
        this.columns = ImmutableSortedSet.copyOf(columns);
      }
  
      @Override public int hashCode() {
        return columnOrdinals.hashCode();
      }
  
      @Override public boolean equals(@Nullable Object o) {
        return o == this
            || o instanceof Space
            && columnOrdinals.equals(((Space) o).columnOrdinals);
      }
  
      /** Returns the distribution created from this space, or null if no
       * distribution has been registered yet. */
      public @Nullable Distribution distribution() {
        return run.distributions.get(columnOrdinals);
      }
  
      double surprise() {
        return SimpleProfiler.surprise(expectedCardinality, cardinality);
      }
    }
  
    /** Builds a {@link org.apache.calcite.profile.ProfilerImpl}. */
    public static class Builder {
      int combinationsPerPass = 100;
      Predicate<Pair<Space, Column>> predicate = p -> true;
  
      public ProfilerImpl build() {
        return new ProfilerImpl(combinationsPerPass, 200, predicate);
      }
  
      public Builder withPassSize(int passSize) {
        this.combinationsPerPass = passSize;
        return this;
      }
  
      public Builder withMinimumSurprise(double v) {
        predicate =
            spaceColumnPair -> {
              @SuppressWarnings("unused")
              final Space space = spaceColumnPair.left;
              return false;
            };
        return this;
      }
    }
  
    /** Collects values of a column or columns. */
    abstract static class Collector {
      protected final Space space;
  
      Collector(Space space) {
        this.space = space;
      }
  
      abstract void add(List<Comparable> row);
      abstract void finish();
  
      /** Creates an initial collector of the appropriate kind. */
      public static Collector create(Space space, int sketchThreshold) {
        final List<Integer> columnOrdinalList = space.columnOrdinals.asList();
        if (columnOrdinalList.size() == 1) {
          return new SingletonCollector(space, columnOrdinalList.get(0),
              sketchThreshold);
        } else {
          return new CompositeCollector(space,
              (int[]) Primitive.INT.toArray(columnOrdinalList), sketchThreshold);
        }
      }
    }
  
    /** Collector that collects values of a single column. */
    static class SingletonCollector extends Collector {
      final NavigableSet<Comparable> values = new TreeSet<>();
      final int columnOrdinal;
      final int sketchThreshold;
      int nullCount = 0;
  
      SingletonCollector(Space space, int columnOrdinal, int sketchThreshold) {
        super(space);
        this.columnOrdinal = columnOrdinal;
        this.sketchThreshold = sketchThreshold;
      }
  
      @Override public void add(List<Comparable> row) {
        final Comparable v = row.get(columnOrdinal);
        if (v == NullSentinel.INSTANCE) {
          nullCount++;
        } else {
          if (values.add(v) && values.size() == sketchThreshold) {
            // Too many values. Switch to a sketch collector.
            final HllSingletonCollector collector =
                new HllSingletonCollector(space, columnOrdinal);
            for (Comparable value : values) {
              collector.add(value);
            }
            space.collector = collector;
          }
        }
      }
  
      @Override public void finish() {
        space.nullCount = nullCount;
        space.cardinality = values.size() + (nullCount > 0 ? 1 : 0);
        space.valueSet = values.size() < 20 ? values : null;
      }
    }
  
    /** Collector that collects two or more column values in a tree set. */
    static class CompositeCollector extends Collector {
      protected static final ImmutableBitSet OF = ImmutableBitSet.of(2, 13);
      final Set<FlatLists.ComparableList> values = new HashSet<>();
      final int[] columnOrdinals;
      final Comparable[] columnValues;
      int nullCount = 0;
      private final int sketchThreshold;
  
      CompositeCollector(Space space, int[] columnOrdinals, int sketchThreshold) {
        super(space);
        this.columnOrdinals = columnOrdinals;
        this.columnValues = new Comparable[columnOrdinals.length];
        this.sketchThreshold = sketchThreshold;
      }
  
      @Override public void add(List<Comparable> row) {
        if (space.columnOrdinals.equals(OF)) {
          Util.discard(0);
        }
        int nullCountThisRow = 0;
        for (int i = 0, length = columnOrdinals.length; i < length; i++) {
          final Comparable value = row.get(columnOrdinals[i]);
          if (value == NullSentinel.INSTANCE) {
            if (nullCountThisRow++ == 0) {
              nullCount++;
            }
          }
          columnValues[i] = value;
        }
        //noinspection unchecked
        if (((Set) values).add(FlatLists.copyOf(columnValues))
            && values.size() == sketchThreshold) {
          // Too many values. Switch to a sketch collector.
          final HllCompositeCollector collector =
              new HllCompositeCollector(space, columnOrdinals);
          final List<Comparable> list =
              new ArrayList<>(
                  Collections.nCopies(columnOrdinals[columnOrdinals.length - 1]
                          + 1,
                      null));
          for (FlatLists.ComparableList value : this.values) {
            for (int i = 0; i < value.size(); i++) {
              Comparable c = (Comparable) value.get(i);
              list.set(columnOrdinals[i], c);
            }
            collector.add(list);
          }
          space.collector = collector;
        }
      }
  
      @Override public void finish() {
        // number of input rows (not distinct values)
        // that were null or partially null
        space.nullCount = nullCount;
        space.cardinality = values.size() + (nullCount > 0 ? 1 : 0);
        space.valueSet = null;
      }
  
    }
  
    /** Collector that collects two or more column values into a HyperLogLog
     * sketch. */
    abstract static class HllCollector extends Collector {
      final HllSketch sketch;
      int nullCount = 0;
  
      static final long[] NULL_BITS = {0x9f77d57e93167a16L};
  
      HllCollector(Space space) {
        super(space);
        this.sketch = HllSketch.builder().build();
      }
  
      protected void add(Comparable value) {
        if (value == NullSentinel.INSTANCE) {
          sketch.update(NULL_BITS);
        } else if (value instanceof String) {
          sketch.update((String) value);
        } else if (value instanceof Double) {
          sketch.update((Double) value);
        } else if (value instanceof Float) {
          sketch.update((Float) value);
        } else if (value instanceof Long) {
          sketch.update((Long) value);
        } else if (value instanceof Number) {
          sketch.update(((Number) value).longValue());
        } else {
          sketch.update(value.toString());
        }
      }
  
      @Override public void finish() {
        space.nullCount = nullCount;
        space.cardinality = (int) sketch.getEstimate();
        space.valueSet = null;
      }
    }
  
    /** Collector that collects one column value into a HyperLogLog sketch. */
    static class HllSingletonCollector extends HllCollector {
      final int columnOrdinal;
  
      HllSingletonCollector(Space space, int columnOrdinal) {
        super(space);
        this.columnOrdinal = columnOrdinal;
      }
  
      @Override public void add(List<Comparable> row) {
        final Comparable value = row.get(columnOrdinal);
        if (value == NullSentinel.INSTANCE) {
          nullCount++;
          sketch.update(NULL_BITS);
        } else {
          add(value);
        }
      }
    }
  
    /** Collector that collects two or more column values into a HyperLogLog
     * sketch. */
    static class HllCompositeCollector extends HllCollector {
      private final int[] columnOrdinals;
      private final ByteBuffer buf = ByteBuffer.allocate(1024);
  
      HllCompositeCollector(Space space, int[] columnOrdinals) {
        super(space);
        this.columnOrdinals = columnOrdinals;
      }
  
      @Override public void add(List<Comparable> row) {
        if (space.columnOrdinals.equals(OF)) {
          Util.discard(0);
        }
        int nullCountThisRow = 0;
        buf.clear();
        for (int columnOrdinal : columnOrdinals) {
          final Comparable value = row.get(columnOrdinal);
          if (value == NullSentinel.INSTANCE) {
            if (nullCountThisRow++ == 0) {
              nullCount++;
            }
            buf.put((byte) 0);
          } else if (value instanceof String) {
            buf.put((byte) 1)
                .put(((String) value).getBytes(StandardCharsets.UTF_8));
          } else if (value instanceof Double) {
            buf.put((byte) 2).putDouble((Double) value);
          } else if (value instanceof Float) {
            buf.put((byte) 3).putFloat((Float) value);
          } else if (value instanceof Long) {
            buf.put((byte) 4).putLong((Long) value);
          } else if (value instanceof Integer) {
            buf.put((byte) 5).putInt((Integer) value);
          } else if (value instanceof Boolean) {
            buf.put((Boolean) value ? (byte) 6 : (byte) 7);
          } else {
            buf.put((byte) 8)
                .put(value.toString().getBytes(StandardCharsets.UTF_8));
          }
        }
        sketch.update(Arrays.copyOf(buf.array(), buf.position()));
      }
    }
  
    /** A priority queue of the last N surprise values. Accepts a new value if
     * the queue is not yet full, or if its value is greater than the median value
     * over the last N. */
    static class SurpriseQueue {
      private final int warmUpCount;
      private final int size;
      int count = 0;
      final Deque<Double> deque = new ArrayDeque<>();
      final PriorityQueue<Double> priorityQueue =
          new PriorityQueue<>(11, Ordering.natural());
  
      SurpriseQueue(int warmUpCount, int size) {
        this.warmUpCount = warmUpCount;
        this.size = size;
        checkArgument(warmUpCount > 3);
        checkArgument(size > 0);
      }
  
      @Override public String toString() {
        return "min: " + priorityQueue.peek()
            + ", contents: " + deque.toString();
      }
  
      boolean isValid() {
        if (CalciteSystemProperty.DEBUG.value()) {
          System.out.println(toString());
        }
        assert deque.size() == priorityQueue.size();
        if (count > size) {
          assert deque.size() == size;
        }
        return true;
      }
  
      boolean offer(double d) {
        boolean b;
        if (count++ < warmUpCount || d > castNonNull(priorityQueue.peek())) {
          if (priorityQueue.size() >= size) {
            priorityQueue.remove(deque.pop());
          }
          priorityQueue.add(d);
          deque.add(d);
          b = true;
        } else {
          b = false;
        }
        if (CalciteSystemProperty.DEBUG.value()) {
          System.out.println("offer " + d
              + " min " + priorityQueue.peek()
              + " accepted " + b);
        }
        return b;
      }
    }
  }
  