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