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.format;
  
  
  import java.io.IOException;
  import java.lang.ref.WeakReference;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.ScheduledFuture;
  import java.util.concurrent.ScheduledThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Preconditions;
  import com.google.common.collect.Iterables;
  import com.google.common.collect.Ordering;
  import com.google.common.primitives.Longs;
  import com.google.common.util.concurrent.RateLimiter;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.clearspring.analytics.stream.cardinality.CardinalityMergeException;
  import com.clearspring.analytics.stream.cardinality.ICardinality;
  import org.apache.cassandra.concurrent.ExecutorPlus;
  import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
  import org.apache.cassandra.concurrent.ScheduledExecutors;
  import org.apache.cassandra.config.CassandraRelevantProperties;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.PartitionPosition;
  import org.apache.cassandra.db.SerializationHeader;
  import org.apache.cassandra.db.SystemKeyspace;
  import org.apache.cassandra.db.rows.Cell;
  import org.apache.cassandra.db.rows.EncodingStats;
  import org.apache.cassandra.db.rows.UnfilteredRowIterator;
  import org.apache.cassandra.db.rows.UnfilteredSource;
  import org.apache.cassandra.dht.AbstractBounds;
  import org.apache.cassandra.dht.Bounds;
  import org.apache.cassandra.dht.Range;
  import org.apache.cassandra.dht.Token;
  import org.apache.cassandra.io.FSError;
  import org.apache.cassandra.io.FSWriteError;
  import org.apache.cassandra.io.compress.CompressionMetadata;
  import org.apache.cassandra.io.sstable.AbstractRowIndexEntry;
  import org.apache.cassandra.io.sstable.Component;
  import org.apache.cassandra.io.sstable.CorruptSSTableException;
  import org.apache.cassandra.io.sstable.Descriptor;
  import org.apache.cassandra.io.sstable.ISSTableScanner;
  import org.apache.cassandra.io.sstable.IVerifier;
  import org.apache.cassandra.io.sstable.KeyIterator;
  import org.apache.cassandra.io.sstable.KeyReader;
  import org.apache.cassandra.io.sstable.SSTable;
  import org.apache.cassandra.io.sstable.SSTableIdFactory;
  import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
  import org.apache.cassandra.io.sstable.SSTableReadsListener;
  import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
  import org.apache.cassandra.io.sstable.metadata.CompactionMetadata;
  import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
  import org.apache.cassandra.io.util.ChannelProxy;
  import org.apache.cassandra.io.util.CheckedFunction;
  import org.apache.cassandra.io.util.DataIntegrityMetadata;
  import org.apache.cassandra.io.util.File;
  import org.apache.cassandra.io.util.FileDataInput;
  import org.apache.cassandra.io.util.FileHandle;
  import org.apache.cassandra.io.util.FileUtils;
  import org.apache.cassandra.io.util.FileUtils.DuplicateHardlinkException;
  import org.apache.cassandra.io.util.RandomAccessReader;
  import org.apache.cassandra.metrics.RestorableMeter;
  import org.apache.cassandra.schema.SchemaConstants;
  import org.apache.cassandra.schema.TableMetadataRef;
  import org.apache.cassandra.service.ActiveRepairService;
  import org.apache.cassandra.utils.EstimatedHistogram;
  import org.apache.cassandra.utils.ExecutorUtils;
  import org.apache.cassandra.utils.FBUtilities;
  import org.apache.cassandra.utils.Interval;
  import org.apache.cassandra.utils.JVMStabilityInspector;
  import org.apache.cassandra.utils.NativeLibrary;
  import org.apache.cassandra.utils.OutputHandler;
  import org.apache.cassandra.utils.Throwables;
  import org.apache.cassandra.utils.TimeUUID;
  import org.apache.cassandra.utils.concurrent.OpOrder;
  import org.apache.cassandra.utils.concurrent.Ref;
  import org.apache.cassandra.utils.concurrent.SelfRefCounted;
  import org.apache.cassandra.utils.concurrent.SharedCloseable;
  import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
  
  import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
  import static org.apache.cassandra.utils.TimeUUID.unixMicrosToRawTimestamp;
  import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
  import static org.apache.cassandra.utils.concurrent.SharedCloseable.sharedCopyOrNull;
  
  /**
   * An SSTableReader can be constructed in a number of places, but typically is either read from disk at startup, or
   * constructed from a flushed memtable, or after compaction to replace some existing sstables. However once created,
   * an sstablereader may also be modified.
   * <p>
   * A reader's {@link OpenReason} describes its current stage in its lifecycle. Note that in parallel to this, there are
   * two different Descriptor types; TMPLINK and FINAL; the latter corresponds to {@link OpenReason#NORMAL} state readers
   * and all readers that replace a {@link OpenReason#NORMAL} one. TMPLINK is used for {@link OpenReason#EARLY} state
   * readers and no others.
   * <p>
   * When a reader is being compacted, if the result is large its replacement may be opened as {@link OpenReason#EARLY}
   * before compaction completes in order to present the result to consumers earlier. In this case the reader will itself
   * be changed to a {@link OpenReason#MOVED_START} state, where its start no longer represents its on-disk minimum key.
   * This is to permit reads to be directed to only one reader when the two represent the same data.
   * The {@link OpenReason#EARLY} file can represent a compaction result that is either partially complete and still
   * in-progress, or a complete and immutable sstable that is part of a larger macro compaction action that has not yet
   * fully completed.
   * <p>
   * Currently ALL compaction results at least briefly go through an {@link OpenReason#EARLY} open state prior to completion,
   * regardless of if early opening is enabled.
   * <p>
   * Since a reader can be created multiple times over the same shared underlying resources, and the exact resources it
   * shares between each instance differ subtly, we track the lifetime of any underlying resource with its own reference
   * count, which each instance takes a {@link Ref} to. Each instance then tracks references to itself, and once these
   * all expire it releases all its {@link Ref} to these underlying resources.
   * <p>
   * There is some shared cleanup behaviour needed only once all readers in a certain stage of their lifecycle
   * (i.e. {@link OpenReason#EARLY} or {@link OpenReason#NORMAL} opening), and some that must only occur once all readers
   * of any kind over a single logical sstable have expired. These are managed by the {@link InstanceTidier} and
   * {@link GlobalTidy} classes at the bottom, and are effectively managed as another resource each instance tracks its
   * own {@link Ref} instance to, to ensure all of these resources are cleaned up safely and can be debugged otherwise.
   * <p>
   * TODO: fill in details about Tracker and lifecycle interactions for tools, and for compaction strategies
   */
  public abstract class SSTableReader extends SSTable implements UnfilteredSource, SelfRefCounted<SSTableReader>, Comparable<SSTableReader>
  {
      private static final Logger logger = LoggerFactory.getLogger(SSTableReader.class);
  
      private static final boolean TRACK_ACTIVITY = CassandraRelevantProperties.DISABLE_SSTABLE_ACTIVITY_TRACKING.getBoolean();
  
      private static final ScheduledExecutorPlus syncExecutor = initSyncExecutor();
  
      private static ScheduledExecutorPlus initSyncExecutor()
      {
          if (DatabaseDescriptor.isClientOrToolInitialized())
              return null;
  
          // Do NOT start this thread pool in client mode
  
          ScheduledExecutorPlus syncExecutor = executorFactory().scheduled("read-hotness-tracker");
          // Immediately remove readMeter sync task when cancelled.
          // TODO: should we set this by default on all scheduled executors?
          if (syncExecutor instanceof ScheduledThreadPoolExecutor)
              ((ScheduledThreadPoolExecutor) syncExecutor).setRemoveOnCancelPolicy(true);
          return syncExecutor;
      }
  
      private static final RateLimiter meterSyncThrottle = RateLimiter.create(100.0);
  
      public static final Comparator<SSTableReader> maxTimestampAscending = Comparator.comparingLong(SSTableReader::getMaxTimestamp);
      public static final Comparator<SSTableReader> maxTimestampDescending = maxTimestampAscending.reversed();
  
      public static final TimeUUID.Generator.Factory<UniqueIdentifier> UNIQUE_IDENTIFIER_FACTORY = new TimeUUID.Generator.Factory<UniqueIdentifier>()
      {
          @Override
          public UniqueIdentifier atUnixMicrosWithLsb(long unixMicros, long clockSeqAndNode)
          {
              return new UniqueIdentifier(unixMicrosToRawTimestamp(unixMicros), clockSeqAndNode);
          }
      };
  
      // it's just an object, which we use regular Object equality on; we introduce a special class just for easy recognition
      // Also includes a TimeUUID to make these sortable
      public static final class UniqueIdentifier extends TimeUUID
      {
          private UniqueIdentifier(long unixMicros, long clockSeqAndNode)
          {
              super(unixMicros, clockSeqAndNode);
          }
      }
  
      public final UniqueIdentifier instanceId = TimeUUID.Generator.nextTimeUUID(UNIQUE_IDENTIFIER_FACTORY);
  
      public static final Comparator<SSTableReader> firstKeyComparator = (o1, o2) -> o1.getFirst().compareTo(o2.getFirst());
      public static final Ordering<SSTableReader> firstKeyOrdering = Ordering.from(firstKeyComparator);
      public static final Comparator<SSTableReader> lastKeyComparator = (o1, o2) -> o1.getLast().compareTo(o2.getLast());
  
      public static final Comparator<SSTableReader> idComparator = Comparator.comparing(t -> t.descriptor.id, SSTableIdFactory.COMPARATOR);
      public static final Comparator<SSTableReader> idReverseComparator = idComparator.reversed();
  
      public static final Comparator<SSTableReader> sizeComparator = (o1, o2) -> Longs.compare(o1.onDiskLength(), o2.onDiskLength());
  
      /**
       * maxDataAge is a timestamp in local server time (e.g. Global.currentTimeMilli) which represents an upper bound
       * to the newest piece of data stored in the sstable. In other words, this sstable does not contain items created
       * later than maxDataAge.
       * <p>
       * The field is not serialized to disk, so relying on it for more than what truncate does is not advised.
       * <p>
       * When a new sstable is flushed, maxDataAge is set to the time of creation.
       * When a sstable is created from compaction, maxDataAge is set to max of all merged sstables.
       * <p>
       * The age is in milliseconds since epoc and is local to this host.
       */
      public final long maxDataAge;
  
      public enum OpenReason
      {
          /**
           * <ul>
           * <li>From {@code None} - Reader has been read from disk, either at startup or from a flushed memtable</li>
           * <li>From {@link #EARLY} - Reader is the final result of a compaction</li>
           * <li>From {@link #MOVED_START} - Reader WAS being compacted, but this failed and it has been restored
           *     to {code NORMAL}status</li>
           * </ul>
           */
          NORMAL,
  
          /**
           * <ul>
           * <li>From {@code None} - Reader is a compaction replacement that is either incomplete and has been opened
           *     to represent its partial result status, or has been finished but the compaction it is a part of has not
           *     yet completed fully</li>
           * <li>From {@link #EARLY} - Same as from {@code None}, only it is not the first time it has been
           * </ul>
           */
          EARLY,
  
          /**
           * From:
           * <ul>
           * <li>From {@link #NORMAL} - Reader has seen low traffic and the amount of memory available for index summaries
           *     is constrained, so its index summary has been downsampled</li>
           * <li>From {@link #METADATA_CHANGE} - Same
           * </ul>
           */
          METADATA_CHANGE,
  
          /**
           * <ul>
           * <li>From {@link #NORMAL} - Reader is being compacted. This compaction has not finished, but the compaction
           *     result is either partially or fully opened, to either partially or fully replace this reader. This reader's
           *     start key has been updated to represent this, so that reads only hit one or the other reader.</li>
           * </ul>
           */
          MOVED_START
      }
  
      public final OpenReason openReason;
  
      protected final FileHandle dfile;
  
      // technically isCompacted is not necessary since it should never be unreferenced unless it is also compacted,
      // but it seems like a good extra layer of protection against reference counting bugs to not delete data based on that alone
      public final AtomicBoolean isSuspect = new AtomicBoolean(false);
  
      // not final since we need to be able to change level on a file.
      protected volatile StatsMetadata sstableMetadata;
  
      public final SerializationHeader header;
  
      private final InstanceTidier tidy;
      private final Ref<SSTableReader> selfRef;
  
      private RestorableMeter readMeter;
  
      private volatile double crcCheckChance;
  
      protected final DecoratedKey first;
      protected final DecoratedKey last;
      public final AbstractBounds<Token> bounds;
      private final Interval<PartitionPosition, SSTableReader> interval;
  
      /**
       * Calculate approximate key count.
       * If cardinality estimator is available on all given sstables, then this method use them to estimate
       * key count.
       * If not, then this uses index summaries.
       *
       * @param sstables SSTables to calculate key count
       * @return estimated key count
       */
      public static long getApproximateKeyCount(Iterable<SSTableReader> sstables)
      {
          long count = -1;
  
          if (Iterables.isEmpty(sstables))
              return count;
  
          boolean failed = false;
          ICardinality cardinality = null;
          for (SSTableReader sstable : sstables)
          {
              if (sstable.openReason == OpenReason.EARLY)
                  continue;
  
              try
              {
                  CompactionMetadata metadata = StatsComponent.load(sstable.descriptor).compactionMetadata();
                  // If we can't load the CompactionMetadata, we are forced to estimate the keys using the index
                  // summary. (CASSANDRA-10676)
                  if (metadata == null)
                  {
                      logger.warn("Reading cardinality from Statistics.db failed for {}", sstable.getFilename());
                      failed = true;
                      break;
                  }
  
                  if (cardinality == null)
                      cardinality = metadata.cardinalityEstimator;
                  else
                      cardinality = cardinality.merge(metadata.cardinalityEstimator);
              }
              catch (IOException e)
              {
                  logger.warn("Reading cardinality from Statistics.db failed.", e);
                  failed = true;
                  break;
              }
              catch (CardinalityMergeException e)
              {
                  logger.warn("Cardinality merge failed.", e);
                  failed = true;
                  break;
              }
          }
          if (cardinality != null && !failed)
              count = cardinality.cardinality();
  
          // if something went wrong above or cardinality is not available, calculate using index summary
          if (count < 0)
          {
              count = 0;
              for (SSTableReader sstable : sstables)
                  count += sstable.estimatedKeys();
          }
          return count;
      }
  
      public static SSTableReader open(SSTable.Owner owner, Descriptor descriptor)
      {
          return open(owner, descriptor, null);
      }
  
      public static SSTableReader open(SSTable.Owner owner, Descriptor desc, TableMetadataRef metadata)
      {
          return open(owner, desc,  null, metadata);
      }
  
      public static SSTableReader open(SSTable.Owner owner, Descriptor descriptor, Set<Component> components, TableMetadataRef metadata)
      {
          return open(owner, descriptor, components, metadata, true, false);
      }
  
      // use only for offline or "Standalone" operations
      public static SSTableReader openNoValidation(Descriptor descriptor, Set<Component> components, ColumnFamilyStore cfs)
      {
          return open(cfs, descriptor, components, cfs.metadata, false, true);
      }
  
      // use only for offline or "Standalone" operations
      public static SSTableReader openNoValidation(SSTable.Owner owner, Descriptor descriptor, TableMetadataRef metadata)
      {
          return open(owner, descriptor, null, metadata, false, true);
      }
  
      /**
       * Open SSTable reader to be used in batch mode(such as sstableloader).
       */
      public static SSTableReader openForBatch(SSTable.Owner owner, Descriptor descriptor, Set<Component> components, TableMetadataRef metadata)
      {
          return open(owner, descriptor, components, metadata, true, true);
      }
  
      /**
       * Open an SSTable for reading
       *
       * @param owner      owning entity
       * @param descriptor SSTable to open
       * @param components Components included with this SSTable
       * @param metadata   for this SSTables CF
       * @param validate   Check SSTable for corruption (limited)
       * @param isOffline  Whether we are opening this SSTable "offline", for example from an external tool or not for inclusion in queries (validations)
       *                   This stops regenerating BF + Summaries and also disables tracking of hotness for the SSTable.
       * @return {@link SSTableReader}
       */
      public static SSTableReader open(Owner owner,
                                       Descriptor descriptor,
                                       Set<Component> components,
                                       TableMetadataRef metadata,
                                       boolean validate,
                                       boolean isOffline)
      {
          SSTableReaderLoadingBuilder<?, ?> builder = descriptor.getFormat().getReaderFactory().loadingBuilder(descriptor, metadata, components);
          return builder.build(owner, validate, !isOffline);
      }
  
      public static Collection<SSTableReader> openAll(SSTable.Owner owner, Set<Map.Entry<Descriptor, Set<Component>>> entries,
                                                      final TableMetadataRef metadata)
      {
          final Collection<SSTableReader> sstables = newBlockingQueue();
  
          ExecutorPlus executor = executorFactory().pooled("SSTableBatchOpen", FBUtilities.getAvailableProcessors());
          try
          {
              for (final Map.Entry<Descriptor, Set<Component>> entry : entries)
              {
                  Runnable runnable = () -> {
                      SSTableReader sstable;
                      try
                      {
                          sstable = open(owner, entry.getKey(), entry.getValue(), metadata);
                      }
                      catch (CorruptSSTableException ex)
                      {
                          JVMStabilityInspector.inspectThrowable(ex);
                          logger.error("Corrupt sstable {}; skipping table", entry, ex);
                          return;
                      }
                      catch (FSError ex)
                      {
                          JVMStabilityInspector.inspectThrowable(ex);
                          logger.error("Cannot read sstable {}; file system error, skipping table", entry, ex);
                          return;
                      }
                      sstables.add(sstable);
                  };
                  executor.submit(runnable);
              }
          }
          finally
          {
              executor.shutdown();
          }
  
          try
          {
              executor.awaitTermination(7, TimeUnit.DAYS);
          }
          catch (InterruptedException e)
          {
              throw new UncheckedInterruptedException(e);
          }
  
          return sstables;
      }
  
      protected SSTableReader(Builder<?, ?> builder, Owner owner)
      {
          super(builder, owner);
  
          this.sstableMetadata = builder.getStatsMetadata();
          this.header = builder.getSerializationHeader();
          this.dfile = builder.getDataFile();
          this.maxDataAge = builder.getMaxDataAge();
          this.openReason = builder.getOpenReason();
          this.first = builder.getFirst();
          this.last = builder.getLast();
          this.interval = first == null || last == null ? null : Interval.create(first, last, this);
          this.bounds = first == null || last == null || AbstractBounds.strictlyWrapsAround(first.getToken(), last.getToken())
                        ? null // this will cause the validation to fail, but the reader is opened with no validation,
                               // e.g. for scrubbing, we should accept screwed bounds
                        : AbstractBounds.bounds(first.getToken(), true, last.getToken(), true);
  
          tidy = new InstanceTidier(descriptor, owner);
          selfRef = new Ref<>(this, tidy);
      }
  
      @Override
      public DecoratedKey getFirst()
      {
          return first;
      }
  
      @Override
      public DecoratedKey getLast()
      {
          return last;
      }
  
      public Interval<PartitionPosition, SSTableReader> getInterval()
      {
          return interval;
      }
  
      @Override
      public AbstractBounds<Token> getBounds()
      {
          return Objects.requireNonNull(bounds, "Bounds were not created because the sstable is out of order");
      }
  
      public DataIntegrityMetadata.ChecksumValidator maybeGetChecksumValidator() throws IOException
      {
          if (descriptor.fileFor(Components.CRC).exists())
              return new DataIntegrityMetadata.ChecksumValidator(descriptor.fileFor(Components.DATA), descriptor.fileFor(Components.CRC));
          else
              return null;
      }
  
      public DataIntegrityMetadata.FileDigestValidator maybeGetDigestValidator() throws IOException
      {
          if (descriptor.fileFor(Components.DIGEST).exists())
              return new DataIntegrityMetadata.FileDigestValidator(descriptor.fileFor(Components.DATA), descriptor.fileFor(Components.DIGEST));
          else
              return null;
      }
  
      public static long getTotalBytes(Iterable<SSTableReader> sstables)
      {
          long sum = 0;
          for (SSTableReader sstable : sstables)
              sum += sstable.onDiskLength();
          return sum;
      }
  
      public static long getTotalUncompressedBytes(Iterable<SSTableReader> sstables)
      {
          long sum = 0;
          for (SSTableReader sstable : sstables)
              sum += sstable.uncompressedLength();
  
          return sum;
      }
  
      public boolean equals(Object that)
      {
          return that instanceof SSTableReader && ((SSTableReader) that).descriptor.equals(this.descriptor);
      }
  
      public int hashCode()
      {
          return this.descriptor.hashCode();
      }
  
      public String getFilename()
      {
          return dfile.path();
      }
  
      public void setupOnline()
      {
          owner().ifPresent(o -> setCrcCheckChance(o.getCrcCheckChance()));
      }
  
      /**
       * Execute provided task with sstable lock to avoid racing with index summary redistribution, SEE CASSANDRA-15861.
       *
       * @param task to be guarded by sstable lock
       */
      public <R, E extends Exception> R runWithLock(CheckedFunction<Descriptor, R, E> task) throws E
      {
          synchronized (tidy.global)
          {
              return task.apply(descriptor);
          }
      }
  
      public void setReplaced()
      {
          synchronized (tidy.global)
          {
              assert !tidy.isReplaced;
              tidy.isReplaced = true;
          }
      }
  
      public boolean isReplaced()
      {
          synchronized (tidy.global)
          {
              return tidy.isReplaced;
          }
      }
  
      /**
       * The runnable passed to this method must not be an anonymous or non-static inner class. It can be a lambda or a
       * method reference provided that it does not retain a reference chain to this reader.
       */
      public void runOnClose(final Runnable runOnClose)
      {
          if (runOnClose == null)
              return;
  
          synchronized (tidy.global)
          {
              final Runnable existing = tidy.runOnClose;
              if (existing == null)
                  tidy.runOnClose = runOnClose;
              else
                  tidy.runOnClose = () -> {
                      existing.run();
                      runOnClose.run();
                  };
          }
      }
  
      /**
       * The method sets fields specific to this {@link SSTableReader} and the parent {@link SSTable} on the provided
       * {@link Builder}. The method is intended to be called from the overloaded {@code unbuildTo} method in subclasses.
       *
       * @param builder    the builder on which the fields should be set
       * @param sharedCopy whether the {@link SharedCloseable} resources should be passed as shared copies or directly;
       *                   note that the method will overwrite the fields representing {@link SharedCloseable} only if
       *                   they are not set in the builder yet (the relevant fields in the builder are {@code null}).
       * @return the same instance of builder as provided
       */
      protected final <B extends Builder<?, B>> B unbuildTo(B builder, boolean sharedCopy)
      {
          B b = super.unbuildTo(builder, sharedCopy);
          if (builder.getDataFile() == null)
              b.setDataFile(sharedCopy ? sharedCopyOrNull(dfile) : dfile);
  
          b.setStatsMetadata(sstableMetadata);
          b.setSerializationHeader(header);
          b.setMaxDataAge(maxDataAge);
          b.setOpenReason(openReason);
          b.setFirst(first);
          b.setLast(last);
          b.setSuspected(isSuspect.get());
          return b;
      }
  
      public abstract SSTableReader cloneWithRestoredStart(DecoratedKey restoredStart);
  
      public abstract SSTableReader cloneWithNewStart(DecoratedKey newStart);
  
      public RestorableMeter getReadMeter()
      {
          return readMeter;
      }
  
      /**
       * All the resources which should be released upon closing this sstable reader are registered with in
       * {@link GlobalTidy}. This method lets close a provided resource explicitly any time and unregister it from
       * {@link GlobalTidy} so that it is not tried to be released twice.
       *
       * @param closeable a resource to be closed
       */
      protected void closeInternalComponent(AutoCloseable closeable)
      {
          synchronized (tidy.global)
          {
              boolean removed = tidy.closeables.remove(closeable);
              Preconditions.checkState(removed);
              try
              {
                  closeable.close();
              }
              catch (Exception ex)
              {
                  throw new RuntimeException("Failed to close " + closeable, ex);
              }
          }
      }
  
      /**
       * This method is expected to close the components which occupy memory but are not needed when we just want to
       * stream the components (for example, when SSTable is opened with SSTableLoader). The method should call
       * {@link #closeInternalComponent(AutoCloseable)} for each such component. Leaving the implementation empty is
       * valid given there are not such resources to release.
       */
      public abstract void releaseInMemoryComponents();
  
      /**
       * Perform any validation needed for the reader upon creation before returning it from the {@link Builder}.
       */
      public void validate()
      {
          if (this.first.compareTo(this.last) > 0 || bounds == null)
          {
              throw new CorruptSSTableException(new IllegalStateException(String.format("SSTable first key %s > last key %s", this.first, this.last)), getFilename());
          }
      }
  
      /**
       * Returns the compression metadata for this sstable. Note that the compression metdata is a resource and should not
       * be closed by the caller.
       * TODO do not return a closeable resource or return a shared copy
       *
       * @throws IllegalStateException if the sstable is not compressed
       */
      public CompressionMetadata getCompressionMetadata()
      {
          if (!compression)
              throw new IllegalStateException(this + " is not compressed");
  
          return dfile.compressionMetadata().get();
      }
  
      /**
       * Returns the amount of memory in bytes used off heap by the compression meta-data.
       *
       * @return the amount of memory in bytes used off heap by the compression meta-data
       */
      public long getCompressionMetadataOffHeapSize()
      {
          if (!compression)
              return 0;
  
          return getCompressionMetadata().offHeapSize();
      }
  
      /**
       * Calculates an estimate of the number of keys in the sstable represented by this reader.
       */
      public abstract long estimatedKeys();
  
      /**
       * Calculates an estimate of the number of keys for the given ranges in the sstable represented by this reader.
       */
      public abstract long estimatedKeysForRanges(Collection<Range<Token>> ranges);
  
      /**
       * Returns whether methods like {@link #estimatedKeys()} or {@link #estimatedKeysForRanges(Collection)} can return
       * sensible estimations.
       */
      public abstract boolean isEstimationInformative();
  
      /**
       * Returns sample keys for the provided token range.
       */
      public abstract Iterable<DecoratedKey> getKeySamples(final Range<Token> range);
  
      /**
       * Determine the minimal set of sections that can be extracted from this SSTable to cover the given ranges.
       *
       * @return A sorted list of [offset,end) pairs that cover the given ranges in the datafile for this SSTable.
       */
      public List<PartitionPositionBounds> getPositionsForRanges(Collection<Range<Token>> ranges)
      {
          // use the index to determine a minimal section for each range
          List<PartitionPositionBounds> positions = new ArrayList<>();
          for (Range<Token> range : Range.normalize(ranges))
          {
              assert !range.isWrapAround() || range.right.isMinimum();
              AbstractBounds<PartitionPosition> bounds = Range.makeRowRange(range);
              PartitionPositionBounds pb = getPositionsForBounds(bounds);
              if (pb != null)
                  positions.add(pb);
          }
          return positions;
      }
  
      /**
       * Get a list of data positions in this SSTable that correspond to the given list of bounds. This method will remove
       * non-covered intervals, but will not correct order or overlap in the supplied list, e.g. if bounds overlap, the
       * result will be sections of the data file that repeat the same positions.
       *
       * @return A sorted list of [offset,end) pairs corresponding to the given boundsList in the datafile for this
       *         SSTable.
       */
      public List<PartitionPositionBounds> getPositionsForBoundsIterator(Iterator<AbstractBounds<PartitionPosition>> boundsList)
      {
          // use the index to determine a minimal section for each range
          List<PartitionPositionBounds> positions = new ArrayList<>();
          while (boundsList.hasNext())
          {
              AbstractBounds<PartitionPosition> bounds = boundsList.next();
              PartitionPositionBounds pb = getPositionsForBounds(bounds);
              if (pb != null)
                  positions.add(pb);
          }
          return positions;
      }
  
      /**
       * Determine the data positions in this SSTable that cover the given bounds.
       *
       * @return An [offset,end) pair that cover the given bounds in the datafile for this SSTable, or null if the range
       *         is not covered by the sstable or is empty.
       */
      public PartitionPositionBounds getPositionsForBounds(AbstractBounds<PartitionPosition> bounds)
      {
          long left = getPosition(bounds.left, bounds.inclusiveLeft() ? Operator.GE : Operator.GT);
          // Note: getPosition will apply a moved start if the sstable is in MOVED_START state.
          if (left < 0) // empty range
              return null;
  
          long right = bounds.right.isMinimum() ? -1
                                                : getPosition(bounds.right, bounds.inclusiveRight() ? Operator.GT
                                                                                                    : Operator.GE);
          if (right < 0) // right is beyond end
              right = uncompressedLength();   // this should also be correct for EARLY readers
  
          if (left >= right) // empty range
              return null;
  
          return new PartitionPositionBounds(left, right);
      }
  
      /**
       * Return an [offset,end) pair that covers the whole file. This could be null if the sstable's moved start has
       * made the sstable effectively empty.
       */
      public PartitionPositionBounds getPositionsForFullRange()
      {
          if (openReason != OpenReason.MOVED_START)
              return new PartitionPositionBounds(0, uncompressedLength());
          else
          {
              // query a full range, so that the required adjustments can be applied
              PartitionPosition minToken = getPartitioner().getMinimumToken().minKeyBound();
              return getPositionsForBounds(new Range<>(minToken, minToken));
          }
      }
  
      /**
       * Calculate a total on-disk (compressed) size for the given partition positions. For uncompressed files this is
       * equal to the sum of the size of the covered ranges. For compressed files this is the sum of the size of the
       * chunks that contain the requested ranges and may be significantly bigger than the size of the requested ranges.
       *
       * @param positionBounds a list of [offset,end) pairs that specify the relevant sections of the data file; this must
       *                       be non-overlapping and in ascending order.
       */
      public long onDiskSizeForPartitionPositions(Collection<PartitionPositionBounds> positionBounds)
      {
          long total = 0;
          if (!compression)
          {
              for (PartitionPositionBounds position : positionBounds)
                  total += position.upperPosition - position.lowerPosition;
          }
          else
          {
              final CompressionMetadata compressionMetadata = getCompressionMetadata();
              long lastEnd = 0;
              for (PartitionPositionBounds position : positionBounds)
              {
                  // The end of the chunk that contains the last required byte from the range.
                  long upperChunkEnd = compressionMetadata.chunkFor(position.upperPosition - 1).chunkEnd();
                  // The start of the chunk that contains the first required byte from the range.
                  long lowerChunkStart = compressionMetadata.chunkFor(position.lowerPosition).offset;
                  if (lowerChunkStart < lastEnd)  // if regions include the same chunk, count it only once
                      lowerChunkStart = lastEnd;
                  total += upperChunkEnd - lowerChunkStart;
                  lastEnd = upperChunkEnd;
              }
          }
          return total;
      }
  
      /**
       * Retrieves the position while updating the key cache and the stats.
       *
       * @param key The key to apply as the rhs to the given Operator. A 'fake' key is allowed to
       *            allow key selection by token bounds but only if op != * EQ
       * @param op  The Operator defining matching keys: the nearest key to the target matching the operator wins.
       */
      public final long getPosition(PartitionPosition key, Operator op)
      {
          return getPosition(key, op, SSTableReadsListener.NOOP_LISTENER);
      }
  
      public final long getPosition(PartitionPosition key, Operator op, SSTableReadsListener listener)
      {
          return getPosition(key, op, true, listener);
      }
  
      public final long getPosition(PartitionPosition key,
                                    Operator op,
                                    boolean updateStats)
      {
          return getPosition(key, op, updateStats, SSTableReadsListener.NOOP_LISTENER);
      }
  
      /**
       * Retrieve a position in data file according to the provided key and operator.
       *
       * @param key         The key to apply as the rhs to the given Operator. A 'fake' key is allowed to
       *                    allow key selection by token bounds but only if op != * EQ
       * @param op          The Operator defining matching keys: the nearest key to the target matching the operator wins.
       * @param updateStats true if updating stats and cache
       * @param listener    a listener used to handle internal events
       * @return The index entry corresponding to the key, or null if the key is not present
       */
      protected long getPosition(PartitionPosition key,
                                 Operator op,
                                 boolean updateStats,
                                 SSTableReadsListener listener)
      {
          AbstractRowIndexEntry rie = getRowIndexEntry(key, op, updateStats, listener);
          return rie != null ? rie.position : -1;
      }
  
      /**
       * Retrieve an index entry for the partition found according to the provided key and operator.
       *
       * @param key         The key to apply as the rhs to the given Operator. A 'fake' key is allowed to
       *                    allow key selection by token bounds but only if op != * EQ
       * @param op          The Operator defining matching keys: the nearest key to the target matching the operator wins.
       * @param updateStats true if updating stats and cache
       * @param listener    a listener used to handle internal events
       * @return The index entry corresponding to the key, or null if the key is not present
       */
      @VisibleForTesting
      protected abstract AbstractRowIndexEntry getRowIndexEntry(PartitionPosition key,
                                                                Operator op,
                                                                boolean updateStats,
                                                                SSTableReadsListener listener);
  
      public UnfilteredRowIterator simpleIterator(FileDataInput file, DecoratedKey key, long dataPosition, boolean tombstoneOnly)
      {
          return SSTableIdentityIterator.create(this, file, dataPosition, key, tombstoneOnly);
      }
  
      /**
       * Returns a {@link KeyReader} over all keys in the sstable.
       */
      public abstract KeyReader keyReader() throws IOException;
  
      /**
       * Returns a {@link KeyReader} over all keys in the sstable after a given key.
       * @param key
       * @return
       * @throws IOException
       */
      public abstract KeyReader keyReader(PartitionPosition key) throws IOException;
  
      /**
       * Returns a {@link KeyIterator} over all keys in the sstable.
       */
      public KeyIterator keyIterator() throws IOException
      {
          return new KeyIterator(null, keyReader(), getPartitioner(), uncompressedLength(), new ReentrantReadWriteLock());
      }
  
      public abstract KeyIterator keyIterator(AbstractBounds<PartitionPosition> range) throws IOException;
  
      /**
       * Finds and returns the first key beyond a given token in this SSTable or null if no such key exists.
       */
      public abstract DecoratedKey firstKeyBeyond(PartitionPosition token);
  
      /**
       * Returns the length in bytes of the (uncompressed) data for this SSTable. For compressed files, this is not
       * the same thing as the on disk size (see {@link #onDiskLength()}).
       */
      public long uncompressedLength()
      {
          return dfile.dataLength();
      }
  
      /**
       * @return the fraction of the token space for which this sstable has content. In the simplest case this is just the
       * size of the interval returned by {@link #getBounds()}, but the sstable may contain "holes" when the locally-owned
       * range is not contiguous (e.g. with vnodes).
       * As this is affected by the local ranges which can change, the token space fraction is calculated at the time of
       * writing the sstable and stored with its metadata.
       * For older sstables that do not contain this metadata field, this method returns NaN.
       */
      public double tokenSpaceCoverage()
      {
          return sstableMetadata.tokenSpaceCoverage;
      }
  
      /**
       * The length in bytes of the on disk size for this SSTable. For compressed files, this is not the same thing
       * as the data length (see {@link #uncompressedLength()}).
       */
      public long onDiskLength()
      {
          return dfile.onDiskLength;
      }
  
      @VisibleForTesting
      public double getCrcCheckChance()
      {
          return crcCheckChance;
      }
  
      /**
       * Set the value of CRC check chance. The argument supplied is obtained from the property of the owning CFS.
       * Called when either the SSTR is initialized, or the CFS's property is updated via JMX
       */
      public void setCrcCheckChance(double crcCheckChance)
      {
          this.crcCheckChance = crcCheckChance;
      }
  
      /**
       * Mark the sstable as obsolete, i.e., compacted into newer sstables.
       * <p>
       * When calling this function, the caller must ensure that the SSTableReader is not referenced anywhere except for
       * threads holding a reference.
       * <p>
       * Calling it multiple times is usually buggy.
       */
      public void markObsolete(Runnable tidier)
      {
          if (logger.isTraceEnabled())
              logger.trace("Marking {} compacted", getFilename());
  
          synchronized (tidy.global)
          {
              assert !tidy.isReplaced;
              assert tidy.global.obsoletion == null : this + " was already marked compacted";
  
              tidy.global.obsoletion = tidier;
              tidy.global.stopReadMeterPersistence();
          }
      }
  
      public boolean isMarkedCompacted()
      {
          return tidy.global.obsoletion != null;
      }
  
      public void markSuspect()
      {
          if (logger.isTraceEnabled())
              logger.trace("Marking {} as a suspect to be excluded from reads.", getFilename());
  
          isSuspect.getAndSet(true);
      }
  
      @VisibleForTesting
      public void unmarkSuspect()
      {
          isSuspect.getAndSet(false);
      }
  
      public boolean isMarkedSuspect()
      {
          return isSuspect.get();
      }
  
      /**
       * Direct I/O SSTableScanner over a defined range of tokens.
       *
       * @param range the range of keys to cover
       * @return A Scanner for seeking over the rows of the SSTable.
       */
      public ISSTableScanner getScanner(Range<Token> range)
      {
          if (range == null)
              return getScanner();
          else
              return getScanner(Collections.singletonList(range));
      }
  
      /**
       * Direct I/O SSTableScanner over the entirety of the sstable.
       *
       * @return A Scanner over the full content of the SSTable.
       */
      public ISSTableScanner getScanner()
      {
          PartitionPositionBounds fullRange = getPositionsForFullRange();
          if (fullRange != null)
              return new SSTableSimpleScanner(this, Collections.singletonList(fullRange));
          else
              return new SSTableSimpleScanner(this, Collections.emptyList());
      }
  
      /**
       * Direct I/O SSTableScanner over a defined collection of ranges of tokens.
       *
       * @param ranges the range of keys to cover
       * @return A Scanner for seeking over the rows of the SSTable.
       */
      public ISSTableScanner getScanner(Collection<Range<Token>> ranges)
      {
          if (ranges != null)
              return new SSTableSimpleScanner(this, getPositionsForRanges(ranges));
          else
              return getScanner();
      }
  
      /**
       * Direct I/O SSTableScanner over an iterator of bounds.
       *
       * @param boundsIterator the keys to cover
       * @return A Scanner for seeking over the rows of the SSTable.
       */
      public ISSTableScanner getScanner(Iterator<AbstractBounds<PartitionPosition>> boundsIterator)
      {
          return new SSTableSimpleScanner(this, getPositionsForBoundsIterator(boundsIterator));
      }
  
      public ISSTableScanner getScanner(AbstractBounds<PartitionPosition> bounds)
      {
          PartitionPositionBounds positionBounds = getPositionsForBounds(bounds);
          return new SSTableSimpleScanner(this, positionBounds == null ? Collections.emptyList() : Collections.singletonList(positionBounds));
      }
  
  
      /**
       * Create a {@link FileDataInput} for the data file of the sstable represented by this reader. This method returns
       * a newly opened resource which must be closed by the caller.
       *
       * @param position the data input will be opened and seek to this position
       */
      public FileDataInput getFileDataInput(long position)
      {
          return dfile.createReader(position);
      }
  
      /**
       * Tests if the sstable contains data newer than the given age param (in localhost currentMillis time).
       * This works in conjunction with maxDataAge which is an upper bound on the data in the sstable represented
       * by this reader.
       *
       * @return {@code true} iff this sstable contains data that's newer than the given timestamp
       */
      public boolean newSince(long timestampMillis)
      {
          return maxDataAge > timestampMillis;
      }
  
      public void createLinks(String snapshotDirectoryPath)
      {
          createLinks(snapshotDirectoryPath, null);
      }
  
      public void createLinks(String snapshotDirectoryPath, RateLimiter rateLimiter)
      {
          createLinks(snapshotDirectoryPath, rateLimiter, false);
      }
  
      public void createLinks(String snapshotDirectoryPath, RateLimiter rateLimiter, boolean ephemeralSnapshot)
      {
          createLinks(descriptor, components, snapshotDirectoryPath, rateLimiter, ephemeralSnapshot);
      }
  
      public static void createLinks(Descriptor descriptor, Set<Component> components, String snapshotDirectoryPath)
      {
          createLinks(descriptor, components, snapshotDirectoryPath, null, false);
      }
  
      /**
       * Create hardlinks for given set of components
       *
       * @param descriptor descriptor to use
       * @param components components to create links for
       * @param snapshotDirectoryPath directory path for snapshot
       * @param limiter rate limiter to use
       * @param force if true, if target link file exists, do not fail, otherwise throw RTE
       */
      public static void createLinks(Descriptor descriptor, Set<Component> components, String snapshotDirectoryPath, RateLimiter limiter, boolean force)
      {
          for (Component component : components)
          {
              File sourceFile = descriptor.fileFor(component);
              if (!sourceFile.exists())
                  continue;
              if (null != limiter)
                  limiter.acquire();
              File targetLink = new File(snapshotDirectoryPath, sourceFile.name());
              try
              {
                  FileUtils.createHardLink(sourceFile, targetLink);
              }
              catch (DuplicateHardlinkException ex)
              {
                  if (!force)
                      throw new RuntimeException(ex.getMessage());
              }
          }
      }
  
      public boolean isRepaired()
      {
          return sstableMetadata.repairedAt != ActiveRepairService.UNREPAIRED_SSTABLE;
      }
  
      /**
       * Reads the key stored at the position saved in SASI.
       * <p>
       * When SASI is created, it uses key locations retrieved from {@link KeyReader#keyPositionForSecondaryIndex()}.
       * This method is to read the key stored at such position. It is up to the concrete SSTable format implementation
       * what that position means and which file it refers. The only requirement is that it is consistent with what
       * {@link KeyReader#keyPositionForSecondaryIndex()} returns.
       *
       * @return key if found, {@code null} otherwise
       */
      public abstract DecoratedKey keyAtPositionFromSecondaryIndex(long keyPositionFromSecondaryIndex) throws IOException;
  
      public boolean isPendingRepair()
      {
          return sstableMetadata.pendingRepair != ActiveRepairService.NO_PENDING_REPAIR;
      }
  
      public TimeUUID getPendingRepair()
      {
          return sstableMetadata.pendingRepair;
      }
  
      public long getRepairedAt()
      {
          return sstableMetadata.repairedAt;
      }
  
      public boolean isTransient()
      {
          return sstableMetadata.isTransient;
      }
  
      public boolean intersects(Collection<Range<Token>> ranges)
      {
          Bounds<Token> range = new Bounds<>(first.getToken(), last.getToken());
          return Iterables.any(ranges, r -> r.intersects(range));
      }
  
      /**
       * TODO: Move someplace reusable
       */
      public abstract static class Operator
      {
          public static final Operator EQ = new Equals();
          public static final Operator GE = new GreaterThanOrEqualTo();
          public static final Operator GT = new GreaterThan();
  
          /**
           * @param comparison The result of a call to compare/compareTo, with the desired field on the rhs.
           * @return less than 0 if the operator cannot match forward, 0 if it matches, greater than 0 if it might match forward.
           */
          public abstract int apply(int comparison);
  
          final static class Equals extends Operator
          {
              public int apply(int comparison)
              {
                  return -comparison;
              }
          }
  
          final static class GreaterThanOrEqualTo extends Operator
          {
              public int apply(int comparison)
              {
                  return comparison >= 0 ? 0 : 1;
              }
          }
  
          final static class GreaterThan extends Operator
          {
              public int apply(int comparison)
              {
                  return comparison > 0 ? 0 : 1;
              }
          }
      }
  
      public EstimatedHistogram getEstimatedPartitionSize()
      {
          return sstableMetadata.estimatedPartitionSize;
      }
  
      public EstimatedHistogram getEstimatedCellPerPartitionCount()
      {
          return sstableMetadata.estimatedCellPerPartitionCount;
      }
  
      public double getEstimatedDroppableTombstoneRatio(long gcBefore)
      {
          return sstableMetadata.getEstimatedDroppableTombstoneRatio(gcBefore);
      }
  
      public double getDroppableTombstonesBefore(long gcBefore)
      {
          return sstableMetadata.getDroppableTombstonesBefore(gcBefore);
      }
  
      public double getCompressionRatio()
      {
          return sstableMetadata.compressionRatio;
      }
  
      public long getMinTimestamp()
      {
          return sstableMetadata.minTimestamp;
      }
  
      public long getMaxTimestamp()
      {
          return sstableMetadata.maxTimestamp;
      }
  
      public long getMinLocalDeletionTime()
      {
          return sstableMetadata.minLocalDeletionTime;
      }
  
      public long getMaxLocalDeletionTime()
      {
          return sstableMetadata.maxLocalDeletionTime;
      }
  
      /**
       * Whether the sstable may contain tombstones or if it is guaranteed to not contain any.
       * <p>
       * Note that having that method return {@code false} guarantees the sstable has no tombstones whatsoever (so no
       * cell tombstone, no range tombstone maker and no expiring columns), but having it return {@code true} doesn't
       * guarantee it contains any as it may simply have non-expired cells.
       */
      public boolean mayHaveTombstones()
      {
          // A sstable is guaranteed to have no tombstones if minLocalDeletionTime is still set to its default,
          // Cell.NO_DELETION_TIME, which is bigger than any valid deletion times.
          return getMinLocalDeletionTime() != Cell.NO_DELETION_TIME;
      }
  
      public int getMinTTL()
      {
          return sstableMetadata.minTTL;
      }
  
      public int getMaxTTL()
      {
          return sstableMetadata.maxTTL;
      }
  
      public long getTotalColumnsSet()
      {
          return sstableMetadata.totalColumnsSet;
      }
  
      public long getTotalRows()
      {
          return sstableMetadata.totalRows;
      }
  
      public int getAvgColumnSetPerRow()
      {
          return sstableMetadata.totalRows < 0
                 ? -1
                 : (sstableMetadata.totalRows == 0 ? 0 : (int) (sstableMetadata.totalColumnsSet / sstableMetadata.totalRows));
      }
  
      public int getSSTableLevel()
      {
          return sstableMetadata.sstableLevel;
      }
  
      /**
       * Mutate sstable level with a lock to avoid racing with entire-sstable-streaming and then reload sstable metadata
       */
      public void mutateLevelAndReload(int newLevel) throws IOException
      {
          synchronized (tidy.global)
          {
              descriptor.getMetadataSerializer().mutateLevel(descriptor, newLevel);
              reloadSSTableMetadata();
          }
      }
  
      /**
       * Mutate sstable repair metadata with a lock to avoid racing with entire-sstable-streaming and then reload sstable metadata
       */
      public void mutateRepairedAndReload(long newRepairedAt, TimeUUID newPendingRepair, boolean isTransient) throws IOException
      {
          synchronized (tidy.global)
          {
              descriptor.getMetadataSerializer().mutateRepairMetadata(descriptor, newRepairedAt, newPendingRepair, isTransient);
              reloadSSTableMetadata();
          }
      }
  
      /**
       * Reloads the sstable metadata from disk.
       * <p>
       * Called after level is changed on sstable, for example if the sstable is dropped to L0
       * <p>
       * Might be possible to remove in future versions
       *
       * @throws IOException
       */
      public void reloadSSTableMetadata() throws IOException
      {
          this.sstableMetadata = StatsComponent.load(descriptor).statsMetadata();
      }
  
      public StatsMetadata getSSTableMetadata()
      {
          return sstableMetadata;
      }
  
      public RandomAccessReader openDataReader(RateLimiter limiter)
      {
          assert limiter != null;
          return dfile.createReader(limiter);
      }
  
      public RandomAccessReader openDataReader()
      {
          return dfile.createReader();
      }
  
      public RandomAccessReader openDataReaderForScan()
      {
          return dfile.createReaderForScan();
      }
  
      public void trySkipFileCacheBefore(DecoratedKey key)
      {
          long position = getPosition(key, SSTableReader.Operator.GE);
          NativeLibrary.trySkipCache(descriptor.fileFor(Components.DATA).absolutePath(), 0, position < 0 ? 0 : position);
      }
  
      public ChannelProxy getDataChannel()
      {
          return dfile.channel;
      }
  
      /**
       * @return last modified time for data component. 0 if given component does not exist or IO error occurs.
       */
      public long getDataCreationTime()
      {
          return descriptor.fileFor(Components.DATA).lastModified();
      }
  
      /**
       * Increment the total read count and read rate for this SSTable.  This should not be incremented for non-query reads,
       * like compaction.
       */
      public void incrementReadCount()
      {
          if (readMeter != null)
              readMeter.mark();
      }
  
      public EncodingStats stats()
      {
          // We could return sstable.header.stats(), but this may not be as accurate than the actual sstable stats (see
          // SerializationHeader.make() for details) so we use the latter instead.
          return sstableMetadata.encodingStats;
      }
  
      public Ref<SSTableReader> tryRef()
      {
          return selfRef.tryRef();
      }
  
      public Ref<SSTableReader> selfRef()
      {
          return selfRef;
      }
  
      public Ref<SSTableReader> ref()
      {
          return selfRef.ref();
      }
  
      protected List<AutoCloseable> setupInstance(boolean trackHotness)
      {
          return Collections.singletonList(dfile);
      }
  
      public void setup(boolean trackHotness)
      {
          assert tidy.closeables == null;
          trackHotness &= TRACK_ACTIVITY;
          tidy.setup(this, trackHotness, setupInstance(trackHotness));
          this.readMeter = tidy.global.readMeter;
      }
  
      @VisibleForTesting
      public void overrideReadMeter(RestorableMeter readMeter)
      {
          this.readMeter = tidy.global.readMeter = readMeter;
      }
  
      public void addTo(Ref.IdentityCollection identities)
      {
          identities.add(this);
          identities.add(tidy.globalRef);
          tidy.closeables.forEach(c -> {
              if (c instanceof SharedCloseable)
                  ((SharedCloseable) c).addTo(identities);
          });
      }
  
      /**
       * The method verifies whether the sstable may contain the provided key. The method does approximation using
       * Bloom filter if it is present and if it is not, performs accurate check in the index.
       */
      public abstract boolean mayContainAssumingKeyIsInRange(DecoratedKey key);
  
      /**
       * One instance per SSTableReader we create.
       * <p>
       * We can create many InstanceTidiers (one for every time we reopen an sstable with MOVED_START for example),
       * but there can only be one GlobalTidy for one single logical sstable.
       * <p>
       * When the InstanceTidier cleansup, it releases its reference to its GlobalTidy; when all InstanceTidiers
       * for that type have run, the GlobalTidy cleans up.
       */
      protected static final class InstanceTidier implements Tidy
      {
          private final Descriptor descriptor;
          private final WeakReference<Owner> owner;
  
          private List<? extends AutoCloseable> closeables;
          private Runnable runOnClose;
  
          private boolean isReplaced = false;
  
          // a reference to our shared tidy instance, that
          // we will release when we are ourselves released
          private Ref<GlobalTidy> globalRef;
          private GlobalTidy global;
  
          private volatile boolean setup;
  
          public void setup(SSTableReader reader, boolean trackHotness, Collection<? extends AutoCloseable> closeables)
          {
              // get a new reference to the shared descriptor-type tidy
              this.globalRef = GlobalTidy.get(reader);
              this.global = globalRef.get();
              if (trackHotness)
                  global.ensureReadMeter();
              this.closeables = new ArrayList<>(closeables);
              // to avoid tidy seeing partial state, set setup=true at the end
              this.setup = true;
          }
  
          private InstanceTidier(Descriptor descriptor, Owner owner)
          {
              this.descriptor = descriptor;
              this.owner = new WeakReference<>(owner);
          }
  
          @Override
          public void tidy()
          {
              logger.trace("Running instance tidier for {} with setup {}", descriptor, setup);
  
              // don't try to cleanup if the sstablereader was never fully constructed
              if (!setup)
                  return;
  
              final OpOrder.Barrier barrier;
              Owner owner = this.owner.get();
              if (owner != null)
              {
                  barrier = owner.newReadOrderingBarrier();
                  barrier.issue();
              }
              else
              {
                  barrier = null;
              }
  
              ScheduledExecutors.nonPeriodicTasks.execute(new Runnable()
              {
                  public void run()
                  {
                      logger.trace("Async instance tidier for {}, before barrier", descriptor);
  
                      if (barrier != null)
                          barrier.await();
  
                      logger.trace("Async instance tidier for {}, after barrier", descriptor);
  
                      Throwable exceptions = null;
                      if (runOnClose != null) try
                      {
                          runOnClose.run();
                      }
                      catch (RuntimeException | Error ex)
                      {
                          logger.error("Failed to run on-close listeners for sstable " + descriptor.baseFile(), ex);
                          exceptions = ex;
                      }
  
                      Throwable closeExceptions = Throwables.close(null, Iterables.filter(closeables, Objects::nonNull));
                      if (closeExceptions != null)
                      {
                          logger.error("Failed to close some sstable components of " + descriptor.baseFile(), closeExceptions);
                          exceptions = Throwables.merge(exceptions, closeExceptions);
                      }
  
                      try
                      {
                          globalRef.release();
                      }
                      catch (RuntimeException | Error ex)
                      {
                          logger.error("Failed to release the global ref of " + descriptor.baseFile(), ex);
                          exceptions = Throwables.merge(exceptions, ex);
                      }
  
                      if (exceptions != null)
                          JVMStabilityInspector.inspectThrowable(exceptions);
  
                      logger.trace("Async instance tidier for {}, completed", descriptor);
                  }
  
                  @Override
                  public String toString()
                  {
                      return "Tidy " + descriptor.ksname + '.' + descriptor.cfname + '-' + descriptor.id;
                  }
              });
          }
  
          @Override
          public String name()
          {
              return descriptor.toString();
          }
      }
  
      /**
       * One instance per logical sstable. This both tracks shared cleanup and some shared state related
       * to the sstable's lifecycle.
       * <p>
       * All InstanceTidiers, on setup(), ask the static get() method for their shared state,
       * and stash a reference to it to be released when they are. Once all such references are
       * released, this shared tidy will be performed.
       */
      static final class GlobalTidy implements Tidy
      {
          static final WeakReference<ScheduledFuture<?>> NULL = new WeakReference<>(null);
          // keyed by descriptor, mapping to the shared GlobalTidy for that descriptor
          static final ConcurrentMap<Descriptor, Ref<GlobalTidy>> lookup = new ConcurrentHashMap<>();
  
          private final Descriptor desc;
          // the readMeter that is shared between all instances of the sstable, and can be overridden in all of them
          // at once also, for testing purposes
          private RestorableMeter readMeter;
          // the scheduled persistence of the readMeter, that we will cancel once all instances of this logical
          // sstable have been released
          private WeakReference<ScheduledFuture<?>> readMeterSyncFuture = NULL;
          // shared state managing if the logical sstable has been compacted; this is used in cleanup
          private volatile Runnable obsoletion;
  
          GlobalTidy(final SSTableReader reader)
          {
              this.desc = reader.descriptor;
          }
  
          void ensureReadMeter()
          {
              if (readMeter != null)
                  return;
  
              // Don't track read rates for tables in the system keyspace and don't bother trying to load or persist
              // the read meter when in client mode.
              // Also, do not track read rates when running in client or tools mode (syncExecuter isn't available in these modes)
              if (!TRACK_ACTIVITY || SchemaConstants.isLocalSystemKeyspace(desc.ksname) || DatabaseDescriptor.isClientOrToolInitialized())
              {
                  readMeter = null;
                  readMeterSyncFuture = NULL;
                  return;
              }
  
              readMeter = SystemKeyspace.getSSTableReadMeter(desc.ksname, desc.cfname, desc.id);
              // sync the average read rate to system.sstable_activity every five minutes, starting one minute from now
              readMeterSyncFuture = new WeakReference<>(syncExecutor.scheduleAtFixedRate(this::maybePersistSSTableReadMeter, 1, 5, TimeUnit.MINUTES));
          }
  
          void maybePersistSSTableReadMeter()
          {
              if (obsoletion == null && DatabaseDescriptor.getSStableReadRatePersistenceEnabled())
              {
                  meterSyncThrottle.acquire();
                  SystemKeyspace.persistSSTableReadMeter(desc.ksname, desc.cfname, desc.id, readMeter);
              }
          }
  
          private void stopReadMeterPersistence()
          {
              ScheduledFuture<?> readMeterSyncFutureLocal = readMeterSyncFuture.get();
              if (readMeterSyncFutureLocal != null)
              {
                  readMeterSyncFutureLocal.cancel(true);
                  readMeterSyncFuture = NULL;
              }
          }
  
          public void tidy()
          {
              lookup.remove(desc);
  
              if (obsoletion != null)
                  obsoletion.run();
  
              // don't ideally want to dropPageCache for the file until all instances have been released
              for (Component c : desc.discoverComponents())
                  NativeLibrary.trySkipCache(desc.fileFor(c).absolutePath(), 0, 0);
          }
  
          public String name()
          {
              return desc.toString();
          }
  
          // get a new reference to the shared GlobalTidy for this sstable
          public static Ref<GlobalTidy> get(SSTableReader sstable)
          {
              Descriptor descriptor = sstable.descriptor;
  
              while (true)
              {
                  Ref<GlobalTidy> ref = lookup.get(descriptor);
                  if (ref == null)
                  {
                      final GlobalTidy tidy = new GlobalTidy(sstable);
                      ref = new Ref<>(tidy, tidy);
                      Ref<GlobalTidy> ex = lookup.putIfAbsent(descriptor, ref);
                      if (ex == null)
                          return ref;
                      ref = ex;
                  }
  
                  Ref<GlobalTidy> newRef = ref.tryRef();
                  if (newRef != null)
                      return newRef;
  
                  // raced with tidy
                  lookup.remove(descriptor, ref);
              }
          }
      }
  
      @VisibleForTesting
      public static void resetTidying()
      {
          GlobalTidy.lookup.clear();
      }
  
      public static class PartitionPositionBounds
      {
          public final long lowerPosition;
          public final long upperPosition;
  
          public PartitionPositionBounds(long lower, long upper)
          {
              this.lowerPosition = lower;
              this.upperPosition = upper;
          }
  
          @Override
          public final int hashCode()
          {
              int hashCode = (int) lowerPosition ^ (int) (lowerPosition >>> 32);
              return 31 * (hashCode ^ (int) ((int) upperPosition ^ (upperPosition >>> 32)));
          }
  
          @Override
          public final boolean equals(Object o)
          {
              if (!(o instanceof PartitionPositionBounds))
                  return false;
              PartitionPositionBounds that = (PartitionPositionBounds) o;
              return lowerPosition == that.lowerPosition && upperPosition == that.upperPosition;
          }
      }
  
      public static class IndexesBounds
      {
          public final int lowerPosition;
          public final int upperPosition;
  
          public IndexesBounds(int lower, int upper)
          {
              this.lowerPosition = lower;
              this.upperPosition = upper;
          }
  
          @Override
          public final int hashCode()
          {
              return 31 * lowerPosition * upperPosition;
          }
  
          @Override
          public final boolean equals(Object o)
          {
              if (!(o instanceof IndexesBounds))
                  return false;
              IndexesBounds that = (IndexesBounds) o;
              return lowerPosition == that.lowerPosition && upperPosition == that.upperPosition;
          }
      }
  
      /**
       * Moves the sstable in oldDescriptor to a new place (with generation etc) in newDescriptor.
       * <p>
       * All components given will be moved/renamed
       */
      public static SSTableReader moveAndOpenSSTable(ColumnFamilyStore cfs, Descriptor oldDescriptor, Descriptor newDescriptor, Set<Component> components, boolean copyData)
      {
          if (!oldDescriptor.isCompatible())
              throw new RuntimeException(String.format("Can't open incompatible SSTable! Current version %s, found file: %s",
                                                       oldDescriptor.getFormat().getLatestVersion(),
                                                       oldDescriptor));
  
          boolean isLive = cfs.getLiveSSTables().stream().anyMatch(r -> r.descriptor.equals(newDescriptor)
                                                                        || r.descriptor.equals(oldDescriptor));
          if (isLive)
          {
              String message = String.format("Can't move and open a file that is already in use in the table %s -> %s", oldDescriptor, newDescriptor);
              logger.error(message);
              throw new RuntimeException(message);
          }
          if (newDescriptor.fileFor(Components.DATA).exists())
          {
              String msg = String.format("File %s already exists, can't move the file there", newDescriptor.fileFor(Components.DATA));
              logger.error(msg);
              throw new RuntimeException(msg);
          }
  
          if (copyData)
          {
              try
              {
                  logger.info("Hardlinking new SSTable {} to {}", oldDescriptor, newDescriptor);
                  hardlink(oldDescriptor, newDescriptor, components);
              }
              catch (FSWriteError ex)
              {
                  logger.warn("Unable to hardlink new SSTable {} to {}, falling back to copying", oldDescriptor, newDescriptor, ex);
                  copy(oldDescriptor, newDescriptor, components);
              }
          }
          else
          {
              logger.info("Moving new SSTable {} to {}", oldDescriptor, newDescriptor);
              rename(oldDescriptor, newDescriptor, components);
          }
  
          SSTableReader reader;
          try
          {
              reader = open(cfs, newDescriptor, components, cfs.metadata);
          }
          catch (Throwable t)
          {
              logger.error("Aborting import of sstables. {} was corrupt", newDescriptor);
              throw new RuntimeException(newDescriptor + " is corrupt, can't import", t);
          }
          return reader;
      }
  
      public static void shutdownBlocking(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
      {
  
          ExecutorUtils.shutdownNowAndWait(timeout, unit, syncExecutor);
          resetTidying();
      }
  
      /**
       * @return the physical size on disk of all components for this SSTable in bytes
       */
      public long bytesOnDisk()
      {
          return bytesOnDisk(false);
      }
  
      /**
       * @return the total logical/uncompressed size in bytes of all components for this SSTable
       */
      public long logicalBytesOnDisk()
      {
          return bytesOnDisk(true);
      }
  
      private long bytesOnDisk(boolean logical)
      {
          long bytes = 0;
          for (Component component : components)
          {
              // Only the data file is compressable.
              bytes += logical && component == Components.DATA && compression
                       ? getCompressionMetadata().dataLength
                       : descriptor.fileFor(component).length();
          }
          return bytes;
      }
  
      @VisibleForTesting
      public void maybePersistSSTableReadMeter()
      {
          tidy.global.maybePersistSSTableReadMeter();
      }
  
      /**
       * Returns a new verifier for this sstable. Note that the reader must match the provided cfs.
       */
      public abstract IVerifier getVerifier(ColumnFamilyStore cfs,
                                            OutputHandler outputHandler,
                                            boolean isOffline,
                                            IVerifier.Options options);
  
      public UniqueIdentifier instanceId()
      {
          return instanceId;
      }
  
      @Override
      public int compareTo(SSTableReader other)
      {
          // Used in IntervalTree with the expectation that compareTo uniquely identifies an SSTableReader
          // Use accessor for instanceId for mocks
          return instanceId().compareTo(other.instanceId());
      }
  
      /**
       * A method to be called by {@link #getPosition(PartitionPosition, Operator, boolean, SSTableReadsListener)}
       * and {@link #getRowIndexEntry(PartitionPosition, Operator, boolean, SSTableReadsListener)} methods when
       * a searched key is found. It adds a trace message and notify the provided listener.
       */
      protected void notifySelected(SSTableReadsListener.SelectionReason reason, SSTableReadsListener localListener, Operator op, boolean updateStats, AbstractRowIndexEntry entry)
      {
          reason.trace(descriptor, entry);
  
          if (localListener != null)
              localListener.onSSTableSelected(this, reason);
      }
  
      /**
       * A method to be called by {@link #getPosition(PartitionPosition, Operator, boolean, SSTableReadsListener)}
       * and {@link #getRowIndexEntry(PartitionPosition, Operator, boolean, SSTableReadsListener)} methods when
       * a searched key is not found. It adds a trace message and notify the provided listener.
       */
      protected void notifySkipped(SSTableReadsListener.SkippingReason reason, SSTableReadsListener localListener, Operator op, boolean updateStats)
      {
          reason.trace(descriptor);
  
          if (localListener != null)
              localListener.onSSTableSkipped(this, reason);
      }
  
      /**
       * A builder of this sstable reader. It should be extended for each implementation of {@link SSTableReader} with
       * the implementation specific fields.
       *
       * @param <R> type of the reader the builder creates
       * @param <B> type of this builder
       */
      public abstract static class Builder<R extends SSTableReader, B extends Builder<R, B>> extends SSTable.Builder<R, B>
      {
          private long maxDataAge;
          private StatsMetadata statsMetadata;
          private OpenReason openReason;
          private SerializationHeader serializationHeader;
          private FileHandle dataFile;
          private DecoratedKey first;
          private DecoratedKey last;
          private boolean suspected;
  
          public Builder(Descriptor descriptor)
          {
              super(descriptor);
          }
  
          public B setMaxDataAge(long maxDataAge)
          {
              Preconditions.checkArgument(maxDataAge >= 0);
              this.maxDataAge = maxDataAge;
              return (B) this;
          }
  
          public B setStatsMetadata(StatsMetadata statsMetadata)
          {
              Preconditions.checkNotNull(statsMetadata);
              this.statsMetadata = statsMetadata;
              return (B) this;
          }
  
          public B setOpenReason(OpenReason openReason)
          {
              Preconditions.checkNotNull(openReason);
              this.openReason = openReason;
              return (B) this;
          }
  
          public B setSerializationHeader(SerializationHeader serializationHeader)
          {
              this.serializationHeader = serializationHeader;
              return (B) this;
          }
  
          public B setDataFile(FileHandle dataFile)
          {
              this.dataFile = dataFile;
              return (B) this;
          }
  
          public B setFirst(DecoratedKey first)
          {
              this.first = first != null ? first.retainable() : null;
              return (B) this;
          }
  
          public B setLast(DecoratedKey last)
          {
              this.last = last != null ? last.retainable() : null;
              return (B) this;
          }
  
          public B setSuspected(boolean suspected)
          {
              this.suspected = suspected;
              return (B) this;
          }
  
          public long getMaxDataAge()
          {
              return maxDataAge;
          }
  
          public StatsMetadata getStatsMetadata()
          {
              return statsMetadata;
          }
  
          public OpenReason getOpenReason()
          {
              return openReason;
          }
  
          public SerializationHeader getSerializationHeader()
          {
              return serializationHeader;
          }
  
          public FileHandle getDataFile()
          {
              return dataFile;
          }
  
          public DecoratedKey getFirst()
          {
              return first;
          }
  
          public DecoratedKey getLast()
          {
              return last;
          }
  
          public boolean isSuspected()
          {
              return suspected;
          }
  
          protected abstract R buildInternal(Owner owner);
  
          public R build(Owner owner, boolean validate, boolean online)
          {
              R reader = buildInternal(owner);
  
              try
              {
                  if (isSuspected())
                      reader.markSuspect();
  
                  reader.setup(online);
  
                  if (validate)
                      reader.validate();
              }
              catch (RuntimeException | Error ex)
              {
                  JVMStabilityInspector.inspectThrowable(ex);
                  reader.selfRef().release();
                  throw ex;
              }
  
              return reader;
          }
      }
  }
  