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