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