1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
107
108
109
110
111
112
113
114
115
116
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
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
139 static int closeCheckInterval = 0;
140 private volatile long storeSize = 0L;
141 private volatile long totalUncompressedBytes = 0L;
142
143
144
145
146
147
148
149
150
151
152 final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
153
154
155
156
157
158
159
160
161
162
163
164 final ReentrantLock archiveLock = new ReentrantLock();
165
166 private final boolean verifyBulkLoads;
167
168 private ScanInfo scanInfo;
169
170
171
172 private final List<StoreFile> filesCompacting = Lists.newArrayList();
173
174
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
182 private ChecksumType checksumType;
183 private int bytesPerChecksum;
184
185
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
212
213
214
215
216
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
225 fs.createStoreDir(family.getNameAsString());
226 this.region = region;
227 this.family = family;
228
229
230
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
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
248 long ttl = determineTTLFromFamily(family);
249
250
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
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
275 }
276
277 this.storeEngine = StoreEngine.create(this, this.conf, this.comparator);
278 List<StoreFile> storeFiles = loadStoreFiles();
279
280
281
282
283 this.storeSize = getStorefilesSize(storeFiles);
284 this.totalUncompressedBytes = getTotalUmcompressedBytes(storeFiles);
285 this.storeEngine.getStoreFileManager().loadFiles(storeFiles);
286
287
288 this.checksumType = getChecksumType(conf);
289
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
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
308 String masterKeyName = conf.get(HConstants.CRYPTO_MASTERKEY_NAME_CONF_KEY,
309 User.getCurrent().getShortName());
310 try {
311
312 key = EncryptionUtil.unwrapKey(conf, masterKeyName, keyBytes);
313 } catch (KeyException e) {
314
315
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
332 cipher = Encryption.getCipher(conf, key.getAlgorithm());
333 if (cipher == null) {
334 throw new RuntimeException("Cipher '" + key.getAlgorithm() + "' is not available");
335 }
336
337
338
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
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
366
367
368 private static long determineTTLFromFamily(final HColumnDescriptor family) {
369
370 long ttl = family.getTimeToLive();
371 if (ttl == HConstants.FOREVER) {
372
373 ttl = Long.MAX_VALUE;
374 } else if (ttl == -1) {
375 ttl = Long.MAX_VALUE;
376 } else {
377
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
403 @Override
404 public long getStoreFileTtl() {
405
406 return (this.scanInfo.getMinVersions() == 0) ? this.scanInfo.getTtl() : Long.MAX_VALUE;
407 }
408
409 @Override
410 public long getMemstoreFlushSize() {
411
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
435
436
437
438
439
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
448
449
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
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
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
487
488
489
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
499
500
501
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
516
517
518 void setDataBlockEncoderInTest(HFileDataBlockEncoder blockEncoder) {
519 this.dataBlockEncoder = blockEncoder;
520 }
521
522
523
524
525
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
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
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
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
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
614
615
616
617
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
636
637
638
639
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
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
679 List<StoreFile> openedFiles = openStoreFiles(toBeAddedFiles);
680
681
682 replaceStoreFiles(toBeRemovedStoreFiles, openedFiles);
683
684
685
686
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
726
727
728
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
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
863 this.lock.writeLock().lock();
864 try {
865 this.storeEngine.getStoreFileManager().insertNewFiles(Lists.newArrayList(sf));
866 } finally {
867
868
869
870
871
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
889 ImmutableCollection<StoreFile> result = storeEngine.getStoreFileManager().clearFiles();
890 Collection<StoreFile> compactedfiles =
891 storeEngine.getStoreFileManager().clearCompactedFiles();
892
893 if (compactedfiles != null && !compactedfiles.isEmpty()) {
894 removeCompactedfiles(compactedfiles);
895 }
896 if (!result.isEmpty()) {
897
898 ThreadPoolExecutor storeFileCloserThreadPool = this.region
899 .getStoreFileOpenAndCloseThreadPool("StoreFileCloserThread-"
900 + this.getColumnFamilyName());
901
902
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
947
948
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
961
962
963
964
965
966
967
968 protected List<Path> flushCache(final long logCacheFlushId, MemStoreSnapshot snapshot,
969 MonitoredTask status, ThroughputController throughputController) throws IOException {
970
971
972
973
974
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
1015
1016
1017
1018
1019
1020 private StoreFile commitFile(final Path path, final long logCacheFlushId, MonitoredTask status)
1021 throws IOException {
1022
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
1051
1052
1053
1054
1055
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
1068
1069
1070
1071
1072
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
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
1134
1135
1136
1137
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
1149
1150
1151
1152
1153 this.lock.writeLock().unlock();
1154 }
1155
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
1172
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
1189
1190
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
1208
1209
1210
1211
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
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
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
1256 this.changedReaderObservers.remove(o);
1257 }
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
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
1320
1321
1322 long compactionStartTime = EnvironmentEdgeManager.currentTime();
1323 assert compaction.hasSelection();
1324 Collection<StoreFile> filesToCompact = cr.getFiles();
1325 assert !filesToCompact.isEmpty();
1326 synchronized (filesCompacting) {
1327
1328
1329 Preconditions.checkArgument(filesCompacting.containsAll(filesToCompact));
1330 }
1331
1332
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
1339 List<Path> newFiles = compaction.compact(throughputController, user);
1340
1341 long outputBytes = 0L;
1342
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
1350 StoreFile sf = createStoreFileAndReader(newFile);
1351 sf.closeReader(evictOnClose);
1352 sfs.add(sf);
1353 }
1354 return sfs;
1355 }
1356
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
1373 completeCompaction(filesToCompact);
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
1423 StoreFile moveFileIntoPlace(final Path newFile) throws IOException {
1424 validateStoreFile(newFile);
1425
1426 Path destPath = fs.commitStoreFile(getColumnFamilyName(), newFile);
1427 return createStoreFileAndReader(destPath);
1428 }
1429
1430
1431
1432
1433
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
1469
1470
1471
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
1511
1512
1513
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
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
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
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
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
1576
1577
1578
1579
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
1591
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
1614 List<Path> newFiles = ((DefaultCompactor)this.storeEngine.getCompactor())
1615 .compactForTesting(filesToCompact, isMajor);
1616 for (Path newFile: newFiles) {
1617
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
1635
1636 this.lock.readLock().lock();
1637 try {
1638
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
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
1679 if (!this.areWritesEnabled()) {
1680 return null;
1681 }
1682
1683
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
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
1716 compaction.forceSelect(new CompactionRequest(candidatesForCoproc));
1717 }
1718 }
1719
1720
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
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
1763 if (baseRequest != null) {
1764
1765
1766 compaction.forceSelect(
1767 baseRequest.combineWith(compaction.getRequest()));
1768 }
1769
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
1779 this.forceMajor = this.forceMajor && !request.isMajor();
1780
1781
1782
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
1798 private void addToCompactingFiles(final Collection<StoreFile> filesToAdd) {
1799 if (filesToAdd == null) return;
1800
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>();
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
1858
1859
1860
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
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892 @VisibleForTesting
1893 protected void completeCompaction(final Collection<StoreFile> compactedFiles)
1894 throws IOException {
1895 completeCompaction(compactedFiles, true);
1896 }
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
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
1932
1933
1934 int versionsToReturn(final int wantedVersions) {
1935 if (wantedVersions <= 0) {
1936 throw new IllegalArgumentException("Number of versions must be > 0");
1937 }
1938
1939 int maxVersions = this.family.getMaxVersions();
1940 return wantedVersions > maxVersions ? maxVersions: wantedVersions;
1941 }
1942
1943
1944
1945
1946
1947
1948 static boolean isCellTTLExpired(final Cell cell, final long oldestTimestamp, final long now) {
1949
1950
1951 if (cell.getTagsLength() > 0) {
1952
1953
1954
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
1961
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
1969
1970 break;
1971 }
1972 }
1973 }
1974 return false;
1975 }
1976
1977 @Override
1978 public Cell getRowKeyAtOrBefore(final byte[] row) throws IOException {
1979
1980
1981
1982
1983
1984
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
1994 this.memstore.getRowKeyAtOrBefore(state);
1995
1996
1997 Iterator<StoreFile> sfIterator = this.storeEngine.getStoreFileManager()
1998 .getCandidateFilesForRowKeyBefore(state.getTargetKey());
1999 while (sfIterator.hasNext()) {
2000 StoreFile sf = sfIterator.next();
2001 sfIterator.remove();
2002 boolean haveNewCandidate = rowAtOrBeforeFromStoreFile(sf, state);
2003 Cell candidate = state.getCandidate();
2004
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
2021
2022
2023
2024
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
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
2047
2048 if (!state.isTargetTable(lastKV)) return false;
2049
2050
2051 firstOnRow = new KeyValue(lastKV.getRow(), HConstants.LATEST_TIMESTAMP);
2052 }
2053
2054 HFileScanner scanner = r.getScanner(true, true, false);
2055
2056 if (!seekToScanner(scanner, firstOnRow, firstKV)) return false;
2057
2058
2059 if (walkForwardInSingleRow(scanner, firstOnRow, state)) return true;
2060
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
2067 firstOnRow = new KeyValue(kv.getRow(), HConstants.LATEST_TIMESTAMP);
2068
2069 if (!seekToScanner(scanner, firstOnRow, firstKV)) return false;
2070
2071 if (walkForwardInSingleRow(scanner, firstOnRow, state)) return true;
2072 }
2073 return false;
2074 }
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084 private boolean seekToScanner(final HFileScanner scanner,
2085 final KeyValue firstOnRow,
2086 final KeyValue firstKV)
2087 throws IOException {
2088 KeyValue kv = firstOnRow;
2089
2090 if (this.comparator.compareRows(firstKV, firstOnRow) == 0) kv = firstKV;
2091 int result = scanner.seekTo(kv);
2092 return result != -1;
2093 }
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
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
2113 if (this.comparator.compareRows(kv, firstOnRow) < 0) continue;
2114
2115 if (state.isTooFar(kv, firstOnRow)) break;
2116 if (state.isExpired(kv)) {
2117 continue;
2118 }
2119
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
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
2148 assert !this.getRegionInfo().isMetaRegion();
2149
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
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
2417
2418
2419
2420
2421
2422
2423
2424
2425
2426
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
2479
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
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
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
2551
2552
2553
2554
2555
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
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;
2577 if (dropMemstoreSnapshot && snapshot != null) {
2578 snapshotId = snapshot.getId();
2579 }
2580 HStore.this.updateStorefiles(storeFiles, snapshotId);
2581 }
2582
2583
2584
2585
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
2637
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
2685
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
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
2710
2711 @Override
2712 public void registerChildren(ConfigurationManager manager) {
2713
2714 }
2715
2716
2717
2718
2719 @Override
2720 public void deregisterChildren(ConfigurationManager manager) {
2721
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
2736
2737
2738 public void preSnapshotOperation() {
2739 archiveLock.lock();
2740 }
2741
2742
2743
2744
2745
2746 public void postSnapshotOperation() {
2747 archiveLock.unlock();
2748 }
2749
2750 @Override
2751 public synchronized void closeAndArchiveCompactedFiles() throws IOException {
2752
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
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
2781
2782
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
2801
2802 if (LOG.isTraceEnabled()) {
2803 LOG.trace("Closing and archiving the file " + file.getPath());
2804 }
2805 r.close(true);
2806
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
2826
2827
2828 if (!filesToRemove.isEmpty()) {
2829 if (LOG.isDebugEnabled()) {
2830 LOG.debug("Moving the files " + filesToRemove + " to archive");
2831 }
2832
2833 try {
2834 this.fs.removeStoreFiles(this.getFamily().getNameAsString(), filesToRemove);
2835 } catch (FailedArchiveException fae) {
2836
2837
2838
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
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 }