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).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);
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 checkLen(final int v) {
976 return v < 0 || v > this.blockBuffer.limit();
977 }
978
979
980
981
982 protected final void checkKeyValueLen() {
983 if (checkLen(this.currKeyLen) || checkLen(this.currValueLen)) {
984 throw new IllegalStateException("Invalid currKeyLen " + this.currKeyLen +
985 " or currValueLen " + this.currValueLen + ". Block offset: " + block.getOffset() +
986 ", block length: " + this.blockBuffer.limit() + ", position: " +
987 this.blockBuffer.position() + " (without header).");
988 }
989 }
990
991 protected void readKeyValueLen() {
992
993
994
995
996
997 int p = blockBuffer.position() + blockBuffer.arrayOffset();
998
999
1000 long ll = Bytes.toLong(blockBuffer.array(), p);
1001
1002 this.currKeyLen = (int)(ll >> Integer.SIZE);
1003 this.currValueLen = (int)(Bytes.MASK_FOR_LOWER_INT_IN_LONG ^ ll);
1004 checkKeyValueLen();
1005
1006 p += (Bytes.SIZEOF_LONG + currKeyLen + currValueLen);
1007 readMvccVersion(p);
1008 }
1009
1010
1011
1012
1013
1014 protected void readMvccVersion(final int position) {
1015
1016 if (!this.reader.shouldIncludeMemstoreTS()) return;
1017 if (!this.reader.decodeMemstoreTS) {
1018 currMemstoreTS = 0;
1019 currMemstoreTSLen = 1;
1020 return;
1021 }
1022 _readMvccVersion(position);
1023 }
1024
1025
1026
1027
1028
1029 private void _readMvccVersion(final int position) {
1030
1031
1032
1033 byte firstByte = blockBuffer.array()[position];
1034 int len = WritableUtils.decodeVIntSize(firstByte);
1035 if (len == 1) {
1036 this.currMemstoreTS = firstByte;
1037 } else {
1038 long i = 0;
1039 for (int idx = 0; idx < len - 1; idx++) {
1040 byte b = blockBuffer.array()[position + 1 + idx];
1041 i = i << 8;
1042 i = i | (b & 0xFF);
1043 }
1044 currMemstoreTS = (WritableUtils.isNegativeVInt(firstByte) ? ~i : i);
1045 }
1046 this.currMemstoreTSLen = len;
1047 }
1048
1049 protected void readMvccVersion() {
1050
1051 readMvccVersion(blockBuffer.arrayOffset() + blockBuffer.position());
1052 }
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071 protected int blockSeek(Cell key, boolean seekBefore) {
1072 int klen, vlen;
1073 long memstoreTS = 0;
1074 int memstoreTSLen = 0;
1075 int lastKeyValueSize = -1;
1076 KeyValue.KeyOnlyKeyValue keyOnlykv = new KeyValue.KeyOnlyKeyValue();
1077 do {
1078 blockBuffer.mark();
1079 klen = blockBuffer.getInt();
1080 vlen = blockBuffer.getInt();
1081 blockBuffer.reset();
1082 if (this.reader.shouldIncludeMemstoreTS()) {
1083 if (this.reader.decodeMemstoreTS) {
1084 int memstoreTSOffset = blockBuffer.arrayOffset() + blockBuffer.position()
1085 + KEY_VALUE_LEN_SIZE + klen + vlen;
1086 memstoreTS = Bytes.readAsVLong(blockBuffer.array(), memstoreTSOffset);
1087 memstoreTSLen = WritableUtils.getVIntSize(memstoreTS);
1088 } else {
1089 memstoreTS = 0;
1090 memstoreTSLen = 1;
1091 }
1092 }
1093
1094 int keyOffset = blockBuffer.arrayOffset() + blockBuffer.position() + KEY_VALUE_LEN_SIZE;
1095 keyOnlykv.setKey(blockBuffer.array(), keyOffset, klen);
1096 int comp = reader.getComparator().compareOnlyKeyPortion(key, keyOnlykv);
1097
1098 if (comp == 0) {
1099 if (seekBefore) {
1100 if (lastKeyValueSize < 0) {
1101 throw new IllegalStateException("blockSeek with seekBefore "
1102 + "at the first key of the block: key="
1103 + CellUtil.getCellKeyAsString(key)
1104 + ", blockOffset=" + block.getOffset() + ", onDiskSize="
1105 + block.getOnDiskSizeWithHeader());
1106 }
1107 blockBuffer.position(blockBuffer.position() - lastKeyValueSize);
1108 readKeyValueLen();
1109 return 1;
1110 }
1111 currKeyLen = klen;
1112 currValueLen = vlen;
1113 if (this.reader.shouldIncludeMemstoreTS()) {
1114 currMemstoreTS = memstoreTS;
1115 currMemstoreTSLen = memstoreTSLen;
1116 }
1117 return 0;
1118 } else if (comp < 0) {
1119 if (lastKeyValueSize > 0)
1120 blockBuffer.position(blockBuffer.position() - lastKeyValueSize);
1121 readKeyValueLen();
1122 if (lastKeyValueSize == -1 && blockBuffer.position() == 0
1123 && this.reader.trailer.getMinorVersion() >= MINOR_VERSION_WITH_FAKED_KEY) {
1124 return HConstants.INDEX_KEY_MAGIC;
1125 }
1126 return 1;
1127 }
1128
1129
1130 lastKeyValueSize = klen + vlen + memstoreTSLen + KEY_VALUE_LEN_SIZE;
1131 blockBuffer.position(blockBuffer.position() + lastKeyValueSize);
1132 } while (blockBuffer.remaining() > 0);
1133
1134
1135
1136
1137 blockBuffer.position(blockBuffer.position() - lastKeyValueSize);
1138 readKeyValueLen();
1139 return 1;
1140 }
1141
1142 @Override
1143 protected ByteBuffer getFirstKeyInBlock(HFileBlock curBlock) {
1144 ByteBuffer buffer = curBlock.getBufferWithoutHeader();
1145
1146 buffer.rewind();
1147 int klen = buffer.getInt();
1148 buffer.getInt();
1149 ByteBuffer keyBuff = buffer.slice();
1150 keyBuff.limit(klen);
1151 keyBuff.rewind();
1152 return keyBuff;
1153 }
1154
1155 @Override
1156 public String getKeyString() {
1157 return Bytes.toStringBinary(blockBuffer.array(),
1158 blockBuffer.arrayOffset() + blockBuffer.position()
1159 + KEY_VALUE_LEN_SIZE, currKeyLen);
1160 }
1161
1162 @Override
1163 public String getValueString() {
1164 return Bytes.toString(blockBuffer.array(), blockBuffer.arrayOffset()
1165 + blockBuffer.position() + KEY_VALUE_LEN_SIZE + currKeyLen,
1166 currValueLen);
1167 }
1168
1169 @Override
1170 public int compareKey(KVComparator comparator, Cell key) {
1171 return comparator.compareOnlyKeyPortion(
1172 key,
1173 new KeyValue.KeyOnlyKeyValue(blockBuffer.array(), blockBuffer.arrayOffset()
1174 + blockBuffer.position() + KEY_VALUE_LEN_SIZE, currKeyLen));
1175 }
1176 }
1177
1178
1179
1180
1181 protected static class EncodedScannerV2 extends AbstractScannerV2 {
1182 private final HFileBlockDecodingContext decodingCtx;
1183 private final DataBlockEncoder.EncodedSeeker seeker;
1184 private final DataBlockEncoder dataBlockEncoder;
1185 protected final HFileContext meta;
1186
1187 public EncodedScannerV2(HFileReaderV2 reader, boolean cacheBlocks,
1188 boolean pread, boolean isCompaction, HFileContext meta) {
1189 super(reader, cacheBlocks, pread, isCompaction);
1190 DataBlockEncoding encoding = reader.dataBlockEncoder.getDataBlockEncoding();
1191 dataBlockEncoder = encoding.getEncoder();
1192 decodingCtx = dataBlockEncoder.newDataBlockDecodingContext(meta);
1193 seeker = dataBlockEncoder.createSeeker(
1194 reader.getComparator(), decodingCtx);
1195 this.meta = meta;
1196 }
1197
1198 @Override
1199 public boolean isSeeked(){
1200 return this.block != null;
1201 }
1202
1203
1204
1205
1206
1207
1208
1209
1210 private void updateCurrentBlock(HFileBlock newBlock) throws CorruptHFileException {
1211 block = newBlock;
1212
1213
1214 if (block.getBlockType() != BlockType.ENCODED_DATA) {
1215 throw new IllegalStateException(
1216 "EncodedScanner works only on encoded data blocks");
1217 }
1218 short dataBlockEncoderId = block.getDataBlockEncodingId();
1219 if (!DataBlockEncoding.isCorrectEncoder(dataBlockEncoder, dataBlockEncoderId)) {
1220 String encoderCls = dataBlockEncoder.getClass().getName();
1221 throw new CorruptHFileException("Encoder " + encoderCls
1222 + " doesn't support data block encoding "
1223 + DataBlockEncoding.getNameFromId(dataBlockEncoderId));
1224 }
1225
1226 seeker.setCurrentBuffer(getEncodedBuffer(newBlock));
1227 blockFetches.incrementAndGet();
1228
1229
1230 this.nextIndexedKey = null;
1231 }
1232
1233 private ByteBuffer getEncodedBuffer(HFileBlock newBlock) {
1234 ByteBuffer origBlock = newBlock.getBufferReadOnly();
1235 ByteBuffer encodedBlock = ByteBuffer.wrap(origBlock.array(),
1236 origBlock.arrayOffset() + newBlock.headerSize() +
1237 DataBlockEncoding.ID_SIZE,
1238 newBlock.getUncompressedSizeWithoutHeader() -
1239 DataBlockEncoding.ID_SIZE).slice();
1240 return encodedBlock;
1241 }
1242
1243 @Override
1244 public boolean seekTo() throws IOException {
1245 if (reader == null) {
1246 return false;
1247 }
1248
1249 if (reader.getTrailer().getEntryCount() == 0) {
1250
1251 return false;
1252 }
1253
1254 long firstDataBlockOffset =
1255 reader.getTrailer().getFirstDataBlockOffset();
1256 if (block != null && block.getOffset() == firstDataBlockOffset) {
1257 seeker.rewind();
1258 return true;
1259 }
1260
1261 block = reader.readBlock(firstDataBlockOffset, -1, cacheBlocks, pread,
1262 isCompaction, true, BlockType.DATA, getEffectiveDataBlockEncoding());
1263 if (block.getOffset() < 0) {
1264 throw new IOException("Invalid block offset: " + block.getOffset());
1265 }
1266 updateCurrentBlock(block);
1267 return true;
1268 }
1269
1270 @Override
1271 public boolean next() throws IOException {
1272 boolean isValid = seeker.next();
1273 if (!isValid) {
1274 block = readNextDataBlock();
1275 isValid = block != null;
1276 if (isValid) {
1277 updateCurrentBlock(block);
1278 }
1279 }
1280 return isValid;
1281 }
1282
1283 @Override
1284 public ByteBuffer getKey() {
1285 assertValidSeek();
1286 return seeker.getKeyDeepCopy();
1287 }
1288
1289 @Override
1290 public int compareKey(KVComparator comparator, byte[] key, int offset, int length) {
1291 return seeker.compareKey(comparator, key, offset, length);
1292 }
1293
1294 @Override
1295 public ByteBuffer getValue() {
1296 assertValidSeek();
1297 return seeker.getValueShallowCopy();
1298 }
1299
1300 @Override
1301 public Cell getKeyValue() {
1302 if (block == null) {
1303 return null;
1304 }
1305 return seeker.getKeyValue();
1306 }
1307
1308 @Override
1309 public String getKeyString() {
1310 ByteBuffer keyBuffer = getKey();
1311 return Bytes.toStringBinary(keyBuffer.array(),
1312 keyBuffer.arrayOffset(), keyBuffer.limit());
1313 }
1314
1315 @Override
1316 public String getValueString() {
1317 ByteBuffer valueBuffer = getValue();
1318 return Bytes.toStringBinary(valueBuffer.array(),
1319 valueBuffer.arrayOffset(), valueBuffer.limit());
1320 }
1321
1322 private void assertValidSeek() {
1323 if (block == null) {
1324 throw new NotSeekedException();
1325 }
1326 }
1327
1328 @Override
1329 protected ByteBuffer getFirstKeyInBlock(HFileBlock curBlock) {
1330 return dataBlockEncoder.getFirstKeyInBlock(getEncodedBuffer(curBlock));
1331 }
1332
1333 @Override
1334 protected int loadBlockAndSeekToKey(HFileBlock seekToBlock, Cell nextIndexedKey,
1335 boolean rewind, Cell key, boolean seekBefore) throws IOException {
1336 if (block == null || block.getOffset() != seekToBlock.getOffset()) {
1337 updateCurrentBlock(seekToBlock);
1338 } else if (rewind) {
1339 seeker.rewind();
1340 }
1341 this.nextIndexedKey = nextIndexedKey;
1342 return seeker.seekToKeyInBlock(key, seekBefore);
1343 }
1344
1345 @Override
1346 public int compareKey(KVComparator comparator, Cell key) {
1347 return seeker.compareKey(comparator, key);
1348 }
1349 }
1350
1351
1352
1353
1354
1355 @Override
1356 public DataInput getGeneralBloomFilterMetadata() throws IOException {
1357 return this.getBloomFilterMetadata(BlockType.GENERAL_BLOOM_META);
1358 }
1359
1360 @Override
1361 public DataInput getDeleteBloomFilterMetadata() throws IOException {
1362 return this.getBloomFilterMetadata(BlockType.DELETE_FAMILY_BLOOM_META);
1363 }
1364
1365 private DataInput getBloomFilterMetadata(BlockType blockType)
1366 throws IOException {
1367 if (blockType != BlockType.GENERAL_BLOOM_META &&
1368 blockType != BlockType.DELETE_FAMILY_BLOOM_META) {
1369 throw new RuntimeException("Block Type: " + blockType.toString() +
1370 " is not supported") ;
1371 }
1372
1373 for (HFileBlock b : loadOnOpenBlocks)
1374 if (b.getBlockType() == blockType)
1375 return b.getByteStream();
1376 return null;
1377 }
1378
1379 @Override
1380 public boolean isFileInfoLoaded() {
1381 return true;
1382 }
1383
1384
1385
1386
1387
1388 private void validateMinorVersion(Path path, int minorVersion) {
1389 if (minorVersion < MIN_MINOR_VERSION ||
1390 minorVersion > MAX_MINOR_VERSION) {
1391 String msg = "Minor version for path " + path +
1392 " is expected to be between " +
1393 MIN_MINOR_VERSION + " and " + MAX_MINOR_VERSION +
1394 " but is found to be " + minorVersion;
1395 LOG.error(msg);
1396 throw new RuntimeException(msg);
1397 }
1398 }
1399
1400 @Override
1401 public int getMajorVersion() {
1402 return 2;
1403 }
1404
1405 @Override
1406 public HFileContext getFileContext() {
1407 return hfileContext;
1408 }
1409
1410
1411
1412
1413
1414 @VisibleForTesting
1415 boolean prefetchComplete() {
1416 return PrefetchExecutor.isCompleted(path);
1417 }
1418
1419 @Override
1420 public void unbufferStream() {
1421 fsBlockReader.unbufferStream();
1422 }
1423 }