View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.regionserver;
20  
21  import java.io.IOException;
22  import java.io.InterruptedIOException;
23  import java.net.InetSocketAddress;
24  import java.security.Key;
25  import java.security.KeyException;
26  import java.security.PrivilegedExceptionAction;
27  import java.util.ArrayList;
28  import java.util.Collection;
29  import java.util.Collections;
30  import java.util.HashMap;
31  import java.util.HashSet;
32  import java.util.Iterator;
33  import java.util.List;
34  import java.util.NavigableSet;
35  import java.util.Set;
36  import java.util.concurrent.Callable;
37  import java.util.concurrent.CompletionService;
38  import java.util.concurrent.ConcurrentHashMap;
39  import java.util.concurrent.ExecutionException;
40  import java.util.concurrent.ExecutorCompletionService;
41  import java.util.concurrent.Future;
42  import java.util.concurrent.ThreadPoolExecutor;
43  import java.util.concurrent.atomic.AtomicBoolean;
44  import java.util.concurrent.locks.ReentrantLock;
45  import java.util.concurrent.locks.ReentrantReadWriteLock;
46  
47  import org.apache.commons.logging.Log;
48  import org.apache.commons.logging.LogFactory;
49  import org.apache.hadoop.conf.Configuration;
50  import org.apache.hadoop.fs.FileSystem;
51  import org.apache.hadoop.fs.Path;
52  import org.apache.hadoop.hbase.Cell;
53  import org.apache.hadoop.hbase.CellComparator;
54  import org.apache.hadoop.hbase.CellUtil;
55  import org.apache.hadoop.hbase.CompoundConfiguration;
56  import org.apache.hadoop.hbase.HColumnDescriptor;
57  import org.apache.hadoop.hbase.HConstants;
58  import org.apache.hadoop.hbase.HRegionInfo;
59  import org.apache.hadoop.hbase.KeyValue;
60  import org.apache.hadoop.hbase.TableName;
61  import org.apache.hadoop.hbase.Tag;
62  import org.apache.hadoop.hbase.TagType;
63  import org.apache.hadoop.hbase.backup.FailedArchiveException;
64  import org.apache.hadoop.hbase.classification.InterfaceAudience;
65  import org.apache.hadoop.hbase.client.Scan;
66  import org.apache.hadoop.hbase.conf.ConfigurationManager;
67  import org.apache.hadoop.hbase.io.compress.Compression;
68  import org.apache.hadoop.hbase.io.crypto.Cipher;
69  import org.apache.hadoop.hbase.io.crypto.Encryption;
70  import org.apache.hadoop.hbase.io.hfile.CacheConfig;
71  import org.apache.hadoop.hbase.io.hfile.HFile;
72  import org.apache.hadoop.hbase.io.hfile.HFileContext;
73  import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
74  import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoder;
75  import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoderImpl;
76  import org.apache.hadoop.hbase.io.hfile.HFileScanner;
77  import org.apache.hadoop.hbase.io.hfile.InvalidHFileException;
78  import org.apache.hadoop.hbase.monitoring.MonitoredTask;
79  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
80  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.CompactionDescriptor;
81  import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
82  import org.apache.hadoop.hbase.regionserver.compactions.CompactionProgress;
83  import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
84  import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
85  import org.apache.hadoop.hbase.regionserver.compactions.OffPeakHours;
86  import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
87  import org.apache.hadoop.hbase.regionserver.wal.WALUtil;
88  import org.apache.hadoop.hbase.security.EncryptionUtil;
89  import org.apache.hadoop.hbase.security.User;
90  import org.apache.hadoop.hbase.util.Bytes;
91  import org.apache.hadoop.hbase.util.ChecksumType;
92  import org.apache.hadoop.hbase.util.ClassSize;
93  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
94  import org.apache.hadoop.hbase.util.ReflectionUtils;
95  import org.apache.hadoop.util.StringUtils;
96  import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix;
97  
98  import com.google.common.annotations.VisibleForTesting;
99  import com.google.common.base.Preconditions;
100 import com.google.common.collect.ImmutableCollection;
101 import com.google.common.collect.ImmutableList;
102 import com.google.common.collect.Lists;
103 import com.google.common.collect.Sets;
104 
105 /**
106  * A Store holds a column family in a Region.  Its a memstore and a set of zero
107  * or more StoreFiles, which stretch backwards over time.
108  *
109  * <p>There's no reason to consider append-logging at this level; all logging
110  * and locking is handled at the HRegion level.  Store just provides
111  * services to manage sets of StoreFiles.  One of the most important of those
112  * services is compaction services where files are aggregated once they pass
113  * a configurable threshold.
114  *
115  * <p>Locking and transactions are handled at a higher level.  This API should
116  * not be called directly but by an HRegion manager.
117  */
118 @InterfaceAudience.Private
119 public class HStore implements Store {
120   private static final String MEMSTORE_CLASS_NAME = "hbase.regionserver.memstore.class";
121   public static final String COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY =
122       "hbase.server.compactchecker.interval.multiplier";
123   public static final String BLOCKING_STOREFILES_KEY = "hbase.hstore.blockingStoreFiles";
124   public static final int DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER = 1000;
125   public static final int DEFAULT_BLOCKING_STOREFILE_COUNT = 7;
126 
127   private static final Log LOG = LogFactory.getLog(HStore.class);
128 
129   protected final MemStore memstore;
130   // This stores directory in the filesystem.
131   private final HRegion region;
132   private final HColumnDescriptor family;
133   private final HRegionFileSystem fs;
134   private Configuration conf;
135   private final CacheConfig cacheConf;
136   private long lastCompactSize = 0;
137   volatile boolean forceMajor = false;
138   /* how many bytes to write between status checks */
139   static int closeCheckInterval = 0;
140   private volatile long storeSize = 0L;
141   private volatile long totalUncompressedBytes = 0L;
142 
143   /**
144    * RWLock for store operations.
145    * Locked in shared mode when the list of component stores is looked at:
146    *   - all reads/writes to table data
147    *   - checking for split
148    * Locked in exclusive mode when the list of component stores is modified:
149    *   - closing
150    *   - completing a compaction
151    */
152   final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
153   /**
154    * Lock specific to archiving compacted store files.  This avoids races around
155    * the combination of retrieving the list of compacted files and moving them to
156    * the archive directory.  Since this is usually a background process (other than
157    * on close), we don't want to handle this with the store write lock, which would
158    * block readers and degrade performance.
159    *
160    * Locked by:
161    *   - CompactedHFilesDispatchHandler via closeAndArchiveCompactedFiles()
162    *   - close()
163    */
164   final ReentrantLock archiveLock = new ReentrantLock();
165 
166   private final boolean verifyBulkLoads;
167 
168   private ScanInfo scanInfo;
169 
170   // All access must be synchronized.
171   // TODO: ideally, this should be part of storeFileManager, as we keep passing this to it.
172   private final List<StoreFile> filesCompacting = Lists.newArrayList();
173 
174   // All access must be synchronized.
175   private final Set<ChangedReadersObserver> changedReaderObservers =
176     Collections.newSetFromMap(new ConcurrentHashMap<ChangedReadersObserver, Boolean>());
177 
178   private final int blocksize;
179   private HFileDataBlockEncoder dataBlockEncoder;
180 
181   /** Checksum configuration */
182   private ChecksumType checksumType;
183   private int bytesPerChecksum;
184 
185   // Comparing KeyValues
186   private final KeyValue.KVComparator comparator;
187 
188   final StoreEngine<?, ?, ?, ?> storeEngine;
189 
190   private static final AtomicBoolean offPeakCompactionTracker = new AtomicBoolean();
191   private volatile OffPeakHours offPeakHours;
192 
193   private static final int DEFAULT_FLUSH_RETRIES_NUMBER = 10;
194   private int flushRetriesNumber;
195   private int pauseTime;
196 
197   private long blockingFileCount;
198   private int compactionCheckMultiplier;
199 
200   private Encryption.Context cryptoContext = Encryption.Context.NONE;
201 
202   private volatile long flushedCellsCount = 0;
203   private volatile long compactedCellsCount = 0;
204   private volatile long majorCompactedCellsCount = 0;
205   private volatile long flushedCellsSize = 0;
206   private volatile long flushedOutputFileSize = 0;
207   private volatile long compactedCellsSize = 0;
208   private volatile long majorCompactedCellsSize = 0;
209 
210   /**
211    * Constructor
212    * @param region
213    * @param family HColumnDescriptor for this column
214    * @param confParam configuration object
215    * failed.  Can be null.
216    * @throws IOException
217    */
218   protected HStore(final HRegion region, final HColumnDescriptor family,
219       final Configuration confParam) throws IOException {
220 
221     HRegionInfo info = region.getRegionInfo();
222     this.fs = region.getRegionFileSystem();
223 
224     // Assemble the store's home directory and Ensure it exists.
225     fs.createStoreDir(family.getNameAsString());
226     this.region = region;
227     this.family = family;
228     // 'conf' renamed to 'confParam' b/c we use this.conf in the constructor
229     // CompoundConfiguration will look for keys in reverse order of addition, so we'd
230     // add global config first, then table and cf overrides, then cf metadata.
231     this.conf = new CompoundConfiguration()
232       .add(confParam)
233       .addStringMap(region.getTableDesc().getConfiguration())
234       .addStringMap(family.getConfiguration())
235       .addWritableMap(family.getValues());
236     this.blocksize = family.getBlocksize();
237 
238     this.dataBlockEncoder =
239         new HFileDataBlockEncoderImpl(family.getDataBlockEncoding());
240 
241     this.comparator = info.getComparator();
242     // used by ScanQueryMatcher
243     long timeToPurgeDeletes =
244         Math.max(conf.getLong("hbase.hstore.time.to.purge.deletes", 0), 0);
245     LOG.trace("Time to purge deletes set to " + timeToPurgeDeletes +
246         "ms in store " + this);
247     // Get TTL
248     long ttl = determineTTLFromFamily(family);
249     // Why not just pass a HColumnDescriptor in here altogether?  Even if have
250     // to clone it?
251     scanInfo = new ScanInfo(conf, family, ttl, timeToPurgeDeletes, this.comparator);
252     String className = conf.get(MEMSTORE_CLASS_NAME, DefaultMemStore.class.getName());
253     this.memstore = ReflectionUtils.instantiateWithCustomCtor(className, new Class[] {
254         Configuration.class, KeyValue.KVComparator.class }, new Object[] { conf, this.comparator });
255     this.offPeakHours = OffPeakHours.getInstance(conf);
256 
257     // Setting up cache configuration for this family
258     this.cacheConf = new CacheConfig(conf, family);
259 
260     this.verifyBulkLoads = conf.getBoolean("hbase.hstore.bulkload.verify", false);
261 
262     this.blockingFileCount =
263         conf.getInt(BLOCKING_STOREFILES_KEY, DEFAULT_BLOCKING_STOREFILE_COUNT);
264     this.compactionCheckMultiplier = conf.getInt(
265         COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY, DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER);
266     if (this.compactionCheckMultiplier <= 0) {
267       LOG.error("Compaction check period multiplier must be positive, setting default: "
268           + DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER);
269       this.compactionCheckMultiplier = DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER;
270     }
271 
272     if (HStore.closeCheckInterval == 0) {
273       HStore.closeCheckInterval = conf.getInt(
274           "hbase.hstore.close.check.interval", 10*1000*1000 /* 10 MB */);
275     }
276 
277     this.storeEngine = StoreEngine.create(this, this.conf, this.comparator);
278     List<StoreFile> storeFiles = loadStoreFiles();
279     // Move the storeSize calculation out of loadStoreFiles() method, because the secondary read
280     // replica's refreshStoreFiles() will also use loadStoreFiles() to refresh its store files and
281     // update the storeSize in the completeCompaction(..) finally (just like compaction) , so
282     // no need calculate the storeSize twice.
283     this.storeSize = getStorefilesSize(storeFiles);
284     this.totalUncompressedBytes = getTotalUmcompressedBytes(storeFiles);
285     this.storeEngine.getStoreFileManager().loadFiles(storeFiles);
286 
287     // Initialize checksum type from name. The names are CRC32, CRC32C, etc.
288     this.checksumType = getChecksumType(conf);
289     // initilize bytes per checksum
290     this.bytesPerChecksum = getBytesPerChecksum(conf);
291     flushRetriesNumber = conf.getInt(
292         "hbase.hstore.flush.retries.number", DEFAULT_FLUSH_RETRIES_NUMBER);
293     pauseTime = conf.getInt(HConstants.HBASE_SERVER_PAUSE, HConstants.DEFAULT_HBASE_SERVER_PAUSE);
294     if (flushRetriesNumber <= 0) {
295       throw new IllegalArgumentException(
296           "hbase.hstore.flush.retries.number must be > 0, not "
297               + flushRetriesNumber);
298     }
299 
300     // Crypto context for new store files
301     String cipherName = family.getEncryptionType();
302     if (cipherName != null) {
303       Cipher cipher;
304       Key key;
305       byte[] keyBytes = family.getEncryptionKey();
306       if (keyBytes != null) {
307         // Family provides specific key material
308         String masterKeyName = conf.get(HConstants.CRYPTO_MASTERKEY_NAME_CONF_KEY,
309           User.getCurrent().getShortName());
310         try {
311           // First try the master key
312           key = EncryptionUtil.unwrapKey(conf, masterKeyName, keyBytes);
313         } catch (KeyException e) {
314           // If the current master key fails to unwrap, try the alternate, if
315           // one is configured
316           if (LOG.isDebugEnabled()) {
317             LOG.debug("Unable to unwrap key with current master key '" + masterKeyName + "'");
318           }
319           String alternateKeyName =
320             conf.get(HConstants.CRYPTO_MASTERKEY_ALTERNATE_NAME_CONF_KEY);
321           if (alternateKeyName != null) {
322             try {
323               key = EncryptionUtil.unwrapKey(conf, alternateKeyName, keyBytes);
324             } catch (KeyException ex) {
325               throw new IOException(ex);
326             }
327           } else {
328             throw new IOException(e);
329           }
330         }
331         // Use the algorithm the key wants
332         cipher = Encryption.getCipher(conf, key.getAlgorithm());
333         if (cipher == null) {
334           throw new RuntimeException("Cipher '" + key.getAlgorithm() + "' is not available");
335         }
336         // Fail if misconfigured
337         // We use the encryption type specified in the column schema as a sanity check on
338         // what the wrapped key is telling us
339         if (!cipher.getName().equalsIgnoreCase(cipherName)) {
340           throw new RuntimeException("Encryption for family '" + family.getNameAsString() +
341             "' configured with type '" + cipherName +
342             "' but key specifies algorithm '" + cipher.getName() + "'");
343         }
344       } else {
345         // Family does not provide key material, create a random key
346         cipher = Encryption.getCipher(conf, cipherName);
347         if (cipher == null) {
348           throw new RuntimeException("Cipher '" + cipherName + "' is not available");
349         }
350         key = cipher.getRandomKey();
351       }
352       cryptoContext = Encryption.newContext(conf);
353       cryptoContext.setCipher(cipher);
354       cryptoContext.setKey(key);
355     }
356 
357     LOG.info("Store=" + getColumnFamilyName() +
358       ", memstore type=" + this.memstore.getClass().getSimpleName() +
359       ", verifyBulkLoads=" + verifyBulkLoads +
360       ", encoding=" + family.getDataBlockEncoding() +
361       ", compression=" + family.getCompressionType());
362   }
363 
364   /**
365    * @param family
366    * @return TTL in seconds of the specified family
367    */
368   private static long determineTTLFromFamily(final HColumnDescriptor family) {
369     // HCD.getTimeToLive returns ttl in seconds.  Convert to milliseconds.
370     long ttl = family.getTimeToLive();
371     if (ttl == HConstants.FOREVER) {
372       // Default is unlimited ttl.
373       ttl = Long.MAX_VALUE;
374     } else if (ttl == -1) {
375       ttl = Long.MAX_VALUE;
376     } else {
377       // Second -> ms adjust for user data
378       ttl *= 1000;
379     }
380     return ttl;
381   }
382 
383   @Override
384   public String getColumnFamilyName() {
385     return this.family.getNameAsString();
386   }
387 
388   @Override
389   public TableName getTableName() {
390     return this.getRegionInfo().getTable();
391   }
392 
393   @Override
394   public FileSystem getFileSystem() {
395     return this.fs.getFileSystem();
396   }
397 
398   public HRegionFileSystem getRegionFileSystem() {
399     return this.fs;
400   }
401 
402   /* Implementation of StoreConfigInformation */
403   @Override
404   public long getStoreFileTtl() {
405     // TTL only applies if there's no MIN_VERSIONs setting on the column.
406     return (this.scanInfo.getMinVersions() == 0) ? this.scanInfo.getTtl() : Long.MAX_VALUE;
407   }
408 
409   @Override
410   public long getMemstoreFlushSize() {
411     // TODO: Why is this in here?  The flushsize of the region rather than the store?  St.Ack
412     return this.region.memstoreFlushSize;
413   }
414 
415   @Override
416   public long getFlushableSize() {
417     return this.memstore.getFlushableSize();
418   }
419 
420   @Override
421   public long getSnapshotSize() {
422     return this.memstore.getSnapshotSize();
423   }
424 
425   @Override
426   public long getCompactionCheckMultiplier() {
427     return this.compactionCheckMultiplier;
428   }
429 
430   @Override
431   public long getBlockingFileCount() {
432     return blockingFileCount;
433   }
434   /* End implementation of StoreConfigInformation */
435 
436   /**
437    * Returns the configured bytesPerChecksum value.
438    * @param conf The configuration
439    * @return The bytesPerChecksum that is set in the configuration
440    */
441   public static int getBytesPerChecksum(Configuration conf) {
442     return conf.getInt(HConstants.BYTES_PER_CHECKSUM,
443                        HFile.DEFAULT_BYTES_PER_CHECKSUM);
444   }
445 
446   /**
447    * Returns the configured checksum algorithm.
448    * @param conf The configuration
449    * @return The checksum algorithm that is set in the configuration
450    */
451   public static ChecksumType getChecksumType(Configuration conf) {
452     String checksumName = conf.get(HConstants.CHECKSUM_TYPE_NAME);
453     if (checksumName == null) {
454       return ChecksumType.getDefaultChecksumType();
455     } else {
456       return ChecksumType.nameToType(checksumName);
457     }
458   }
459 
460   /**
461    * @return how many bytes to write between status checks
462    */
463   public static int getCloseCheckInterval() {
464     return closeCheckInterval;
465   }
466 
467   @Override
468   public HColumnDescriptor getFamily() {
469     return this.family;
470   }
471 
472   /**
473    * @return The maximum sequence id in all store files. Used for log replay.
474    */
475   @Override
476   public long getMaxSequenceId() {
477     return StoreFile.getMaxSequenceIdInList(this.getStorefiles());
478   }
479 
480   @Override
481   public long getMaxMemstoreTS() {
482     return StoreFile.getMaxMemstoreTSInList(this.getStorefiles());
483   }
484 
485   /**
486    * @param tabledir {@link Path} to where the table is being stored
487    * @param hri {@link HRegionInfo} for the region.
488    * @param family {@link HColumnDescriptor} describing the column family
489    * @return Path to family/Store home directory.
490    */
491   @Deprecated
492   public static Path getStoreHomedir(final Path tabledir,
493       final HRegionInfo hri, final byte[] family) {
494     return getStoreHomedir(tabledir, hri.getEncodedName(), family);
495   }
496 
497   /**
498    * @param tabledir {@link Path} to where the table is being stored
499    * @param encodedName Encoded region name.
500    * @param family {@link HColumnDescriptor} describing the column family
501    * @return Path to family/Store home directory.
502    */
503   @Deprecated
504   public static Path getStoreHomedir(final Path tabledir,
505       final String encodedName, final byte[] family) {
506     return new Path(tabledir, new Path(encodedName, Bytes.toString(family)));
507   }
508 
509   @Override
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   /**
523    * Creates an unsorted list of StoreFile loaded in parallel
524    * from the given directory.
525    * @throws IOException
526    */
527   private List<StoreFile> loadStoreFiles() throws IOException {
528     Collection<StoreFileInfo> files = fs.getStoreFiles(getColumnFamilyName());
529     return openStoreFiles(files);
530   }
531 
532   private List<StoreFile> openStoreFiles(Collection<StoreFileInfo> files) throws IOException {
533     if (files == null || files.size() == 0) {
534       return new ArrayList<StoreFile>();
535     }
536     // initialize the thread pool for opening store files in parallel..
537     ThreadPoolExecutor storeFileOpenerThreadPool =
538       this.region.getStoreFileOpenAndCloseThreadPool("StoreFileOpenerThread-" +
539           this.getColumnFamilyName());
540     CompletionService<StoreFile> completionService =
541       new ExecutorCompletionService<StoreFile>(storeFileOpenerThreadPool);
542 
543     int totalValidStoreFile = 0;
544     for (final StoreFileInfo storeFileInfo: files) {
545       // open each store file in parallel
546       completionService.submit(new Callable<StoreFile>() {
547         @Override
548         public StoreFile call() throws IOException {
549           StoreFile storeFile = createStoreFileAndReader(storeFileInfo);
550           return storeFile;
551         }
552       });
553       totalValidStoreFile++;
554     }
555 
556     Set<String> compactedStoreFiles = new HashSet<>();
557     ArrayList<StoreFile> results = new ArrayList<StoreFile>(files.size());
558     IOException ioe = null;
559     try {
560       for (int i = 0; i < totalValidStoreFile; i++) {
561         try {
562           Future<StoreFile> future = completionService.take();
563           StoreFile storeFile = future.get();
564           if (storeFile != null) {
565             if (LOG.isDebugEnabled()) {
566               LOG.debug("loaded " + storeFile.toStringDetailed());
567             }
568             results.add(storeFile);
569             compactedStoreFiles.addAll(storeFile.getCompactedStoreFiles());
570           }
571         } catch (InterruptedException e) {
572           if (ioe == null) ioe = new InterruptedIOException(e.getMessage());
573         } catch (ExecutionException e) {
574           if (ioe == null) ioe = new IOException(e.getCause());
575         }
576       }
577     } finally {
578       storeFileOpenerThreadPool.shutdownNow();
579     }
580     if (ioe != null) {
581       // close StoreFile readers
582       boolean evictOnClose =
583           cacheConf != null? cacheConf.shouldEvictOnClose(): true;
584       for (StoreFile file : results) {
585         try {
586           if (file != null) file.closeReader(evictOnClose);
587         } catch (IOException e) {
588           LOG.warn(e.getMessage());
589         }
590       }
591       throw ioe;
592     }
593 
594     // Remove the compacted files from result
595     List<StoreFile> filesToRemove = new ArrayList<>(compactedStoreFiles.size());
596     for (StoreFile storeFile : results) {
597       if (compactedStoreFiles.contains(storeFile.getPath().getName())) {
598         LOG.warn("Clearing the compacted storefile " + storeFile + " from this store");
599         storeFile.getReader().close(true);
600         filesToRemove.add(storeFile);
601       }
602     }
603     results.removeAll(filesToRemove);
604     if (!filesToRemove.isEmpty() && this.isPrimaryReplicaStore()) {
605       LOG.debug("Moving the files " + filesToRemove + " to archive");
606       this.fs.removeStoreFiles(this.getColumnFamilyName(), filesToRemove);
607     }
608 
609     return results;
610   }
611 
612   /**
613    * Checks the underlying store files, and opens the files that  have not
614    * been opened, and removes the store file readers for store files no longer
615    * available. Mainly used by secondary region replicas to keep up to date with
616    * the primary region files.
617    * @throws IOException
618    */
619   @Override
620   public void refreshStoreFiles() throws IOException {
621     Collection<StoreFileInfo> newFiles = fs.getStoreFiles(getColumnFamilyName());
622     refreshStoreFilesInternal(newFiles);
623   }
624 
625   @Override
626   public void refreshStoreFiles(Collection<String> newFiles) throws IOException {
627     List<StoreFileInfo> storeFiles = new ArrayList<StoreFileInfo>(newFiles.size());
628     for (String file : newFiles) {
629       storeFiles.add(fs.getStoreFileInfo(getColumnFamilyName(), file));
630     }
631     refreshStoreFilesInternal(storeFiles);
632   }
633 
634   /**
635    * Checks the underlying store files, and opens the files that  have not
636    * been opened, and removes the store file readers for store files no longer
637    * available. Mainly used by secondary region replicas to keep up to date with
638    * the primary region files.
639    * @throws IOException
640    */
641   private void refreshStoreFilesInternal(Collection<StoreFileInfo> newFiles) throws IOException {
642     StoreFileManager sfm = storeEngine.getStoreFileManager();
643     Collection<StoreFile> currentFiles = sfm.getStorefiles();
644     Collection<StoreFile> compactedFiles = sfm.getCompactedfiles();
645     if (currentFiles == null) currentFiles = Collections.emptySet();
646     if (newFiles == null) newFiles = Collections.emptySet();
647     if (compactedFiles == null) compactedFiles = Collections.emptySet();
648 
649     HashMap<StoreFileInfo, StoreFile> currentFilesSet =
650         new HashMap<StoreFileInfo, StoreFile>(currentFiles.size());
651     for (StoreFile sf : currentFiles) {
652       currentFilesSet.put(sf.getFileInfo(), sf);
653     }
654     HashMap<StoreFileInfo, StoreFile> compactedFilesSet =
655         new HashMap<StoreFileInfo, StoreFile>(compactedFiles.size());
656     for (StoreFile sf : compactedFiles) {
657       compactedFilesSet.put(sf.getFileInfo(), sf);
658     }
659 
660     Set<StoreFileInfo> newFilesSet = new HashSet<StoreFileInfo>(newFiles);
661     //Exclude the files that have already been compacted
662     newFilesSet = Sets.difference(newFilesSet, compactedFilesSet.keySet());
663     Set<StoreFileInfo> toBeAddedFiles = Sets.difference(newFilesSet, currentFilesSet.keySet());
664     Set<StoreFileInfo> toBeRemovedFiles = Sets.difference(currentFilesSet.keySet(), newFilesSet);
665 
666     if (toBeAddedFiles.isEmpty() && toBeRemovedFiles.isEmpty()) {
667       return;
668     }
669 
670     LOG.info("Refreshing store files for region " + this.getRegionInfo().getRegionNameAsString()
671       + " files to add: " + toBeAddedFiles + " files to remove: " + toBeRemovedFiles);
672 
673     Set<StoreFile> toBeRemovedStoreFiles = new HashSet<StoreFile>(toBeRemovedFiles.size());
674     for (StoreFileInfo sfi : toBeRemovedFiles) {
675       toBeRemovedStoreFiles.add(currentFilesSet.get(sfi));
676     }
677 
678     // try to open the files
679     List<StoreFile> openedFiles = openStoreFiles(toBeAddedFiles);
680 
681     // propogate the file changes to the underlying store file manager
682     replaceStoreFiles(toBeRemovedStoreFiles, openedFiles); //won't throw an exception
683 
684     // Advance the memstore read point to be at least the new store files seqIds so that
685     // readers might pick it up. This assumes that the store is not getting any writes (otherwise
686     // in-flight transactions might be made visible)
687     if (!toBeAddedFiles.isEmpty()) {
688       region.getMVCC().advanceTo(this.getMaxSequenceId());
689     }
690 
691     completeCompaction(toBeRemovedStoreFiles);
692   }
693 
694   private StoreFile createStoreFileAndReader(final Path p) throws IOException {
695     StoreFileInfo info = new StoreFileInfo(conf, this.getFileSystem(), p);
696     return createStoreFileAndReader(info);
697   }
698 
699   private StoreFile createStoreFileAndReader(final StoreFileInfo info)
700       throws IOException {
701     info.setRegionCoprocessorHost(this.region.getCoprocessorHost());
702     StoreFile storeFile = new StoreFile(this.getFileSystem(), info, this.conf, this.cacheConf,
703       this.family.getBloomFilterType());
704     StoreFile.Reader r = storeFile.createReader();
705     r.setReplicaStoreFile(isPrimaryReplicaStore());
706     return storeFile;
707   }
708 
709   @Override
710   public long add(final Cell cell) {
711     lock.readLock().lock();
712     try {
713        return this.memstore.add(cell);
714     } finally {
715       lock.readLock().unlock();
716     }
717   }
718 
719   @Override
720   public long timeOfOldestEdit() {
721     return memstore.timeOfOldestEdit();
722   }
723 
724   /**
725    * Adds a value to the memstore
726    *
727    * @param kv
728    * @return memstore size delta
729    */
730   protected long delete(final KeyValue kv) {
731     lock.readLock().lock();
732     try {
733       return this.memstore.delete(kv);
734     } finally {
735       lock.readLock().unlock();
736     }
737   }
738 
739   @Override
740   public void rollback(final Cell cell) {
741     lock.readLock().lock();
742     try {
743       this.memstore.rollback(cell);
744     } finally {
745       lock.readLock().unlock();
746     }
747   }
748 
749   /**
750    * @return All store files.
751    */
752   @Override
753   public Collection<StoreFile> getStorefiles() {
754     return this.storeEngine.getStoreFileManager().getStorefiles();
755   }
756 
757   public Collection<StoreFile> getCompactedfiles() {
758     Collection<StoreFile> compactedFiles =
759         this.storeEngine.getStoreFileManager().getCompactedfiles();
760     return compactedFiles == null ? new ArrayList<StoreFile>() : compactedFiles;
761   }
762 
763   @Override
764   public void assertBulkLoadHFileOk(Path srcPath) throws IOException {
765     HFile.Reader reader  = null;
766     try {
767       LOG.info("Validating hfile at " + srcPath + " for inclusion in "
768           + "store " + this + " region " + this.getRegionInfo().getRegionNameAsString());
769       reader = HFile.createReader(srcPath.getFileSystem(conf),
770           srcPath, cacheConf, conf);
771       reader.loadFileInfo();
772 
773       byte[] firstKey = reader.getFirstRowKey();
774       Preconditions.checkState(firstKey != null, "First key can not be null");
775       byte[] lk = reader.getLastKey();
776       Preconditions.checkState(lk != null, "Last key can not be null");
777       byte[] lastKey =  KeyValue.createKeyValueFromKey(lk).getRow();
778 
779       LOG.debug("HFile bounds: first=" + Bytes.toStringBinary(firstKey) +
780           " last=" + Bytes.toStringBinary(lastKey));
781       LOG.debug("Region bounds: first=" +
782           Bytes.toStringBinary(getRegionInfo().getStartKey()) +
783           " last=" + Bytes.toStringBinary(getRegionInfo().getEndKey()));
784 
785       if (!this.getRegionInfo().containsRange(firstKey, lastKey)) {
786         throw new WrongRegionException(
787             "Bulk load file " + srcPath.toString() + " does not fit inside region "
788             + this.getRegionInfo().getRegionNameAsString());
789       }
790 
791       if(reader.length() > conf.getLong(HConstants.HREGION_MAX_FILESIZE,
792           HConstants.DEFAULT_MAX_FILE_SIZE)) {
793         LOG.warn("Trying to bulk load hfile " + srcPath.toString() + " with size: " +
794             reader.length() + " bytes can be problematic as it may lead to oversplitting.");
795       }
796 
797       if (verifyBulkLoads) {
798         long verificationStartTime = EnvironmentEdgeManager.currentTime();
799         LOG.info("Full verification started for bulk load hfile: " + srcPath.toString());
800         Cell prevCell = null;
801         HFileScanner scanner = reader.getScanner(false, false, false);
802         scanner.seekTo();
803         do {
804           Cell cell = scanner.getKeyValue();
805           if (prevCell != null) {
806             if (CellComparator.compareRows(prevCell, cell) > 0) {
807               throw new InvalidHFileException("Previous row is greater than"
808                   + " current row: path=" + srcPath + " previous="
809                   + CellUtil.getCellKeyAsString(prevCell) + " current="
810                   + CellUtil.getCellKeyAsString(cell));
811             }
812             if (CellComparator.compareFamilies(prevCell, cell) != 0) {
813               throw new InvalidHFileException("Previous key had different"
814                   + " family compared to current key: path=" + srcPath
815                   + " previous="
816                   + Bytes.toStringBinary(prevCell.getFamilyArray(), prevCell.getFamilyOffset(),
817                       prevCell.getFamilyLength())
818                   + " current="
819                   + Bytes.toStringBinary(cell.getFamilyArray(), cell.getFamilyOffset(),
820                       cell.getFamilyLength()));
821             }
822           }
823           prevCell = cell;
824         } while (scanner.next());
825       LOG.info("Full verification complete for bulk load hfile: " + srcPath.toString()
826          + " took " + (EnvironmentEdgeManager.currentTime() - verificationStartTime)
827          + " ms");
828       }
829     } finally {
830       if (reader != null) reader.close();
831     }
832   }
833 
834   @Override
835   public Path bulkLoadHFile(String srcPathStr, long seqNum) throws IOException {
836     Path srcPath = new Path(srcPathStr);
837     Path dstPath = fs.bulkLoadStoreFile(getColumnFamilyName(), srcPath, seqNum);
838 
839     LOG.info("Loaded HFile " + srcPath + " into store '" + getColumnFamilyName() + "' as "
840         + dstPath + " - updating store file list.");
841 
842     StoreFile sf = createStoreFileAndReader(dstPath);
843     bulkLoadHFile(sf);
844 
845     LOG.info("Successfully loaded store file " + srcPath + " into store " + this
846         + " (new location: " + dstPath + ")");
847 
848     return dstPath;
849   }
850 
851   @Override
852   public void bulkLoadHFile(StoreFileInfo fileInfo) throws IOException {
853     StoreFile sf = createStoreFileAndReader(fileInfo);
854     bulkLoadHFile(sf);
855   }
856 
857   private void bulkLoadHFile(StoreFile sf) throws IOException {
858     StoreFile.Reader r = sf.getReader();
859     this.storeSize += r.length();
860     this.totalUncompressedBytes += r.getTotalUncompressedBytes();
861 
862     // Append the new storefile into the list
863     this.lock.writeLock().lock();
864     try {
865       this.storeEngine.getStoreFileManager().insertNewFiles(Lists.newArrayList(sf));
866     } finally {
867       // We need the lock, as long as we are updating the storeFiles
868       // or changing the memstore. Let us release it before calling
869       // notifyChangeReadersObservers. See HBASE-4485 for a possible
870       // deadlock scenario that could have happened if continue to hold
871       // the lock.
872       this.lock.writeLock().unlock();
873     }
874     LOG.info("Loaded HFile " + sf.getFileInfo() + " into store '" + getColumnFamilyName());
875     if (LOG.isTraceEnabled()) {
876       String traceMessage = "BULK LOAD time,size,store size,store files ["
877           + EnvironmentEdgeManager.currentTime() + "," + r.length() + "," + storeSize
878           + "," + storeEngine.getStoreFileManager().getStorefileCount() + "]";
879       LOG.trace(traceMessage);
880     }
881   }
882 
883   @Override
884   public ImmutableCollection<StoreFile> close() throws IOException {
885     this.archiveLock.lock();
886     this.lock.writeLock().lock();
887     try {
888       // Clear so metrics doesn't find them.
889       ImmutableCollection<StoreFile> result = storeEngine.getStoreFileManager().clearFiles();
890       Collection<StoreFile> compactedfiles =
891           storeEngine.getStoreFileManager().clearCompactedFiles();
892       // clear the compacted files
893       if (compactedfiles != null && !compactedfiles.isEmpty()) {
894         removeCompactedfiles(compactedfiles);
895       }
896       if (!result.isEmpty()) {
897         // initialize the thread pool for closing store files in parallel.
898         ThreadPoolExecutor storeFileCloserThreadPool = this.region
899             .getStoreFileOpenAndCloseThreadPool("StoreFileCloserThread-"
900                 + this.getColumnFamilyName());
901 
902         // close each store file in parallel
903         CompletionService<Void> completionService =
904           new ExecutorCompletionService<Void>(storeFileCloserThreadPool);
905         for (final StoreFile f : result) {
906           completionService.submit(new Callable<Void>() {
907             @Override
908             public Void call() throws IOException {
909               boolean evictOnClose =
910                   cacheConf != null? cacheConf.shouldEvictOnClose(): true;
911               f.closeReader(evictOnClose);
912               return null;
913             }
914           });
915         }
916 
917         IOException ioe = null;
918         try {
919           for (int i = 0; i < result.size(); i++) {
920             try {
921               Future<Void> future = completionService.take();
922               future.get();
923             } catch (InterruptedException e) {
924               if (ioe == null) {
925                 ioe = new InterruptedIOException();
926                 ioe.initCause(e);
927               }
928             } catch (ExecutionException e) {
929               if (ioe == null) ioe = new IOException(e.getCause());
930             }
931           }
932         } finally {
933           storeFileCloserThreadPool.shutdownNow();
934         }
935         if (ioe != null) throw ioe;
936       }
937       LOG.info("Closed " + this);
938       return result;
939     } finally {
940       this.lock.writeLock().unlock();
941       this.archiveLock.unlock();
942     }
943   }
944 
945   /**
946    * Snapshot this stores memstore. Call before running
947    * {@link #flushCache(long, MemStoreSnapshot, MonitoredTask, ThroughputController)}
948    *  so it has some work to do.
949    */
950   void snapshot() {
951     this.lock.writeLock().lock();
952     try {
953       this.memstore.snapshot();
954     } finally {
955       this.lock.writeLock().unlock();
956     }
957   }
958 
959   /**
960    * Write out current snapshot. Presumes {@link #snapshot()} has been called previously.
961    * @param logCacheFlushId flush sequence number
962    * @param snapshot
963    * @param status
964    * @param throughputController
965    * @return The path name of the tmp file to which the store was flushed
966    * @throws IOException if exception occurs during process
967    */
968   protected List<Path> flushCache(final long logCacheFlushId, MemStoreSnapshot snapshot,
969       MonitoredTask status, ThroughputController throughputController) throws IOException {
970     // If an exception happens flushing, we let it out without clearing
971     // the memstore snapshot.  The old snapshot will be returned when we say
972     // 'snapshot', the next time flush comes around.
973     // Retry after catching exception when flushing, otherwise server will abort
974     // itself
975     StoreFlusher flusher = storeEngine.getStoreFlusher();
976     IOException lastException = null;
977     for (int i = 0; i < flushRetriesNumber; i++) {
978       try {
979         List<Path> pathNames =
980             flusher.flushSnapshot(snapshot, logCacheFlushId, status, throughputController);
981         Path lastPathName = null;
982         try {
983           for (Path pathName : pathNames) {
984             lastPathName = pathName;
985             validateStoreFile(pathName);
986           }
987           return pathNames;
988         } catch (Exception e) {
989           LOG.warn("Failed validating store file " + lastPathName + ", retrying num=" + i, e);
990           if (e instanceof IOException) {
991             lastException = (IOException) e;
992           } else {
993             lastException = new IOException(e);
994           }
995         }
996       } catch (IOException e) {
997         LOG.warn("Failed flushing store file, retrying num=" + i, e);
998         lastException = e;
999       }
1000       if (lastException != null && i < (flushRetriesNumber - 1)) {
1001         try {
1002           Thread.sleep(pauseTime);
1003         } catch (InterruptedException e) {
1004           IOException iie = new InterruptedIOException();
1005           iie.initCause(e);
1006           throw iie;
1007         }
1008       }
1009     }
1010     throw lastException;
1011   }
1012 
1013   /*
1014    * @param path The pathname of the tmp file into which the store was flushed
1015    * @param logCacheFlushId
1016    * @param status
1017    * @return StoreFile created.
1018    * @throws IOException
1019    */
1020   private StoreFile commitFile(final Path path, final long logCacheFlushId, MonitoredTask status)
1021       throws IOException {
1022     // Write-out finished successfully, move into the right spot
1023     Path dstPath = fs.commitStoreFile(getColumnFamilyName(), path);
1024 
1025     status.setStatus("Flushing " + this + ": reopening flushed file");
1026     StoreFile sf = createStoreFileAndReader(dstPath);
1027 
1028     StoreFile.Reader r = sf.getReader();
1029     this.storeSize += r.length();
1030     this.totalUncompressedBytes += r.getTotalUncompressedBytes();
1031 
1032     if (LOG.isInfoEnabled()) {
1033       LOG.info("Added " + sf + ", entries=" + r.getEntries() +
1034         ", sequenceid=" + logCacheFlushId +
1035         ", filesize=" + TraditionalBinaryPrefix.long2String(r.length(), "", 1));
1036     }
1037     return sf;
1038   }
1039 
1040   @Override
1041   public StoreFile.Writer createWriterInTmp(long maxKeyCount, Compression.Algorithm compression,
1042                                             boolean isCompaction, boolean includeMVCCReadpoint,
1043                                             boolean includesTag)
1044       throws IOException {
1045     return createWriterInTmp(maxKeyCount, compression, isCompaction, includeMVCCReadpoint,
1046         includesTag, false);
1047   }
1048 
1049   /*
1050    * @param maxKeyCount
1051    * @param compression Compression algorithm to use
1052    * @param isCompaction whether we are creating a new file in a compaction
1053    * @param includesMVCCReadPoint - whether to include MVCC or not
1054    * @param includesTag - includesTag or not
1055    * @return Writer for a new StoreFile in the tmp dir.
1056    */
1057   @Override
1058   public StoreFile.Writer createWriterInTmp(long maxKeyCount, Compression.Algorithm compression,
1059       boolean isCompaction, boolean includeMVCCReadpoint, boolean includesTag,
1060       boolean shouldDropBehind)
1061   throws IOException {
1062     return createWriterInTmp(maxKeyCount, compression, isCompaction, includeMVCCReadpoint,
1063         includesTag, shouldDropBehind, null);
1064   }
1065 
1066   /*
1067    * @param maxKeyCount
1068    * @param compression Compression algorithm to use
1069    * @param isCompaction whether we are creating a new file in a compaction
1070    * @param includesMVCCReadPoint - whether to include MVCC or not
1071    * @param includesTag - includesTag or not
1072    * @return Writer for a new StoreFile in the tmp dir.
1073    */
1074   @Override
1075   public StoreFile.Writer createWriterInTmp(long maxKeyCount, Compression.Algorithm compression,
1076       boolean isCompaction, boolean includeMVCCReadpoint, boolean includesTag,
1077       boolean shouldDropBehind, final TimeRangeTracker trt)
1078   throws IOException {
1079     final CacheConfig writerCacheConf;
1080     if (isCompaction) {
1081       // Don't cache data on write on compactions.
1082       writerCacheConf = new CacheConfig(cacheConf);
1083       writerCacheConf.setCacheDataOnWrite(false);
1084     } else {
1085       writerCacheConf = cacheConf;
1086     }
1087     InetSocketAddress[] favoredNodes = null;
1088     if (region.getRegionServerServices() != null) {
1089       favoredNodes = region.getRegionServerServices().getFavoredNodesForRegion(
1090           region.getRegionInfo().getEncodedName());
1091     }
1092     HFileContext hFileContext = createFileContext(compression, includeMVCCReadpoint, includesTag,
1093       cryptoContext);
1094     StoreFile.WriterBuilder builder = new StoreFile.WriterBuilder(conf, writerCacheConf,
1095         this.getFileSystem())
1096             .withFilePath(fs.createTempName())
1097             .withComparator(comparator)
1098             .withBloomType(family.getBloomFilterType())
1099             .withMaxKeyCount(maxKeyCount)
1100             .withFavoredNodes(favoredNodes)
1101             .withFileContext(hFileContext)
1102             .withShouldDropCacheBehind(shouldDropBehind)
1103             .withCompactedFiles(this.getCompactedfiles());
1104     if (trt != null) {
1105       builder.withTimeRangeTracker(trt);
1106     }
1107     return builder.build();
1108   }
1109 
1110   private HFileContext createFileContext(Compression.Algorithm compression,
1111       boolean includeMVCCReadpoint, boolean includesTag, Encryption.Context cryptoContext) {
1112     if (compression == null) {
1113       compression = HFile.DEFAULT_COMPRESSION_ALGORITHM;
1114     }
1115     HFileContext hFileContext = new HFileContextBuilder()
1116                                 .withIncludesMvcc(includeMVCCReadpoint)
1117                                 .withIncludesTags(includesTag)
1118                                 .withCompression(compression)
1119                                 .withCompressTags(family.isCompressTags())
1120                                 .withChecksumType(checksumType)
1121                                 .withBytesPerCheckSum(bytesPerChecksum)
1122                                 .withBlockSize(blocksize)
1123                                 .withHBaseCheckSum(true)
1124                                 .withDataBlockEncoding(family.getDataBlockEncoding())
1125                                 .withEncryptionContext(cryptoContext)
1126                                 .withCreateTime(EnvironmentEdgeManager.currentTime())
1127                                 .build();
1128     return hFileContext;
1129   }
1130 
1131 
1132   /*
1133    * Change storeFiles adding into place the Reader produced by this new flush.
1134    * @param sfs Store files
1135    * @param snapshotId
1136    * @throws IOException
1137    * @return Whether compaction is required.
1138    */
1139   private boolean updateStorefiles(final List<StoreFile> sfs, final long snapshotId)
1140       throws IOException {
1141     this.lock.writeLock().lock();
1142     try {
1143       this.storeEngine.getStoreFileManager().insertNewFiles(sfs);
1144       if (snapshotId > 0) {
1145         this.memstore.clearSnapshot(snapshotId);
1146       }
1147     } finally {
1148       // We need the lock, as long as we are updating the storeFiles
1149       // or changing the memstore. Let us release it before calling
1150       // notifyChangeReadersObservers. See HBASE-4485 for a possible
1151       // deadlock scenario that could have happened if continue to hold
1152       // the lock.
1153       this.lock.writeLock().unlock();
1154     }
1155     // notify to be called here - only in case of flushes
1156     notifyChangedReadersObservers(sfs);
1157     if (LOG.isTraceEnabled()) {
1158       long totalSize = 0;
1159       for (StoreFile sf : sfs) {
1160         totalSize += sf.getReader().length();
1161       }
1162       String traceMessage = "FLUSH time,count,size,store size,store files ["
1163           + EnvironmentEdgeManager.currentTime() + "," + sfs.size() + "," + totalSize
1164           + "," + storeSize + "," + storeEngine.getStoreFileManager().getStorefileCount() + "]";
1165       LOG.trace(traceMessage);
1166     }
1167     return needsCompaction();
1168   }
1169 
1170   /*
1171    * Notify all observers that set of Readers has changed.
1172    * @throws IOException
1173    */
1174   private void notifyChangedReadersObservers(List<StoreFile> sfs) throws IOException {
1175     for (ChangedReadersObserver o : this.changedReaderObservers) {
1176       List<KeyValueScanner> memStoreScanners;
1177       this.lock.readLock().lock();
1178       try {
1179         memStoreScanners = this.memstore.getScanners(o.getReadPoint());
1180       } finally {
1181         this.lock.readLock().unlock();
1182       }
1183       o.updateReaders(sfs, memStoreScanners);
1184     }
1185   }
1186 
1187   /**
1188    * Get all scanners with no filtering based on TTL (that happens further down
1189    * the line).
1190    * @return all scanners for this store
1191    */
1192   @Override
1193   public List<KeyValueScanner> getScanners(boolean cacheBlocks, boolean isGet,
1194       boolean usePread, boolean isCompaction, ScanQueryMatcher matcher, byte[] startRow,
1195       byte[] stopRow, long readPt) throws IOException {
1196     Collection<StoreFile> storeFilesToScan;
1197     List<KeyValueScanner> memStoreScanners;
1198     this.lock.readLock().lock();
1199     try {
1200       storeFilesToScan =
1201           this.storeEngine.getStoreFileManager().getFilesForScanOrGet(isGet, startRow, stopRow);
1202       memStoreScanners = this.memstore.getScanners(readPt);
1203     } finally {
1204       this.lock.readLock().unlock();
1205     }
1206 
1207     // First the store file scanners
1208 
1209     // TODO this used to get the store files in descending order,
1210     // but now we get them in ascending order, which I think is
1211     // actually more correct, since memstore get put at the end.
1212     List<StoreFileScanner> sfScanners = StoreFileScanner.getScannersForStoreFiles(storeFilesToScan,
1213         cacheBlocks, usePread, isCompaction, false, matcher, readPt, isPrimaryReplicaStore());
1214     List<KeyValueScanner> scanners =
1215       new ArrayList<KeyValueScanner>(sfScanners.size()+1);
1216     scanners.addAll(sfScanners);
1217     // Then the memstore scanners
1218     if (memStoreScanners != null) {
1219       scanners.addAll(memStoreScanners);
1220     }
1221     return scanners;
1222   }
1223 
1224   @Override
1225   public List<KeyValueScanner> getScanners(List<StoreFile> files, boolean cacheBlocks,
1226       boolean isGet, boolean usePread, boolean isCompaction, ScanQueryMatcher matcher,
1227       byte[] startRow, byte[] stopRow, long readPt, boolean includeMemstoreScanner) throws IOException {
1228     List<KeyValueScanner> memStoreScanners = null;
1229     if (includeMemstoreScanner) {
1230       this.lock.readLock().lock();
1231       try {
1232         memStoreScanners = this.memstore.getScanners(readPt);
1233       } finally {
1234         this.lock.readLock().unlock();
1235       }
1236     }
1237     List<StoreFileScanner> sfScanners = StoreFileScanner.getScannersForStoreFiles(files,
1238       cacheBlocks, usePread, isCompaction, false, matcher, readPt, isPrimaryReplicaStore());
1239     List<KeyValueScanner> scanners = new ArrayList<KeyValueScanner>(sfScanners.size() + 1);
1240     scanners.addAll(sfScanners);
1241     // Then the memstore scanners
1242     if (memStoreScanners != null) {
1243       scanners.addAll(memStoreScanners);
1244     }
1245     return scanners;
1246   }
1247 
1248   @Override
1249   public void addChangedReaderObserver(ChangedReadersObserver o) {
1250     this.changedReaderObservers.add(o);
1251   }
1252 
1253   @Override
1254   public void deleteChangedReaderObserver(ChangedReadersObserver o) {
1255     // We don't check if observer present; it may not be (legitimately)
1256     this.changedReaderObservers.remove(o);
1257   }
1258 
1259   //////////////////////////////////////////////////////////////////////////////
1260   // Compaction
1261   //////////////////////////////////////////////////////////////////////////////
1262 
1263   /**
1264    * Compact the StoreFiles.  This method may take some time, so the calling
1265    * thread must be able to block for long periods.
1266    *
1267    * <p>During this time, the Store can work as usual, getting values from
1268    * StoreFiles and writing new StoreFiles from the memstore.
1269    *
1270    * Existing StoreFiles are not destroyed until the new compacted StoreFile is
1271    * completely written-out to disk.
1272    *
1273    * <p>The compactLock prevents multiple simultaneous compactions.
1274    * The structureLock prevents us from interfering with other write operations.
1275    *
1276    * <p>We don't want to hold the structureLock for the whole time, as a compact()
1277    * can be lengthy and we want to allow cache-flushes during this period.
1278    *
1279    * <p> Compaction event should be idempotent, since there is no IO Fencing for
1280    * the region directory in hdfs. A region server might still try to complete the
1281    * compaction after it lost the region. That is why the following events are carefully
1282    * ordered for a compaction:
1283    *  1. Compaction writes new files under region/.tmp directory (compaction output)
1284    *  2. Compaction atomically moves the temporary file under region directory
1285    *  3. Compaction appends a WAL edit containing the compaction input and output files.
1286    *  Forces sync on WAL.
1287    *  4. Compaction deletes the input files from the region directory.
1288    *
1289    * Failure conditions are handled like this:
1290    *  - If RS fails before 2, compaction wont complete. Even if RS lives on and finishes
1291    *  the compaction later, it will only write the new data file to the region directory.
1292    *  Since we already have this data, this will be idempotent but we will have a redundant
1293    *  copy of the data.
1294    *  - If RS fails between 2 and 3, the region will have a redundant copy of the data. The
1295    *  RS that failed won't be able to finish snyc() for WAL because of lease recovery in WAL.
1296    *  - If RS fails after 3, the region region server who opens the region will pick up the
1297    *  the compaction marker from the WAL and replay it by removing the compaction input files.
1298    *  Failed RS can also attempt to delete those files, but the operation will be idempotent
1299    *
1300    * See HBASE-2231 for details.
1301    *
1302    * @param compaction compaction details obtained from requestCompaction()
1303    * @throws IOException
1304    * @return Storefile we compacted into or null if we failed or opted out early.
1305    */
1306   @Override
1307   public List<StoreFile> compact(CompactionContext compaction,
1308       ThroughputController throughputController) throws IOException {
1309     return compact(compaction, throughputController, null);
1310   }
1311 
1312   @Override
1313   public List<StoreFile> compact(CompactionContext compaction,
1314     ThroughputController throughputController, User user) throws IOException {
1315     assert compaction != null;
1316     List<StoreFile> sfs = null;
1317     CompactionRequest cr = compaction.getRequest();
1318     try {
1319       // Do all sanity checking in here if we have a valid CompactionRequest
1320       // because we need to clean up after it on the way out in a finally
1321       // block below
1322       long compactionStartTime = EnvironmentEdgeManager.currentTime();
1323       assert compaction.hasSelection();
1324       Collection<StoreFile> filesToCompact = cr.getFiles();
1325       assert !filesToCompact.isEmpty();
1326       synchronized (filesCompacting) {
1327         // sanity check: we're compacting files that this store knows about
1328         // TODO: change this to LOG.error() after more debugging
1329         Preconditions.checkArgument(filesCompacting.containsAll(filesToCompact));
1330       }
1331 
1332       // Ready to go. Have list of files to compact.
1333       LOG.info("Starting compaction of " + filesToCompact.size() + " file(s) in "
1334           + this + " of " + this.getRegionInfo().getRegionNameAsString()
1335           + " into tmpdir=" + fs.getTempDir() + ", totalSize="
1336           + TraditionalBinaryPrefix.long2String(cr.getSize(), "", 1));
1337 
1338       // Commence the compaction.
1339       List<Path> newFiles = compaction.compact(throughputController, user);
1340 
1341       long outputBytes = 0L;
1342       // TODO: get rid of this!
1343       if (!this.conf.getBoolean("hbase.hstore.compaction.complete", true)) {
1344         LOG.warn("hbase.hstore.compaction.complete is set to false");
1345         sfs = new ArrayList<StoreFile>(newFiles.size());
1346         final boolean evictOnClose =
1347             cacheConf != null? cacheConf.shouldEvictOnClose(): true;
1348         for (Path newFile : newFiles) {
1349           // Create storefile around what we wrote with a reader on it.
1350           StoreFile sf = createStoreFileAndReader(newFile);
1351           sf.closeReader(evictOnClose);
1352           sfs.add(sf);
1353         }
1354         return sfs;
1355       }
1356       // Do the steps necessary to complete the compaction.
1357       sfs = moveCompatedFilesIntoPlace(cr, newFiles, user);
1358       writeCompactionWalRecord(filesToCompact, sfs);
1359       replaceStoreFiles(filesToCompact, sfs);
1360       if (cr.isMajor()) {
1361         majorCompactedCellsCount += getCompactionProgress().totalCompactingKVs;
1362         majorCompactedCellsSize += getCompactionProgress().totalCompactedSize;
1363       } else {
1364         compactedCellsCount += getCompactionProgress().totalCompactingKVs;
1365         compactedCellsSize += getCompactionProgress().totalCompactedSize;
1366       }
1367 
1368       for (StoreFile sf : sfs) {
1369         outputBytes += sf.getReader().length();
1370       }
1371 
1372       // At this point the store will use new files for all new scanners.
1373       completeCompaction(filesToCompact); // update store size.
1374 
1375       long now = EnvironmentEdgeManager.currentTime();
1376       if (region.getRegionServerServices() != null
1377           && region.getRegionServerServices().getMetrics() != null) {
1378         region.getRegionServerServices().getMetrics().updateCompaction(cr.isMajor(),
1379           now - compactionStartTime, cr.getFiles().size(), newFiles.size(), cr.getSize(),
1380           outputBytes);
1381       }
1382 
1383       logCompactionEndMessage(cr, sfs, now, compactionStartTime);
1384       return sfs;
1385     } finally {
1386       finishCompactionRequest(cr);
1387     }
1388   }
1389 
1390   private List<StoreFile> moveCompatedFilesIntoPlace(
1391       final CompactionRequest cr, List<Path> newFiles, User user) throws IOException {
1392     List<StoreFile> sfs = new ArrayList<StoreFile>(newFiles.size());
1393     for (Path newFile : newFiles) {
1394       assert newFile != null;
1395       final StoreFile sf = moveFileIntoPlace(newFile);
1396       if (this.getCoprocessorHost() != null) {
1397         final Store thisStore = this;
1398         if (user == null) {
1399           getCoprocessorHost().postCompact(thisStore, sf, cr);
1400         } else {
1401           try {
1402             user.getUGI().doAs(new PrivilegedExceptionAction<Void>() {
1403               @Override
1404               public Void run() throws Exception {
1405                 getCoprocessorHost().postCompact(thisStore, sf, cr);
1406                 return null;
1407               }
1408             });
1409           } catch (InterruptedException ie) {
1410             InterruptedIOException iioe = new InterruptedIOException();
1411             iioe.initCause(ie);
1412             throw iioe;
1413           }
1414         }
1415       }
1416       assert sf != null;
1417       sfs.add(sf);
1418     }
1419     return sfs;
1420   }
1421 
1422   // Package-visible for tests
1423   StoreFile moveFileIntoPlace(final Path newFile) throws IOException {
1424     validateStoreFile(newFile);
1425     // Move the file into the right spot
1426     Path destPath = fs.commitStoreFile(getColumnFamilyName(), newFile);
1427     return createStoreFileAndReader(destPath);
1428   }
1429 
1430   /**
1431    * Writes the compaction WAL record.
1432    * @param filesCompacted Files compacted (input).
1433    * @param newFiles Files from compaction.
1434    */
1435   private void writeCompactionWalRecord(Collection<StoreFile> filesCompacted,
1436       Collection<StoreFile> newFiles) throws IOException {
1437     if (region.getWAL() == null) return;
1438     List<Path> inputPaths = new ArrayList<Path>(filesCompacted.size());
1439     for (StoreFile f : filesCompacted) {
1440       inputPaths.add(f.getPath());
1441     }
1442     List<Path> outputPaths = new ArrayList<Path>(newFiles.size());
1443     for (StoreFile f : newFiles) {
1444       outputPaths.add(f.getPath());
1445     }
1446     HRegionInfo info = this.region.getRegionInfo();
1447     CompactionDescriptor compactionDescriptor = ProtobufUtil.toCompactionDescriptor(info,
1448         family.getName(), inputPaths, outputPaths, fs.getStoreDir(getFamily().getNameAsString()));
1449     WALUtil.writeCompactionMarker(region.getWAL(), this.region.getTableDesc(),
1450         this.region.getRegionInfo(), compactionDescriptor, region.getMVCC());
1451   }
1452 
1453   @VisibleForTesting
1454   void replaceStoreFiles(final Collection<StoreFile> compactedFiles,
1455       final Collection<StoreFile> result) throws IOException {
1456     this.lock.writeLock().lock();
1457     try {
1458       this.storeEngine.getStoreFileManager().addCompactionResults(compactedFiles, result);
1459       synchronized (filesCompacting) {
1460         filesCompacting.removeAll(compactedFiles);
1461       }
1462     } finally {
1463       this.lock.writeLock().unlock();
1464     }
1465   }
1466 
1467   /**
1468    * Log a very elaborate compaction completion message.
1469    * @param cr Request.
1470    * @param sfs Resulting files.
1471    * @param compactionStartTime Start time.
1472    */
1473   private void logCompactionEndMessage(
1474       CompactionRequest cr, List<StoreFile> sfs, long now, long compactionStartTime) {
1475     StringBuilder message = new StringBuilder(
1476       "Completed" + (cr.isMajor() ? " major" : "") + " compaction of "
1477       + cr.getFiles().size() + (cr.isAllFiles() ? " (all)" : "") + " file(s) in "
1478       + this + " of " + this.getRegionInfo().getRegionNameAsString() + " into ");
1479     if (sfs.isEmpty()) {
1480       message.append("none, ");
1481     } else {
1482       for (StoreFile sf: sfs) {
1483         message.append(sf.getPath().getName());
1484         message.append("(size=");
1485         message.append(TraditionalBinaryPrefix.long2String(sf.getReader().length(), "", 1));
1486         message.append("), ");
1487       }
1488     }
1489     message.append("total size for store is ")
1490       .append(StringUtils.TraditionalBinaryPrefix.long2String(storeSize, "", 1))
1491       .append(". This selection was in queue for ")
1492       .append(StringUtils.formatTimeDiff(compactionStartTime, cr.getSelectionTime()))
1493       .append(", and took ").append(StringUtils.formatTimeDiff(now, compactionStartTime))
1494       .append(" to execute.");
1495     LOG.info(message.toString());
1496     if (LOG.isTraceEnabled()) {
1497       int fileCount = storeEngine.getStoreFileManager().getStorefileCount();
1498       long resultSize = 0;
1499       for (StoreFile sf : sfs) {
1500         resultSize += sf.getReader().length();
1501       }
1502       String traceMessage = "COMPACTION start,end,size out,files in,files out,store size,"
1503         + "store files [" + compactionStartTime + "," + now + "," + resultSize + ","
1504           + cr.getFiles().size() + "," + sfs.size() + "," +  storeSize + "," + fileCount + "]";
1505       LOG.trace(traceMessage);
1506     }
1507   }
1508 
1509   /**
1510    * Call to complete a compaction. Its for the case where we find in the WAL a compaction
1511    * that was not finished.  We could find one recovering a WAL after a regionserver crash.
1512    * See HBASE-2231.
1513    * @param compaction
1514    */
1515   @Override
1516   public void replayCompactionMarker(CompactionDescriptor compaction,
1517       boolean pickCompactionFiles, boolean removeFiles)
1518       throws IOException {
1519     LOG.debug("Completing compaction from the WAL marker");
1520     List<String> compactionInputs = compaction.getCompactionInputList();
1521     List<String> compactionOutputs = Lists.newArrayList(compaction.getCompactionOutputList());
1522 
1523     // The Compaction Marker is written after the compaction is completed,
1524     // and the files moved into the region/family folder.
1525     //
1526     // If we crash after the entry is written, we may not have removed the
1527     // input files, but the output file is present.
1528     // (The unremoved input files will be removed by this function)
1529     //
1530     // If we scan the directory and the file is not present, it can mean that:
1531     //   - The file was manually removed by the user
1532     //   - The file was removed as consequence of subsequent compaction
1533     // so, we can't do anything with the "compaction output list" because those
1534     // files have already been loaded when opening the region (by virtue of
1535     // being in the store's folder) or they may be missing due to a compaction.
1536 
1537     String familyName = this.getColumnFamilyName();
1538     List<String> inputFiles = new ArrayList<String>(compactionInputs.size());
1539     for (String compactionInput : compactionInputs) {
1540       Path inputPath = fs.getStoreFilePath(familyName, compactionInput);
1541       inputFiles.add(inputPath.getName());
1542     }
1543 
1544     //some of the input files might already be deleted
1545     List<StoreFile> inputStoreFiles = new ArrayList<StoreFile>(compactionInputs.size());
1546     for (StoreFile sf : this.getStorefiles()) {
1547       if (inputFiles.contains(sf.getPath().getName())) {
1548         inputStoreFiles.add(sf);
1549       }
1550     }
1551 
1552     // check whether we need to pick up the new files
1553     List<StoreFile> outputStoreFiles = new ArrayList<StoreFile>(compactionOutputs.size());
1554 
1555     if (pickCompactionFiles) {
1556       for (StoreFile sf : this.getStorefiles()) {
1557         compactionOutputs.remove(sf.getPath().getName());
1558       }
1559       for (String compactionOutput : compactionOutputs) {
1560         StoreFileInfo storeFileInfo = fs.getStoreFileInfo(getColumnFamilyName(), compactionOutput);
1561         StoreFile storeFile = createStoreFileAndReader(storeFileInfo);
1562         outputStoreFiles.add(storeFile);
1563       }
1564     }
1565 
1566     if (!inputStoreFiles.isEmpty() || !outputStoreFiles.isEmpty()) {
1567       LOG.info("Replaying compaction marker, replacing input files: " +
1568           inputStoreFiles + " with output files : " + outputStoreFiles);
1569       this.replaceStoreFiles(inputStoreFiles, outputStoreFiles);
1570       this.completeCompaction(inputStoreFiles);
1571     }
1572   }
1573 
1574   /**
1575    * This method tries to compact N recent files for testing.
1576    * Note that because compacting "recent" files only makes sense for some policies,
1577    * e.g. the default one, it assumes default policy is used. It doesn't use policy,
1578    * but instead makes a compaction candidate list by itself.
1579    * @param N Number of files.
1580    */
1581   public void compactRecentForTestingAssumingDefaultPolicy(int N) throws IOException {
1582     List<StoreFile> filesToCompact;
1583     boolean isMajor;
1584 
1585     this.lock.readLock().lock();
1586     try {
1587       synchronized (filesCompacting) {
1588         filesToCompact = Lists.newArrayList(storeEngine.getStoreFileManager().getStorefiles());
1589         if (!filesCompacting.isEmpty()) {
1590           // exclude all files older than the newest file we're currently
1591           // compacting. this allows us to preserve contiguity (HBASE-2856)
1592           StoreFile last = filesCompacting.get(filesCompacting.size() - 1);
1593           int idx = filesToCompact.indexOf(last);
1594           Preconditions.checkArgument(idx != -1);
1595           filesToCompact.subList(0, idx + 1).clear();
1596         }
1597         int count = filesToCompact.size();
1598         if (N > count) {
1599           throw new RuntimeException("Not enough files");
1600         }
1601 
1602         filesToCompact = filesToCompact.subList(count - N, count);
1603         isMajor = (filesToCompact.size() == storeEngine.getStoreFileManager().getStorefileCount());
1604         filesCompacting.addAll(filesToCompact);
1605         Collections.sort(filesCompacting, storeEngine.getStoreFileManager()
1606             .getStoreFileComparator());
1607       }
1608     } finally {
1609       this.lock.readLock().unlock();
1610     }
1611 
1612     try {
1613       // Ready to go. Have list of files to compact.
1614       List<Path> newFiles = ((DefaultCompactor)this.storeEngine.getCompactor())
1615           .compactForTesting(filesToCompact, isMajor);
1616       for (Path newFile: newFiles) {
1617         // Move the compaction into place.
1618         StoreFile sf = moveFileIntoPlace(newFile);
1619         if (this.getCoprocessorHost() != null) {
1620           this.getCoprocessorHost().postCompact(this, sf, null);
1621         }
1622         replaceStoreFiles(filesToCompact, Lists.newArrayList(sf));
1623         completeCompaction(filesToCompact);
1624       }
1625     } finally {
1626       synchronized (filesCompacting) {
1627         filesCompacting.removeAll(filesToCompact);
1628       }
1629     }
1630   }
1631 
1632   @Override
1633   public boolean hasReferences() {
1634     // Grab the read lock here, because we need to ensure that: only when the atomic
1635     // replaceStoreFiles(..) finished, we can get all the complete store file list.
1636     this.lock.readLock().lock();
1637     try {
1638       // Merge the current store files with compacted files here due to HBASE-20940.
1639       Collection<StoreFile> allStoreFiles = new ArrayList<>(getStorefiles());
1640       allStoreFiles.addAll(getCompactedfiles());
1641       return StoreUtils.hasReferences(allStoreFiles);
1642     } finally {
1643       this.lock.readLock().unlock();
1644     }
1645   }
1646 
1647   @Override
1648   public CompactionProgress getCompactionProgress() {
1649     return this.storeEngine.getCompactor().getProgress();
1650   }
1651 
1652   @Override
1653   public boolean isMajorCompaction() throws IOException {
1654     for (StoreFile sf : this.storeEngine.getStoreFileManager().getStorefiles()) {
1655       // TODO: what are these reader checks all over the place?
1656       if (sf.getReader() == null) {
1657         LOG.debug("StoreFile " + sf + " has null Reader");
1658         return false;
1659       }
1660     }
1661     return storeEngine.getCompactionPolicy().shouldPerformMajorCompaction(
1662         this.storeEngine.getStoreFileManager().getStorefiles());
1663   }
1664 
1665   @Override
1666   public CompactionContext requestCompaction() throws IOException {
1667     return requestCompaction(Store.NO_PRIORITY, null);
1668   }
1669 
1670   @Override
1671   public CompactionContext requestCompaction(int priority, CompactionRequest baseRequest)
1672       throws IOException {
1673     return requestCompaction(priority, baseRequest, null);
1674   }
1675   @Override
1676   public CompactionContext requestCompaction(int priority, final CompactionRequest baseRequest,
1677       User user) throws IOException {
1678     // don't even select for compaction if writes are disabled
1679     if (!this.areWritesEnabled()) {
1680       return null;
1681     }
1682 
1683     // Before we do compaction, try to get rid of unneeded files to simplify things.
1684     removeUnneededFiles();
1685 
1686     final CompactionContext compaction = storeEngine.createCompaction();
1687     CompactionRequest request = null;
1688     this.lock.readLock().lock();
1689     try {
1690       synchronized (filesCompacting) {
1691         final Store thisStore = this;
1692         // First, see if coprocessor would want to override selection.
1693         if (this.getCoprocessorHost() != null) {
1694           final List<StoreFile> candidatesForCoproc = compaction.preSelect(this.filesCompacting);
1695           boolean override = false;
1696           if (user == null) {
1697             override = getCoprocessorHost().preCompactSelection(this, candidatesForCoproc,
1698               baseRequest);
1699           } else {
1700             try {
1701               override = user.getUGI().doAs(new PrivilegedExceptionAction<Boolean>() {
1702                 @Override
1703                 public Boolean run() throws Exception {
1704                   return getCoprocessorHost().preCompactSelection(thisStore, candidatesForCoproc,
1705                     baseRequest);
1706                 }
1707               });
1708             } catch (InterruptedException ie) {
1709               InterruptedIOException iioe = new InterruptedIOException();
1710               iioe.initCause(ie);
1711               throw iioe;
1712             }
1713           }
1714           if (override) {
1715             // Coprocessor is overriding normal file selection.
1716             compaction.forceSelect(new CompactionRequest(candidatesForCoproc));
1717           }
1718         }
1719 
1720         // Normal case - coprocessor is not overriding file selection.
1721         if (!compaction.hasSelection()) {
1722           boolean isUserCompaction = priority == Store.PRIORITY_USER;
1723           boolean mayUseOffPeak = offPeakHours.isOffPeakHour() &&
1724               offPeakCompactionTracker.compareAndSet(false, true);
1725           try {
1726             compaction.select(this.filesCompacting, isUserCompaction,
1727               mayUseOffPeak, forceMajor && filesCompacting.isEmpty());
1728           } catch (IOException e) {
1729             if (mayUseOffPeak) {
1730               offPeakCompactionTracker.set(false);
1731             }
1732             throw e;
1733           }
1734           assert compaction.hasSelection();
1735           if (mayUseOffPeak && !compaction.getRequest().isOffPeak()) {
1736             // Compaction policy doesn't want to take advantage of off-peak.
1737             offPeakCompactionTracker.set(false);
1738           }
1739         }
1740         if (this.getCoprocessorHost() != null) {
1741           if (user == null) {
1742             this.getCoprocessorHost().postCompactSelection(
1743               this, ImmutableList.copyOf(compaction.getRequest().getFiles()), baseRequest);
1744           } else {
1745             try {
1746               user.getUGI().doAs(new PrivilegedExceptionAction<Void>() {
1747                 @Override
1748                 public Void run() throws Exception {
1749                   getCoprocessorHost().postCompactSelection(
1750                     thisStore,ImmutableList.copyOf(compaction.getRequest().getFiles()),baseRequest);
1751                   return null;
1752                 }
1753               });
1754             } catch (InterruptedException ie) {
1755               InterruptedIOException iioe = new InterruptedIOException();
1756               iioe.initCause(ie);
1757               throw iioe;
1758             }
1759           }
1760         }
1761 
1762         // Selected files; see if we have a compaction with some custom base request.
1763         if (baseRequest != null) {
1764           // Update the request with what the system thinks the request should be;
1765           // its up to the request if it wants to listen.
1766           compaction.forceSelect(
1767               baseRequest.combineWith(compaction.getRequest()));
1768         }
1769         // Finally, we have the resulting files list. Check if we have any files at all.
1770         request = compaction.getRequest();
1771         final Collection<StoreFile> selectedFiles = request.getFiles();
1772         if (selectedFiles.isEmpty()) {
1773           return null;
1774         }
1775 
1776         addToCompactingFiles(selectedFiles);
1777 
1778         // If we're enqueuing a major, clear the force flag.
1779         this.forceMajor = this.forceMajor && !request.isMajor();
1780 
1781         // Set common request properties.
1782         // Set priority, either override value supplied by caller or from store.
1783         request.setPriority((priority != Store.NO_PRIORITY) ? priority : getCompactPriority());
1784         request.setDescription(getRegionInfo().getRegionNameAsString(), getColumnFamilyName());
1785       }
1786     } finally {
1787       this.lock.readLock().unlock();
1788     }
1789 
1790     LOG.debug(getRegionInfo().getEncodedName() + " - "  + getColumnFamilyName()
1791         + ": Initiating " + (request.isMajor() ? "major" : "minor") + " compaction"
1792         + (request.isAllFiles() ? " (all files)" : ""));
1793     this.region.reportCompactionRequestStart(request.isMajor());
1794     return compaction;
1795   }
1796 
1797   /** Adds the files to compacting files. filesCompacting must be locked. */
1798   private void addToCompactingFiles(final Collection<StoreFile> filesToAdd) {
1799     if (filesToAdd == null) return;
1800     // Check that we do not try to compact the same StoreFile twice.
1801     if (!Collections.disjoint(filesCompacting, filesToAdd)) {
1802       Preconditions.checkArgument(false, "%s overlaps with %s", filesToAdd, filesCompacting);
1803     }
1804     filesCompacting.addAll(filesToAdd);
1805     Collections.sort(filesCompacting, storeEngine.getStoreFileManager().getStoreFileComparator());
1806   }
1807 
1808   private void removeUnneededFiles() throws IOException {
1809     if (!conf.getBoolean("hbase.store.delete.expired.storefile", true)) return;
1810     if (getFamily().getMinVersions() > 0) {
1811       LOG.debug("Skipping expired store file removal due to min version being " +
1812           getFamily().getMinVersions());
1813       return;
1814     }
1815     this.lock.readLock().lock();
1816     Collection<StoreFile> delSfs = null;
1817     try {
1818       synchronized (filesCompacting) {
1819         long cfTtl = getStoreFileTtl();
1820         if (cfTtl != Long.MAX_VALUE) {
1821           delSfs = storeEngine.getStoreFileManager().getUnneededFiles(
1822               EnvironmentEdgeManager.currentTime() - cfTtl, filesCompacting);
1823           addToCompactingFiles(delSfs);
1824         }
1825       }
1826     } finally {
1827       this.lock.readLock().unlock();
1828     }
1829     if (delSfs == null || delSfs.isEmpty()) return;
1830 
1831     Collection<StoreFile> newFiles = new ArrayList<StoreFile>(); // No new files.
1832     writeCompactionWalRecord(delSfs, newFiles);
1833     replaceStoreFiles(delSfs, newFiles);
1834     completeCompaction(delSfs);
1835     LOG.info("Completed removal of " + delSfs.size() + " unnecessary (expired) file(s) in "
1836         + this + " of " + this.getRegionInfo().getRegionNameAsString()
1837         + "; total size for store is " + TraditionalBinaryPrefix.long2String(storeSize, "", 1));
1838   }
1839 
1840   @Override
1841   public void cancelRequestedCompaction(CompactionContext compaction) {
1842     finishCompactionRequest(compaction.getRequest());
1843   }
1844 
1845   private void finishCompactionRequest(CompactionRequest cr) {
1846     this.region.reportCompactionRequestEnd(cr.isMajor(), cr.getFiles().size(), cr.getSize());
1847     if (cr.isOffPeak()) {
1848       offPeakCompactionTracker.set(false);
1849       cr.setOffPeak(false);
1850     }
1851     synchronized (filesCompacting) {
1852       filesCompacting.removeAll(cr.getFiles());
1853     }
1854   }
1855 
1856   /**
1857    * Validates a store file by opening and closing it. In HFileV2 this should
1858    * not be an expensive operation.
1859    *
1860    * @param path the path to the store file
1861    */
1862   private void validateStoreFile(Path path)
1863       throws IOException {
1864     StoreFile storeFile = null;
1865     try {
1866       storeFile = createStoreFileAndReader(path);
1867     } catch (IOException e) {
1868       LOG.error("Failed to open store file : " + path
1869           + ", keeping it in tmp location", e);
1870       throw e;
1871     } finally {
1872       if (storeFile != null) {
1873         storeFile.closeReader(false);
1874       }
1875     }
1876   }
1877 
1878   /**
1879    * <p>It works by processing a compaction that's been written to disk.
1880    *
1881    * <p>It is usually invoked at the end of a compaction, but might also be
1882    * invoked at HStore startup, if the prior execution died midway through.
1883    *
1884    * <p>Moving the compacted TreeMap into place means:
1885    * <pre>
1886    * 1) Unload all replaced StoreFile, close and collect list to delete.
1887    * 2) Compute new store size
1888    * </pre>
1889    *
1890    * @param compactedFiles list of files that were compacted
1891    */
1892   @VisibleForTesting
1893   protected void completeCompaction(final Collection<StoreFile> compactedFiles)
1894     throws IOException {
1895     completeCompaction(compactedFiles, true);
1896   }
1897 
1898 
1899   /**
1900    * <p>It works by processing a compaction that's been written to disk.
1901    *
1902    * <p>It is usually invoked at the end of a compaction, but might also be
1903    * invoked at HStore startup, if the prior execution died midway through.
1904    *
1905    * <p>Moving the compacted TreeMap into place means:
1906    * <pre>
1907    * 1) Unload all replaced StoreFile, close and collect list to delete.
1908    * 2) Compute new store size
1909    * </pre>
1910    *
1911    * @param compactedFiles list of files that were compacted
1912    */
1913   @VisibleForTesting
1914   protected void completeCompaction(final Collection<StoreFile> compactedFiles, boolean removeFiles)
1915       throws IOException {
1916     LOG.debug("Completing compaction...");
1917     this.storeSize = 0L;
1918     this.totalUncompressedBytes = 0L;
1919     for (StoreFile hsf : this.storeEngine.getStoreFileManager().getStorefiles()) {
1920       StoreFile.Reader r = hsf.getReader();
1921       if (r == null) {
1922         LOG.warn("StoreFile " + hsf + " has a null Reader");
1923         continue;
1924       }
1925       this.storeSize += r.length();
1926       this.totalUncompressedBytes += r.getTotalUncompressedBytes();
1927     }
1928   }
1929 
1930   /*
1931    * @param wantedVersions How many versions were asked for.
1932    * @return wantedVersions or this families' {@link HConstants#VERSIONS}.
1933    */
1934   int versionsToReturn(final int wantedVersions) {
1935     if (wantedVersions <= 0) {
1936       throw new IllegalArgumentException("Number of versions must be > 0");
1937     }
1938     // Make sure we do not return more than maximum versions for this store.
1939     int maxVersions = this.family.getMaxVersions();
1940     return wantedVersions > maxVersions ? maxVersions: wantedVersions;
1941   }
1942 
1943   /**
1944    * @param cell
1945    * @param oldestTimestamp
1946    * @return true if the cell is expired
1947    */
1948   static boolean isCellTTLExpired(final Cell cell, final long oldestTimestamp, final long now) {
1949     // Do not create an Iterator or Tag objects unless the cell actually has tags.
1950     // TODO: This check for tags is really expensive. We decode an int for key and value. Costs.
1951     if (cell.getTagsLength() > 0) {
1952       // Look for a TTL tag first. Use it instead of the family setting if
1953       // found. If a cell has multiple TTLs, resolve the conflict by using the
1954       // first tag encountered.
1955       Iterator<Tag> i = CellUtil.tagsIterator(cell.getTagsArray(), cell.getTagsOffset(),
1956         cell.getTagsLength());
1957       while (i.hasNext()) {
1958         Tag t = i.next();
1959         if (TagType.TTL_TAG_TYPE == t.getType()) {
1960           // Unlike in schema cell TTLs are stored in milliseconds, no need
1961           // to convert
1962           long ts = cell.getTimestamp();
1963           assert t.getTagLength() == Bytes.SIZEOF_LONG;
1964           long ttl = Bytes.toLong(t.getBuffer(), t.getTagOffset(), t.getTagLength());
1965           if (ts + ttl < now) {
1966             return true;
1967           }
1968           // Per cell TTLs cannot extend lifetime beyond family settings, so
1969           // fall through to check that
1970           break;
1971         }
1972       }
1973     }
1974     return false;
1975   }
1976 
1977   @Override
1978   public Cell getRowKeyAtOrBefore(final byte[] row) throws IOException {
1979     // If minVersions is set, we will not ignore expired KVs.
1980     // As we're only looking for the latest matches, that should be OK.
1981     // With minVersions > 0 we guarantee that any KV that has any version
1982     // at all (expired or not) has at least one version that will not expire.
1983     // Note that this method used to take a KeyValue as arguments. KeyValue
1984     // can be back-dated, a row key cannot.
1985     long ttlToUse = scanInfo.getMinVersions() > 0 ? Long.MAX_VALUE : this.scanInfo.getTtl();
1986 
1987     KeyValue kv = new KeyValue(row, HConstants.LATEST_TIMESTAMP);
1988 
1989     GetClosestRowBeforeTracker state = new GetClosestRowBeforeTracker(
1990       this.comparator, kv, ttlToUse, this.getRegionInfo().isMetaRegion());
1991     this.lock.readLock().lock();
1992     try {
1993       // First go to the memstore.  Pick up deletes and candidates.
1994       this.memstore.getRowKeyAtOrBefore(state);
1995       // Check if match, if we got a candidate on the asked for 'kv' row.
1996       // Process each relevant store file. Run through from newest to oldest.
1997       Iterator<StoreFile> sfIterator = this.storeEngine.getStoreFileManager()
1998           .getCandidateFilesForRowKeyBefore(state.getTargetKey());
1999       while (sfIterator.hasNext()) {
2000         StoreFile sf = sfIterator.next();
2001         sfIterator.remove(); // Remove sf from iterator.
2002         boolean haveNewCandidate = rowAtOrBeforeFromStoreFile(sf, state);
2003         Cell candidate = state.getCandidate();
2004         // we have an optimization here which stops the search if we find exact match.
2005         if (candidate != null && CellUtil.matchingRow(candidate, row)) {
2006           return candidate;
2007         }
2008         if (haveNewCandidate) {
2009           sfIterator = this.storeEngine.getStoreFileManager().updateCandidateFilesForRowKeyBefore(
2010               sfIterator, state.getTargetKey(), candidate);
2011         }
2012       }
2013       return state.getCandidate();
2014     } finally {
2015       this.lock.readLock().unlock();
2016     }
2017   }
2018 
2019   /*
2020    * Check an individual MapFile for the row at or before a given row.
2021    * @param f
2022    * @param state
2023    * @throws IOException
2024    * @return True iff the candidate has been updated in the state.
2025    */
2026   private boolean rowAtOrBeforeFromStoreFile(final StoreFile f,
2027                                           final GetClosestRowBeforeTracker state)
2028       throws IOException {
2029     StoreFile.Reader r = f.getReader();
2030     if (r == null) {
2031       LOG.warn("StoreFile " + f + " has a null Reader");
2032       return false;
2033     }
2034     if (r.getEntries() == 0) {
2035       LOG.warn("StoreFile " + f + " is a empty store file");
2036       return false;
2037     }
2038     // TODO: Cache these keys rather than make each time?
2039     byte [] fk = r.getFirstKey();
2040     if (fk == null) return false;
2041     KeyValue firstKV = KeyValue.createKeyValueFromKey(fk, 0, fk.length);
2042     byte [] lk = r.getLastKey();
2043     KeyValue lastKV = KeyValue.createKeyValueFromKey(lk, 0, lk.length);
2044     KeyValue firstOnRow = state.getTargetKey();
2045     if (this.comparator.compareRows(lastKV, firstOnRow) < 0) {
2046       // If last key in file is not of the target table, no candidates in this
2047       // file.  Return.
2048       if (!state.isTargetTable(lastKV)) return false;
2049       // If the row we're looking for is past the end of file, set search key to
2050       // last key. TODO: Cache last and first key rather than make each time.
2051       firstOnRow = new KeyValue(lastKV.getRow(), HConstants.LATEST_TIMESTAMP);
2052     }
2053     // Get a scanner that caches blocks and that uses pread.
2054     HFileScanner scanner = r.getScanner(true, true, false);
2055     // Seek scanner.  If can't seek it, return.
2056     if (!seekToScanner(scanner, firstOnRow, firstKV)) return false;
2057     // If we found candidate on firstOnRow, just return. THIS WILL NEVER HAPPEN!
2058     // Unlikely that there'll be an instance of actual first row in table.
2059     if (walkForwardInSingleRow(scanner, firstOnRow, state)) return true;
2060     // If here, need to start backing up.
2061     while (scanner.seekBefore(firstOnRow.getBuffer(), firstOnRow.getKeyOffset(),
2062        firstOnRow.getKeyLength())) {
2063       Cell kv = scanner.getKeyValue();
2064       if (!state.isTargetTable(kv)) break;
2065       if (!state.isBetterCandidate(kv)) break;
2066       // Make new first on row.
2067       firstOnRow = new KeyValue(kv.getRow(), HConstants.LATEST_TIMESTAMP);
2068       // Seek scanner.  If can't seek it, break.
2069       if (!seekToScanner(scanner, firstOnRow, firstKV)) return false;
2070       // If we find something, break;
2071       if (walkForwardInSingleRow(scanner, firstOnRow, state)) return true;
2072     }
2073     return false;
2074   }
2075 
2076   /*
2077    * Seek the file scanner to firstOnRow or first entry in file.
2078    * @param scanner
2079    * @param firstOnRow
2080    * @param firstKV
2081    * @return True if we successfully seeked scanner.
2082    * @throws IOException
2083    */
2084   private boolean seekToScanner(final HFileScanner scanner,
2085                                 final KeyValue firstOnRow,
2086                                 final KeyValue firstKV)
2087       throws IOException {
2088     KeyValue kv = firstOnRow;
2089     // If firstOnRow < firstKV, set to firstKV
2090     if (this.comparator.compareRows(firstKV, firstOnRow) == 0) kv = firstKV;
2091     int result = scanner.seekTo(kv);
2092     return result != -1;
2093   }
2094 
2095   /*
2096    * When we come in here, we are probably at the kv just before we break into
2097    * the row that firstOnRow is on.  Usually need to increment one time to get
2098    * on to the row we are interested in.
2099    * @param scanner
2100    * @param firstOnRow
2101    * @param state
2102    * @return True we found a candidate.
2103    * @throws IOException
2104    */
2105   private boolean walkForwardInSingleRow(final HFileScanner scanner,
2106                                          final KeyValue firstOnRow,
2107                                          final GetClosestRowBeforeTracker state)
2108       throws IOException {
2109     boolean foundCandidate = false;
2110     do {
2111       Cell kv = scanner.getKeyValue();
2112       // If we are not in the row, skip.
2113       if (this.comparator.compareRows(kv, firstOnRow) < 0) continue;
2114       // Did we go beyond the target row? If so break.
2115       if (state.isTooFar(kv, firstOnRow)) break;
2116       if (state.isExpired(kv)) {
2117         continue;
2118       }
2119       // If we added something, this row is a contender. break.
2120       if (state.handle(kv)) {
2121         foundCandidate = true;
2122         break;
2123       }
2124     } while(scanner.next());
2125     return foundCandidate;
2126   }
2127 
2128   @Override
2129   public boolean canSplit() {
2130     this.lock.readLock().lock();
2131     try {
2132       // Not split-able if we find a reference store file present in the store.
2133       boolean result = !hasReferences();
2134       if (!result && LOG.isDebugEnabled()) {
2135         LOG.debug("Cannot split region due to reference files being there");
2136       }
2137       return result;
2138     } finally {
2139       this.lock.readLock().unlock();
2140     }
2141   }
2142 
2143   @Override
2144   public byte[] getSplitPoint() {
2145     this.lock.readLock().lock();
2146     try {
2147       // Should already be enforced by the split policy!
2148       assert !this.getRegionInfo().isMetaRegion();
2149       // Not split-able if we find a reference store file present in the store.
2150       if (hasReferences()) {
2151         if (LOG.isTraceEnabled()) {
2152           LOG.trace("Not splittable; has references: " + this);
2153         }
2154         return null;
2155       }
2156       return this.storeEngine.getStoreFileManager().getSplitPoint();
2157     } catch(IOException e) {
2158       LOG.warn("Failed getting store size for " + this, e);
2159     } finally {
2160       this.lock.readLock().unlock();
2161     }
2162     return null;
2163   }
2164 
2165   @Override
2166   public long getLastCompactSize() {
2167     return this.lastCompactSize;
2168   }
2169 
2170   @Override
2171   public long getSize() {
2172     return storeSize;
2173   }
2174 
2175   @Override
2176   public void triggerMajorCompaction() {
2177     this.forceMajor = true;
2178   }
2179 
2180 
2181   //////////////////////////////////////////////////////////////////////////////
2182   // File administration
2183   //////////////////////////////////////////////////////////////////////////////
2184 
2185   @Override
2186   public KeyValueScanner getScanner(Scan scan,
2187       final NavigableSet<byte []> targetCols, long readPt) throws IOException {
2188     lock.readLock().lock();
2189     try {
2190       KeyValueScanner scanner = null;
2191       if (this.getCoprocessorHost() != null) {
2192         scanner = this.getCoprocessorHost().preStoreScannerOpen(this, scan, targetCols);
2193       }
2194       if (scanner == null) {
2195         scanner = scan.isReversed() ? new ReversedStoreScanner(this,
2196             getScanInfo(), scan, targetCols, readPt) : new StoreScanner(this,
2197             getScanInfo(), scan, targetCols, readPt);
2198       }
2199       return scanner;
2200     } finally {
2201       lock.readLock().unlock();
2202     }
2203   }
2204 
2205   @Override
2206   public String toString() {
2207     return this.getColumnFamilyName();
2208   }
2209 
2210   @Override
2211   public int getStorefilesCount() {
2212     return this.storeEngine.getStoreFileManager().getStorefileCount();
2213   }
2214 
2215   @Override
2216   public long getMaxStoreFileAge() {
2217     long earliestTS = Long.MAX_VALUE;
2218     for (StoreFile s: this.storeEngine.getStoreFileManager().getStorefiles()) {
2219       StoreFile.Reader r = s.getReader();
2220       if (r == null) {
2221         LOG.warn("StoreFile " + s + " has a null Reader");
2222         continue;
2223       }
2224       if (!s.isHFile()) {
2225         continue;
2226       }
2227       long createdTS = s.getFileInfo().getCreatedTimestamp();
2228       earliestTS = (createdTS < earliestTS) ? createdTS : earliestTS;
2229     }
2230     long now = EnvironmentEdgeManager.currentTime();
2231     return now - earliestTS;
2232   }
2233 
2234   @Override
2235   public long getMinStoreFileAge() {
2236     long latestTS = 0;
2237     for (StoreFile s: this.storeEngine.getStoreFileManager().getStorefiles()) {
2238       StoreFile.Reader r = s.getReader();
2239       if (r == null) {
2240         LOG.warn("StoreFile " + s + " has a null Reader");
2241         continue;
2242       }
2243       if (!s.isHFile()) {
2244         continue;
2245       }
2246       long createdTS = s.getFileInfo().getCreatedTimestamp();
2247       latestTS = (createdTS > latestTS) ? createdTS : latestTS;
2248     }
2249     long now = EnvironmentEdgeManager.currentTime();
2250     return now - latestTS;
2251   }
2252 
2253   @Override
2254   public long getAvgStoreFileAge() {
2255     long sum = 0, count = 0;
2256     for (StoreFile s: this.storeEngine.getStoreFileManager().getStorefiles()) {
2257       StoreFile.Reader r = s.getReader();
2258       if (r == null) {
2259         LOG.warn("StoreFile " + s + " has a null Reader");
2260         continue;
2261       }
2262       if (!s.isHFile()) {
2263         continue;
2264       }
2265       sum += s.getFileInfo().getCreatedTimestamp();
2266       count++;
2267     }
2268     if (count == 0) {
2269       return 0;
2270     }
2271     long avgTS = sum / count;
2272     long now = EnvironmentEdgeManager.currentTime();
2273     return now - avgTS;
2274   }
2275 
2276   @Override
2277   public long getNumReferenceFiles() {
2278     long numRefFiles = 0;
2279     for (StoreFile s : this.storeEngine.getStoreFileManager().getStorefiles()) {
2280       if (s.isReference()) {
2281         numRefFiles++;
2282       }
2283     }
2284     return numRefFiles;
2285   }
2286 
2287   @Override
2288   public long getNumHFiles() {
2289     long numHFiles = 0;
2290     for (StoreFile s : this.storeEngine.getStoreFileManager().getStorefiles()) {
2291       if (s.isHFile()) {
2292         numHFiles++;
2293       }
2294     }
2295     return numHFiles;
2296   }
2297 
2298   @Override
2299   public long getStoreSizeUncompressed() {
2300     return this.totalUncompressedBytes;
2301   }
2302 
2303   private long getTotalUmcompressedBytes(Collection<StoreFile> files) {
2304     long size = 0;
2305     for (StoreFile sf : files) {
2306       if (sf != null && sf.getReader() != null) {
2307         size += sf.getReader().getTotalUncompressedBytes();
2308       }
2309     }
2310     return size;
2311   }
2312 
2313   private long getStorefilesSize(Collection<StoreFile> files) {
2314     long size = 0;
2315     for (StoreFile sf : files) {
2316       if (sf != null) {
2317         if (sf.getReader() == null) {
2318           LOG.warn("StoreFile " + sf + " has a null Reader");
2319           continue;
2320         }
2321         size += sf.getReader().length();
2322       }
2323     }
2324     return size;
2325   }
2326 
2327   @Override
2328   public long getStorefilesSize() {
2329     return getStorefilesSize(storeEngine.getStoreFileManager().getStorefiles());
2330   }
2331 
2332   @Override
2333   public long getStorefilesIndexSize() {
2334     long size = 0;
2335     for (StoreFile s: this.storeEngine.getStoreFileManager().getStorefiles()) {
2336       StoreFile.Reader r = s.getReader();
2337       if (r == null) {
2338         LOG.warn("StoreFile " + s + " has a null Reader");
2339         continue;
2340       }
2341       size += r.indexSize();
2342     }
2343     return size;
2344   }
2345 
2346   @Override
2347   public long getTotalStaticIndexSize() {
2348     long size = 0;
2349     for (StoreFile s : this.storeEngine.getStoreFileManager().getStorefiles()) {
2350       StoreFile.Reader r = s.getReader();
2351       if (r == null) {
2352         continue;
2353       }
2354       size += r.getUncompressedDataIndexSize();
2355     }
2356     return size;
2357   }
2358 
2359   @Override
2360   public long getTotalStaticBloomSize() {
2361     long size = 0;
2362     for (StoreFile s : this.storeEngine.getStoreFileManager().getStorefiles()) {
2363       StoreFile.Reader r = s.getReader();
2364       if (r == null) {
2365         continue;
2366       }
2367       size += r.getTotalBloomSize();
2368     }
2369     return size;
2370   }
2371 
2372   @Override
2373   public long getMemStoreSize() {
2374     return this.memstore.size();
2375   }
2376 
2377   @Override
2378   public int getCompactPriority() {
2379     int priority = this.storeEngine.getStoreFileManager().getStoreCompactionPriority();
2380     if (priority == PRIORITY_USER) {
2381       LOG.warn("Compaction priority is USER despite there being no user compaction");
2382     }
2383     return priority;
2384   }
2385 
2386   @Override
2387   public boolean throttleCompaction(long compactionSize) {
2388     return storeEngine.getCompactionPolicy().throttleCompaction(compactionSize);
2389   }
2390 
2391   public HRegion getHRegion() {
2392     return this.region;
2393   }
2394 
2395   @Override
2396   public RegionCoprocessorHost getCoprocessorHost() {
2397     return this.region.getCoprocessorHost();
2398   }
2399 
2400   @Override
2401   public HRegionInfo getRegionInfo() {
2402     return this.fs.getRegionInfo();
2403   }
2404 
2405   @Override
2406   public boolean areWritesEnabled() {
2407     return this.region.areWritesEnabled();
2408   }
2409 
2410   @Override
2411   public long getSmallestReadPoint() {
2412     return this.region.getSmallestReadPoint();
2413   }
2414 
2415   /**
2416    * Used in tests. TODO: Remove
2417    *
2418    * Updates the value for the given row/family/qualifier. This function will always be seen as
2419    * atomic by other readers because it only puts a single KV to memstore. Thus no read/write
2420    * control necessary.
2421    * @param row row to update
2422    * @param f family to update
2423    * @param qualifier qualifier to update
2424    * @param newValue the new value to set into memstore
2425    * @return memstore size delta
2426    * @throws IOException
2427    */
2428   public long updateColumnValue(byte [] row, byte [] f,
2429                                 byte [] qualifier, long newValue)
2430       throws IOException {
2431 
2432     this.lock.readLock().lock();
2433     try {
2434       long now = EnvironmentEdgeManager.currentTime();
2435 
2436       return this.memstore.updateColumnValue(row,
2437           f,
2438           qualifier,
2439           newValue,
2440           now);
2441 
2442     } finally {
2443       this.lock.readLock().unlock();
2444     }
2445   }
2446 
2447   @Override
2448   public long upsert(Iterable<Cell> cells, long readpoint,
2449                 List<Cell> removedCells) throws IOException {
2450     this.lock.readLock().lock();
2451     try {
2452       return this.memstore.upsert(cells, readpoint, removedCells);
2453     } finally {
2454       this.lock.readLock().unlock();
2455     }
2456   }
2457 
2458   @Override
2459   public StoreFlushContext createFlushContext(long cacheFlushId) {
2460     return new StoreFlusherImpl(cacheFlushId);
2461   }
2462 
2463   private final class StoreFlusherImpl implements StoreFlushContext {
2464 
2465     private long cacheFlushSeqNum;
2466     private MemStoreSnapshot snapshot;
2467     private List<Path> tempFiles;
2468     private List<Path> committedFiles;
2469     private long cacheFlushCount;
2470     private long cacheFlushSize;
2471     private long outputFileSize;
2472 
2473     private StoreFlusherImpl(long cacheFlushSeqNum) {
2474       this.cacheFlushSeqNum = cacheFlushSeqNum;
2475     }
2476 
2477     /**
2478      * This is not thread safe. The caller should have a lock on the region or the store.
2479      * If necessary, the lock can be added with the patch provided in HBASE-10087
2480      */
2481     @Override
2482     public void prepare() {
2483       this.snapshot = memstore.snapshot();
2484       this.cacheFlushCount = snapshot.getCellsCount();
2485       this.cacheFlushSize = snapshot.getSize();
2486       committedFiles = new ArrayList<Path>(1);
2487     }
2488 
2489     @Override
2490     public void flushCache(MonitoredTask status) throws IOException {
2491       RegionServerServices rsService = region.getRegionServerServices();
2492       ThroughputController throughputController =
2493           rsService == null ? null : rsService.getFlushThroughputController();
2494       tempFiles = HStore.this.flushCache(cacheFlushSeqNum, snapshot, status, throughputController);
2495     }
2496 
2497     @Override
2498     public boolean commit(MonitoredTask status) throws IOException {
2499       if (this.tempFiles == null || this.tempFiles.isEmpty()) {
2500         return false;
2501       }
2502       List<StoreFile> storeFiles = new ArrayList<StoreFile>(this.tempFiles.size());
2503       for (Path storeFilePath : tempFiles) {
2504         try {
2505           StoreFile sf = HStore.this.commitFile(storeFilePath, cacheFlushSeqNum, status);
2506           outputFileSize += sf.getReader().length();
2507           storeFiles.add(sf);
2508         } catch (IOException ex) {
2509           LOG.error("Failed to commit store file " + storeFilePath, ex);
2510           // Try to delete the files we have committed before.
2511           for (StoreFile sf : storeFiles) {
2512             Path pathToDelete = sf.getPath();
2513             try {
2514               sf.deleteReader();
2515             } catch (IOException deleteEx) {
2516               LOG.fatal("Failed to delete store file we committed, halting " + pathToDelete, ex);
2517               Runtime.getRuntime().halt(1);
2518             }
2519           }
2520           throw new IOException("Failed to commit the flush", ex);
2521         }
2522       }
2523 
2524       for (StoreFile sf : storeFiles) {
2525         if (HStore.this.getCoprocessorHost() != null) {
2526           HStore.this.getCoprocessorHost().postFlush(HStore.this, sf);
2527         }
2528         committedFiles.add(sf.getPath());
2529       }
2530 
2531       HStore.this.flushedCellsCount += cacheFlushCount;
2532       HStore.this.flushedCellsSize += cacheFlushSize;
2533       HStore.this.flushedOutputFileSize += outputFileSize;
2534 
2535       // Add new file to store files.  Clear snapshot too while we have the Store write lock.
2536       return HStore.this.updateStorefiles(storeFiles, snapshot.getId());
2537     }
2538 
2539     @Override
2540     public long getOutputFileSize() {
2541       return outputFileSize;
2542     }
2543 
2544     @Override
2545     public List<Path> getCommittedFiles() {
2546       return committedFiles;
2547     }
2548 
2549     /**
2550      * Similar to commit, but called in secondary region replicas for replaying the
2551      * flush cache from primary region. Adds the new files to the store, and drops the
2552      * snapshot depending on dropMemstoreSnapshot argument.
2553      * @param fileNames names of the flushed files
2554      * @param dropMemstoreSnapshot whether to drop the prepared memstore snapshot
2555      * @throws IOException
2556      */
2557     @Override
2558     public void replayFlush(List<String> fileNames, boolean dropMemstoreSnapshot)
2559         throws IOException {
2560       List<StoreFile> storeFiles = new ArrayList<StoreFile>(fileNames.size());
2561       for (String file : fileNames) {
2562         // open the file as a store file (hfile link, etc)
2563         StoreFileInfo storeFileInfo = fs.getStoreFileInfo(getColumnFamilyName(), file);
2564         StoreFile storeFile = createStoreFileAndReader(storeFileInfo);
2565         storeFiles.add(storeFile);
2566         HStore.this.storeSize += storeFile.getReader().length();
2567         HStore.this.totalUncompressedBytes += storeFile.getReader().getTotalUncompressedBytes();
2568         if (LOG.isInfoEnabled()) {
2569           LOG.info("Region: " + HStore.this.getRegionInfo().getEncodedName() +
2570             " added " + storeFile + ", entries=" + storeFile.getReader().getEntries() +
2571             ", sequenceid=" + storeFile.getReader().getSequenceID() +
2572             ", filesize=" + StringUtils.humanReadableInt(storeFile.getReader().length()));
2573         }
2574       }
2575 
2576       long snapshotId = -1; // -1 means do not drop
2577       if (dropMemstoreSnapshot && snapshot != null) {
2578         snapshotId = snapshot.getId();
2579       }
2580       HStore.this.updateStorefiles(storeFiles, snapshotId);
2581     }
2582 
2583     /**
2584      * Abort the snapshot preparation. Drops the snapshot if any.
2585      * @throws IOException
2586      */
2587     @Override
2588     public void abort() throws IOException {
2589       if (snapshot == null) {
2590         return;
2591       }
2592       HStore.this.updateStorefiles(new ArrayList<StoreFile>(0), snapshot.getId());
2593     }
2594   }
2595 
2596   @Override
2597   public boolean needsCompaction() {
2598     List<StoreFile> filesCompactingClone = null;
2599     synchronized (filesCompacting) {
2600       filesCompactingClone = Lists.newArrayList(filesCompacting);
2601     }
2602     return this.storeEngine.needsCompaction(filesCompactingClone);
2603   }
2604 
2605   @Override
2606   public CacheConfig getCacheConfig() {
2607     return this.cacheConf;
2608   }
2609 
2610   public static final long FIXED_OVERHEAD =
2611       ClassSize.align(ClassSize.OBJECT + (17 * ClassSize.REFERENCE) + (11 * Bytes.SIZEOF_LONG)
2612               + (5 * Bytes.SIZEOF_INT) + (2 * Bytes.SIZEOF_BOOLEAN));
2613 
2614   public static final long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD
2615       + ClassSize.OBJECT + ClassSize.REENTRANT_LOCK
2616       + ClassSize.CONCURRENT_SKIPLISTMAP
2617       + ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY + ClassSize.OBJECT
2618       + ScanInfo.FIXED_OVERHEAD);
2619 
2620   @Override
2621   public long heapSize() {
2622     return DEEP_OVERHEAD + this.memstore.heapSize();
2623   }
2624 
2625   @Override
2626   public KeyValue.KVComparator getComparator() {
2627     return comparator;
2628   }
2629 
2630   @Override
2631   public ScanInfo getScanInfo() {
2632     return scanInfo;
2633   }
2634 
2635   /**
2636    * Set scan info, used by test
2637    * @param scanInfo new scan info to use for test
2638    */
2639   void setScanInfo(ScanInfo scanInfo) {
2640     this.scanInfo = scanInfo;
2641   }
2642 
2643   @Override
2644   public boolean hasTooManyStoreFiles() {
2645     return getStorefilesCount() > this.blockingFileCount;
2646   }
2647 
2648   @Override
2649   public long getFlushedCellsCount() {
2650     return flushedCellsCount;
2651   }
2652 
2653   @Override
2654   public long getFlushedCellsSize() {
2655     return flushedCellsSize;
2656   }
2657 
2658   @Override
2659   public long getFlushedOutputFileSize() {
2660     return flushedOutputFileSize;
2661   }
2662 
2663   @Override
2664   public long getCompactedCellsCount() {
2665     return compactedCellsCount;
2666   }
2667 
2668   @Override
2669   public long getCompactedCellsSize() {
2670     return compactedCellsSize;
2671   }
2672 
2673   @Override
2674   public long getMajorCompactedCellsCount() {
2675     return majorCompactedCellsCount;
2676   }
2677 
2678   @Override
2679   public long getMajorCompactedCellsSize() {
2680     return majorCompactedCellsSize;
2681   }
2682 
2683   /**
2684    * Returns the StoreEngine that is backing this concrete implementation of Store.
2685    * @return Returns the {@link StoreEngine} object used internally inside this HStore object.
2686    */
2687   @VisibleForTesting
2688   public StoreEngine<?, ?, ?, ?> getStoreEngine() {
2689     return this.storeEngine;
2690   }
2691 
2692   protected OffPeakHours getOffPeakHours() {
2693     return this.offPeakHours;
2694   }
2695 
2696   /**
2697    * {@inheritDoc}
2698    */
2699   @Override
2700   public void onConfigurationChange(Configuration conf) {
2701     this.conf = new CompoundConfiguration()
2702             .add(conf)
2703             .addWritableMap(family.getValues());
2704     this.storeEngine.compactionPolicy.setConf(conf);
2705     this.offPeakHours = OffPeakHours.getInstance(conf);
2706   }
2707 
2708   /**
2709    * {@inheritDoc}
2710    */
2711   @Override
2712   public void registerChildren(ConfigurationManager manager) {
2713     // No children to register
2714   }
2715 
2716   /**
2717    * {@inheritDoc}
2718    */
2719   @Override
2720   public void deregisterChildren(ConfigurationManager manager) {
2721     // No children to deregister
2722   }
2723 
2724   @Override
2725   public double getCompactionPressure() {
2726     return storeEngine.getStoreFileManager().getCompactionPressure();
2727   }
2728 
2729   @Override
2730   public boolean isPrimaryReplicaStore() {
2731 	   return getRegionInfo().getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID;
2732   }
2733 
2734   /**
2735    * Sets the store up for a region level snapshot operation.
2736    * @see #postSnapshotOperation()
2737    */
2738   public void preSnapshotOperation() {
2739     archiveLock.lock();
2740   }
2741 
2742   /**
2743    * Perform tasks needed after the completion of snapshot operation.
2744    * @see #preSnapshotOperation()
2745    */
2746   public void postSnapshotOperation() {
2747     archiveLock.unlock();
2748   }
2749 
2750   @Override
2751   public synchronized void closeAndArchiveCompactedFiles() throws IOException {
2752     // ensure other threads do not attempt to archive the same files on close()
2753     archiveLock.lock();
2754     try {
2755       lock.readLock().lock();
2756       Collection<StoreFile> copyCompactedfiles = null;
2757       try {
2758         Collection<StoreFile> compactedfiles = getCompactedfiles();
2759         if (compactedfiles != null && compactedfiles.size() != 0) {
2760           // Do a copy under read lock
2761           copyCompactedfiles = new ArrayList<StoreFile>(compactedfiles);
2762         } else {
2763           if (LOG.isTraceEnabled()) {
2764             LOG.trace("No compacted files to archive");
2765             return;
2766           }
2767         }
2768       } finally {
2769         lock.readLock().unlock();
2770       }
2771       if (copyCompactedfiles != null && !copyCompactedfiles.isEmpty()) {
2772         removeCompactedfiles(copyCompactedfiles);
2773       }
2774     } finally {
2775       archiveLock.unlock();
2776     }
2777   }
2778 
2779   /**
2780    * Archives and removes the compacted files
2781    * @param compactedfiles The compacted files in this store that are not active in reads
2782    * @throws IOException
2783    */
2784   private void removeCompactedfiles(Collection<StoreFile> compactedfiles)
2785       throws IOException {
2786     final List<StoreFile> filesToRemove = new ArrayList<StoreFile>(compactedfiles.size());
2787     for (final StoreFile file : compactedfiles) {
2788       synchronized (file) {
2789         try {
2790           StoreFile.Reader r = file.getReader();
2791 
2792           if (r == null) {
2793             if (LOG.isDebugEnabled()) {
2794               LOG.debug("The file " + file + " was closed but still not archived.");
2795             }
2796             filesToRemove.add(file);
2797           }
2798 
2799           if (r != null && r.isCompactedAway() && !r.isReferencedInReads()) {
2800             // Even if deleting fails we need not bother as any new scanners won't be
2801             // able to use the compacted file as the status is already compactedAway
2802             if (LOG.isTraceEnabled()) {
2803               LOG.trace("Closing and archiving the file " + file.getPath());
2804             }
2805             r.close(true);
2806             // Just close and return
2807             filesToRemove.add(file);
2808           } else {
2809             if (r != null) {
2810               LOG.info("Can't archive compacted file " + file.getPath()
2811                 + " because of either isCompactedAway=" + r.isCompactedAway()
2812                 + " or file has reference, isReferencedInReads=" + r.isReferencedInReads()
2813                 + ", refCount=" + r.getRefCount() + ", skipping for now.");
2814             } else {
2815               LOG.info("Can't archive compacted file " + file.getPath() + ", skipping for now.");
2816             }
2817           }
2818         } catch (Exception e) {
2819           LOG.error(
2820             "Exception while trying to close the compacted store file " + file.getPath().getName());
2821         }
2822       }
2823     }
2824     if (this.isPrimaryReplicaStore()) {
2825       // Only the primary region is allowed to move the file to archive.
2826       // The secondary region does not move the files to archive. Any active reads from
2827       // the secondary region will still work because the file as such has active readers on it.
2828       if (!filesToRemove.isEmpty()) {
2829         if (LOG.isDebugEnabled()) {
2830           LOG.debug("Moving the files " + filesToRemove + " to archive");
2831         }
2832         // Only if this is successful it has to be removed
2833         try {
2834           this.fs.removeStoreFiles(this.getFamily().getNameAsString(), filesToRemove);
2835         } catch (FailedArchiveException fae) {
2836           // Even if archiving some files failed, we still need to clear out any of the
2837           // files which were successfully archived.  Otherwise we will receive a
2838           // FileNotFoundException when we attempt to re-archive them in the next go around.
2839           Collection<Path> failedFiles = fae.getFailedFiles();
2840           Iterator<StoreFile> iter = filesToRemove.iterator();
2841           while (iter.hasNext()) {
2842             if (failedFiles.contains(iter.next().getPath())) {
2843               iter.remove();
2844             }
2845           }
2846           if (!filesToRemove.isEmpty()) {
2847             clearCompactedfiles(filesToRemove);
2848           }
2849           throw fae;
2850         }
2851       }
2852     }
2853     if (!filesToRemove.isEmpty()) {
2854       // Clear the compactedfiles from the store file manager
2855       clearCompactedfiles(filesToRemove);
2856     }
2857   }
2858 
2859   private void clearCompactedfiles(final List<StoreFile> filesToRemove) throws IOException {
2860     if (LOG.isTraceEnabled()) {
2861       LOG.trace("Clearing the compacted file " + filesToRemove + " from this store");
2862     }
2863     try {
2864       lock.writeLock().lock();
2865       this.getStoreEngine().getStoreFileManager().removeCompactedFiles(filesToRemove);
2866     } finally {
2867       lock.writeLock().unlock();
2868     }
2869   }
2870 }