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.cassandra.io.sstable.metadata;
  
  import java.nio.ByteBuffer;
  import java.util.Collections;
  import java.util.EnumMap;
  import java.util.Map;
  import java.util.UUID;
  
  import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
  import com.clearspring.analytics.stream.cardinality.ICardinality;
  import org.apache.cassandra.db.Clustering;
  import org.apache.cassandra.db.ClusteringBound;
  import org.apache.cassandra.db.ClusteringBoundOrBoundary;
  import org.apache.cassandra.db.ClusteringComparator;
  import org.apache.cassandra.db.ClusteringPrefix;
  import org.apache.cassandra.db.DeletionTime;
  import org.apache.cassandra.db.LivenessInfo;
  import org.apache.cassandra.db.SerializationHeader;
  import org.apache.cassandra.db.Slice;
  import org.apache.cassandra.db.commitlog.CommitLogPosition;
  import org.apache.cassandra.db.commitlog.IntervalSet;
  import org.apache.cassandra.db.partitions.PartitionStatisticsCollector;
  import org.apache.cassandra.db.rows.Cell;
  import org.apache.cassandra.db.rows.Unfiltered;
  import org.apache.cassandra.io.sstable.SSTable;
  import org.apache.cassandra.io.sstable.format.SSTableReader;
  import org.apache.cassandra.service.ActiveRepairService;
  import org.apache.cassandra.service.StorageService;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.utils.EstimatedHistogram;
  import org.apache.cassandra.utils.FBUtilities;
  import org.apache.cassandra.utils.MurmurHash;
  import org.apache.cassandra.utils.TimeUUID;
  import org.apache.cassandra.utils.streamhist.StreamingTombstoneHistogramBuilder;
  import org.apache.cassandra.utils.streamhist.TombstoneHistogram;
  
  public class MetadataCollector implements PartitionStatisticsCollector
  {
      public static final double NO_COMPRESSION_RATIO = -1.0;
  
      private long currentPartitionCells = 0;
  
      static EstimatedHistogram defaultCellPerPartitionCountHistogram()
      {
          // EH of 118 can track a max value of 4139110981, i.e., > 4B cells
          return new EstimatedHistogram(118);
      }
  
      static EstimatedHistogram defaultPartitionSizeHistogram()
      {
          // EH of 155 can track a max value of 3520571548412 i.e. 3.5TB
          return new EstimatedHistogram(155);
  
      }
  
      static TombstoneHistogram defaultTombstoneDropTimeHistogram()
      {
          return TombstoneHistogram.createDefault();
      }
  
      public static StatsMetadata defaultStatsMetadata()
      {
          return new StatsMetadata(defaultPartitionSizeHistogram(),
                                   defaultCellPerPartitionCountHistogram(),
                                   IntervalSet.empty(),
                                   Long.MIN_VALUE,
                                   Long.MAX_VALUE,
                                   Integer.MAX_VALUE,
                                   Integer.MAX_VALUE,
                                   0,
                                   Integer.MAX_VALUE,
                                   NO_COMPRESSION_RATIO,
                                   defaultTombstoneDropTimeHistogram(),
                                   0,
                                   Collections.emptyList(),
                                   Slice.ALL,
                                   true,
                                   ActiveRepairService.UNREPAIRED_SSTABLE,
                                   -1,
                                   -1,
                                   Double.NaN,
                                   null,
                                   null,
                                   false,
                                   true,
                                   ByteBufferUtil.EMPTY_BYTE_BUFFER,
                                   ByteBufferUtil.EMPTY_BYTE_BUFFER);
      }
  
      protected EstimatedHistogram estimatedPartitionSize = defaultPartitionSizeHistogram();
      // TODO: cound the number of row per partition (either with the number of cells, or instead)
      protected EstimatedHistogram estimatedCellPerPartitionCount = defaultCellPerPartitionCountHistogram();
      protected IntervalSet<CommitLogPosition> commitLogIntervals = IntervalSet.empty();
      protected final MinMaxLongTracker timestampTracker = new MinMaxLongTracker();
      protected final MinMaxLongTracker localDeletionTimeTracker = new MinMaxLongTracker(Cell.NO_DELETION_TIME, Cell.NO_DELETION_TIME);
      protected final MinMaxIntTracker ttlTracker = new MinMaxIntTracker(Cell.NO_TTL, Cell.NO_TTL);
      protected double compressionRatio = NO_COMPRESSION_RATIO;
      protected StreamingTombstoneHistogramBuilder estimatedTombstoneDropTime = new StreamingTombstoneHistogramBuilder(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE, SSTable.TOMBSTONE_HISTOGRAM_SPOOL_SIZE, SSTable.TOMBSTONE_HISTOGRAM_TTL_ROUND_SECONDS);
      protected int sstableLevel;
  
      /**
       * The smallest clustering prefix for any {@link Unfiltered} in the sstable.
       *
       * <p>This is always either a Clustering, or a start bound (since for any end range tombstone bound, there should
       * be a corresponding start bound that is smaller).
       */
      private ClusteringPrefix<?> minClustering = ClusteringBound.MAX_START;
      /**
       * The largest clustering prefix for any {@link Unfiltered} in the sstable.
       *
       * <p>This is always either a Clustering, or an end bound (since for any start range tombstone bound, there should
       * be a corresponding end bound that is bigger).
       */
      private ClusteringPrefix<?> maxClustering = ClusteringBound.MIN_END;
  
      protected boolean hasLegacyCounterShards = false;
      private boolean hasPartitionLevelDeletions = false;
      protected long totalColumnsSet;
      protected long totalRows;
      public int totalTombstones;
  
      protected double tokenSpaceCoverage = Double.NaN;
  
      /**
       * Default cardinality estimation method is to use HyperLogLog++.
       * Parameter here(p=13, sp=25) should give reasonable estimation
       * while lowering bytes required to hold information.
       * See CASSANDRA-5906 for detail.
       */
      protected ICardinality cardinality = new HyperLogLogPlus(13, 25);
      private final ClusteringComparator comparator;
      private final long nowInSec = FBUtilities.nowInSeconds();
  
      private final UUID originatingHostId;
  
      public MetadataCollector(ClusteringComparator comparator)
      {
          this(comparator, StorageService.instance.getLocalHostUUID());
      }
  
      public MetadataCollector(ClusteringComparator comparator, UUID originatingHostId)
      {
          this.comparator = comparator;
          this.originatingHostId = originatingHostId;
      }
  
      public MetadataCollector(Iterable<SSTableReader> sstables, ClusteringComparator comparator)
      {
          this(comparator);
  
          IntervalSet.Builder<CommitLogPosition> intervals = new IntervalSet.Builder<>();
          if (originatingHostId != null)
          {
              for (SSTableReader sstable : sstables)
              {
                  if (originatingHostId.equals(sstable.getSSTableMetadata().originatingHostId))
                      intervals.addAll(sstable.getSSTableMetadata().commitLogIntervals);
              }
          }
          commitLogIntervals(intervals.build());
      }
  
      public MetadataCollector addKey(ByteBuffer key)
      {
          long hashed = MurmurHash.hash2_64(key, key.position(), key.remaining(), 0);
          cardinality.offerHashed(hashed);
          totalTombstones = 0;
          return this;
      }
  
      public MetadataCollector addPartitionSizeInBytes(long partitionSize)
      {
          estimatedPartitionSize.add(partitionSize);
          return this;
      }
  
      public MetadataCollector addCellPerPartitionCount(long cellCount)
      {
          estimatedCellPerPartitionCount.add(cellCount);
          return this;
      }
  
      public MetadataCollector addCellPerPartitionCount()
      {
          estimatedCellPerPartitionCount.add(currentPartitionCells);
          currentPartitionCells = 0;
          return this;
      }
  
      /**
       * Ratio is compressed/uncompressed and it is
       * if you have 1.x then compression isn't helping
       */
      public MetadataCollector addCompressionRatio(long compressed, long uncompressed)
      {
          compressionRatio = (double) compressed/uncompressed;
          return this;
      }
  
      public void update(LivenessInfo newInfo)
      {
          if (newInfo.isEmpty())
              return;
  
          updateTimestamp(newInfo.timestamp());
          updateTTL(newInfo.ttl());
          updateLocalDeletionTime(newInfo.localExpirationTime());
          if (!newInfo.isLive(nowInSec))
              updateTombstoneCount();
      }
  
      public void update(Cell<?> cell)
      {
          ++currentPartitionCells;
          updateTimestamp(cell.timestamp());
          updateTTL(cell.ttl());
          updateLocalDeletionTime(cell.localDeletionTime());
          if (!cell.isLive(nowInSec))
              updateTombstoneCount();
      }
  
      public void updatePartitionDeletion(DeletionTime dt)
      {
          if (!dt.isLive())
              hasPartitionLevelDeletions = true;
          update(dt);
      }
  
      public void update(DeletionTime dt)
      {
          if (!dt.isLive())
          {
              updateTimestamp(dt.markedForDeleteAt());
              updateLocalDeletionTime(dt.localDeletionTime());
              updateTombstoneCount();
          }
      }
  
      public void updateColumnSetPerRow(long columnSetInRow)
      {
          totalColumnsSet += columnSetInRow;
          ++totalRows;
      }
  
      private void updateTimestamp(long newTimestamp)
      {
          timestampTracker.update(newTimestamp);
      }
  
      private void updateLocalDeletionTime(long newLocalDeletionTime)
      {
          localDeletionTimeTracker.update(newLocalDeletionTime);
          if (newLocalDeletionTime != Cell.NO_DELETION_TIME)
              estimatedTombstoneDropTime.update(newLocalDeletionTime);
      }
  
      private void updateTombstoneCount()
      {
          ++totalTombstones;
      }
  
      private void updateTTL(int newTTL)
      {
          ttlTracker.update(newTTL);
      }
  
      public MetadataCollector commitLogIntervals(IntervalSet<CommitLogPosition> commitLogIntervals)
      {
          this.commitLogIntervals = commitLogIntervals;
          return this;
      }
  
      public MetadataCollector sstableLevel(int sstableLevel)
      {
          this.sstableLevel = sstableLevel;
          return this;
      }
  
      public MetadataCollector tokenSpaceCoverage(double coverage)
      {
          tokenSpaceCoverage = coverage;
          return this;
      }
  
      public void updateClusteringValues(Clustering<?> clustering)
      {
          if (clustering == Clustering.STATIC_CLUSTERING)
              return;
  
          // In case of monotonically growing stream of clusterings, we will usually require only one comparison
          // because if we detected X is greater than the current MAX, then it cannot be lower than the current MIN
          // at the same time. The only case when we need to update MIN when the current MAX was detected to be updated
          // is the case when MIN was not yet initialized and still point the ClusteringBound.MAX_START
          if (comparator.compare(clustering, maxClustering) > 0)
          {
              maxClustering = clustering;
              if (minClustering == ClusteringBound.MAX_START)
                  minClustering = clustering;
          }
          else if (comparator.compare(clustering, minClustering) < 0)
          {
              minClustering = clustering;
          }
      }
  
      public void updateClusteringValuesByBoundOrBoundary(ClusteringBoundOrBoundary<?> clusteringBoundOrBoundary)
      {
          // In a SSTable, every opening marker will be closed, so the start of a range tombstone marker will never be
          // the maxClustering (the corresponding close might though) and there is no point in doing the comparison
          // (and vice-versa for the close). By the same reasoning, a boundary will never be either the min or max
          // clustering, and we can save on comparisons.
          if (clusteringBoundOrBoundary.isBoundary())
              return;
  
          // see the comment in updateClusteringValues(Clustering)
          if (comparator.compare(clusteringBoundOrBoundary, maxClustering) > 0)
          {
              if (clusteringBoundOrBoundary.kind().isEnd())
                  maxClustering = clusteringBoundOrBoundary;
  
              // note that since we excluded boundaries above, there is no way that the provided clustering prefix is
              // a start and en end at the same time
              else if (minClustering == ClusteringBound.MAX_START)
                  minClustering = clusteringBoundOrBoundary;
          }
          else if (comparator.compare(clusteringBoundOrBoundary, minClustering) < 0)
          {
              if (clusteringBoundOrBoundary.kind().isStart())
                  minClustering = clusteringBoundOrBoundary;
              else if (maxClustering == ClusteringBound.MIN_END)
                  maxClustering = clusteringBoundOrBoundary;
          }
      }
  
      public void updateHasLegacyCounterShards(boolean hasLegacyCounterShards)
      {
          this.hasLegacyCounterShards = this.hasLegacyCounterShards || hasLegacyCounterShards;
      }
  
      public Map<MetadataType, MetadataComponent> finalizeMetadata(String partitioner, double bloomFilterFPChance, long repairedAt, TimeUUID pendingRepair, boolean isTransient, SerializationHeader header, ByteBuffer firstKey, ByteBuffer lastKey)
      {
          assert minClustering.kind() == ClusteringPrefix.Kind.CLUSTERING || minClustering.kind().isStart();
          assert maxClustering.kind() == ClusteringPrefix.Kind.CLUSTERING || maxClustering.kind().isEnd();
  
          Map<MetadataType, MetadataComponent> components = new EnumMap<>(MetadataType.class);
          components.put(MetadataType.VALIDATION, new ValidationMetadata(partitioner, bloomFilterFPChance));
          components.put(MetadataType.STATS, new StatsMetadata(estimatedPartitionSize,
                                                               estimatedCellPerPartitionCount,
                                                               commitLogIntervals,
                                                               timestampTracker.min(),
                                                               timestampTracker.max(),
                                                               localDeletionTimeTracker.min(),
                                                               localDeletionTimeTracker.max(),
                                                               ttlTracker.min(),
                                                               ttlTracker.max(),
                                                               compressionRatio,
                                                               estimatedTombstoneDropTime.build(),
                                                               sstableLevel,
                                                               comparator.subtypes(),
                                                               Slice.make(minClustering.retainable().asStartBound(), maxClustering.retainable().asEndBound()),
                                                               hasLegacyCounterShards,
                                                               repairedAt,
                                                               totalColumnsSet,
                                                               totalRows,
                                                               tokenSpaceCoverage,
                                                               originatingHostId,
                                                               pendingRepair,
                                                               isTransient,
                                                               hasPartitionLevelDeletions,
                                                               firstKey,
                                                               lastKey));
          components.put(MetadataType.COMPACTION, new CompactionMetadata(cardinality));
          components.put(MetadataType.HEADER, header.toComponent());
          return components;
      }
  
      /**
       * Release large memory objects while keeping metrics intact
       */
      public void release()
      {
          estimatedTombstoneDropTime.releaseBuffers();
      }
  
      public static class MinMaxLongTracker
      {
          private final long defaultMin;
          private final long defaultMax;
  
          private boolean isSet = false;
          private long min;
          private long max;
  
          public MinMaxLongTracker()
          {
              this(Long.MIN_VALUE, Long.MAX_VALUE);
          }
  
          public MinMaxLongTracker(long defaultMin, long defaultMax)
          {
              this.defaultMin = defaultMin;
              this.defaultMax = defaultMax;
          }
  
          public void update(long value)
          {
              if (!isSet)
              {
                  min = max = value;
                  isSet = true;
              }
              else
              {
                  if (value < min)
                      min = value;
                  if (value > max)
                      max = value;
              }
          }
  
          public long min()
          {
              return isSet ? min : defaultMin;
          }
  
          public long max()
          {
              return isSet ? max : defaultMax;
          }
      }
  
      public static class MinMaxIntTracker
      {
          private final int defaultMin;
          private final int defaultMax;
  
          private boolean isSet = false;
          private int min;
          private int max;
  
          public MinMaxIntTracker()
          {
              this(Integer.MIN_VALUE, Integer.MAX_VALUE);
          }
  
          public MinMaxIntTracker(int defaultMin, int defaultMax)
          {
              this.defaultMin = defaultMin;
              this.defaultMax = defaultMax;
          }
  
          public void update(int value)
          {
              if (!isSet)
              {
                  min = max = value;
                  isSet = true;
              }
              else
              {
                  if (value < min)
                      min = value;
                  if (value > max)
                      max = value;
              }
          }
  
          public int min()
          {
              return isSet ? min : defaultMin;
          }
  
          public int max()
          {
              return isSet ? max : defaultMax;
          }
      }
  }