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