1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.io.hfile;
19
20 import java.io.DataInput;
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23 import java.util.ArrayList;
24 import java.util.List;
25
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.apache.hadoop.conf.Configuration;
29 import org.apache.hadoop.fs.Path;
30 import org.apache.hadoop.hbase.Cell;
31 import org.apache.hadoop.hbase.CellUtil;
32 import org.apache.hadoop.hbase.HConstants;
33 import org.apache.hadoop.hbase.KeyValue;
34 import org.apache.hadoop.hbase.KeyValue.KVComparator;
35 import org.apache.hadoop.hbase.NoTagsKeyValue;
36 import org.apache.hadoop.hbase.classification.InterfaceAudience;
37 import org.apache.hadoop.hbase.fs.HFileSystem;
38 import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper;
39 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoder;
40 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
41 import org.apache.hadoop.hbase.io.encoding.HFileBlockDecodingContext;
42 import org.apache.hadoop.hbase.io.hfile.HFile.FileInfo;
43 import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
44 import org.apache.hadoop.hbase.util.Bytes;
45 import org.apache.hadoop.hbase.util.IdLock;
46 import org.apache.hadoop.io.WritableUtils;
47 import org.apache.htrace.Trace;
48 import org.apache.htrace.TraceScope;
49
50 import com.google.common.annotations.VisibleForTesting;
51
52
53
54
55 @InterfaceAudience.Private
56 public class HFileReaderV2 extends AbstractHFileReader {
57
58 private static final Log LOG = LogFactory.getLog(HFileReaderV2.class);
59
60
61 public static final int MINOR_VERSION_WITH_CHECKSUM = 1;
62
63 public static final int MINOR_VERSION_NO_CHECKSUM = 0;
64
65
66 public static final int PBUF_TRAILER_MINOR_VERSION = 2;
67
68
69
70
71
72 public final static int KEY_VALUE_LEN_SIZE = 2 * Bytes.SIZEOF_INT;
73
74 private boolean includesMemstoreTS = false;
75 protected boolean decodeMemstoreTS = false;
76
77 protected boolean shouldIncludeMemstoreTS() {
78 return includesMemstoreTS;
79 }
80
81
82 private HFileBlock.FSReader fsBlockReader;
83
84
85
86
87
88
89
90 private IdLock offsetLock = new IdLock();
91
92
93
94
95
96 private List<HFileBlock> loadOnOpenBlocks = new ArrayList<HFileBlock>();
97
98
99 static final int MIN_MINOR_VERSION = 0;
100
101
102
103
104 static final int MAX_MINOR_VERSION = 3;
105
106
107 static final int MINOR_VERSION_WITH_FAKED_KEY = 3;
108
109 HFileContext hfileContext;
110
111
112
113
114
115
116
117
118
119
120
121
122
123 public HFileReaderV2(final Path path, final FixedFileTrailer trailer,
124 final FSDataInputStreamWrapper fsdis, final long size, final CacheConfig cacheConf,
125 final HFileSystem hfs, final Configuration conf) throws IOException {
126 super(path, trailer, size, cacheConf, hfs, conf);
127 this.conf = conf;
128 trailer.expectMajorVersion(getMajorVersion());
129 validateMinorVersion(path, trailer.getMinorVersion());
130 this.hfileContext = createHFileContext(fsdis, fileSize, hfs, path, trailer);
131 HFileBlock.FSReaderImpl fsBlockReaderV2 =
132 new HFileBlock.FSReaderImpl(fsdis, fileSize, hfs, path, hfileContext);
133 this.fsBlockReader = fsBlockReaderV2;
134
135
136 comparator = trailer.createComparator();
137 dataBlockIndexReader = new HFileBlockIndex.BlockIndexReader(comparator,
138 trailer.getNumDataIndexLevels(), this);
139 metaBlockIndexReader = new HFileBlockIndex.BlockIndexReader(
140 KeyValue.RAW_COMPARATOR, 1);
141
142
143
144 HFileBlock.BlockIterator blockIter = fsBlockReaderV2.blockRange(
145 trailer.getLoadOnOpenDataOffset(),
146 fileSize - trailer.getTrailerSize());
147
148
149
150 dataBlockIndexReader.readMultiLevelIndexRoot(
151 blockIter.nextBlockWithBlockType(BlockType.ROOT_INDEX),
152 trailer.getDataIndexCount());
153
154
155 metaBlockIndexReader.readRootIndex(
156 blockIter.nextBlockWithBlockType(BlockType.ROOT_INDEX),
157 trailer.getMetaIndexCount());
158
159
160 fileInfo = new FileInfo();
161 fileInfo.read(blockIter.nextBlockWithBlockType(BlockType.FILE_INFO).getByteStream());
162 byte[] creationTimeBytes = fileInfo.get(FileInfo.CREATE_TIME_TS);
163 this.hfileContext.setFileCreateTime(creationTimeBytes == null? 0:
164 Bytes.toLong(creationTimeBytes));
165 lastKey = fileInfo.get(FileInfo.LASTKEY);
166 avgKeyLen = Bytes.toInt(fileInfo.get(FileInfo.AVG_KEY_LEN));
167 avgValueLen = Bytes.toInt(fileInfo.get(FileInfo.AVG_VALUE_LEN));
168 byte [] keyValueFormatVersion =
169 fileInfo.get(HFileWriterV2.KEY_VALUE_VERSION);
170 includesMemstoreTS = keyValueFormatVersion != null &&
171 Bytes.toInt(keyValueFormatVersion) ==
172 HFileWriterV2.KEY_VALUE_VER_WITH_MEMSTORE;
173 fsBlockReaderV2.setIncludesMemstoreTS(includesMemstoreTS);
174 if (includesMemstoreTS) {
175 decodeMemstoreTS = Bytes.toLong(fileInfo.get(HFileWriterV2.MAX_MEMSTORE_TS_KEY)) > 0;
176 }
177
178
179 dataBlockEncoder = HFileDataBlockEncoderImpl.createFromFileInfo(fileInfo);
180 fsBlockReaderV2.setDataBlockEncoder(dataBlockEncoder);
181
182
183 HFileBlock b;
184 while ((b = blockIter.nextBlock()) != null) {
185 loadOnOpenBlocks.add(b);
186 }
187
188
189 if (cacheConf.shouldPrefetchOnOpen()) {
190 PrefetchExecutor.request(path, new Runnable() {
191 public void run() {
192 long offset = 0;
193 long end = 0;
194 try {
195 end = getTrailer().getLoadOnOpenDataOffset();
196 HFileBlock prevBlock = null;
197 if (LOG.isTraceEnabled()) {
198 LOG.trace("Prefetch start " + getPathOffsetEndStr(path, offset, end));
199 }
200 while (offset < end) {
201 if (Thread.interrupted()) {
202 break;
203 }
204
205
206
207
208 long onDiskSize = prevBlock != null? prevBlock.getNextBlockOnDiskSize(): -1;
209 HFileBlock block = readBlock(offset, onDiskSize, true, false, false, false,
210 null, null);
211 prevBlock = block;
212 offset += block.getOnDiskSizeWithHeader();
213 }
214 } catch (IOException e) {
215
216 if (LOG.isTraceEnabled()) {
217 LOG.trace("Prefetch " + getPathOffsetEndStr(path, offset, end), e);
218 }
219 } catch (NullPointerException e) {
220 LOG.warn("Stream moved/closed or prefetch cancelled?" +
221 getPathOffsetEndStr(path, offset, end), e);
222 } catch (Exception e) {
223
224 LOG.warn("Prefetch " + getPathOffsetEndStr(path, offset, end), e);
225 } finally {
226 PrefetchExecutor.complete(path);
227 }
228 }
229 });
230 }
231 }
232
233 protected HFileContext createHFileContext(FSDataInputStreamWrapper fsdis, long fileSize,
234 HFileSystem hfs, Path path, FixedFileTrailer trailer) throws IOException {
235 return new HFileContextBuilder()
236 .withIncludesMvcc(this.includesMemstoreTS)
237 .withCompression(this.compressAlgo)
238 .withHBaseCheckSum(trailer.getMinorVersion() >= MINOR_VERSION_WITH_CHECKSUM)
239 .build();
240 }
241
242 private static String getPathOffsetEndStr(final Path path, final long offset, final long end) {
243 return "path=" + path.toString() + ", offset=" + offset + ", end=" + end;
244 }
245
246
247
248
249
250
251
252
253
254
255
256
257
258 @Override
259 public HFileScanner getScanner(boolean cacheBlocks, final boolean pread,
260 final boolean isCompaction) {
261 if (dataBlockEncoder.useEncodedScanner()) {
262 return new EncodedScannerV2(this, cacheBlocks, pread, isCompaction,
263 hfileContext);
264 }
265
266 return new ScannerV2(this, cacheBlocks, pread, isCompaction);
267 }
268
269
270
271
272
273 private HFileBlock getCachedBlock(BlockCacheKey cacheKey, boolean cacheBlock, boolean useLock,
274 boolean isCompaction, boolean updateCacheMetrics, BlockType expectedBlockType,
275 DataBlockEncoding expectedDataBlockEncoding) throws IOException {
276
277 if (cacheConf.isBlockCacheEnabled()) {
278 BlockCache cache = cacheConf.getBlockCache();
279 HFileBlock cachedBlock = (HFileBlock) cache.getBlock(cacheKey, cacheBlock, useLock,
280 updateCacheMetrics);
281 if (cachedBlock != null) {
282 if (cacheConf.shouldCacheCompressed(cachedBlock.getBlockType().getCategory())) {
283 cachedBlock = cachedBlock.unpack(hfileContext, fsBlockReader);
284 }
285 validateBlockType(cachedBlock, expectedBlockType);
286
287 if (expectedDataBlockEncoding == null) {
288 return cachedBlock;
289 }
290 DataBlockEncoding actualDataBlockEncoding =
291 cachedBlock.getDataBlockEncoding();
292
293
294
295 if (cachedBlock.getBlockType().isData() &&
296 !actualDataBlockEncoding.equals(expectedDataBlockEncoding)) {
297
298
299
300
301
302
303
304
305
306
307 if (!expectedDataBlockEncoding.equals(DataBlockEncoding.NONE) &&
308 !actualDataBlockEncoding.equals(DataBlockEncoding.NONE)) {
309
310
311
312
313
314
315 LOG.info("Evicting cached block with key " + cacheKey +
316 " because of a data block encoding mismatch" +
317 "; expected: " + expectedDataBlockEncoding +
318 ", actual: " + actualDataBlockEncoding);
319 cache.evictBlock(cacheKey);
320 }
321 return null;
322 }
323 return cachedBlock;
324 }
325 }
326 return null;
327 }
328
329
330
331
332
333
334 @Override
335 public ByteBuffer getMetaBlock(String metaBlockName, boolean cacheBlock)
336 throws IOException {
337 if (trailer.getMetaIndexCount() == 0) {
338 return null;
339 }
340 if (metaBlockIndexReader == null) {
341 throw new IOException("Meta index not loaded");
342 }
343
344 byte[] mbname = Bytes.toBytes(metaBlockName);
345 int block = metaBlockIndexReader.rootBlockContainingKey(mbname,
346 0, mbname.length);
347 if (block == -1)
348 return null;
349 long blockSize = metaBlockIndexReader.getRootBlockDataSize(block);
350
351
352
353
354 synchronized (metaBlockIndexReader.getRootBlockKey(block)) {
355
356 long metaBlockOffset = metaBlockIndexReader.getRootBlockOffset(block);
357 BlockCacheKey cacheKey = new BlockCacheKey(name, metaBlockOffset,
358 this.isPrimaryReplicaReader(), BlockType.META);
359
360 cacheBlock &= cacheConf.shouldCacheDataOnRead();
361 if (cacheConf.isBlockCacheEnabled()) {
362 HFileBlock cachedBlock = getCachedBlock(cacheKey, cacheBlock, false, true, true,
363 BlockType.META, null);
364 if (cachedBlock != null) {
365 assert cachedBlock.isUnpacked() : "Packed block leak.";
366
367
368 return cachedBlock.getBufferWithoutHeader();
369 }
370
371 }
372
373 HFileBlock metaBlock = fsBlockReader.readBlockData(metaBlockOffset,
374 blockSize, true, false).unpack(hfileContext, fsBlockReader);
375
376
377 if (cacheBlock) {
378 cacheConf.getBlockCache().cacheBlock(cacheKey, metaBlock,
379 cacheConf.isInMemory(), this.cacheConf.isCacheDataInL1());
380 }
381
382 return metaBlock.getBufferWithoutHeader();
383 }
384 }
385
386 @Override
387 public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize,
388 final boolean cacheBlock, boolean pread, final boolean isCompaction,
389 boolean updateCacheMetrics, BlockType expectedBlockType,
390 DataBlockEncoding expectedDataBlockEncoding)
391 throws IOException {
392 if (dataBlockIndexReader == null) {
393 throw new IOException("Block index not loaded");
394 }
395 long trailerOffset = trailer.getLoadOnOpenDataOffset();
396 if (dataBlockOffset < 0 || dataBlockOffset >= trailerOffset) {
397 throw new IOException("Requested block is out of range: " + dataBlockOffset +
398 ", lastDataBlockOffset: " + trailer.getLastDataBlockOffset() +
399 ", trailer.getLoadOnOpenDataOffset: " + trailerOffset);
400 }
401
402
403
404
405
406 BlockCacheKey cacheKey =
407 new BlockCacheKey(name, dataBlockOffset, this.isPrimaryReplicaReader(), expectedBlockType);
408 boolean useLock = false;
409 IdLock.Entry lockEntry = null;
410 TraceScope traceScope = Trace.startSpan("HFileReaderV2.readBlock");
411 try {
412 while (true) {
413
414 if (cacheConf.shouldReadBlockFromCache(expectedBlockType)) {
415 if (useLock) {
416 lockEntry = offsetLock.getLockEntry(dataBlockOffset);
417 }
418
419
420 HFileBlock cachedBlock = getCachedBlock(cacheKey, cacheBlock, useLock, isCompaction,
421 updateCacheMetrics, expectedBlockType, expectedDataBlockEncoding);
422 if (cachedBlock != null) {
423 if (Trace.isTracing()) {
424 traceScope.getSpan().addTimelineAnnotation("blockCacheHit");
425 }
426 assert cachedBlock.isUnpacked() : "Packed block leak.";
427 if (cachedBlock.getBlockType().isData()) {
428 if (updateCacheMetrics) {
429 HFile.DATABLOCK_READ_COUNT.increment();
430 }
431
432
433 if (cachedBlock.getDataBlockEncoding() != dataBlockEncoder.getDataBlockEncoding()) {
434 throw new IOException("Cached block under key " + cacheKey + " "
435 + "has wrong encoding: " + cachedBlock.getDataBlockEncoding() + " (expected: "
436 + dataBlockEncoder.getDataBlockEncoding() + ")");
437 }
438 }
439
440 return cachedBlock;
441 }
442 if (!useLock && cacheBlock && cacheConf.shouldLockOnCacheMiss(expectedBlockType)) {
443
444 useLock = true;
445 continue;
446 }
447
448 }
449
450 if (Trace.isTracing()) {
451 traceScope.getSpan().addTimelineAnnotation("blockCacheMiss");
452 }
453
454 HFileBlock hfileBlock = fsBlockReader.readBlockData(dataBlockOffset, onDiskBlockSize,
455 pread, !isCompaction);
456 validateBlockType(hfileBlock, expectedBlockType);
457 HFileBlock unpacked = hfileBlock.unpack(hfileContext, fsBlockReader);
458 BlockType.BlockCategory category = hfileBlock.getBlockType().getCategory();
459
460
461 if (cacheBlock && cacheConf.shouldCacheBlockOnRead(category)) {
462 cacheConf.getBlockCache().cacheBlock(cacheKey,
463 cacheConf.shouldCacheCompressed(category) ? hfileBlock : unpacked,
464 cacheConf.isInMemory(), this.cacheConf.isCacheDataInL1());
465 }
466
467 if (updateCacheMetrics && hfileBlock.getBlockType().isData()) {
468 HFile.DATABLOCK_READ_COUNT.increment();
469 }
470
471 return unpacked;
472 }
473 } finally {
474 traceScope.close();
475 if (lockEntry != null) {
476 offsetLock.releaseLockEntry(lockEntry);
477 }
478 }
479 }
480
481 @Override
482 public boolean hasMVCCInfo() {
483 return includesMemstoreTS && decodeMemstoreTS;
484 }
485
486
487
488
489
490
491
492
493
494
495 private void validateBlockType(HFileBlock block,
496 BlockType expectedBlockType) throws IOException {
497 if (expectedBlockType == null) {
498 return;
499 }
500 BlockType actualBlockType = block.getBlockType();
501 if (expectedBlockType.isData() && actualBlockType.isData()) {
502
503
504 return;
505 }
506 if (actualBlockType != expectedBlockType) {
507 throw new IOException("Expected block type " + expectedBlockType + ", " +
508 "but got " + actualBlockType + ": " + block);
509 }
510 }
511
512
513
514
515
516
517 @Override
518 public byte[] getLastKey() {
519 return dataBlockIndexReader.isEmpty() ? null : lastKey;
520 }
521
522
523
524
525
526
527 @Override
528 public byte[] midkey() throws IOException {
529 return dataBlockIndexReader.midkey();
530 }
531
532 @Override
533 public void close() throws IOException {
534 close(cacheConf.shouldEvictOnClose());
535 }
536
537 public void close(boolean evictOnClose) throws IOException {
538 PrefetchExecutor.cancel(path);
539 if (evictOnClose && cacheConf.isBlockCacheEnabled()) {
540 int numEvicted = cacheConf.getBlockCache().evictBlocksByHfileName(name);
541 if (LOG.isTraceEnabled()) {
542 LOG.trace("On close, file=" + name + " evicted=" + numEvicted
543 + " block(s)");
544 }
545 }
546 fsBlockReader.closeStreams();
547 }
548
549 public DataBlockEncoding getEffectiveEncodingInCache(boolean isCompaction) {
550 return dataBlockEncoder.getEffectiveEncodingInCache(isCompaction);
551 }
552
553
554 @Override
555 HFileBlock.FSReader getUncachedBlockReader() {
556 return fsBlockReader;
557 }
558
559
560 protected abstract static class AbstractScannerV2
561 extends AbstractHFileReader.Scanner {
562 protected HFileBlock block;
563
564 @Override
565 public Cell getNextIndexedKey() {
566 return nextIndexedKey;
567 }
568
569
570
571
572
573
574
575 protected Cell nextIndexedKey;
576
577 public AbstractScannerV2(HFileReaderV2 r, boolean cacheBlocks,
578 final boolean pread, final boolean isCompaction) {
579 super(r, cacheBlocks, pread, isCompaction);
580 }
581
582 protected abstract ByteBuffer getFirstKeyInBlock(HFileBlock curBlock);
583
584 protected abstract int loadBlockAndSeekToKey(HFileBlock seekToBlock, Cell nextIndexedKey,
585 boolean rewind, Cell key, boolean seekBefore) throws IOException;
586
587 @Override
588 public int seekTo(byte[] key, int offset, int length) throws IOException {
589
590
591 return seekTo(new KeyValue.KeyOnlyKeyValue(key, offset, length));
592 }
593
594 @Override
595 public int reseekTo(byte[] key, int offset, int length) throws IOException {
596 return reseekTo(new KeyValue.KeyOnlyKeyValue(key, offset, length));
597 }
598
599 @Override
600 public int seekTo(Cell key) throws IOException {
601 return seekTo(key, true);
602 }
603
604 @Override
605 public int reseekTo(Cell key) throws IOException {
606 int compared;
607 if (isSeeked()) {
608 compared = compareKey(reader.getComparator(), key);
609 if (compared < 1) {
610
611
612 return compared;
613 } else {
614
615 if (this.nextIndexedKey != null &&
616 (this.nextIndexedKey == KeyValueScanner.NO_NEXT_INDEXED_KEY || reader
617 .getComparator()
618 .compareOnlyKeyPortion(key, nextIndexedKey) < 0)) {
619
620
621
622
623
624
625 return loadBlockAndSeekToKey(this.block, nextIndexedKey, false, key, false);
626 }
627 }
628 }
629
630
631 return seekTo(key, false);
632 }
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649 public int seekTo(Cell key, boolean rewind) throws IOException {
650 HFileBlockIndex.BlockIndexReader indexReader = reader.getDataBlockIndexReader();
651 BlockWithScanInfo blockWithScanInfo = indexReader.loadDataBlockWithScanInfo(key, block,
652 cacheBlocks, pread, isCompaction, getEffectiveDataBlockEncoding());
653 if (blockWithScanInfo == null || blockWithScanInfo.getHFileBlock() == null) {
654
655 return -1;
656 }
657 return loadBlockAndSeekToKey(blockWithScanInfo.getHFileBlock(),
658 blockWithScanInfo.getNextIndexedKey(), rewind, key, false);
659 }
660
661 @Override
662 public boolean seekBefore(byte[] key, int offset, int length) throws IOException {
663 return seekBefore(new KeyValue.KeyOnlyKeyValue(key, offset, length));
664 }
665
666 @Override
667 public boolean seekBefore(Cell key) throws IOException {
668 HFileBlock seekToBlock = reader.getDataBlockIndexReader().seekToDataBlock(key, block,
669 cacheBlocks, pread, isCompaction,
670 ((HFileReaderV2) reader).getEffectiveEncodingInCache(isCompaction));
671 if (seekToBlock == null) {
672 return false;
673 }
674 ByteBuffer firstKey = getFirstKeyInBlock(seekToBlock);
675
676 if (reader.getComparator()
677 .compareOnlyKeyPortion(
678 new KeyValue.KeyOnlyKeyValue(firstKey.array(), firstKey.arrayOffset(),
679 firstKey.limit()), key) >= 0) {
680 long previousBlockOffset = seekToBlock.getPrevBlockOffset();
681
682 if (previousBlockOffset == -1) {
683
684 return false;
685 }
686
687
688
689
690
691
692 int prevBlockSize = -1;
693 seekToBlock = reader.readBlock(previousBlockOffset,
694 prevBlockSize, cacheBlocks,
695 pread, isCompaction, true, BlockType.DATA, getEffectiveDataBlockEncoding());
696
697
698 }
699 Cell firstKeyInCurrentBlock = new KeyValue.KeyOnlyKeyValue(Bytes.getBytes(firstKey));
700 loadBlockAndSeekToKey(seekToBlock, firstKeyInCurrentBlock, true, key, true);
701 return true;
702 }
703
704
705
706
707
708
709
710
711 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NP_NULL_ON_SOME_PATH",
712 justification="Yeah, unnecessary null check; could do w/ clean up")
713 protected HFileBlock readNextDataBlock() throws IOException {
714 long lastDataBlockOffset = reader.getTrailer().getLastDataBlockOffset();
715 if (block == null)
716 return null;
717
718 HFileBlock curBlock = block;
719
720 do {
721 if (curBlock.getOffset() >= lastDataBlockOffset) {
722 return null;
723 }
724
725 if (curBlock.getOffset() < 0) {
726 throw new IOException("Invalid block file offset: " + block);
727 }
728
729
730
731 curBlock = reader.readBlock(curBlock.getOffset()
732 + curBlock.getOnDiskSizeWithHeader(),
733 curBlock.getNextBlockOnDiskSize(), cacheBlocks, pread,
734 isCompaction, true, null, getEffectiveDataBlockEncoding());
735 } while (!curBlock.getBlockType().isData());
736
737 return curBlock;
738 }
739
740 public DataBlockEncoding getEffectiveDataBlockEncoding() {
741 return ((HFileReaderV2)reader).getEffectiveEncodingInCache(isCompaction);
742 }
743
744
745
746
747
748
749
750
751 public abstract int compareKey(KVComparator comparator, byte[] key, int offset,
752 int length);
753
754 public abstract int compareKey(KVComparator comparator, Cell kv);
755 }
756
757
758
759
760 protected static class ScannerV2 extends AbstractScannerV2 {
761 private HFileReaderV2 reader;
762
763 public ScannerV2(HFileReaderV2 r, boolean cacheBlocks,
764 final boolean pread, final boolean isCompaction) {
765 super(r, cacheBlocks, pread, isCompaction);
766 this.reader = r;
767 }
768
769 @Override
770 public Cell getKeyValue() {
771 if (!isSeeked())
772 return null;
773
774 return formNoTagsKeyValue();
775 }
776
777 protected Cell formNoTagsKeyValue() {
778 NoTagsKeyValue ret = new NoTagsKeyValue(blockBuffer.array(), blockBuffer.arrayOffset()
779 + blockBuffer.position(), getCellBufSize());
780 if (this.reader.shouldIncludeMemstoreTS()) {
781 ret.setSequenceId(currMemstoreTS);
782 }
783 return ret;
784 }
785
786 protected int getCellBufSize() {
787 return KEY_VALUE_LEN_SIZE + currKeyLen + currValueLen;
788 }
789
790 @Override
791 public ByteBuffer getKey() {
792 assertSeeked();
793 return ByteBuffer.wrap(
794 blockBuffer.array(),
795 blockBuffer.arrayOffset() + blockBuffer.position()
796 + KEY_VALUE_LEN_SIZE, currKeyLen).slice();
797 }
798
799 @Override
800 public int compareKey(KVComparator comparator, byte[] key, int offset, int length) {
801 return comparator.compareFlatKey(key, offset, length, blockBuffer.array(),
802 blockBuffer.arrayOffset() + blockBuffer.position() + KEY_VALUE_LEN_SIZE, currKeyLen);
803 }
804
805 @Override
806 public ByteBuffer getValue() {
807 assertSeeked();
808 return ByteBuffer.wrap(
809 blockBuffer.array(),
810 blockBuffer.arrayOffset() + blockBuffer.position()
811 + KEY_VALUE_LEN_SIZE + currKeyLen, currValueLen).slice();
812 }
813
814 protected void setNonSeekedState() {
815 block = null;
816 blockBuffer = null;
817 currKeyLen = 0;
818 currValueLen = 0;
819 currMemstoreTS = 0;
820 currMemstoreTSLen = 0;
821 }
822
823
824
825
826 private void positionThisBlockBuffer() {
827 try {
828 blockBuffer.position(getNextCellStartPosition());
829 } catch (IllegalArgumentException e) {
830 LOG.error("Current pos = " + blockBuffer.position()
831 + "; currKeyLen = " + currKeyLen + "; currValLen = "
832 + currValueLen + "; block limit = " + blockBuffer.limit()
833 + "; HFile name = " + reader.getName()
834 + "; currBlock currBlockOffset = " + block.getOffset());
835 throw e;
836 }
837 }
838
839
840
841
842
843
844 private boolean positionForNextBlock() throws IOException {
845
846 long lastDataBlockOffset = reader.getTrailer().getLastDataBlockOffset();
847 if (block.getOffset() >= lastDataBlockOffset) {
848 setNonSeekedState();
849 return false;
850 }
851 return isNextBlock();
852 }
853
854 private boolean isNextBlock() throws IOException {
855
856 HFileBlock nextBlock = readNextDataBlock();
857 if (nextBlock == null) {
858 setNonSeekedState();
859 return false;
860 }
861 updateCurrBlock(nextBlock);
862 return true;
863 }
864
865 private final boolean _next() throws IOException {
866
867 if (blockBuffer.remaining() <= 0) {
868 return positionForNextBlock();
869 }
870
871 readKeyValueLen();
872 return true;
873 }
874
875
876
877
878
879
880
881
882 @Override
883 public boolean next() throws IOException {
884
885
886 assertSeeked();
887 positionThisBlockBuffer();
888 return _next();
889 }
890
891 protected int getNextCellStartPosition() {
892 return blockBuffer.position() + KEY_VALUE_LEN_SIZE + currKeyLen + currValueLen
893 + currMemstoreTSLen;
894 }
895
896
897
898
899
900
901
902
903 @Override
904 public boolean seekTo() throws IOException {
905 if (reader == null) {
906 return false;
907 }
908
909 if (reader.getTrailer().getEntryCount() == 0) {
910
911 return false;
912 }
913
914 long firstDataBlockOffset =
915 reader.getTrailer().getFirstDataBlockOffset();
916 if (block != null && block.getOffset() == firstDataBlockOffset) {
917 blockBuffer.rewind();
918 readKeyValueLen();
919 return true;
920 }
921
922 block = reader.readBlock(firstDataBlockOffset, -1, cacheBlocks, pread,
923 isCompaction, true, BlockType.DATA, getEffectiveDataBlockEncoding());
924 if (block.getOffset() < 0) {
925 throw new IOException("Invalid block offset: " + block.getOffset());
926 }
927 updateCurrBlock(block);
928 return true;
929 }
930
931 @Override
932 protected int loadBlockAndSeekToKey(HFileBlock seekToBlock, Cell nextIndexedKey,
933 boolean rewind, Cell key, boolean seekBefore) throws IOException {
934 if (block == null || block.getOffset() != seekToBlock.getOffset()) {
935 updateCurrBlock(seekToBlock);
936 } else if (rewind) {
937 blockBuffer.rewind();
938 }
939
940
941 this.nextIndexedKey = nextIndexedKey;
942 return blockSeek(key, seekBefore);
943 }
944
945
946
947
948
949
950
951 protected void updateCurrBlock(HFileBlock newBlock) {
952 block = newBlock;
953
954
955 if (block.getBlockType() != BlockType.DATA) {
956 throw new IllegalStateException("ScannerV2 works only on data " +
957 "blocks, got " + block.getBlockType() + "; " +
958 "fileName=" + reader.name + ", " +
959 "dataBlockEncoder=" + reader.dataBlockEncoder + ", " +
960 "isCompaction=" + isCompaction);
961 }
962
963 blockBuffer = block.getBufferWithoutHeader();
964 readKeyValueLen();
965 blockFetches.incrementAndGet();
966
967
968 this.nextIndexedKey = null;
969 }
970
971
972
973
974
975 protected final boolean checkKeyLen(final int v) {
976 return v <= 0 || v > this.blockBuffer.limit();
977 }
978
979
980
981
982
983 protected final boolean checkLen(final int v) {
984 return v < 0 || v > this.blockBuffer.limit();
985 }
986
987
988
989
990 protected final void checkKeyValueLen() {
991 if (checkKeyLen(this.currKeyLen) || checkLen(this.currValueLen)) {
992 throw new IllegalStateException("Invalid currKeyLen " + this.currKeyLen +
993 " or currValueLen " + this.currValueLen + ". Block offset: " + block.getOffset() +
994 ", block length: " + this.blockBuffer.limit() + ", position: " +
995 this.blockBuffer.position() + " (without header).");
996 }
997 }
998
999 protected void readKeyValueLen() {
1000
1001
1002
1003
1004
1005 int p = blockBuffer.position() + blockBuffer.arrayOffset();
1006
1007
1008 long ll = Bytes.toLong(blockBuffer.array(), p);
1009
1010 this.currKeyLen = (int)(ll >> Integer.SIZE);
1011 this.currValueLen = (int)(Bytes.MASK_FOR_LOWER_INT_IN_LONG ^ ll);
1012 checkKeyValueLen();
1013
1014 p += (Bytes.SIZEOF_LONG + currKeyLen + currValueLen);
1015 readMvccVersion(p);
1016 }
1017
1018
1019
1020
1021
1022 protected void readMvccVersion(final int position) {
1023
1024 if (!this.reader.shouldIncludeMemstoreTS()) return;
1025 if (!this.reader.decodeMemstoreTS) {
1026 currMemstoreTS = 0;
1027 currMemstoreTSLen = 1;
1028 return;
1029 }
1030 _readMvccVersion(position);
1031 }
1032
1033
1034
1035
1036
1037 private void _readMvccVersion(final int position) {
1038
1039
1040
1041 byte firstByte = blockBuffer.array()[position];
1042 int len = WritableUtils.decodeVIntSize(firstByte);
1043 if (len == 1) {
1044 this.currMemstoreTS = firstByte;
1045 } else {
1046 long i = 0;
1047 for (int idx = 0; idx < len - 1; idx++) {
1048 byte b = blockBuffer.array()[position + 1 + idx];
1049 i = i << 8;
1050 i = i | (b & 0xFF);
1051 }
1052 currMemstoreTS = (WritableUtils.isNegativeVInt(firstByte) ? ~i : i);
1053 }
1054 this.currMemstoreTSLen = len;
1055 }
1056
1057 protected void readMvccVersion() {
1058
1059 readMvccVersion(blockBuffer.arrayOffset() + blockBuffer.position());
1060 }
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079 protected int blockSeek(Cell key, boolean seekBefore) {
1080 int klen, vlen;
1081 long memstoreTS = 0;
1082 int memstoreTSLen = 0;
1083 int lastKeyValueSize = -1;
1084 KeyValue.KeyOnlyKeyValue keyOnlykv = new KeyValue.KeyOnlyKeyValue();
1085 do {
1086 blockBuffer.mark();
1087 klen = blockBuffer.getInt();
1088 vlen = blockBuffer.getInt();
1089 if (checkKeyLen(klen) || checkLen(vlen)) {
1090 throw new IllegalStateException("Invalid klen " + klen + " or vlen "
1091 + vlen + ". Block offset: "
1092 + block.getOffset() + ", block length: " + blockBuffer.limit() + ", position: "
1093 + blockBuffer.position() + " (without header).");
1094 }
1095 blockBuffer.reset();
1096 if (this.reader.shouldIncludeMemstoreTS()) {
1097 if (this.reader.decodeMemstoreTS) {
1098 int memstoreTSOffset = blockBuffer.arrayOffset() + blockBuffer.position()
1099 + KEY_VALUE_LEN_SIZE + klen + vlen;
1100 memstoreTS = Bytes.readAsVLong(blockBuffer.array(), memstoreTSOffset);
1101 memstoreTSLen = WritableUtils.getVIntSize(memstoreTS);
1102 } else {
1103 memstoreTS = 0;
1104 memstoreTSLen = 1;
1105 }
1106 }
1107
1108 int keyOffset = blockBuffer.arrayOffset() + blockBuffer.position() + KEY_VALUE_LEN_SIZE;
1109 keyOnlykv.setKey(blockBuffer.array(), keyOffset, klen);
1110 int comp = reader.getComparator().compareOnlyKeyPortion(key, keyOnlykv);
1111
1112 if (comp == 0) {
1113 if (seekBefore) {
1114 if (lastKeyValueSize < 0) {
1115 throw new IllegalStateException("blockSeek with seekBefore "
1116 + "at the first key of the block: key="
1117 + CellUtil.getCellKeyAsString(key)
1118 + ", blockOffset=" + block.getOffset() + ", onDiskSize="
1119 + block.getOnDiskSizeWithHeader());
1120 }
1121 blockBuffer.position(blockBuffer.position() - lastKeyValueSize);
1122 readKeyValueLen();
1123 return 1;
1124 }
1125 currKeyLen = klen;
1126 currValueLen = vlen;
1127 if (this.reader.shouldIncludeMemstoreTS()) {
1128 currMemstoreTS = memstoreTS;
1129 currMemstoreTSLen = memstoreTSLen;
1130 }
1131 return 0;
1132 } else if (comp < 0) {
1133 if (lastKeyValueSize > 0)
1134 blockBuffer.position(blockBuffer.position() - lastKeyValueSize);
1135 readKeyValueLen();
1136 if (lastKeyValueSize == -1 && blockBuffer.position() == 0
1137 && this.reader.trailer.getMinorVersion() >= MINOR_VERSION_WITH_FAKED_KEY) {
1138 return HConstants.INDEX_KEY_MAGIC;
1139 }
1140 return 1;
1141 }
1142
1143
1144 lastKeyValueSize = klen + vlen + memstoreTSLen + KEY_VALUE_LEN_SIZE;
1145 blockBuffer.position(blockBuffer.position() + lastKeyValueSize);
1146 } while (blockBuffer.remaining() > 0);
1147
1148
1149
1150
1151 blockBuffer.position(blockBuffer.position() - lastKeyValueSize);
1152 readKeyValueLen();
1153 return 1;
1154 }
1155
1156 @Override
1157 protected ByteBuffer getFirstKeyInBlock(HFileBlock curBlock) {
1158 ByteBuffer buffer = curBlock.getBufferWithoutHeader();
1159
1160 buffer.rewind();
1161 int klen = buffer.getInt();
1162 buffer.getInt();
1163 ByteBuffer keyBuff = buffer.slice();
1164 keyBuff.limit(klen);
1165 keyBuff.rewind();
1166 return keyBuff;
1167 }
1168
1169 @Override
1170 public String getKeyString() {
1171 return Bytes.toStringBinary(blockBuffer.array(),
1172 blockBuffer.arrayOffset() + blockBuffer.position()
1173 + KEY_VALUE_LEN_SIZE, currKeyLen);
1174 }
1175
1176 @Override
1177 public String getValueString() {
1178 return Bytes.toString(blockBuffer.array(), blockBuffer.arrayOffset()
1179 + blockBuffer.position() + KEY_VALUE_LEN_SIZE + currKeyLen,
1180 currValueLen);
1181 }
1182
1183 @Override
1184 public int compareKey(KVComparator comparator, Cell key) {
1185 return comparator.compareOnlyKeyPortion(
1186 key,
1187 new KeyValue.KeyOnlyKeyValue(blockBuffer.array(), blockBuffer.arrayOffset()
1188 + blockBuffer.position() + KEY_VALUE_LEN_SIZE, currKeyLen));
1189 }
1190 }
1191
1192
1193
1194
1195 protected static class EncodedScannerV2 extends AbstractScannerV2 {
1196 private final HFileBlockDecodingContext decodingCtx;
1197 private final DataBlockEncoder.EncodedSeeker seeker;
1198 private final DataBlockEncoder dataBlockEncoder;
1199 protected final HFileContext meta;
1200
1201 public EncodedScannerV2(HFileReaderV2 reader, boolean cacheBlocks,
1202 boolean pread, boolean isCompaction, HFileContext meta) {
1203 super(reader, cacheBlocks, pread, isCompaction);
1204 DataBlockEncoding encoding = reader.dataBlockEncoder.getDataBlockEncoding();
1205 dataBlockEncoder = encoding.getEncoder();
1206 decodingCtx = dataBlockEncoder.newDataBlockDecodingContext(meta);
1207 seeker = dataBlockEncoder.createSeeker(
1208 reader.getComparator(), decodingCtx);
1209 this.meta = meta;
1210 }
1211
1212 @Override
1213 public boolean isSeeked(){
1214 return this.block != null;
1215 }
1216
1217
1218
1219
1220
1221
1222
1223
1224 private void updateCurrentBlock(HFileBlock newBlock) throws CorruptHFileException {
1225 block = newBlock;
1226
1227
1228 if (block.getBlockType() != BlockType.ENCODED_DATA) {
1229 throw new IllegalStateException(
1230 "EncodedScanner works only on encoded data blocks");
1231 }
1232 short dataBlockEncoderId = block.getDataBlockEncodingId();
1233 if (!DataBlockEncoding.isCorrectEncoder(dataBlockEncoder, dataBlockEncoderId)) {
1234 String encoderCls = dataBlockEncoder.getClass().getName();
1235 throw new CorruptHFileException("Encoder " + encoderCls
1236 + " doesn't support data block encoding "
1237 + DataBlockEncoding.getNameFromId(dataBlockEncoderId));
1238 }
1239
1240 seeker.setCurrentBuffer(getEncodedBuffer(newBlock));
1241 blockFetches.incrementAndGet();
1242
1243
1244 this.nextIndexedKey = null;
1245 }
1246
1247 private ByteBuffer getEncodedBuffer(HFileBlock newBlock) {
1248 ByteBuffer origBlock = newBlock.getBufferReadOnly();
1249 ByteBuffer encodedBlock = ByteBuffer.wrap(origBlock.array(),
1250 origBlock.arrayOffset() + newBlock.headerSize() +
1251 DataBlockEncoding.ID_SIZE,
1252 newBlock.getUncompressedSizeWithoutHeader() -
1253 DataBlockEncoding.ID_SIZE).slice();
1254 return encodedBlock;
1255 }
1256
1257 @Override
1258 public boolean seekTo() throws IOException {
1259 if (reader == null) {
1260 return false;
1261 }
1262
1263 if (reader.getTrailer().getEntryCount() == 0) {
1264
1265 return false;
1266 }
1267
1268 long firstDataBlockOffset =
1269 reader.getTrailer().getFirstDataBlockOffset();
1270 if (block != null && block.getOffset() == firstDataBlockOffset) {
1271 seeker.rewind();
1272 return true;
1273 }
1274
1275 block = reader.readBlock(firstDataBlockOffset, -1, cacheBlocks, pread,
1276 isCompaction, true, BlockType.DATA, getEffectiveDataBlockEncoding());
1277 if (block.getOffset() < 0) {
1278 throw new IOException("Invalid block offset: " + block.getOffset());
1279 }
1280 updateCurrentBlock(block);
1281 return true;
1282 }
1283
1284 @Override
1285 public boolean next() throws IOException {
1286 boolean isValid = seeker.next();
1287 if (!isValid) {
1288 block = readNextDataBlock();
1289 isValid = block != null;
1290 if (isValid) {
1291 updateCurrentBlock(block);
1292 }
1293 }
1294 return isValid;
1295 }
1296
1297 @Override
1298 public ByteBuffer getKey() {
1299 assertValidSeek();
1300 return seeker.getKeyDeepCopy();
1301 }
1302
1303 @Override
1304 public int compareKey(KVComparator comparator, byte[] key, int offset, int length) {
1305 return seeker.compareKey(comparator, key, offset, length);
1306 }
1307
1308 @Override
1309 public ByteBuffer getValue() {
1310 assertValidSeek();
1311 return seeker.getValueShallowCopy();
1312 }
1313
1314 @Override
1315 public Cell getKeyValue() {
1316 if (block == null) {
1317 return null;
1318 }
1319 return seeker.getKeyValue();
1320 }
1321
1322 @Override
1323 public String getKeyString() {
1324 ByteBuffer keyBuffer = getKey();
1325 return Bytes.toStringBinary(keyBuffer.array(),
1326 keyBuffer.arrayOffset(), keyBuffer.limit());
1327 }
1328
1329 @Override
1330 public String getValueString() {
1331 ByteBuffer valueBuffer = getValue();
1332 return Bytes.toStringBinary(valueBuffer.array(),
1333 valueBuffer.arrayOffset(), valueBuffer.limit());
1334 }
1335
1336 private void assertValidSeek() {
1337 if (block == null) {
1338 throw new NotSeekedException();
1339 }
1340 }
1341
1342 @Override
1343 protected ByteBuffer getFirstKeyInBlock(HFileBlock curBlock) {
1344 return dataBlockEncoder.getFirstKeyInBlock(getEncodedBuffer(curBlock));
1345 }
1346
1347 @Override
1348 protected int loadBlockAndSeekToKey(HFileBlock seekToBlock, Cell nextIndexedKey,
1349 boolean rewind, Cell key, boolean seekBefore) throws IOException {
1350 if (block == null || block.getOffset() != seekToBlock.getOffset()) {
1351 updateCurrentBlock(seekToBlock);
1352 } else if (rewind) {
1353 seeker.rewind();
1354 }
1355 this.nextIndexedKey = nextIndexedKey;
1356 return seeker.seekToKeyInBlock(key, seekBefore);
1357 }
1358
1359 @Override
1360 public int compareKey(KVComparator comparator, Cell key) {
1361 return seeker.compareKey(comparator, key);
1362 }
1363 }
1364
1365
1366
1367
1368
1369 @Override
1370 public DataInput getGeneralBloomFilterMetadata() throws IOException {
1371 return this.getBloomFilterMetadata(BlockType.GENERAL_BLOOM_META);
1372 }
1373
1374 @Override
1375 public DataInput getDeleteBloomFilterMetadata() throws IOException {
1376 return this.getBloomFilterMetadata(BlockType.DELETE_FAMILY_BLOOM_META);
1377 }
1378
1379 private DataInput getBloomFilterMetadata(BlockType blockType)
1380 throws IOException {
1381 if (blockType != BlockType.GENERAL_BLOOM_META &&
1382 blockType != BlockType.DELETE_FAMILY_BLOOM_META) {
1383 throw new RuntimeException("Block Type: " + blockType.toString() +
1384 " is not supported") ;
1385 }
1386
1387 for (HFileBlock b : loadOnOpenBlocks)
1388 if (b.getBlockType() == blockType)
1389 return b.getByteStream();
1390 return null;
1391 }
1392
1393 @Override
1394 public boolean isFileInfoLoaded() {
1395 return true;
1396 }
1397
1398
1399
1400
1401
1402 private void validateMinorVersion(Path path, int minorVersion) {
1403 if (minorVersion < MIN_MINOR_VERSION ||
1404 minorVersion > MAX_MINOR_VERSION) {
1405 String msg = "Minor version for path " + path +
1406 " is expected to be between " +
1407 MIN_MINOR_VERSION + " and " + MAX_MINOR_VERSION +
1408 " but is found to be " + minorVersion;
1409 LOG.error(msg);
1410 throw new RuntimeException(msg);
1411 }
1412 }
1413
1414 @Override
1415 public int getMajorVersion() {
1416 return 2;
1417 }
1418
1419 @Override
1420 public HFileContext getFileContext() {
1421 return hfileContext;
1422 }
1423
1424
1425
1426
1427
1428 @VisibleForTesting
1429 boolean prefetchComplete() {
1430 return PrefetchExecutor.isCompleted(path);
1431 }
1432
1433 @Override
1434 public void unbufferStream() {
1435 fsBlockReader.unbufferStream();
1436 }
1437 }