001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.regionserver;
019
020import com.google.errorprone.annotations.RestrictedApi;
021import java.io.IOException;
022import java.io.InterruptedIOException;
023import java.net.InetSocketAddress;
024import java.util.ArrayList;
025import java.util.Collection;
026import java.util.Collections;
027import java.util.HashMap;
028import java.util.HashSet;
029import java.util.Iterator;
030import java.util.List;
031import java.util.Map;
032import java.util.Map.Entry;
033import java.util.NavigableSet;
034import java.util.Optional;
035import java.util.OptionalDouble;
036import java.util.OptionalInt;
037import java.util.OptionalLong;
038import java.util.Set;
039import java.util.concurrent.Callable;
040import java.util.concurrent.CompletionService;
041import java.util.concurrent.ConcurrentHashMap;
042import java.util.concurrent.ExecutionException;
043import java.util.concurrent.ExecutorCompletionService;
044import java.util.concurrent.Future;
045import java.util.concurrent.ThreadPoolExecutor;
046import java.util.concurrent.atomic.AtomicBoolean;
047import java.util.concurrent.atomic.AtomicInteger;
048import java.util.concurrent.atomic.AtomicLong;
049import java.util.concurrent.atomic.LongAdder;
050import java.util.concurrent.locks.ReentrantLock;
051import java.util.function.Consumer;
052import java.util.function.Supplier;
053import java.util.function.ToLongFunction;
054import java.util.stream.Collectors;
055import java.util.stream.LongStream;
056import org.apache.hadoop.conf.Configuration;
057import org.apache.hadoop.fs.FileSystem;
058import org.apache.hadoop.fs.Path;
059import org.apache.hadoop.fs.permission.FsAction;
060import org.apache.hadoop.hbase.Cell;
061import org.apache.hadoop.hbase.CellComparator;
062import org.apache.hadoop.hbase.CellUtil;
063import org.apache.hadoop.hbase.ExtendedCell;
064import org.apache.hadoop.hbase.HConstants;
065import org.apache.hadoop.hbase.InnerStoreCellComparator;
066import org.apache.hadoop.hbase.MemoryCompactionPolicy;
067import org.apache.hadoop.hbase.MetaCellComparator;
068import org.apache.hadoop.hbase.TableName;
069import org.apache.hadoop.hbase.backup.FailedArchiveException;
070import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
071import org.apache.hadoop.hbase.client.RegionInfo;
072import org.apache.hadoop.hbase.client.Scan;
073import org.apache.hadoop.hbase.conf.ConfigKey;
074import org.apache.hadoop.hbase.conf.ConfigurationManager;
075import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
076import org.apache.hadoop.hbase.coprocessor.ReadOnlyConfiguration;
077import org.apache.hadoop.hbase.io.HeapSize;
078import org.apache.hadoop.hbase.io.hfile.CacheConfig;
079import org.apache.hadoop.hbase.io.hfile.HFile;
080import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoder;
081import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoderImpl;
082import org.apache.hadoop.hbase.io.hfile.HFileScanner;
083import org.apache.hadoop.hbase.io.hfile.InvalidHFileException;
084import org.apache.hadoop.hbase.monitoring.MonitoredTask;
085import org.apache.hadoop.hbase.quotas.RegionSizeStore;
086import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
087import org.apache.hadoop.hbase.regionserver.compactions.CompactionLifeCycleTracker;
088import org.apache.hadoop.hbase.regionserver.compactions.CompactionProgress;
089import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
090import org.apache.hadoop.hbase.regionserver.compactions.OffPeakHours;
091import org.apache.hadoop.hbase.regionserver.querymatcher.ScanQueryMatcher;
092import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
093import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
094import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
095import org.apache.hadoop.hbase.regionserver.wal.WALUtil;
096import org.apache.hadoop.hbase.security.EncryptionUtil;
097import org.apache.hadoop.hbase.security.User;
098import org.apache.hadoop.hbase.util.Bytes;
099import org.apache.hadoop.hbase.util.ClassSize;
100import org.apache.hadoop.hbase.util.CommonFSUtils;
101import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
102import org.apache.hadoop.hbase.util.Pair;
103import org.apache.hadoop.hbase.util.ReflectionUtils;
104import org.apache.hadoop.util.StringUtils;
105import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix;
106import org.apache.yetus.audience.InterfaceAudience;
107import org.slf4j.Logger;
108import org.slf4j.LoggerFactory;
109
110import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
111import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableCollection;
112import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableList;
113import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
114import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
115import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;
116import org.apache.hbase.thirdparty.org.apache.commons.collections4.IterableUtils;
117
118import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
119import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.CompactionDescriptor;
120
121/**
122 * A Store holds a column family in a Region. Its a memstore and a set of zero or more StoreFiles,
123 * which stretch backwards over time.
124 * <p>
125 * There's no reason to consider append-logging at this level; all logging and locking is handled at
126 * the HRegion level. Store just provides services to manage sets of StoreFiles. One of the most
127 * important of those services is compaction services where files are aggregated once they pass a
128 * configurable threshold.
129 * <p>
130 * Locking and transactions are handled at a higher level. This API should not be called directly
131 * but by an HRegion manager.
132 */
133@InterfaceAudience.Private
134public class HStore
135  implements Store, HeapSize, StoreConfigInformation, PropagatingConfigurationObserver {
136  public static final String MEMSTORE_CLASS_NAME =
137    ConfigKey.CLASS("hbase.regionserver.memstore.class", MemStore.class);
138  public static final String COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY =
139    ConfigKey.INT("hbase.server.compactchecker.interval.multiplier");
140  public static final String BLOCKING_STOREFILES_KEY =
141    ConfigKey.INT("hbase.hstore.blockingStoreFiles");
142  public static final String BLOCK_STORAGE_POLICY_KEY = "hbase.hstore.block.storage.policy";
143  // "NONE" is not a valid storage policy and means we defer the policy to HDFS
144  public static final String DEFAULT_BLOCK_STORAGE_POLICY = "NONE";
145  public static final int DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER = 1000;
146  public static final int DEFAULT_BLOCKING_STOREFILE_COUNT = 16;
147
148  // HBASE-24428 : Update compaction priority for recently split daughter regions
149  // so as to prioritize their compaction.
150  // Any compaction candidate with higher priority than compaction of newly split daugher regions
151  // should have priority value < (Integer.MIN_VALUE + 1000)
152  private static final int SPLIT_REGION_COMPACTION_PRIORITY = Integer.MIN_VALUE + 1000;
153
154  private static final Logger LOG = LoggerFactory.getLogger(HStore.class);
155
156  protected final MemStore memstore;
157  // This stores directory in the filesystem.
158  private final HRegion region;
159  protected Configuration conf;
160  private long lastCompactSize = 0;
161  volatile boolean forceMajor = false;
162  private AtomicLong storeSize = new AtomicLong();
163  private AtomicLong totalUncompressedBytes = new AtomicLong();
164  private LongAdder memstoreOnlyRowReadsCount = new LongAdder();
165  // rows that has cells from both memstore and files (or only files)
166  private LongAdder mixedRowReadsCount = new LongAdder();
167
168  /**
169   * Lock specific to archiving compacted store files. This avoids races around the combination of
170   * retrieving the list of compacted files and moving them to the archive directory. Since this is
171   * usually a background process (other than on close), we don't want to handle this with the store
172   * write lock, which would block readers and degrade performance. Locked by: -
173   * CompactedHFilesDispatchHandler via closeAndArchiveCompactedFiles() - close()
174   */
175  final ReentrantLock archiveLock = new ReentrantLock();
176
177  private final boolean verifyBulkLoads;
178
179  /**
180   * Use this counter to track concurrent puts. If TRACE-log is enabled, if we are over the
181   * threshold set by hbase.region.store.parallel.put.print.threshold (Default is 50) we will log a
182   * message that identifies the Store experience this high-level of concurrency.
183   */
184  private final AtomicInteger currentParallelPutCount = new AtomicInteger(0);
185  private final int parallelPutCountPrintThreshold;
186
187  private ScanInfo scanInfo;
188
189  // All access must be synchronized.
190  // TODO: ideally, this should be part of storeFileManager, as we keep passing this to it.
191  private final List<HStoreFile> filesCompacting = Lists.newArrayList();
192
193  // All access must be synchronized.
194  private final Set<ChangedReadersObserver> changedReaderObservers =
195    Collections.newSetFromMap(new ConcurrentHashMap<ChangedReadersObserver, Boolean>());
196
197  private HFileDataBlockEncoder dataBlockEncoder;
198
199  final StoreEngine<?, ?, ?, ?> storeEngine;
200
201  private static final AtomicBoolean offPeakCompactionTracker = new AtomicBoolean();
202  private volatile OffPeakHours offPeakHours;
203
204  private static final int DEFAULT_FLUSH_RETRIES_NUMBER = 10;
205  private int flushRetriesNumber;
206  private int pauseTime;
207
208  private long blockingFileCount;
209  private int compactionCheckMultiplier;
210
211  private AtomicLong flushedCellsCount = new AtomicLong();
212  private AtomicLong compactedCellsCount = new AtomicLong();
213  private AtomicLong majorCompactedCellsCount = new AtomicLong();
214  private AtomicLong flushedCellsSize = new AtomicLong();
215  private AtomicLong flushedOutputFileSize = new AtomicLong();
216  private AtomicLong compactedCellsSize = new AtomicLong();
217  private AtomicLong majorCompactedCellsSize = new AtomicLong();
218
219  private final StoreContext storeContext;
220
221  // Used to track the store files which are currently being written. For compaction, if we want to
222  // compact store file [a, b, c] to [d], then here we will record 'd'. And we will also use it to
223  // track the store files being written when flushing.
224  // Notice that the creation is in the background compaction or flush thread and we will get the
225  // files in other thread, so it needs to be thread safe.
226  private static final class StoreFileWriterCreationTracker implements Consumer<Path> {
227
228    private final Set<Path> files = Collections.newSetFromMap(new ConcurrentHashMap<>());
229
230    @Override
231    public void accept(Path t) {
232      files.add(t);
233    }
234
235    public Set<Path> get() {
236      return Collections.unmodifiableSet(files);
237    }
238  }
239
240  // We may have multiple compaction running at the same time, and flush can also happen at the same
241  // time, so here we need to use a collection, and the collection needs to be thread safe.
242  // The implementation of StoreFileWriterCreationTracker is very simple and we will not likely to
243  // implement hashCode or equals for it, so here we just use ConcurrentHashMap. Changed to
244  // IdentityHashMap if later we want to implement hashCode or equals.
245  private final Set<StoreFileWriterCreationTracker> storeFileWriterCreationTrackers =
246    Collections.newSetFromMap(new ConcurrentHashMap<>());
247
248  // For the SFT implementation which we will write tmp store file first, we do not need to clean up
249  // the broken store files under the data directory, which means we do not need to track the store
250  // file writer creation. So here we abstract a factory to return different trackers for different
251  // SFT implementations.
252  private final Supplier<StoreFileWriterCreationTracker> storeFileWriterCreationTrackerFactory;
253
254  private final boolean warmup;
255
256  /**
257   * Constructor
258   * @param family    HColumnDescriptor for this column
259   * @param confParam configuration object failed. Can be null.
260   */
261  protected HStore(final HRegion region, final ColumnFamilyDescriptor family,
262    final Configuration confParam, boolean warmup) throws IOException {
263    this.conf = StoreUtils.createStoreConfiguration(confParam, region.getTableDescriptor(), family);
264
265    this.region = region;
266    this.storeContext = initializeStoreContext(family);
267
268    // Assemble the store's home directory and Ensure it exists.
269    region.getRegionFileSystem().createStoreDir(family.getNameAsString());
270
271    // set block storage policy for store directory
272    String policyName = family.getStoragePolicy();
273    if (null == policyName) {
274      policyName = this.conf.get(BLOCK_STORAGE_POLICY_KEY, DEFAULT_BLOCK_STORAGE_POLICY);
275    }
276    region.getRegionFileSystem().setStoragePolicy(family.getNameAsString(), policyName.trim());
277
278    this.dataBlockEncoder = new HFileDataBlockEncoderImpl(family.getDataBlockEncoding());
279
280    // used by ScanQueryMatcher
281    long timeToPurgeDeletes = Math.max(conf.getLong("hbase.hstore.time.to.purge.deletes", 0), 0);
282    LOG.trace("Time to purge deletes set to {}ms in {}", timeToPurgeDeletes, this);
283    // Get TTL
284    long ttl = determineTTLFromFamily(family);
285    // Why not just pass a HColumnDescriptor in here altogether? Even if have
286    // to clone it?
287    scanInfo =
288      new ScanInfo(conf, family, ttl, timeToPurgeDeletes, this.storeContext.getComparator());
289    this.memstore = getMemstore();
290
291    this.offPeakHours = OffPeakHours.getInstance(conf);
292
293    this.verifyBulkLoads = conf.getBoolean("hbase.hstore.bulkload.verify", false);
294
295    this.blockingFileCount = conf.getInt(BLOCKING_STOREFILES_KEY, DEFAULT_BLOCKING_STOREFILE_COUNT);
296    this.compactionCheckMultiplier = conf.getInt(COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY,
297      DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER);
298    if (this.compactionCheckMultiplier <= 0) {
299      LOG.error("Compaction check period multiplier must be positive, setting default: {}",
300        DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER);
301      this.compactionCheckMultiplier = DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER;
302    }
303
304    this.warmup = warmup;
305    this.storeEngine = createStoreEngine(this, this.conf, region.getCellComparator());
306    storeEngine.initialize(warmup);
307    // if require writing to tmp dir first, then we just return null, which indicate that we do not
308    // need to track the creation of store file writer, otherwise we return a new
309    // StoreFileWriterCreationTracker.
310    this.storeFileWriterCreationTrackerFactory = storeEngine.requireWritingToTmpDirFirst()
311      ? () -> null
312      : () -> new StoreFileWriterCreationTracker();
313    refreshStoreSizeAndTotalBytes();
314
315    flushRetriesNumber =
316      conf.getInt("hbase.hstore.flush.retries.number", DEFAULT_FLUSH_RETRIES_NUMBER);
317    pauseTime = conf.getInt(HConstants.HBASE_SERVER_PAUSE, HConstants.DEFAULT_HBASE_SERVER_PAUSE);
318    if (flushRetriesNumber <= 0) {
319      throw new IllegalArgumentException(
320        "hbase.hstore.flush.retries.number must be > 0, not " + flushRetriesNumber);
321    }
322
323    int confPrintThreshold =
324      this.conf.getInt("hbase.region.store.parallel.put.print.threshold", 50);
325    if (confPrintThreshold < 10) {
326      confPrintThreshold = 10;
327    }
328    this.parallelPutCountPrintThreshold = confPrintThreshold;
329
330    LOG.info(
331      "Store={},  memstore type={}, storagePolicy={}, verifyBulkLoads={}, "
332        + "parallelPutCountPrintThreshold={}, encoding={}, compression={}",
333      this, memstore.getClass().getSimpleName(), policyName, verifyBulkLoads,
334      parallelPutCountPrintThreshold, family.getDataBlockEncoding(), family.getCompressionType());
335  }
336
337  private StoreContext initializeStoreContext(ColumnFamilyDescriptor family) throws IOException {
338    return new StoreContext.Builder().withBlockSize(family.getBlocksize())
339      .withEncryptionContext(EncryptionUtil.createEncryptionContext(conf, family))
340      .withBloomType(family.getBloomFilterType()).withCacheConfig(createCacheConf(family))
341      .withCellComparator(region.getTableDescriptor().isMetaTable() || conf
342        .getBoolean(HRegion.USE_META_CELL_COMPARATOR, HRegion.DEFAULT_USE_META_CELL_COMPARATOR)
343          ? MetaCellComparator.META_COMPARATOR
344          : InnerStoreCellComparator.INNER_STORE_COMPARATOR)
345      .withColumnFamilyDescriptor(family).withCompactedFilesSupplier(this::getCompactedFiles)
346      .withRegionFileSystem(region.getRegionFileSystem())
347      .withFavoredNodesSupplier(this::getFavoredNodes)
348      .withFamilyStoreDirectoryPath(
349        region.getRegionFileSystem().getStoreDir(family.getNameAsString()))
350      .withRegionCoprocessorHost(region.getCoprocessorHost()).build();
351  }
352
353  private InetSocketAddress[] getFavoredNodes() {
354    InetSocketAddress[] favoredNodes = null;
355    if (region.getRegionServerServices() != null) {
356      favoredNodes = region.getRegionServerServices()
357        .getFavoredNodesForRegion(region.getRegionInfo().getEncodedName());
358    }
359    return favoredNodes;
360  }
361
362  /** Returns MemStore Instance to use in this store. */
363  private MemStore getMemstore() {
364    MemStore ms = null;
365    // Check if in-memory-compaction configured. Note MemoryCompactionPolicy is an enum!
366    MemoryCompactionPolicy inMemoryCompaction = null;
367    if (this.getTableName().isSystemTable()) {
368      inMemoryCompaction = MemoryCompactionPolicy
369        .valueOf(conf.get("hbase.systemtables.compacting.memstore.type", "NONE").toUpperCase());
370    } else {
371      inMemoryCompaction = getColumnFamilyDescriptor().getInMemoryCompaction();
372    }
373    if (inMemoryCompaction == null) {
374      inMemoryCompaction =
375        MemoryCompactionPolicy.valueOf(conf.get(CompactingMemStore.COMPACTING_MEMSTORE_TYPE_KEY,
376          CompactingMemStore.COMPACTING_MEMSTORE_TYPE_DEFAULT).toUpperCase());
377    }
378
379    switch (inMemoryCompaction) {
380      case NONE:
381        Class<? extends MemStore> memStoreClass =
382          conf.getClass(MEMSTORE_CLASS_NAME, DefaultMemStore.class, MemStore.class);
383        ms = ReflectionUtils.newInstance(memStoreClass,
384          new Object[] { conf, getComparator(), this.getHRegion().getRegionServicesForStores() });
385        break;
386      default:
387        Class<? extends CompactingMemStore> compactingMemStoreClass =
388          conf.getClass(MEMSTORE_CLASS_NAME, CompactingMemStore.class, CompactingMemStore.class);
389        ms =
390          ReflectionUtils.newInstance(compactingMemStoreClass, new Object[] { conf, getComparator(),
391            this, this.getHRegion().getRegionServicesForStores(), inMemoryCompaction });
392    }
393    return ms;
394  }
395
396  /**
397   * Creates the cache config.
398   * @param family The current column family.
399   */
400  protected CacheConfig createCacheConf(final ColumnFamilyDescriptor family) {
401    CacheConfig cacheConf = new CacheConfig(conf, family, region.getBlockCache(),
402      region.getRegionServicesForStores().getByteBuffAllocator());
403    LOG.info("Created cacheConfig: {}, for column family {} of region {} ", cacheConf,
404      family.getNameAsString(), region.getRegionInfo().getEncodedName());
405    return cacheConf;
406  }
407
408  /**
409   * Creates the store engine configured for the given Store.
410   * @param store        The store. An unfortunate dependency needed due to it being passed to
411   *                     coprocessors via the compactor.
412   * @param conf         Store configuration.
413   * @param kvComparator KVComparator for storeFileManager.
414   * @return StoreEngine to use.
415   */
416  protected StoreEngine<?, ?, ?, ?> createStoreEngine(HStore store, Configuration conf,
417    CellComparator kvComparator) throws IOException {
418    return StoreEngine.create(store, conf, kvComparator);
419  }
420
421  /** Returns TTL in seconds of the specified family */
422  public static long determineTTLFromFamily(final ColumnFamilyDescriptor family) {
423    // HCD.getTimeToLive returns ttl in seconds. Convert to milliseconds.
424    long ttl = family.getTimeToLive();
425    if (ttl == HConstants.FOREVER) {
426      // Default is unlimited ttl.
427      ttl = Long.MAX_VALUE;
428    } else if (ttl == -1) {
429      ttl = Long.MAX_VALUE;
430    } else {
431      // Second -> ms adjust for user data
432      ttl *= 1000;
433    }
434    return ttl;
435  }
436
437  public StoreContext getStoreContext() {
438    return storeContext;
439  }
440
441  @Override
442  public String getColumnFamilyName() {
443    return this.storeContext.getFamily().getNameAsString();
444  }
445
446  @Override
447  public TableName getTableName() {
448    return this.getRegionInfo().getTable();
449  }
450
451  @Override
452  public FileSystem getFileSystem() {
453    return storeContext.getRegionFileSystem().getFileSystem();
454  }
455
456  public HRegionFileSystem getRegionFileSystem() {
457    return storeContext.getRegionFileSystem();
458  }
459
460  /* Implementation of StoreConfigInformation */
461  @Override
462  public long getStoreFileTtl() {
463    // TTL only applies if there's no MIN_VERSIONs setting on the column.
464    return (this.scanInfo.getMinVersions() == 0) ? this.scanInfo.getTtl() : Long.MAX_VALUE;
465  }
466
467  @Override
468  public long getMemStoreFlushSize() {
469    // TODO: Why is this in here? The flushsize of the region rather than the store? St.Ack
470    return this.region.memstoreFlushSize;
471  }
472
473  @Override
474  public MemStoreSize getFlushableSize() {
475    return this.memstore.getFlushableSize();
476  }
477
478  @Override
479  public MemStoreSize getSnapshotSize() {
480    return this.memstore.getSnapshotSize();
481  }
482
483  @Override
484  public long getCompactionCheckMultiplier() {
485    return this.compactionCheckMultiplier;
486  }
487
488  @Override
489  public long getBlockingFileCount() {
490    return blockingFileCount;
491  }
492  /* End implementation of StoreConfigInformation */
493
494  @Override
495  public ColumnFamilyDescriptor getColumnFamilyDescriptor() {
496    return this.storeContext.getFamily();
497  }
498
499  @Override
500  public OptionalLong getMaxSequenceId() {
501    return StoreUtils.getMaxSequenceIdInList(this.getStorefiles());
502  }
503
504  @Override
505  public OptionalLong getMaxMemStoreTS() {
506    return StoreUtils.getMaxMemStoreTSInList(this.getStorefiles());
507  }
508
509  /** Returns the data block encoder */
510  public HFileDataBlockEncoder getDataBlockEncoder() {
511    return dataBlockEncoder;
512  }
513
514  /**
515   * Should be used only in tests.
516   * @param blockEncoder the block delta encoder to use
517   */
518  void setDataBlockEncoderInTest(HFileDataBlockEncoder blockEncoder) {
519    this.dataBlockEncoder = blockEncoder;
520  }
521
522  private void postRefreshStoreFiles() throws IOException {
523    // Advance the memstore read point to be at least the new store files seqIds so that
524    // readers might pick it up. This assumes that the store is not getting any writes (otherwise
525    // in-flight transactions might be made visible)
526    getMaxSequenceId().ifPresent(region.getMVCC()::advanceTo);
527    refreshStoreSizeAndTotalBytes();
528  }
529
530  @Override
531  public void refreshStoreFiles() throws IOException {
532    storeEngine.refreshStoreFiles();
533    postRefreshStoreFiles();
534  }
535
536  /**
537   * Replaces the store files that the store has with the given files. Mainly used by secondary
538   * region replicas to keep up to date with the primary region files.
539   */
540  public void refreshStoreFiles(Collection<String> newFiles) throws IOException {
541    storeEngine.refreshStoreFiles(newFiles);
542    postRefreshStoreFiles();
543  }
544
545  /**
546   * This message intends to inform the MemStore that next coming updates are going to be part of
547   * the replaying edits from WAL
548   */
549  public void startReplayingFromWAL() {
550    this.memstore.startReplayingFromWAL();
551  }
552
553  /**
554   * This message intends to inform the MemStore that the replaying edits from WAL are done
555   */
556  public void stopReplayingFromWAL() {
557    this.memstore.stopReplayingFromWAL();
558  }
559
560  /**
561   * Adds a value to the memstore
562   */
563  public void add(final ExtendedCell cell, MemStoreSizing memstoreSizing) {
564    storeEngine.readLock();
565    try {
566      if (this.currentParallelPutCount.getAndIncrement() > this.parallelPutCountPrintThreshold) {
567        LOG.trace("tableName={}, encodedName={}, columnFamilyName={} is too busy!",
568          this.getTableName(), this.getRegionInfo().getEncodedName(), this.getColumnFamilyName());
569      }
570      this.memstore.add(cell, memstoreSizing);
571    } finally {
572      storeEngine.readUnlock();
573      currentParallelPutCount.decrementAndGet();
574    }
575  }
576
577  /**
578   * Adds the specified value to the memstore
579   */
580  public void add(final Iterable<ExtendedCell> cells, MemStoreSizing memstoreSizing) {
581    storeEngine.readLock();
582    try {
583      if (this.currentParallelPutCount.getAndIncrement() > this.parallelPutCountPrintThreshold) {
584        LOG.trace("tableName={}, encodedName={}, columnFamilyName={} is too busy!",
585          this.getTableName(), this.getRegionInfo().getEncodedName(), this.getColumnFamilyName());
586      }
587      memstore.add(cells, memstoreSizing);
588    } finally {
589      storeEngine.readUnlock();
590      currentParallelPutCount.decrementAndGet();
591    }
592  }
593
594  @Override
595  public long timeOfOldestEdit() {
596    return memstore.timeOfOldestEdit();
597  }
598
599  /** Returns All store files. */
600  @Override
601  public Collection<HStoreFile> getStorefiles() {
602    return this.storeEngine.getStoreFileManager().getStoreFiles();
603  }
604
605  @Override
606  public Collection<HStoreFile> getCompactedFiles() {
607    return this.storeEngine.getStoreFileManager().getCompactedfiles();
608  }
609
610  /**
611   * This throws a WrongRegionException if the HFile does not fit in this region, or an
612   * InvalidHFileException if the HFile is not valid.
613   */
614  public void assertBulkLoadHFileOk(Path srcPath) throws IOException {
615    HFile.Reader reader = null;
616    try {
617      LOG.info("Validating hfile at " + srcPath + " for inclusion in " + this);
618      FileSystem srcFs = srcPath.getFileSystem(conf);
619      srcFs.access(srcPath, FsAction.READ_WRITE);
620      reader = HFile.createReader(srcFs, srcPath, getCacheConfig(), isPrimaryReplicaStore(), conf);
621
622      Optional<byte[]> firstKey = reader.getFirstRowKey();
623      Preconditions.checkState(firstKey.isPresent(), "First key can not be null");
624      Optional<ExtendedCell> lk = reader.getLastKey();
625      Preconditions.checkState(lk.isPresent(), "Last key can not be null");
626      byte[] lastKey = CellUtil.cloneRow(lk.get());
627
628      if (LOG.isDebugEnabled()) {
629        LOG.debug("HFile bounds: first=" + Bytes.toStringBinary(firstKey.get()) + " last="
630          + Bytes.toStringBinary(lastKey));
631        LOG.debug("Region bounds: first=" + Bytes.toStringBinary(getRegionInfo().getStartKey())
632          + " last=" + Bytes.toStringBinary(getRegionInfo().getEndKey()));
633      }
634
635      if (!this.getRegionInfo().containsRange(firstKey.get(), lastKey)) {
636        throw new WrongRegionException("Bulk load file " + srcPath.toString()
637          + " does not fit inside region " + this.getRegionInfo().getRegionNameAsString());
638      }
639
640      if (
641        reader.length()
642            > conf.getLong(HConstants.HREGION_MAX_FILESIZE, HConstants.DEFAULT_MAX_FILE_SIZE)
643      ) {
644        LOG.warn("Trying to bulk load hfile " + srcPath + " with size: " + reader.length()
645          + " bytes can be problematic as it may lead to oversplitting.");
646      }
647
648      if (verifyBulkLoads) {
649        long verificationStartTime = EnvironmentEdgeManager.currentTime();
650        LOG.info("Full verification started for bulk load hfile: {}", srcPath);
651        Cell prevCell = null;
652        HFileScanner scanner = reader.getScanner(conf, false, false, false);
653        scanner.seekTo();
654        do {
655          Cell cell = scanner.getCell();
656          if (prevCell != null) {
657            if (getComparator().compareRows(prevCell, cell) > 0) {
658              throw new InvalidHFileException("Previous row is greater than" + " current row: path="
659                + srcPath + " previous=" + CellUtil.getCellKeyAsString(prevCell) + " current="
660                + CellUtil.getCellKeyAsString(cell));
661            }
662            if (CellComparator.getInstance().compareFamilies(prevCell, cell) != 0) {
663              throw new InvalidHFileException("Previous key had different"
664                + " family compared to current key: path=" + srcPath + " previous="
665                + Bytes.toStringBinary(prevCell.getFamilyArray(), prevCell.getFamilyOffset(),
666                  prevCell.getFamilyLength())
667                + " current=" + Bytes.toStringBinary(cell.getFamilyArray(), cell.getFamilyOffset(),
668                  cell.getFamilyLength()));
669            }
670          }
671          prevCell = cell;
672        } while (scanner.next());
673        LOG.info("Full verification complete for bulk load hfile: " + srcPath.toString() + " took "
674          + (EnvironmentEdgeManager.currentTime() - verificationStartTime) + " ms");
675      }
676    } finally {
677      if (reader != null) {
678        reader.close();
679      }
680    }
681  }
682
683  /**
684   * This method should only be called from Region. It is assumed that the ranges of values in the
685   * HFile fit within the stores assigned region. (assertBulkLoadHFileOk checks this)
686   * @param seqNum sequence Id associated with the HFile
687   */
688  public Pair<Path, Path> preBulkLoadHFile(String srcPathStr, long seqNum) throws IOException {
689    Path srcPath = new Path(srcPathStr);
690    return getRegionFileSystem().bulkLoadStoreFile(getColumnFamilyName(), srcPath, seqNum);
691  }
692
693  public Path bulkLoadHFile(byte[] family, String srcPathStr, Path dstPath) throws IOException {
694    Path srcPath = new Path(srcPathStr);
695    try {
696      getRegionFileSystem().commitStoreFile(srcPath, dstPath);
697    } finally {
698      if (this.getCoprocessorHost() != null) {
699        this.getCoprocessorHost().postCommitStoreFile(family, srcPath, dstPath);
700      }
701    }
702
703    LOG.info("Loaded HFile " + srcPath + " into " + this + " as " + dstPath
704      + " - updating store file list.");
705
706    HStoreFile sf = storeEngine.createStoreFileAndReader(dstPath);
707    bulkLoadHFile(sf);
708
709    LOG.info("Successfully loaded {} into {} (new location: {})", srcPath, this, dstPath);
710
711    return dstPath;
712  }
713
714  public void bulkLoadHFile(StoreFileInfo fileInfo) throws IOException {
715    HStoreFile sf = storeEngine.createStoreFileAndReader(fileInfo);
716    bulkLoadHFile(sf);
717  }
718
719  private void bulkLoadHFile(HStoreFile sf) throws IOException {
720    StoreFileReader r = sf.getReader();
721    this.storeSize.addAndGet(r.length());
722    this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
723    storeEngine.addStoreFiles(Lists.newArrayList(sf), () -> {
724    });
725    LOG.info("Loaded HFile " + sf.getFileInfo() + " into " + this);
726    if (LOG.isTraceEnabled()) {
727      String traceMessage = "BULK LOAD time,size,store size,store files ["
728        + EnvironmentEdgeManager.currentTime() + "," + r.length() + "," + storeSize + ","
729        + storeEngine.getStoreFileManager().getStorefileCount() + "]";
730      LOG.trace(traceMessage);
731    }
732  }
733
734  private ImmutableCollection<HStoreFile> closeWithoutLock() throws IOException {
735    memstore.close();
736    // Clear so metrics doesn't find them.
737    ImmutableCollection<HStoreFile> result = storeEngine.getStoreFileManager().clearFiles();
738    Collection<HStoreFile> compactedfiles = storeEngine.getStoreFileManager().clearCompactedFiles();
739    // clear the compacted files
740    if (CollectionUtils.isNotEmpty(compactedfiles)) {
741      removeCompactedfiles(compactedfiles,
742        getCacheConfig() != null ? getCacheConfig().shouldEvictOnClose() : true);
743    }
744    if (!result.isEmpty()) {
745      // initialize the thread pool for closing store files in parallel.
746      ThreadPoolExecutor storeFileCloserThreadPool =
747        this.region.getStoreFileOpenAndCloseThreadPool("StoreFileCloser-"
748          + this.region.getRegionInfo().getEncodedName() + "-" + this.getColumnFamilyName());
749
750      // close each store file in parallel
751      CompletionService<Void> completionService =
752        new ExecutorCompletionService<>(storeFileCloserThreadPool);
753      for (HStoreFile f : result) {
754        completionService.submit(new Callable<Void>() {
755          @Override
756          public Void call() throws IOException {
757            boolean evictOnClose =
758              getCacheConfig() != null ? getCacheConfig().shouldEvictOnClose() : true;
759            f.closeStoreFile(!warmup && evictOnClose);
760            return null;
761          }
762        });
763      }
764
765      IOException ioe = null;
766      try {
767        for (int i = 0; i < result.size(); i++) {
768          try {
769            Future<Void> future = completionService.take();
770            future.get();
771          } catch (InterruptedException e) {
772            if (ioe == null) {
773              ioe = new InterruptedIOException();
774              ioe.initCause(e);
775            }
776          } catch (ExecutionException e) {
777            if (ioe == null) {
778              ioe = new IOException(e.getCause());
779            }
780          }
781        }
782      } finally {
783        storeFileCloserThreadPool.shutdownNow();
784      }
785      if (ioe != null) {
786        throw ioe;
787      }
788    }
789    LOG.trace("Closed {}", this);
790    return result;
791  }
792
793  /**
794   * Close all the readers We don't need to worry about subsequent requests because the Region holds
795   * a write lock that will prevent any more reads or writes.
796   * @return the {@link StoreFile StoreFiles} that were previously being used.
797   * @throws IOException on failure
798   */
799  public ImmutableCollection<HStoreFile> close() throws IOException {
800    // findbugs can not recognize storeEngine.writeLock is just a lock operation so it will report
801    // UL_UNRELEASED_LOCK_EXCEPTION_PATH, so here we have to use two try finally...
802    // Change later if findbugs becomes smarter in the future.
803    this.archiveLock.lock();
804    try {
805      this.storeEngine.writeLock();
806      try {
807        return closeWithoutLock();
808      } finally {
809        this.storeEngine.writeUnlock();
810      }
811    } finally {
812      this.archiveLock.unlock();
813    }
814  }
815
816  /**
817   * Write out current snapshot. Presumes {@code StoreFlusherImpl.prepare()} has been called
818   * previously.
819   * @param logCacheFlushId flush sequence number
820   * @return The path name of the tmp file to which the store was flushed
821   * @throws IOException if exception occurs during process
822   */
823  protected List<Path> flushCache(final long logCacheFlushId, MemStoreSnapshot snapshot,
824    MonitoredTask status, ThroughputController throughputController, FlushLifeCycleTracker tracker,
825    Consumer<Path> writerCreationTracker) throws IOException {
826    // If an exception happens flushing, we let it out without clearing
827    // the memstore snapshot. The old snapshot will be returned when we say
828    // 'snapshot', the next time flush comes around.
829    // Retry after catching exception when flushing, otherwise server will abort
830    // itself
831    StoreFlusher flusher = storeEngine.getStoreFlusher();
832    IOException lastException = null;
833    for (int i = 0; i < flushRetriesNumber; i++) {
834      try {
835        List<Path> pathNames = flusher.flushSnapshot(snapshot, logCacheFlushId, status,
836          throughputController, tracker, writerCreationTracker);
837        Path lastPathName = null;
838        try {
839          for (Path pathName : pathNames) {
840            lastPathName = pathName;
841            storeEngine.validateStoreFile(pathName, false);
842          }
843          return pathNames;
844        } catch (Exception e) {
845          LOG.warn("Failed validating store file {}, retrying num={}", lastPathName, i, e);
846          if (e instanceof IOException) {
847            lastException = (IOException) e;
848          } else {
849            lastException = new IOException(e);
850          }
851        }
852      } catch (IOException e) {
853        LOG.warn("Failed flushing store file for {}, retrying num={}", this, i, e);
854        lastException = e;
855      }
856      if (lastException != null && i < (flushRetriesNumber - 1)) {
857        try {
858          Thread.sleep(pauseTime);
859        } catch (InterruptedException e) {
860          IOException iie = new InterruptedIOException();
861          iie.initCause(e);
862          throw iie;
863        }
864      }
865    }
866    throw lastException;
867  }
868
869  public HStoreFile tryCommitRecoveredHFile(Path path) throws IOException {
870    LOG.info("Validating recovered hfile at {} for inclusion in store {}", path, this);
871    FileSystem srcFs = path.getFileSystem(conf);
872    srcFs.access(path, FsAction.READ_WRITE);
873    try (HFile.Reader reader =
874      HFile.createReader(srcFs, path, getCacheConfig(), isPrimaryReplicaStore(), conf)) {
875      Optional<byte[]> firstKey = reader.getFirstRowKey();
876      Preconditions.checkState(firstKey.isPresent(), "First key can not be null");
877      Optional<ExtendedCell> lk = reader.getLastKey();
878      Preconditions.checkState(lk.isPresent(), "Last key can not be null");
879      byte[] lastKey = CellUtil.cloneRow(lk.get());
880      if (!this.getRegionInfo().containsRange(firstKey.get(), lastKey)) {
881        throw new WrongRegionException("Recovered hfile " + path.toString()
882          + " does not fit inside region " + this.getRegionInfo().getRegionNameAsString());
883      }
884    }
885
886    Path dstPath = getRegionFileSystem().commitStoreFile(getColumnFamilyName(), path);
887    HStoreFile sf = storeEngine.createStoreFileAndReader(dstPath);
888    StoreFileReader r = sf.getReader();
889    this.storeSize.addAndGet(r.length());
890    this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
891
892    storeEngine.addStoreFiles(Lists.newArrayList(sf), () -> {
893    });
894
895    LOG.info("Loaded recovered hfile to {}, entries={}, sequenceid={}, filesize={}", sf,
896      r.getEntries(), r.getSequenceID(), TraditionalBinaryPrefix.long2String(r.length(), "B", 1));
897    return sf;
898  }
899
900  private long getTotalSize(Collection<HStoreFile> sfs) {
901    return sfs.stream().mapToLong(sf -> sf.getReader().length()).sum();
902  }
903
904  private boolean completeFlush(final List<HStoreFile> sfs, long snapshotId) throws IOException {
905    // NOTE:we should keep clearSnapshot method inside the write lock because clearSnapshot may
906    // close {@link DefaultMemStore#snapshot}, which may be used by
907    // {@link DefaultMemStore#getScanners}.
908    storeEngine.addStoreFiles(sfs,
909      // NOTE: here we must increase the refCount for storeFiles because we would open the
910      // storeFiles and get the StoreFileScanners for them in HStore.notifyChangedReadersObservers.
911      // If we don't increase the refCount here, HStore.closeAndArchiveCompactedFiles called by
912      // CompactedHFilesDischarger may archive the storeFiles after a concurrent compaction.Because
913      // HStore.requestCompaction is under storeEngine lock, so here we increase the refCount under
914      // storeEngine lock. see HBASE-27519 for more details.
915      snapshotId > 0 ? () -> {
916        this.memstore.clearSnapshot(snapshotId);
917        HStoreFile.increaseStoreFilesRefeCount(sfs);
918      } : () -> {
919        HStoreFile.increaseStoreFilesRefeCount(sfs);
920      });
921    // notify to be called here - only in case of flushes
922    try {
923      notifyChangedReadersObservers(sfs);
924    } finally {
925      HStoreFile.decreaseStoreFilesRefeCount(sfs);
926    }
927    if (LOG.isTraceEnabled()) {
928      long totalSize = getTotalSize(sfs);
929      String traceMessage = "FLUSH time,count,size,store size,store files ["
930        + EnvironmentEdgeManager.currentTime() + "," + sfs.size() + "," + totalSize + ","
931        + storeSize + "," + storeEngine.getStoreFileManager().getStorefileCount() + "]";
932      LOG.trace(traceMessage);
933    }
934    return needsCompaction();
935  }
936
937  /**
938   * Notify all observers that set of Readers has changed.
939   */
940  private void notifyChangedReadersObservers(List<HStoreFile> sfs) throws IOException {
941    for (ChangedReadersObserver o : this.changedReaderObservers) {
942      List<KeyValueScanner> memStoreScanners;
943      this.storeEngine.readLock();
944      try {
945        memStoreScanners = this.memstore.getScanners(o.getReadPoint());
946      } finally {
947        this.storeEngine.readUnlock();
948      }
949      o.updateReaders(sfs, memStoreScanners);
950    }
951  }
952
953  /**
954   * Get all scanners with no filtering based on TTL (that happens further down the line).
955   * @param cacheBlocks  cache the blocks or not
956   * @param usePread     true to use pread, false if not
957   * @param isCompaction true if the scanner is created for compaction
958   * @param matcher      the scan query matcher
959   * @param startRow     the start row
960   * @param stopRow      the stop row
961   * @param readPt       the read point of the current scan
962   * @return all scanners for this store
963   */
964  public List<KeyValueScanner> getScanners(boolean cacheBlocks, boolean isGet, boolean usePread,
965    boolean isCompaction, ScanQueryMatcher matcher, byte[] startRow, byte[] stopRow, long readPt,
966    boolean onlyLatestVersion) throws IOException {
967    return getScanners(cacheBlocks, usePread, isCompaction, matcher, startRow, true, stopRow, false,
968      readPt, onlyLatestVersion);
969  }
970
971  /**
972   * Get all scanners with no filtering based on TTL (that happens further down the line).
973   * @param cacheBlocks     cache the blocks or not
974   * @param usePread        true to use pread, false if not
975   * @param isCompaction    true if the scanner is created for compaction
976   * @param matcher         the scan query matcher
977   * @param startRow        the start row
978   * @param includeStartRow true to include start row, false if not
979   * @param stopRow         the stop row
980   * @param includeStopRow  true to include stop row, false if not
981   * @param readPt          the read point of the current scan
982   * @return all scanners for this store
983   */
984  public List<KeyValueScanner> getScanners(boolean cacheBlocks, boolean usePread,
985    boolean isCompaction, ScanQueryMatcher matcher, byte[] startRow, boolean includeStartRow,
986    byte[] stopRow, boolean includeStopRow, long readPt, boolean onlyLatestVersion)
987    throws IOException {
988    Collection<HStoreFile> storeFilesToScan;
989    List<KeyValueScanner> memStoreScanners;
990    this.storeEngine.readLock();
991    try {
992      storeFilesToScan = this.storeEngine.getStoreFileManager().getFilesForScan(startRow,
993        includeStartRow, stopRow, includeStopRow, onlyLatestVersion);
994      memStoreScanners = this.memstore.getScanners(readPt);
995      // NOTE: here we must increase the refCount for storeFiles because we would open the
996      // storeFiles and get the StoreFileScanners for them.If we don't increase the refCount here,
997      // HStore.closeAndArchiveCompactedFiles called by CompactedHFilesDischarger may archive the
998      // storeFiles after a concurrent compaction.Because HStore.requestCompaction is under
999      // storeEngine lock, so here we increase the refCount under storeEngine lock. see HBASE-27484
1000      // for more details.
1001      HStoreFile.increaseStoreFilesRefeCount(storeFilesToScan);
1002    } finally {
1003      this.storeEngine.readUnlock();
1004    }
1005    try {
1006      // First the store file scanners
1007
1008      // TODO this used to get the store files in descending order,
1009      // but now we get them in ascending order, which I think is
1010      // actually more correct, since memstore get put at the end.
1011      List<StoreFileScanner> sfScanners = StoreFileScanner.getScannersForStoreFiles(
1012        storeFilesToScan, cacheBlocks, usePread, isCompaction, false, matcher, readPt);
1013      List<KeyValueScanner> scanners = new ArrayList<>(sfScanners.size() + 1);
1014      scanners.addAll(sfScanners);
1015      // Then the memstore scanners
1016      scanners.addAll(memStoreScanners);
1017      return scanners;
1018    } catch (Throwable t) {
1019      clearAndClose(memStoreScanners);
1020      throw t instanceof IOException ? (IOException) t : new IOException(t);
1021    } finally {
1022      HStoreFile.decreaseStoreFilesRefeCount(storeFilesToScan);
1023    }
1024  }
1025
1026  private static void clearAndClose(List<KeyValueScanner> scanners) {
1027    if (scanners == null) {
1028      return;
1029    }
1030    for (KeyValueScanner s : scanners) {
1031      s.close();
1032    }
1033    scanners.clear();
1034  }
1035
1036  /**
1037   * Create scanners on the given files and if needed on the memstore with no filtering based on TTL
1038   * (that happens further down the line).
1039   * @param files                  the list of files on which the scanners has to be created
1040   * @param cacheBlocks            cache the blocks or not
1041   * @param usePread               true to use pread, false if not
1042   * @param isCompaction           true if the scanner is created for compaction
1043   * @param matcher                the scan query matcher
1044   * @param startRow               the start row
1045   * @param stopRow                the stop row
1046   * @param readPt                 the read point of the current scan
1047   * @param includeMemstoreScanner true if memstore has to be included
1048   * @return scanners on the given files and on the memstore if specified
1049   */
1050  public List<KeyValueScanner> getScanners(List<HStoreFile> files, boolean cacheBlocks,
1051    boolean isGet, boolean usePread, boolean isCompaction, ScanQueryMatcher matcher,
1052    byte[] startRow, byte[] stopRow, long readPt, boolean includeMemstoreScanner,
1053    boolean onlyLatestVersion) throws IOException {
1054    return getScanners(files, cacheBlocks, usePread, isCompaction, matcher, startRow, true, stopRow,
1055      false, readPt, includeMemstoreScanner, onlyLatestVersion);
1056  }
1057
1058  /**
1059   * Create scanners on the given files and if needed on the memstore with no filtering based on TTL
1060   * (that happens further down the line).
1061   * @param files                  the list of files on which the scanners has to be created
1062   * @param cacheBlocks            ache the blocks or not
1063   * @param usePread               true to use pread, false if not
1064   * @param isCompaction           true if the scanner is created for compaction
1065   * @param matcher                the scan query matcher
1066   * @param startRow               the start row
1067   * @param includeStartRow        true to include start row, false if not
1068   * @param stopRow                the stop row
1069   * @param includeStopRow         true to include stop row, false if not
1070   * @param readPt                 the read point of the current scan
1071   * @param includeMemstoreScanner true if memstore has to be included
1072   * @return scanners on the given files and on the memstore if specified
1073   */
1074  public List<KeyValueScanner> getScanners(List<HStoreFile> files, boolean cacheBlocks,
1075    boolean usePread, boolean isCompaction, ScanQueryMatcher matcher, byte[] startRow,
1076    boolean includeStartRow, byte[] stopRow, boolean includeStopRow, long readPt,
1077    boolean includeMemstoreScanner, boolean onlyLatestVersion) throws IOException {
1078    List<KeyValueScanner> memStoreScanners = null;
1079    if (includeMemstoreScanner) {
1080      this.storeEngine.readLock();
1081      try {
1082        memStoreScanners = this.memstore.getScanners(readPt);
1083      } finally {
1084        this.storeEngine.readUnlock();
1085      }
1086    }
1087    try {
1088      List<StoreFileScanner> sfScanners = StoreFileScanner.getScannersForStoreFiles(files,
1089        cacheBlocks, usePread, isCompaction, false, matcher, readPt);
1090      List<KeyValueScanner> scanners = new ArrayList<>(sfScanners.size() + 1);
1091      scanners.addAll(sfScanners);
1092      // Then the memstore scanners
1093      if (memStoreScanners != null) {
1094        scanners.addAll(memStoreScanners);
1095      }
1096      return scanners;
1097    } catch (Throwable t) {
1098      clearAndClose(memStoreScanners);
1099      throw t instanceof IOException ? (IOException) t : new IOException(t);
1100    }
1101  }
1102
1103  /**
1104   * @param o Observer who wants to know about changes in set of Readers
1105   */
1106  public void addChangedReaderObserver(ChangedReadersObserver o) {
1107    this.changedReaderObservers.add(o);
1108  }
1109
1110  /**
1111   * @param o Observer no longer interested in changes in set of Readers.
1112   */
1113  public void deleteChangedReaderObserver(ChangedReadersObserver o) {
1114    // We don't check if observer present; it may not be (legitimately)
1115    this.changedReaderObservers.remove(o);
1116  }
1117
1118  //////////////////////////////////////////////////////////////////////////////
1119  // Compaction
1120  //////////////////////////////////////////////////////////////////////////////
1121
1122  /**
1123   * Compact the StoreFiles. This method may take some time, so the calling thread must be able to
1124   * block for long periods.
1125   * <p>
1126   * During this time, the Store can work as usual, getting values from StoreFiles and writing new
1127   * StoreFiles from the MemStore. Existing StoreFiles are not destroyed until the new compacted
1128   * StoreFile is completely written-out to disk.
1129   * <p>
1130   * The compactLock prevents multiple simultaneous compactions. The structureLock prevents us from
1131   * interfering with other write operations.
1132   * <p>
1133   * We don't want to hold the structureLock for the whole time, as a compact() can be lengthy and
1134   * we want to allow cache-flushes during this period.
1135   * <p>
1136   * Compaction event should be idempotent, since there is no IO Fencing for the region directory in
1137   * hdfs. A region server might still try to complete the compaction after it lost the region. That
1138   * is why the following events are carefully ordered for a compaction:
1139   * <ol>
1140   * <li>Compaction writes new files under region/.tmp directory (compaction output)</li>
1141   * <li>Compaction atomically moves the temporary file under region directory</li>
1142   * <li>Compaction appends a WAL edit containing the compaction input and output files. Forces sync
1143   * on WAL.</li>
1144   * <li>Compaction deletes the input files from the region directory.</li>
1145   * </ol>
1146   * Failure conditions are handled like this:
1147   * <ul>
1148   * <li>If RS fails before 2, compaction won't complete. Even if RS lives on and finishes the
1149   * compaction later, it will only write the new data file to the region directory. Since we
1150   * already have this data, this will be idempotent, but we will have a redundant copy of the
1151   * data.</li>
1152   * <li>If RS fails between 2 and 3, the region will have a redundant copy of the data. The RS that
1153   * failed won't be able to finish sync() for WAL because of lease recovery in WAL.</li>
1154   * <li>If RS fails after 3, the region server who opens the region will pick up the compaction
1155   * marker from the WAL and replay it by removing the compaction input files. Failed RS can also
1156   * attempt to delete those files, but the operation will be idempotent</li>
1157   * </ul>
1158   * See HBASE-2231 for details.
1159   * @param compaction compaction details obtained from requestCompaction()
1160   * @return The storefiles that we compacted into or null if we failed or opted out early.
1161   */
1162  public List<HStoreFile> compact(CompactionContext compaction,
1163    ThroughputController throughputController, User user) throws IOException {
1164    assert compaction != null;
1165    CompactionRequestImpl cr = compaction.getRequest();
1166    StoreFileWriterCreationTracker writerCreationTracker =
1167      storeFileWriterCreationTrackerFactory.get();
1168    if (writerCreationTracker != null) {
1169      cr.setWriterCreationTracker(writerCreationTracker);
1170      storeFileWriterCreationTrackers.add(writerCreationTracker);
1171    }
1172    try {
1173      // Do all sanity checking in here if we have a valid CompactionRequestImpl
1174      // because we need to clean up after it on the way out in a finally
1175      // block below
1176      long compactionStartTime = EnvironmentEdgeManager.currentTime();
1177      assert compaction.hasSelection();
1178      Collection<HStoreFile> filesToCompact = cr.getFiles();
1179      assert !filesToCompact.isEmpty();
1180      synchronized (filesCompacting) {
1181        // sanity check: we're compacting files that this store knows about
1182        // TODO: change this to LOG.error() after more debugging
1183        Preconditions.checkArgument(filesCompacting.containsAll(filesToCompact));
1184      }
1185
1186      // Ready to go. Have list of files to compact.
1187      LOG.info("Starting compaction of " + filesToCompact + " into tmpdir="
1188        + getRegionFileSystem().getTempDir() + ", totalSize="
1189        + TraditionalBinaryPrefix.long2String(cr.getSize(), "", 1));
1190
1191      return doCompaction(cr, filesToCompact, user, compactionStartTime,
1192        compaction.compact(throughputController, user));
1193    } finally {
1194      finishCompactionRequest(cr);
1195    }
1196  }
1197
1198  protected List<HStoreFile> doCompaction(CompactionRequestImpl cr,
1199    Collection<HStoreFile> filesToCompact, User user, long compactionStartTime, List<Path> newFiles)
1200    throws IOException {
1201    // Do the steps necessary to complete the compaction.
1202    setStoragePolicyFromFileName(newFiles);
1203    List<HStoreFile> sfs = storeEngine.commitStoreFiles(newFiles, true, true);
1204    if (this.getCoprocessorHost() != null) {
1205      for (HStoreFile sf : sfs) {
1206        getCoprocessorHost().postCompact(this, sf, cr.getTracker(), cr, user);
1207      }
1208    }
1209    replaceStoreFiles(filesToCompact, sfs, true);
1210
1211    long outputBytes = getTotalSize(sfs);
1212
1213    // At this point the store will use new files for all new scanners.
1214    refreshStoreSizeAndTotalBytes(); // update store size.
1215
1216    long now = EnvironmentEdgeManager.currentTime();
1217    if (
1218      region.getRegionServerServices() != null
1219        && region.getRegionServerServices().getMetrics() != null
1220    ) {
1221      region.getRegionServerServices().getMetrics().updateCompaction(
1222        region.getTableDescriptor().getTableName().getNameAsString(), cr.isMajor(),
1223        now - compactionStartTime, cr.getFiles().size(), newFiles.size(), cr.getSize(),
1224        outputBytes);
1225
1226    }
1227
1228    logCompactionEndMessage(cr, sfs, now, compactionStartTime);
1229    return sfs;
1230  }
1231
1232  // Set correct storage policy from the file name of DTCP.
1233  // Rename file will not change the storage policy.
1234  private void setStoragePolicyFromFileName(List<Path> newFiles) throws IOException {
1235    String prefix = HConstants.STORAGE_POLICY_PREFIX;
1236    for (Path newFile : newFiles) {
1237      if (newFile.getParent().getName().startsWith(prefix)) {
1238        CommonFSUtils.setStoragePolicy(getRegionFileSystem().getFileSystem(), newFile,
1239          newFile.getParent().getName().substring(prefix.length()));
1240      }
1241    }
1242  }
1243
1244  /**
1245   * Writes the compaction WAL record.
1246   * @param filesCompacted Files compacted (input).
1247   * @param newFiles       Files from compaction.
1248   */
1249  private void writeCompactionWalRecord(Collection<HStoreFile> filesCompacted,
1250    Collection<HStoreFile> newFiles) throws IOException {
1251    if (region.getWAL() == null) {
1252      return;
1253    }
1254    List<Path> inputPaths =
1255      filesCompacted.stream().map(HStoreFile::getPath).collect(Collectors.toList());
1256    List<Path> outputPaths =
1257      newFiles.stream().map(HStoreFile::getPath).collect(Collectors.toList());
1258    RegionInfo info = this.region.getRegionInfo();
1259    CompactionDescriptor compactionDescriptor = ProtobufUtil.toCompactionDescriptor(info,
1260      getColumnFamilyDescriptor().getName(), inputPaths, outputPaths,
1261      getRegionFileSystem().getStoreDir(getColumnFamilyDescriptor().getNameAsString()));
1262    // Fix reaching into Region to get the maxWaitForSeqId.
1263    // Does this method belong in Region altogether given it is making so many references up there?
1264    // Could be Region#writeCompactionMarker(compactionDescriptor);
1265    WALUtil.writeCompactionMarker(this.region.getWAL(), this.region.getReplicationScope(),
1266      this.region.getRegionInfo(), compactionDescriptor, this.region.getMVCC(),
1267      region.getRegionReplicationSink().orElse(null));
1268  }
1269
1270  @RestrictedApi(explanation = "Should only be called in TestHStore", link = "",
1271      allowedOnPath = ".*/(HStore|TestHStore).java")
1272  void replaceStoreFiles(Collection<HStoreFile> compactedFiles, Collection<HStoreFile> result,
1273    boolean writeCompactionMarker) throws IOException {
1274    storeEngine.replaceStoreFiles(compactedFiles, result, () -> {
1275      if (writeCompactionMarker) {
1276        writeCompactionWalRecord(compactedFiles, result);
1277      }
1278    }, () -> {
1279      synchronized (filesCompacting) {
1280        filesCompacting.removeAll(compactedFiles);
1281      }
1282    });
1283    // These may be null when the RS is shutting down. The space quota Chores will fix the Region
1284    // sizes later so it's not super-critical if we miss these.
1285    RegionServerServices rsServices = region.getRegionServerServices();
1286    if (rsServices != null && rsServices.getRegionServerSpaceQuotaManager() != null) {
1287      updateSpaceQuotaAfterFileReplacement(
1288        rsServices.getRegionServerSpaceQuotaManager().getRegionSizeStore(), getRegionInfo(),
1289        compactedFiles, result);
1290    }
1291  }
1292
1293  /**
1294   * Updates the space quota usage for this region, removing the size for files compacted away and
1295   * adding in the size for new files.
1296   * @param sizeStore  The object tracking changes in region size for space quotas.
1297   * @param regionInfo The identifier for the region whose size is being updated.
1298   * @param oldFiles   Files removed from this store's region.
1299   * @param newFiles   Files added to this store's region.
1300   */
1301  void updateSpaceQuotaAfterFileReplacement(RegionSizeStore sizeStore, RegionInfo regionInfo,
1302    Collection<HStoreFile> oldFiles, Collection<HStoreFile> newFiles) {
1303    long delta = 0;
1304    if (oldFiles != null) {
1305      for (HStoreFile compactedFile : oldFiles) {
1306        if (compactedFile.isHFile()) {
1307          delta -= compactedFile.getReader().length();
1308        }
1309      }
1310    }
1311    if (newFiles != null) {
1312      for (HStoreFile newFile : newFiles) {
1313        if (newFile.isHFile()) {
1314          delta += newFile.getReader().length();
1315        }
1316      }
1317    }
1318    sizeStore.incrementRegionSize(regionInfo, delta);
1319  }
1320
1321  /**
1322   * Log a very elaborate compaction completion message.
1323   * @param cr                  Request.
1324   * @param sfs                 Resulting files.
1325   * @param compactionStartTime Start time.
1326   */
1327  private void logCompactionEndMessage(CompactionRequestImpl cr, List<HStoreFile> sfs, long now,
1328    long compactionStartTime) {
1329    StringBuilder message = new StringBuilder("Completed" + (cr.isMajor() ? " major" : "")
1330      + " compaction of " + cr.getFiles().size() + (cr.isAllFiles() ? " (all)" : "")
1331      + " file(s) in " + this + " of " + this.getRegionInfo().getShortNameToLog() + " into ");
1332    if (sfs.isEmpty()) {
1333      message.append("none, ");
1334    } else {
1335      for (HStoreFile sf : sfs) {
1336        message.append(sf.getPath().getName());
1337        message.append("(size=");
1338        message.append(TraditionalBinaryPrefix.long2String(sf.getReader().length(), "", 1));
1339        message.append("), ");
1340      }
1341    }
1342    message.append("total size for store is ")
1343      .append(StringUtils.TraditionalBinaryPrefix.long2String(storeSize.get(), "", 1))
1344      .append(". This selection was in queue for ")
1345      .append(StringUtils.formatTimeDiff(compactionStartTime, cr.getSelectionTime()))
1346      .append(", and took ").append(StringUtils.formatTimeDiff(now, compactionStartTime))
1347      .append(" to execute.");
1348    LOG.info(message.toString());
1349    if (LOG.isTraceEnabled()) {
1350      int fileCount = storeEngine.getStoreFileManager().getStorefileCount();
1351      long resultSize = getTotalSize(sfs);
1352      String traceMessage = "COMPACTION start,end,size out,files in,files out,store size,"
1353        + "store files [" + compactionStartTime + "," + now + "," + resultSize + ","
1354        + cr.getFiles().size() + "," + sfs.size() + "," + storeSize + "," + fileCount + "]";
1355      LOG.trace(traceMessage);
1356    }
1357  }
1358
1359  /**
1360   * Call to complete a compaction. Its for the case where we find in the WAL a compaction that was
1361   * not finished. We could find one recovering a WAL after a regionserver crash. See HBASE-2231.
1362   */
1363  public void replayCompactionMarker(CompactionDescriptor compaction, boolean pickCompactionFiles,
1364    boolean removeFiles) throws IOException {
1365    LOG.debug("Completing compaction from the WAL marker");
1366    List<String> compactionInputs = compaction.getCompactionInputList();
1367    List<String> compactionOutputs = Lists.newArrayList(compaction.getCompactionOutputList());
1368
1369    // The Compaction Marker is written after the compaction is completed,
1370    // and the files moved into the region/family folder.
1371    //
1372    // If we crash after the entry is written, we may not have removed the
1373    // input files, but the output file is present.
1374    // (The unremoved input files will be removed by this function)
1375    //
1376    // If we scan the directory and the file is not present, it can mean that:
1377    // - The file was manually removed by the user
1378    // - The file was removed as consequence of subsequent compaction
1379    // so, we can't do anything with the "compaction output list" because those
1380    // files have already been loaded when opening the region (by virtue of
1381    // being in the store's folder) or they may be missing due to a compaction.
1382
1383    String familyName = this.getColumnFamilyName();
1384    Set<String> inputFiles = new HashSet<>();
1385    for (String compactionInput : compactionInputs) {
1386      Path inputPath = getRegionFileSystem().getStoreFilePath(familyName, compactionInput);
1387      inputFiles.add(inputPath.getName());
1388    }
1389
1390    // some of the input files might already be deleted
1391    List<HStoreFile> inputStoreFiles = new ArrayList<>(compactionInputs.size());
1392    for (HStoreFile sf : this.getStorefiles()) {
1393      if (inputFiles.contains(sf.getPath().getName())) {
1394        inputStoreFiles.add(sf);
1395      }
1396    }
1397
1398    // check whether we need to pick up the new files
1399    List<HStoreFile> outputStoreFiles = new ArrayList<>(compactionOutputs.size());
1400
1401    if (pickCompactionFiles) {
1402      for (HStoreFile sf : this.getStorefiles()) {
1403        compactionOutputs.remove(sf.getPath().getName());
1404      }
1405      for (String compactionOutput : compactionOutputs) {
1406        StoreFileTracker sft = StoreFileTrackerFactory.create(conf, false, storeContext);
1407        StoreFileInfo storeFileInfo =
1408          getRegionFileSystem().getStoreFileInfo(getColumnFamilyName(), compactionOutput, sft);
1409        HStoreFile storeFile = storeEngine.createStoreFileAndReader(storeFileInfo);
1410        outputStoreFiles.add(storeFile);
1411      }
1412    }
1413
1414    if (!inputStoreFiles.isEmpty() || !outputStoreFiles.isEmpty()) {
1415      LOG.info("Replaying compaction marker, replacing input files: " + inputStoreFiles
1416        + " with output files : " + outputStoreFiles);
1417      this.replaceStoreFiles(inputStoreFiles, outputStoreFiles, false);
1418      this.refreshStoreSizeAndTotalBytes();
1419    }
1420  }
1421
1422  @Override
1423  public boolean hasReferences() {
1424    // Grab the read lock here, because we need to ensure that: only when the atomic
1425    // replaceStoreFiles(..) finished, we can get all the complete store file list.
1426    this.storeEngine.readLock();
1427    try {
1428      // Merge the current store files with compacted files here due to HBASE-20940.
1429      Collection<HStoreFile> allStoreFiles = new ArrayList<>(getStorefiles());
1430      allStoreFiles.addAll(getCompactedFiles());
1431      return StoreUtils.hasReferences(allStoreFiles);
1432    } finally {
1433      this.storeEngine.readUnlock();
1434    }
1435  }
1436
1437  /**
1438   * getter for CompactionProgress object
1439   * @return CompactionProgress object; can be null
1440   */
1441  public CompactionProgress getCompactionProgress() {
1442    return this.storeEngine.getCompactor().getProgress();
1443  }
1444
1445  @Override
1446  public boolean shouldPerformMajorCompaction() throws IOException {
1447    for (HStoreFile sf : this.storeEngine.getStoreFileManager().getStoreFiles()) {
1448      // TODO: what are these reader checks all over the place?
1449      if (sf.getReader() == null) {
1450        LOG.debug("StoreFile {} has null Reader", sf);
1451        return false;
1452      }
1453    }
1454    return storeEngine.getCompactionPolicy()
1455      .shouldPerformMajorCompaction(this.storeEngine.getStoreFileManager().getStoreFiles());
1456  }
1457
1458  public Optional<CompactionContext> requestCompaction() throws IOException {
1459    return requestCompaction(NO_PRIORITY, CompactionLifeCycleTracker.DUMMY, null);
1460  }
1461
1462  public Optional<CompactionContext> requestCompaction(int priority,
1463    CompactionLifeCycleTracker tracker, User user) throws IOException {
1464    // don't even select for compaction if writes are disabled
1465    if (!this.areWritesEnabled()) {
1466      return Optional.empty();
1467    }
1468    // Before we do compaction, try to get rid of unneeded files to simplify things.
1469    removeUnneededFiles();
1470
1471    final CompactionContext compaction = storeEngine.createCompaction();
1472    CompactionRequestImpl request = null;
1473    this.storeEngine.readLock();
1474    try {
1475      synchronized (filesCompacting) {
1476        // First, see if coprocessor would want to override selection.
1477        if (this.getCoprocessorHost() != null) {
1478          final List<HStoreFile> candidatesForCoproc = compaction.preSelect(this.filesCompacting);
1479          boolean override =
1480            getCoprocessorHost().preCompactSelection(this, candidatesForCoproc, tracker, user);
1481          if (override) {
1482            // Coprocessor is overriding normal file selection.
1483            compaction.forceSelect(new CompactionRequestImpl(candidatesForCoproc));
1484          }
1485        }
1486
1487        // Normal case - coprocessor is not overriding file selection.
1488        if (!compaction.hasSelection()) {
1489          boolean isUserCompaction = priority == Store.PRIORITY_USER;
1490          boolean mayUseOffPeak =
1491            offPeakHours.isOffPeakHour() && offPeakCompactionTracker.compareAndSet(false, true);
1492          try {
1493            compaction.select(this.filesCompacting, isUserCompaction, mayUseOffPeak,
1494              forceMajor && filesCompacting.isEmpty());
1495          } catch (IOException e) {
1496            if (mayUseOffPeak) {
1497              offPeakCompactionTracker.set(false);
1498            }
1499            throw e;
1500          }
1501          assert compaction.hasSelection();
1502          if (mayUseOffPeak && !compaction.getRequest().isOffPeak()) {
1503            // Compaction policy doesn't want to take advantage of off-peak.
1504            offPeakCompactionTracker.set(false);
1505          }
1506        }
1507        if (this.getCoprocessorHost() != null) {
1508          this.getCoprocessorHost().postCompactSelection(this,
1509            ImmutableList.copyOf(compaction.getRequest().getFiles()), tracker,
1510            compaction.getRequest(), user);
1511        }
1512        // Finally, we have the resulting files list. Check if we have any files at all.
1513        request = compaction.getRequest();
1514        Collection<HStoreFile> selectedFiles = request.getFiles();
1515        if (selectedFiles.isEmpty()) {
1516          return Optional.empty();
1517        }
1518
1519        addToCompactingFiles(selectedFiles);
1520
1521        // If we're enqueuing a major, clear the force flag.
1522        this.forceMajor = this.forceMajor && !request.isMajor();
1523
1524        // Set common request properties.
1525        // Set priority, either override value supplied by caller or from store.
1526        final int compactionPriority =
1527          (priority != Store.NO_PRIORITY) ? priority : getCompactPriority();
1528        request.setPriority(compactionPriority);
1529
1530        if (request.isAfterSplit()) {
1531          // If the store belongs to recently splitted daughter regions, better we consider
1532          // them with the higher priority in the compaction queue.
1533          // Override priority if it is lower (higher int value) than
1534          // SPLIT_REGION_COMPACTION_PRIORITY
1535          final int splitHousekeepingPriority =
1536            Math.min(compactionPriority, SPLIT_REGION_COMPACTION_PRIORITY);
1537          request.setPriority(splitHousekeepingPriority);
1538          LOG.info(
1539            "Keeping/Overriding Compaction request priority to {} for CF {} since it"
1540              + " belongs to recently split daughter region {}",
1541            splitHousekeepingPriority, this.getColumnFamilyName(),
1542            getRegionInfo().getRegionNameAsString());
1543        }
1544        request.setDescription(getRegionInfo().getRegionNameAsString(), getColumnFamilyName());
1545        request.setTracker(tracker);
1546      }
1547    } finally {
1548      this.storeEngine.readUnlock();
1549    }
1550
1551    if (LOG.isDebugEnabled()) {
1552      LOG.debug(this + " is initiating " + (request.isMajor() ? "major" : "minor") + " compaction"
1553        + (request.isAllFiles() ? " (all files)" : ""));
1554    }
1555    this.region.reportCompactionRequestStart(request.isMajor());
1556    return Optional.of(compaction);
1557  }
1558
1559  /** Adds the files to compacting files. filesCompacting must be locked. */
1560  private void addToCompactingFiles(Collection<HStoreFile> filesToAdd) {
1561    if (CollectionUtils.isEmpty(filesToAdd)) {
1562      return;
1563    }
1564    // Check that we do not try to compact the same StoreFile twice.
1565    if (!Collections.disjoint(filesCompacting, filesToAdd)) {
1566      Preconditions.checkArgument(false, "%s overlaps with %s", filesToAdd, filesCompacting);
1567    }
1568    filesCompacting.addAll(filesToAdd);
1569    Collections.sort(filesCompacting, storeEngine.getStoreFileManager().getStoreFileComparator());
1570  }
1571
1572  private void removeUnneededFiles() throws IOException {
1573    if (!conf.getBoolean("hbase.store.delete.expired.storefile", true)) {
1574      return;
1575    }
1576    if (getColumnFamilyDescriptor().getMinVersions() > 0) {
1577      LOG.debug("Skipping expired store file removal due to min version of {} being {}", this,
1578        getColumnFamilyDescriptor().getMinVersions());
1579      return;
1580    }
1581    this.storeEngine.readLock();
1582    Collection<HStoreFile> delSfs = null;
1583    try {
1584      synchronized (filesCompacting) {
1585        long cfTtl = getStoreFileTtl();
1586        if (cfTtl != Long.MAX_VALUE) {
1587          delSfs = storeEngine.getStoreFileManager()
1588            .getUnneededFiles(EnvironmentEdgeManager.currentTime() - cfTtl, filesCompacting);
1589          addToCompactingFiles(delSfs);
1590        }
1591      }
1592    } finally {
1593      this.storeEngine.readUnlock();
1594    }
1595
1596    if (CollectionUtils.isEmpty(delSfs)) {
1597      return;
1598    }
1599
1600    Collection<HStoreFile> newFiles = Collections.emptyList(); // No new files.
1601    replaceStoreFiles(delSfs, newFiles, true);
1602    refreshStoreSizeAndTotalBytes();
1603    LOG.info("Completed removal of " + delSfs.size() + " unnecessary (expired) file(s) in " + this
1604      + "; total size is " + TraditionalBinaryPrefix.long2String(storeSize.get(), "", 1));
1605  }
1606
1607  public void cancelRequestedCompaction(CompactionContext compaction) {
1608    finishCompactionRequest(compaction.getRequest());
1609  }
1610
1611  private void finishCompactionRequest(CompactionRequestImpl cr) {
1612    this.region.reportCompactionRequestEnd(cr.isMajor(), cr.getFiles().size(), cr.getSize());
1613    if (cr.isOffPeak()) {
1614      offPeakCompactionTracker.set(false);
1615      cr.setOffPeak(false);
1616    }
1617    synchronized (filesCompacting) {
1618      filesCompacting.removeAll(cr.getFiles());
1619    }
1620    // The tracker could be null, for example, we do not need to track the creation of store file
1621    // writer due to different implementation of SFT, or the compaction is canceled.
1622    if (cr.getWriterCreationTracker() != null) {
1623      storeFileWriterCreationTrackers.remove(cr.getWriterCreationTracker());
1624    }
1625  }
1626
1627  /**
1628   * Update counts.
1629   */
1630  protected void refreshStoreSizeAndTotalBytes() throws IOException {
1631    this.storeSize.set(0L);
1632    this.totalUncompressedBytes.set(0L);
1633    for (HStoreFile hsf : this.storeEngine.getStoreFileManager().getStoreFiles()) {
1634      StoreFileReader r = hsf.getReader();
1635      if (r == null) {
1636        LOG.debug("StoreFile {} has a null Reader", hsf);
1637        continue;
1638      }
1639      this.storeSize.addAndGet(r.length());
1640      this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
1641    }
1642  }
1643
1644  /*
1645   * @param wantedVersions How many versions were asked for.
1646   * @return wantedVersions or this families' {@link HConstants#VERSIONS}.
1647   */
1648  int versionsToReturn(final int wantedVersions) {
1649    if (wantedVersions <= 0) {
1650      throw new IllegalArgumentException("Number of versions must be > 0");
1651    }
1652    // Make sure we do not return more than maximum versions for this store.
1653    int maxVersions = getColumnFamilyDescriptor().getMaxVersions();
1654    return wantedVersions > maxVersions ? maxVersions : wantedVersions;
1655  }
1656
1657  @Override
1658  public boolean canSplit() {
1659    // Not split-able if we find a reference store file present in the store.
1660    boolean result = !hasReferences();
1661    if (!result) {
1662      LOG.trace("Not splittable; has references: {}", this);
1663    }
1664    return result;
1665  }
1666
1667  /**
1668   * Determines if Store should be split.
1669   */
1670  public Optional<byte[]> getSplitPoint() {
1671    this.storeEngine.readLock();
1672    try {
1673      // Should already be enforced by the split policy!
1674      assert !this.getRegionInfo().isMetaRegion();
1675      // Not split-able if we find a reference store file present in the store.
1676      if (hasReferences()) {
1677        LOG.trace("Not splittable; has references: {}", this);
1678        return Optional.empty();
1679      }
1680      return this.storeEngine.getStoreFileManager().getSplitPoint();
1681    } catch (IOException e) {
1682      LOG.warn("Failed getting store size for {}", this, e);
1683    } finally {
1684      this.storeEngine.readUnlock();
1685    }
1686    return Optional.empty();
1687  }
1688
1689  @Override
1690  public long getLastCompactSize() {
1691    return this.lastCompactSize;
1692  }
1693
1694  @Override
1695  public long getSize() {
1696    return storeSize.get();
1697  }
1698
1699  public void triggerMajorCompaction() {
1700    this.forceMajor = true;
1701  }
1702
1703  //////////////////////////////////////////////////////////////////////////////
1704  // File administration
1705  //////////////////////////////////////////////////////////////////////////////
1706
1707  /**
1708   * Return a scanner for both the memstore and the HStore files. Assumes we are not in a
1709   * compaction.
1710   * @param scan       Scan to apply when scanning the stores
1711   * @param targetCols columns to scan
1712   * @return a scanner over the current key values
1713   * @throws IOException on failure
1714   */
1715  public KeyValueScanner getScanner(Scan scan, final NavigableSet<byte[]> targetCols, long readPt)
1716    throws IOException {
1717    storeEngine.readLock();
1718    try {
1719      ScanInfo scanInfo;
1720      if (this.getCoprocessorHost() != null) {
1721        scanInfo = this.getCoprocessorHost().preStoreScannerOpen(this, scan);
1722      } else {
1723        scanInfo = getScanInfo();
1724      }
1725      return createScanner(scan, scanInfo, targetCols, readPt);
1726    } finally {
1727      storeEngine.readUnlock();
1728    }
1729  }
1730
1731  // HMobStore will override this method to return its own implementation.
1732  protected KeyValueScanner createScanner(Scan scan, ScanInfo scanInfo,
1733    NavigableSet<byte[]> targetCols, long readPt) throws IOException {
1734    return scan.isReversed()
1735      ? new ReversedStoreScanner(this, scanInfo, scan, targetCols, readPt)
1736      : new StoreScanner(this, scanInfo, scan, targetCols, readPt);
1737  }
1738
1739  /**
1740   * Recreates the scanners on the current list of active store file scanners
1741   * @param currentFileScanners    the current set of active store file scanners
1742   * @param cacheBlocks            cache the blocks or not
1743   * @param usePread               use pread or not
1744   * @param isCompaction           is the scanner for compaction
1745   * @param matcher                the scan query matcher
1746   * @param startRow               the scan's start row
1747   * @param includeStartRow        should the scan include the start row
1748   * @param stopRow                the scan's stop row
1749   * @param includeStopRow         should the scan include the stop row
1750   * @param readPt                 the read point of the current scane
1751   * @param includeMemstoreScanner whether the current scanner should include memstorescanner
1752   * @return list of scanners recreated on the current Scanners
1753   */
1754  public List<KeyValueScanner> recreateScanners(List<KeyValueScanner> currentFileScanners,
1755    boolean cacheBlocks, boolean usePread, boolean isCompaction, ScanQueryMatcher matcher,
1756    byte[] startRow, boolean includeStartRow, byte[] stopRow, boolean includeStopRow, long readPt,
1757    boolean includeMemstoreScanner) throws IOException {
1758    this.storeEngine.readLock();
1759    try {
1760      Map<String, HStoreFile> name2File =
1761        new HashMap<>(getStorefilesCount() + getCompactedFilesCount());
1762      for (HStoreFile file : getStorefiles()) {
1763        name2File.put(file.getFileInfo().getActiveFileName(), file);
1764      }
1765      Collection<HStoreFile> compactedFiles = getCompactedFiles();
1766      for (HStoreFile file : IterableUtils.emptyIfNull(compactedFiles)) {
1767        name2File.put(file.getFileInfo().getActiveFileName(), file);
1768      }
1769      List<HStoreFile> filesToReopen = new ArrayList<>();
1770      for (KeyValueScanner kvs : currentFileScanners) {
1771        assert kvs.isFileScanner();
1772        if (kvs.peek() == null) {
1773          continue;
1774        }
1775        filesToReopen.add(name2File.get(kvs.getFilePath().getName()));
1776      }
1777      if (filesToReopen.isEmpty()) {
1778        return null;
1779      }
1780      return getScanners(filesToReopen, cacheBlocks, false, false, matcher, startRow,
1781        includeStartRow, stopRow, includeStopRow, readPt, false, false);
1782    } finally {
1783      this.storeEngine.readUnlock();
1784    }
1785  }
1786
1787  @Override
1788  public String toString() {
1789    return this.getRegionInfo().getShortNameToLog() + "/" + this.getColumnFamilyName();
1790  }
1791
1792  @Override
1793  public int getStorefilesCount() {
1794    return this.storeEngine.getStoreFileManager().getStorefileCount();
1795  }
1796
1797  @Override
1798  public int getCompactedFilesCount() {
1799    return this.storeEngine.getStoreFileManager().getCompactedFilesCount();
1800  }
1801
1802  private LongStream getStoreFileAgeStream() {
1803    return this.storeEngine.getStoreFileManager().getStoreFiles().stream().filter(sf -> {
1804      if (sf.getReader() == null) {
1805        LOG.debug("StoreFile {} has a null Reader", sf);
1806        return false;
1807      } else {
1808        return true;
1809      }
1810    }).filter(HStoreFile::isHFile).mapToLong(sf -> sf.getFileInfo().getCreatedTimestamp())
1811      .map(t -> EnvironmentEdgeManager.currentTime() - t);
1812  }
1813
1814  @Override
1815  public OptionalLong getMaxStoreFileAge() {
1816    return getStoreFileAgeStream().max();
1817  }
1818
1819  @Override
1820  public OptionalLong getMinStoreFileAge() {
1821    return getStoreFileAgeStream().min();
1822  }
1823
1824  @Override
1825  public OptionalDouble getAvgStoreFileAge() {
1826    return getStoreFileAgeStream().average();
1827  }
1828
1829  @Override
1830  public long getNumReferenceFiles() {
1831    return this.storeEngine.getStoreFileManager().getStoreFiles().stream()
1832      .filter(HStoreFile::isReference).count();
1833  }
1834
1835  @Override
1836  public long getNumHFiles() {
1837    return this.storeEngine.getStoreFileManager().getStoreFiles().stream()
1838      .filter(HStoreFile::isHFile).count();
1839  }
1840
1841  @Override
1842  public long getStoreSizeUncompressed() {
1843    return this.totalUncompressedBytes.get();
1844  }
1845
1846  @Override
1847  public long getStorefilesSize() {
1848    // Include all StoreFiles
1849    return StoreUtils.getStorefilesSize(this.storeEngine.getStoreFileManager().getStoreFiles(),
1850      sf -> true);
1851  }
1852
1853  @Override
1854  public long getHFilesSize() {
1855    // Include only StoreFiles which are HFiles
1856    return StoreUtils.getStorefilesSize(this.storeEngine.getStoreFileManager().getStoreFiles(),
1857      HStoreFile::isHFile);
1858  }
1859
1860  private long getStorefilesFieldSize(ToLongFunction<StoreFileReader> f) {
1861    return this.storeEngine.getStoreFileManager().getStoreFiles().stream()
1862      .mapToLong(file -> StoreUtils.getStorefileFieldSize(file, f)).sum();
1863  }
1864
1865  @Override
1866  public long getStorefilesRootLevelIndexSize() {
1867    return getStorefilesFieldSize(StoreFileReader::indexSize);
1868  }
1869
1870  @Override
1871  public long getTotalStaticIndexSize() {
1872    return getStorefilesFieldSize(StoreFileReader::getUncompressedDataIndexSize);
1873  }
1874
1875  @Override
1876  public long getTotalStaticBloomSize() {
1877    return getStorefilesFieldSize(StoreFileReader::getTotalBloomSize);
1878  }
1879
1880  @Override
1881  public MemStoreSize getMemStoreSize() {
1882    return this.memstore.size();
1883  }
1884
1885  @Override
1886  public int getCompactPriority() {
1887    int priority = this.storeEngine.getStoreFileManager().getStoreCompactionPriority();
1888    if (priority == PRIORITY_USER) {
1889      LOG.warn("Compaction priority is USER despite there being no user compaction");
1890    }
1891    return priority;
1892  }
1893
1894  public boolean throttleCompaction(long compactionSize) {
1895    return storeEngine.getCompactionPolicy().throttleCompaction(compactionSize);
1896  }
1897
1898  public HRegion getHRegion() {
1899    return this.region;
1900  }
1901
1902  public RegionCoprocessorHost getCoprocessorHost() {
1903    return this.region.getCoprocessorHost();
1904  }
1905
1906  @Override
1907  public RegionInfo getRegionInfo() {
1908    return getRegionFileSystem().getRegionInfo();
1909  }
1910
1911  @Override
1912  public boolean areWritesEnabled() {
1913    return this.region.areWritesEnabled();
1914  }
1915
1916  @Override
1917  public long getSmallestReadPoint() {
1918    return this.region.getSmallestReadPoint();
1919  }
1920
1921  /**
1922   * Adds or replaces the specified KeyValues.
1923   * <p>
1924   * For each KeyValue specified, if a cell with the same row, family, and qualifier exists in
1925   * MemStore, it will be replaced. Otherwise, it will just be inserted to MemStore.
1926   * <p>
1927   * This operation is atomic on each KeyValue (row/family/qualifier) but not necessarily atomic
1928   * across all of them.
1929   * @param readpoint readpoint below which we can safely remove duplicate KVs
1930   */
1931  public void upsert(Iterable<ExtendedCell> cells, long readpoint, MemStoreSizing memstoreSizing) {
1932    this.storeEngine.readLock();
1933    try {
1934      this.memstore.upsert(cells, readpoint, memstoreSizing);
1935    } finally {
1936      this.storeEngine.readUnlock();
1937    }
1938  }
1939
1940  public StoreFlushContext createFlushContext(long cacheFlushId, FlushLifeCycleTracker tracker) {
1941    return new StoreFlusherImpl(cacheFlushId, tracker);
1942  }
1943
1944  private final class StoreFlusherImpl implements StoreFlushContext {
1945
1946    private final FlushLifeCycleTracker tracker;
1947    private final StoreFileWriterCreationTracker writerCreationTracker;
1948    private final long cacheFlushSeqNum;
1949    private MemStoreSnapshot snapshot;
1950    private List<Path> tempFiles;
1951    private List<Path> committedFiles;
1952    private long cacheFlushCount;
1953    private long cacheFlushSize;
1954    private long outputFileSize;
1955
1956    private StoreFlusherImpl(long cacheFlushSeqNum, FlushLifeCycleTracker tracker) {
1957      this.cacheFlushSeqNum = cacheFlushSeqNum;
1958      this.tracker = tracker;
1959      this.writerCreationTracker = storeFileWriterCreationTrackerFactory.get();
1960    }
1961
1962    /**
1963     * This is not thread safe. The caller should have a lock on the region or the store. If
1964     * necessary, the lock can be added with the patch provided in HBASE-10087
1965     */
1966    @Override
1967    public MemStoreSize prepare() {
1968      // passing the current sequence number of the wal - to allow bookkeeping in the memstore
1969      this.snapshot = memstore.snapshot();
1970      this.cacheFlushCount = snapshot.getCellsCount();
1971      this.cacheFlushSize = snapshot.getDataSize();
1972      committedFiles = new ArrayList<>(1);
1973      return snapshot.getMemStoreSize();
1974    }
1975
1976    @Override
1977    public void flushCache(MonitoredTask status) throws IOException {
1978      RegionServerServices rsService = region.getRegionServerServices();
1979      ThroughputController throughputController =
1980        rsService == null ? null : rsService.getFlushThroughputController();
1981      // it could be null if we do not need to track the creation of store file writer due to
1982      // different SFT implementation.
1983      if (writerCreationTracker != null) {
1984        HStore.this.storeFileWriterCreationTrackers.add(writerCreationTracker);
1985      }
1986      tempFiles = HStore.this.flushCache(cacheFlushSeqNum, snapshot, status, throughputController,
1987        tracker, writerCreationTracker);
1988    }
1989
1990    @Override
1991    public boolean commit(MonitoredTask status) throws IOException {
1992      try {
1993        if (CollectionUtils.isEmpty(this.tempFiles)) {
1994          return false;
1995        }
1996        status.setStatus("Flushing " + this + ": reopening flushed file");
1997        List<HStoreFile> storeFiles = storeEngine.commitStoreFiles(tempFiles, false, false);
1998        for (HStoreFile sf : storeFiles) {
1999          StoreFileReader r = sf.getReader();
2000          if (LOG.isInfoEnabled()) {
2001            LOG.info("Added {}, entries={}, sequenceid={}, filesize={}", sf, r.getEntries(),
2002              cacheFlushSeqNum, TraditionalBinaryPrefix.long2String(r.length(), "", 1));
2003          }
2004          outputFileSize += r.length();
2005          storeSize.addAndGet(r.length());
2006          totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
2007          committedFiles.add(sf.getPath());
2008        }
2009
2010        flushedCellsCount.addAndGet(cacheFlushCount);
2011        flushedCellsSize.addAndGet(cacheFlushSize);
2012        flushedOutputFileSize.addAndGet(outputFileSize);
2013        // call coprocessor after we have done all the accounting above
2014        for (HStoreFile sf : storeFiles) {
2015          if (getCoprocessorHost() != null) {
2016            getCoprocessorHost().postFlush(HStore.this, sf, tracker);
2017          }
2018        }
2019        // Add new file to store files. Clear snapshot too while we have the Store write lock.
2020        return completeFlush(storeFiles, snapshot.getId());
2021      } finally {
2022        if (writerCreationTracker != null) {
2023          HStore.this.storeFileWriterCreationTrackers.remove(writerCreationTracker);
2024        }
2025      }
2026    }
2027
2028    @Override
2029    public long getOutputFileSize() {
2030      return outputFileSize;
2031    }
2032
2033    @Override
2034    public List<Path> getCommittedFiles() {
2035      return committedFiles;
2036    }
2037
2038    /**
2039     * Similar to commit, but called in secondary region replicas for replaying the flush cache from
2040     * primary region. Adds the new files to the store, and drops the snapshot depending on
2041     * dropMemstoreSnapshot argument.
2042     * @param fileNames            names of the flushed files
2043     * @param dropMemstoreSnapshot whether to drop the prepared memstore snapshot
2044     */
2045    @Override
2046    public void replayFlush(List<String> fileNames, boolean dropMemstoreSnapshot)
2047      throws IOException {
2048      List<HStoreFile> storeFiles = new ArrayList<>(fileNames.size());
2049      for (String file : fileNames) {
2050        // open the file as a store file (hfile link, etc)
2051        StoreFileTracker sft = StoreFileTrackerFactory.create(conf, false, storeContext);
2052        StoreFileInfo storeFileInfo =
2053          getRegionFileSystem().getStoreFileInfo(getColumnFamilyName(), file, sft);
2054        HStoreFile storeFile = storeEngine.createStoreFileAndReader(storeFileInfo);
2055        storeFiles.add(storeFile);
2056        HStore.this.storeSize.addAndGet(storeFile.getReader().length());
2057        HStore.this.totalUncompressedBytes
2058          .addAndGet(storeFile.getReader().getTotalUncompressedBytes());
2059        if (LOG.isInfoEnabled()) {
2060          LOG.info(this + " added " + storeFile + ", entries=" + storeFile.getReader().getEntries()
2061            + ", sequenceid=" + storeFile.getReader().getSequenceID() + ", filesize="
2062            + TraditionalBinaryPrefix.long2String(storeFile.getReader().length(), "", 1));
2063        }
2064      }
2065
2066      long snapshotId = -1; // -1 means do not drop
2067      if (dropMemstoreSnapshot && snapshot != null) {
2068        snapshotId = snapshot.getId();
2069      }
2070      HStore.this.completeFlush(storeFiles, snapshotId);
2071    }
2072
2073    /**
2074     * Abort the snapshot preparation. Drops the snapshot if any.
2075     */
2076    @Override
2077    public void abort() throws IOException {
2078      if (snapshot != null) {
2079        HStore.this.completeFlush(Collections.emptyList(), snapshot.getId());
2080      }
2081    }
2082  }
2083
2084  @Override
2085  public boolean needsCompaction() {
2086    List<HStoreFile> filesCompactingClone = null;
2087    synchronized (filesCompacting) {
2088      filesCompactingClone = Lists.newArrayList(filesCompacting);
2089    }
2090    return this.storeEngine.needsCompaction(filesCompactingClone);
2091  }
2092
2093  /**
2094   * Used for tests.
2095   * @return cache configuration for this Store.
2096   */
2097  public CacheConfig getCacheConfig() {
2098    return storeContext.getCacheConf();
2099  }
2100
2101  public static final long FIXED_OVERHEAD = ClassSize.estimateBase(HStore.class, false);
2102
2103  public static final long DEEP_OVERHEAD = ClassSize.align(
2104    FIXED_OVERHEAD + ClassSize.OBJECT + ClassSize.REENTRANT_LOCK + ClassSize.CONCURRENT_SKIPLISTMAP
2105      + ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY + ClassSize.OBJECT + ScanInfo.FIXED_OVERHEAD);
2106
2107  @Override
2108  public long heapSize() {
2109    MemStoreSize memstoreSize = this.memstore.size();
2110    return DEEP_OVERHEAD + memstoreSize.getHeapSize() + storeContext.heapSize();
2111  }
2112
2113  @Override
2114  public CellComparator getComparator() {
2115    return storeContext.getComparator();
2116  }
2117
2118  public ScanInfo getScanInfo() {
2119    return scanInfo;
2120  }
2121
2122  /**
2123   * Set scan info, used by test
2124   * @param scanInfo new scan info to use for test
2125   */
2126  void setScanInfo(ScanInfo scanInfo) {
2127    this.scanInfo = scanInfo;
2128  }
2129
2130  @Override
2131  public boolean hasTooManyStoreFiles() {
2132    return getStorefilesCount() > this.blockingFileCount;
2133  }
2134
2135  @Override
2136  public long getFlushedCellsCount() {
2137    return flushedCellsCount.get();
2138  }
2139
2140  @Override
2141  public long getFlushedCellsSize() {
2142    return flushedCellsSize.get();
2143  }
2144
2145  @Override
2146  public long getFlushedOutputFileSize() {
2147    return flushedOutputFileSize.get();
2148  }
2149
2150  @Override
2151  public long getCompactedCellsCount() {
2152    return compactedCellsCount.get();
2153  }
2154
2155  @Override
2156  public long getCompactedCellsSize() {
2157    return compactedCellsSize.get();
2158  }
2159
2160  @Override
2161  public long getMajorCompactedCellsCount() {
2162    return majorCompactedCellsCount.get();
2163  }
2164
2165  @Override
2166  public long getMajorCompactedCellsSize() {
2167    return majorCompactedCellsSize.get();
2168  }
2169
2170  public void updateCompactedMetrics(boolean isMajor, CompactionProgress progress) {
2171    if (isMajor) {
2172      majorCompactedCellsCount.addAndGet(progress.getTotalCompactingKVs());
2173      majorCompactedCellsSize.addAndGet(progress.totalCompactedSize);
2174    } else {
2175      compactedCellsCount.addAndGet(progress.getTotalCompactingKVs());
2176      compactedCellsSize.addAndGet(progress.totalCompactedSize);
2177    }
2178  }
2179
2180  /**
2181   * Returns the StoreEngine that is backing this concrete implementation of Store.
2182   * @return Returns the {@link StoreEngine} object used internally inside this HStore object.
2183   */
2184  public StoreEngine<?, ?, ?, ?> getStoreEngine() {
2185    return this.storeEngine;
2186  }
2187
2188  protected OffPeakHours getOffPeakHours() {
2189    return this.offPeakHours;
2190  }
2191
2192  @Override
2193  public void onConfigurationChange(Configuration conf) {
2194    Configuration storeConf = StoreUtils.createStoreConfiguration(conf, region.getTableDescriptor(),
2195      getColumnFamilyDescriptor());
2196    this.conf = storeConf;
2197    this.storeEngine.compactionPolicy.setConf(storeConf);
2198    this.offPeakHours = OffPeakHours.getInstance(storeConf);
2199  }
2200
2201  /**
2202   * {@inheritDoc}
2203   */
2204  @Override
2205  public void registerChildren(ConfigurationManager manager) {
2206    CacheConfig cacheConfig = this.storeContext.getCacheConf();
2207    if (cacheConfig != null) {
2208      manager.registerObserver(cacheConfig);
2209    }
2210  }
2211
2212  /**
2213   * {@inheritDoc}
2214   */
2215  @Override
2216  public void deregisterChildren(ConfigurationManager manager) {
2217    // No children to deregister
2218  }
2219
2220  @Override
2221  public double getCompactionPressure() {
2222    return storeEngine.getStoreFileManager().getCompactionPressure();
2223  }
2224
2225  @Override
2226  public boolean isPrimaryReplicaStore() {
2227    return getRegionInfo().getReplicaId() == RegionInfo.DEFAULT_REPLICA_ID;
2228  }
2229
2230  /**
2231   * Sets the store up for a region level snapshot operation.
2232   * @see #postSnapshotOperation()
2233   */
2234  public void preSnapshotOperation() {
2235    archiveLock.lock();
2236  }
2237
2238  /**
2239   * Perform tasks needed after the completion of snapshot operation.
2240   * @see #preSnapshotOperation()
2241   */
2242  public void postSnapshotOperation() {
2243    archiveLock.unlock();
2244  }
2245
2246  /**
2247   * Closes and archives the compacted files under this store
2248   */
2249  public synchronized void closeAndArchiveCompactedFiles() throws IOException {
2250    // ensure other threads do not attempt to archive the same files on close()
2251    archiveLock.lock();
2252    try {
2253      storeEngine.readLock();
2254      Collection<HStoreFile> copyCompactedfiles = null;
2255      try {
2256        Collection<HStoreFile> compactedfiles =
2257          this.getStoreEngine().getStoreFileManager().getCompactedfiles();
2258        if (CollectionUtils.isNotEmpty(compactedfiles)) {
2259          // Do a copy under read lock
2260          copyCompactedfiles = new ArrayList<>(compactedfiles);
2261        } else {
2262          LOG.trace("No compacted files to archive");
2263        }
2264      } finally {
2265        storeEngine.readUnlock();
2266      }
2267      if (CollectionUtils.isNotEmpty(copyCompactedfiles)) {
2268        removeCompactedfiles(copyCompactedfiles, true);
2269      }
2270    } finally {
2271      archiveLock.unlock();
2272    }
2273  }
2274
2275  /**
2276   * Archives and removes the compacted files
2277   * @param compactedfiles The compacted files in this store that are not active in reads
2278   * @param evictOnClose   true if blocks should be evicted from the cache when an HFile reader is
2279   *                       closed, false if not
2280   */
2281  private void removeCompactedfiles(Collection<HStoreFile> compactedfiles, boolean evictOnClose)
2282    throws IOException {
2283    final List<HStoreFile> filesToRemove = new ArrayList<>(compactedfiles.size());
2284    final List<Long> storeFileSizes = new ArrayList<>(compactedfiles.size());
2285    for (final HStoreFile file : compactedfiles) {
2286      synchronized (file) {
2287        try {
2288          StoreFileReader r = file.getReader();
2289          if (r == null) {
2290            LOG.debug("The file {} was closed but still not archived", file);
2291            // HACK: Temporarily re-open the reader so we can get the size of the file. Ideally,
2292            // we should know the size of an HStoreFile without having to ask the HStoreFileReader
2293            // for that.
2294            long length = getStoreFileSize(file);
2295            filesToRemove.add(file);
2296            storeFileSizes.add(length);
2297            continue;
2298          }
2299
2300          if (file.isCompactedAway() && !file.isReferencedInReads()) {
2301            // Even if deleting fails we need not bother as any new scanners won't be
2302            // able to use the compacted file as the status is already compactedAway
2303            LOG.trace("Closing and archiving the file {}", file);
2304            // Copy the file size before closing the reader
2305            final long length = r.length();
2306            r.close(evictOnClose);
2307            // Just close and return
2308            filesToRemove.add(file);
2309            // Only add the length if we successfully added the file to `filesToRemove`
2310            storeFileSizes.add(length);
2311          } else {
2312            LOG.info("Can't archive compacted file " + file.getPath()
2313              + " because of either isCompactedAway=" + file.isCompactedAway()
2314              + " or file has reference, isReferencedInReads=" + file.isReferencedInReads()
2315              + ", refCount=" + r.getRefCount() + ", skipping for now.");
2316          }
2317        } catch (Exception e) {
2318          LOG.error("Exception while trying to close the compacted store file {}", file.getPath(),
2319            e);
2320        }
2321      }
2322    }
2323    if (this.isPrimaryReplicaStore()) {
2324      // Only the primary region is allowed to move the file to archive.
2325      // The secondary region does not move the files to archive. Any active reads from
2326      // the secondary region will still work because the file as such has active readers on it.
2327      if (!filesToRemove.isEmpty()) {
2328        LOG.debug("Moving the files {} to archive", filesToRemove);
2329        // Only if this is successful it has to be removed
2330        try {
2331          getRegionFileSystem().removeStoreFiles(this.getColumnFamilyDescriptor().getNameAsString(),
2332            filesToRemove);
2333        } catch (FailedArchiveException fae) {
2334          // Even if archiving some files failed, we still need to clear out any of the
2335          // files which were successfully archived. Otherwise we will receive a
2336          // FileNotFoundException when we attempt to re-archive them in the next go around.
2337          Collection<Path> failedFiles = fae.getFailedFiles();
2338          Iterator<HStoreFile> iter = filesToRemove.iterator();
2339          Iterator<Long> sizeIter = storeFileSizes.iterator();
2340          while (iter.hasNext()) {
2341            sizeIter.next();
2342            if (failedFiles.contains(iter.next().getPath())) {
2343              iter.remove();
2344              sizeIter.remove();
2345            }
2346          }
2347          if (!filesToRemove.isEmpty()) {
2348            clearCompactedfiles(filesToRemove);
2349          }
2350          throw fae;
2351        }
2352      }
2353    }
2354    if (!filesToRemove.isEmpty()) {
2355      // Clear the compactedfiles from the store file manager
2356      clearCompactedfiles(filesToRemove);
2357      // Try to send report of this archival to the Master for updating quota usage faster
2358      reportArchivedFilesForQuota(filesToRemove, storeFileSizes);
2359    }
2360  }
2361
2362  /**
2363   * Computes the length of a store file without succumbing to any errors along the way. If an error
2364   * is encountered, the implementation returns {@code 0} instead of the actual size.
2365   * @param file The file to compute the size of.
2366   * @return The size in bytes of the provided {@code file}.
2367   */
2368  long getStoreFileSize(HStoreFile file) {
2369    long length = 0;
2370    try {
2371      file.initReader();
2372      length = file.getReader().length();
2373    } catch (IOException e) {
2374      LOG.trace("Failed to open reader when trying to compute store file size for {}, ignoring",
2375        file, e);
2376    } finally {
2377      try {
2378        file.closeStoreFile(
2379          file.getCacheConf() != null ? file.getCacheConf().shouldEvictOnClose() : true);
2380      } catch (IOException e) {
2381        LOG.trace("Failed to close reader after computing store file size for {}, ignoring", file,
2382          e);
2383      }
2384    }
2385    return length;
2386  }
2387
2388  public Long preFlushSeqIDEstimation() {
2389    return memstore.preFlushSeqIDEstimation();
2390  }
2391
2392  @Override
2393  public boolean isSloppyMemStore() {
2394    return this.memstore.isSloppy();
2395  }
2396
2397  private void clearCompactedfiles(List<HStoreFile> filesToRemove) throws IOException {
2398    LOG.trace("Clearing the compacted file {} from this store", filesToRemove);
2399    storeEngine.removeCompactedFiles(filesToRemove);
2400  }
2401
2402  void reportArchivedFilesForQuota(List<? extends StoreFile> archivedFiles, List<Long> fileSizes) {
2403    // Sanity check from the caller
2404    if (archivedFiles.size() != fileSizes.size()) {
2405      throw new RuntimeException("Coding error: should never see lists of varying size");
2406    }
2407    RegionServerServices rss = this.region.getRegionServerServices();
2408    if (rss == null) {
2409      return;
2410    }
2411    List<Entry<String, Long>> filesWithSizes = new ArrayList<>(archivedFiles.size());
2412    Iterator<Long> fileSizeIter = fileSizes.iterator();
2413    for (StoreFile storeFile : archivedFiles) {
2414      final long fileSize = fileSizeIter.next();
2415      if (storeFile.isHFile() && fileSize != 0) {
2416        filesWithSizes.add(Maps.immutableEntry(storeFile.getPath().getName(), fileSize));
2417      }
2418    }
2419    if (LOG.isTraceEnabled()) {
2420      LOG.trace("Files archived: " + archivedFiles + ", reporting the following to the Master: "
2421        + filesWithSizes);
2422    }
2423    boolean success = rss.reportFileArchivalForQuotas(getTableName(), filesWithSizes);
2424    if (!success) {
2425      LOG.warn("Failed to report archival of files: " + filesWithSizes);
2426    }
2427  }
2428
2429  @Override
2430  public int getCurrentParallelPutCount() {
2431    return currentParallelPutCount.get();
2432  }
2433
2434  public int getStoreRefCount() {
2435    return this.storeEngine.getStoreFileManager().getStoreFiles().stream()
2436      .filter(sf -> sf.getReader() != null).filter(HStoreFile::isHFile)
2437      .mapToInt(HStoreFile::getRefCount).sum();
2438  }
2439
2440  /** Returns get maximum ref count of storeFile among all compacted HStore Files for the HStore */
2441  public int getMaxCompactedStoreFileRefCount() {
2442    OptionalInt maxCompactedStoreFileRefCount = this.storeEngine.getStoreFileManager()
2443      .getCompactedfiles().stream().filter(sf -> sf.getReader() != null).filter(HStoreFile::isHFile)
2444      .mapToInt(HStoreFile::getRefCount).max();
2445    return maxCompactedStoreFileRefCount.isPresent() ? maxCompactedStoreFileRefCount.getAsInt() : 0;
2446  }
2447
2448  @Override
2449  public long getMemstoreOnlyRowReadsCount() {
2450    return memstoreOnlyRowReadsCount.sum();
2451  }
2452
2453  @Override
2454  public long getMixedRowReadsCount() {
2455    return mixedRowReadsCount.sum();
2456  }
2457
2458  @Override
2459  public Configuration getReadOnlyConfiguration() {
2460    return new ReadOnlyConfiguration(this.conf);
2461  }
2462
2463  void updateMetricsStore(boolean memstoreRead) {
2464    if (memstoreRead) {
2465      memstoreOnlyRowReadsCount.increment();
2466    } else {
2467      mixedRowReadsCount.increment();
2468    }
2469  }
2470
2471  /**
2472   * Return the storefiles which are currently being written to. Mainly used by
2473   * {@link BrokenStoreFileCleaner} to prevent deleting the these files as they are not present in
2474   * SFT yet.
2475   */
2476  public Set<Path> getStoreFilesBeingWritten() {
2477    return storeFileWriterCreationTrackers.stream().flatMap(t -> t.get().stream())
2478      .collect(Collectors.toSet());
2479  }
2480
2481  @Override
2482  public long getBloomFilterRequestsCount() {
2483    return storeEngine.getBloomFilterMetrics().getRequestsCount();
2484  }
2485
2486  @Override
2487  public long getBloomFilterNegativeResultsCount() {
2488    return storeEngine.getBloomFilterMetrics().getNegativeResultsCount();
2489  }
2490
2491  @Override
2492  public long getBloomFilterEligibleRequestsCount() {
2493    return storeEngine.getBloomFilterMetrics().getEligibleRequestsCount();
2494  }
2495}