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.linq4j.Ord;
  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.PartiallyOrderedSet;
  import org.apache.calcite.util.Util;
  
  import com.google.common.collect.ImmutableSortedSet;
  import com.google.common.collect.Iterables;
  
  import org.checkerframework.checker.initialization.qual.UnknownInitialization;
  import org.checkerframework.checker.nullness.qual.Nullable;
  import org.checkerframework.checker.nullness.qual.RequiresNonNull;
  
  import java.util.ArrayList;
  import java.util.BitSet;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.NavigableSet;
  import java.util.Set;
  import java.util.SortedSet;
  import java.util.TreeSet;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * Basic implementation of {@link Profiler}.
   */
  public class SimpleProfiler implements Profiler {
  
    @Override public Profile profile(Iterable<List<Comparable>> rows,
        final List<Column> columns, Collection<ImmutableBitSet> initialGroups) {
      Util.discard(initialGroups); // this profiler ignores initial groups
      return new Run(columns).profile(rows);
    }
  
    /** Returns a measure of how much an actual value differs from expected.
     * The formula is {@code abs(expected - actual) / (expected + actual)}.
     *
     * <p>Examples:<ul>
     *   <li>surprise(e, a) is always between 0 and 1;
     *   <li>surprise(e, a) is 0 if e = a;
     *   <li>surprise(e, 0) is 1 if e &gt; 0;
     *   <li>surprise(0, a) is 1 if a &gt; 0;
     *   <li>surprise(5, 0) is 100%;
     *   <li>surprise(5, 3) is 25%;
     *   <li>surprise(5, 4) is 11%;
     *   <li>surprise(5, 5) is 0%;
     *   <li>surprise(5, 6) is 9%;
     *   <li>surprise(5, 16) is 52%;
     *   <li>surprise(5, 100) is 90%;
     * </ul>
     *
     * @param expected Expected value
     * @param actual Actual value
     * @return Measure of how much expected deviates from actual
     */
    public static double surprise(double expected, double actual) {
      if (expected == actual) {
        return 0d;
      }
      final double sum = expected + actual;
      if (sum <= 0d) {
        return 1d;
      }
      return Math.abs(expected - actual) / sum;
    }
  
    /** A run of the profiler. */
    static class Run {
      private final List<Column> columns;
      final List<Space> spaces = new ArrayList<>();
      final List<@Nullable Space> singletonSpaces;
      final List<Statistic> statistics = new ArrayList<>();
      final PartiallyOrderedSet.Ordering<Space> ordering =
          (e1, e2) -> e2.columnOrdinals.contains(e1.columnOrdinals);
      final PartiallyOrderedSet<Space> results =
          new PartiallyOrderedSet<>(ordering);
      final PartiallyOrderedSet<Space> keyResults =
          new PartiallyOrderedSet<>(ordering);
      private final List<ImmutableBitSet> keyOrdinalLists =
          new ArrayList<>();
  
      Run(final List<Column> columns) {
        for (Ord<Column> column : Ord.zip(columns)) {
          if (column.e.ordinal != column.i) {
            throw new IllegalArgumentException();
          }
        }
        this.columns = columns;
        this.singletonSpaces =
            new ArrayList<>(Collections.nCopies(columns.size(), null));
        for (ImmutableBitSet ordinals
            : ImmutableBitSet.range(columns.size()).powerSet()) {
          final Space space = new Space(ordinals, toColumns(ordinals));
          spaces.add(space);
          if (ordinals.cardinality() == 1) {
            singletonSpaces.set(ordinals.nth(0), space);
          }
        }
      }
  
      Profile profile(Iterable<List<Comparable>> rows) {
        final List<Comparable> values = new ArrayList<>();
        int rowCount = 0;
        for (final List<Comparable> row : rows) {
          ++rowCount;
        joint:
          for (Space space : spaces) {
            values.clear();
            for (Column column : space.columns) {
              final Comparable value = row.get(column.ordinal);
              values.add(value);
              if (value == NullSentinel.INSTANCE) {
                space.nullCount++;
                continue joint;
              }
            }
            space.values.add(FlatLists.ofComparable(values));
          }
        }
  
        // Populate unique keys
        // If [x, y] is a key,
        // then [x, y, z] is a key but not intersecting,
        // and [x, y] => [a] is a functional dependency but not interesting,
        // and [x, y, z] is not an interesting distribution.
        final Map<ImmutableBitSet, Distribution> distributions = new HashMap<>();
        for (Space space : spaces) {
          if (space.values.size() == rowCount
              && !containsKey(space.columnOrdinals, false)) {
            // We have discovered a new key.
            // It is not an existing key or a super-set of a key.
            statistics.add(new Unique(space.columns));
            space.unique = true;
            keyOrdinalLists.add(space.columnOrdinals);
          }
  
          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) is null for " + d);
                spaceD.dependents.add(s.columnOrdinals);
              }
            }
          }
  
          int nullCount;
          final SortedSet<Comparable> valueSet;
          if (space.columns.size() == 1) {
            nullCount = space.nullCount;
            valueSet =
                ImmutableSortedSet.copyOf(
                    Util.transform(space.values, Iterables::getOnlyElement));
          } else {
            nullCount = -1;
            valueSet = null;
          }
          double expectedCardinality;
          final double cardinality = space.cardinality();
          switch (space.columns.size()) {
          case 0:
            expectedCardinality = 1d;
            break;
          case 1:
            expectedCardinality = rowCount;
            break;
          default:
            expectedCardinality = rowCount;
            for (Column column : space.columns) {
              final Distribution d1 =
                  distributions.get(ImmutableBitSet.of(column.ordinal));
              final Distribution d2 =
                  distributions.get(space.columnOrdinals.clear(column.ordinal));
              final double d =
                  Lattice.getRowCount(rowCount,
                      requireNonNull(d1, "d1").cardinality,
                      requireNonNull(d2, "d2").cardinality);
              expectedCardinality = Math.min(expectedCardinality, d);
            }
          }
          final boolean minimal = nonMinimal == 0
              && !space.unique
              && !containsKey(space.columnOrdinals, true);
          final Distribution distribution =
              new Distribution(space.columns, valueSet, cardinality, nullCount,
                  expectedCardinality, minimal);
          statistics.add(distribution);
          distributions.put(space.columnOrdinals, distribution);
  
          if (distribution.minimal) {
            results.add(space);
          }
        }
  
        for (Space s : singletonSpaces) {
          for (ImmutableBitSet dependent : requireNonNull(s, "s").dependents) {
            if (!containsKey(dependent, false)
                && !hasNull(dependent)) {
              statistics.add(
                  new FunctionalDependency(toColumns(dependent),
                      Iterables.getOnlyElement(s.columns)));
            }
          }
        }
        return new Profile(columns, new RowCount(rowCount),
            Iterables.filter(statistics, FunctionalDependency.class),
            Iterables.filter(statistics, Distribution.class),
            Iterables.filter(statistics, Unique.class));
      }
  
      /** Returns whether a set of column ordinals
       * matches or contains a unique key.
       * If {@code strict}, it must contain a unique key. */
      private boolean containsKey(ImmutableBitSet ordinals, boolean strict) {
        for (ImmutableBitSet keyOrdinals : keyOrdinalLists) {
          if (ordinals.contains(keyOrdinals)) {
            return !(strict && keyOrdinals.equals(ordinals));
          }
        }
        return false;
      }
  
      private boolean hasNull(ImmutableBitSet columnOrdinals) {
        for (Integer columnOrdinal : columnOrdinals) {
          Space space =
              requireNonNull(singletonSpaces.get(columnOrdinal),
                  () -> "singletonSpaces.get(columnOrdinal) is null for "
                      + columnOrdinal);
          if (space.nullCount > 0) {
            return true;
          }
        }
        return false;
      }
  
      @RequiresNonNull("columns")
      private ImmutableSortedSet<Column> toColumns(
          @UnknownInitialization Run this,
          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 implements Comparable<Space> {
      final ImmutableBitSet columnOrdinals;
      final ImmutableSortedSet<Column> columns;
      int nullCount;
      final NavigableSet<FlatLists.ComparableList<Comparable>> values =
          new TreeSet<>();
      boolean unique;
      final BitSet dependencies = new BitSet();
      final Set<ImmutableBitSet> dependents = new HashSet<>();
  
      Space(ImmutableBitSet columnOrdinals, Iterable<Column> columns) {
        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);
      }
  
      @Override public int compareTo(Space o) {
        return columnOrdinals.equals(o.columnOrdinals) ? 0
            : columnOrdinals.contains(o.columnOrdinals) ? 1
                : -1;
      }
  
      /** Number of distinct values. Null is counted as a value, if present. */
      public double cardinality() {
        return values.size() + (nullCount > 0 ? 1 : 0);
      }
    }
  }
  