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.DataInputStream;
21 import java.io.DataOutput;
22 import java.io.DataOutputStream;
23 import java.io.IOException;
24 import java.io.InputStream;
25 import java.nio.ByteBuffer;
26 import java.util.concurrent.atomic.AtomicReference;
27 import java.util.concurrent.locks.Lock;
28 import java.util.concurrent.locks.ReentrantLock;
29
30 import org.apache.commons.logging.Log;
31 import org.apache.commons.logging.LogFactory;
32 import org.apache.hadoop.fs.FSDataInputStream;
33 import org.apache.hadoop.fs.FSDataOutputStream;
34 import org.apache.hadoop.fs.Path;
35 import org.apache.hadoop.hbase.Cell;
36 import org.apache.hadoop.hbase.HConstants;
37 import org.apache.hadoop.hbase.classification.InterfaceAudience;
38 import org.apache.hadoop.hbase.fs.HFileSystem;
39 import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
40 import org.apache.hadoop.hbase.io.ByteBufferInputStream;
41 import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper;
42 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
43 import org.apache.hadoop.hbase.io.encoding.HFileBlockDecodingContext;
44 import org.apache.hadoop.hbase.io.encoding.HFileBlockDefaultDecodingContext;
45 import org.apache.hadoop.hbase.io.encoding.HFileBlockDefaultEncodingContext;
46 import org.apache.hadoop.hbase.io.encoding.HFileBlockEncodingContext;
47 import org.apache.hadoop.hbase.util.ByteBufferUtils;
48 import org.apache.hadoop.hbase.util.Bytes;
49 import org.apache.hadoop.hbase.util.ChecksumType;
50 import org.apache.hadoop.hbase.util.ClassSize;
51 import org.apache.hadoop.io.IOUtils;
52
53 import com.google.common.annotations.VisibleForTesting;
54 import com.google.common.base.Preconditions;
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107 @InterfaceAudience.Private
108 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="HE_EQUALS_USE_HASHCODE",
109 justification="Fix!!! Fine for now bug FIXXXXXXX!!!!")
110 public class HFileBlock implements Cacheable {
111 private static final Log LOG = LogFactory.getLog(HFileBlock.class);
112
113
114 private BlockType blockType;
115
116
117
118
119
120 private int onDiskSizeWithoutHeader;
121
122
123
124
125
126 private int uncompressedSizeWithoutHeader;
127
128
129
130
131
132 private long prevBlockOffset;
133
134
135
136
137
138
139 private int onDiskDataSizeWithHeader;
140
141
142
143
144
145
146
147
148
149
150
151 private ByteBuffer buf;
152
153
154
155 private HFileContext fileContext;
156
157
158
159
160
161 private long offset = UNSET;
162
163
164
165
166
167
168
169
170
171
172
173
174 private int nextBlockOnDiskSize = UNSET;
175
176
177
178
179
180 static final int CHECKSUM_VERIFICATION_NUM_IO_THRESHOLD = 3;
181
182 private static int UNSET = -1;
183 public static final boolean FILL_HEADER = true;
184 public static final boolean DONT_FILL_HEADER = false;
185
186 public static final int BYTE_BUFFER_HEAP_SIZE = (int) ClassSize.estimateBase(
187 ByteBuffer.wrap(new byte[0], 0, 0).getClass(), false);
188
189
190
191
192
193
194
195
196
197
198
199
200
201 static final int BLOCK_METADATA_SPACE = Bytes.SIZEOF_BYTE + Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT;
202
203
204
205
206 static final int CHECKSUM_SIZE = Bytes.SIZEOF_INT;
207
208 static final byte[] DUMMY_HEADER_NO_CHECKSUM =
209 new byte[HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM];
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225 static final CacheableDeserializer<Cacheable> BLOCK_DESERIALIZER =
226 new CacheableDeserializer<Cacheable>() {
227 @Override
228 public HFileBlock deserialize(ByteBuffer buf, boolean reuse) throws IOException{
229
230
231 buf.limit(buf.limit() - BLOCK_METADATA_SPACE).rewind();
232
233 ByteBuffer newByteBuff;
234 if (reuse) {
235 newByteBuff = buf.slice();
236 } else {
237 int len = buf.limit();
238 newByteBuff = ByteBuffer.allocate(len);
239 ByteBufferUtils.copyFromBufferToBuffer(newByteBuff, buf, buf.position(), 0, len);
240 }
241
242 buf.position(buf.limit());
243 buf.limit(buf.limit() + HFileBlock.BLOCK_METADATA_SPACE);
244 boolean usesChecksum = buf.get() == (byte)1;
245 long offset = buf.getLong();
246 int nextBlockOnDiskSize = buf.getInt();
247 HFileBlock hFileBlock =
248 new HFileBlock(newByteBuff, usesChecksum, offset, nextBlockOnDiskSize, null);
249 return hFileBlock;
250 }
251
252 @Override
253 public int getDeserialiserIdentifier() {
254 return DESERIALIZER_IDENTIFIER;
255 }
256
257 @Override
258 public HFileBlock deserialize(ByteBuffer b) throws IOException {
259 return deserialize(b, false);
260 }
261 };
262 private static final int DESERIALIZER_IDENTIFIER;
263 static {
264 DESERIALIZER_IDENTIFIER =
265 CacheableDeserializerIdManager.registerDeserializer(BLOCK_DESERIALIZER);
266 }
267
268
269 static class Header {
270
271
272
273
274
275
276
277
278
279 static int BLOCK_MAGIC_INDEX = 0;
280 static int ON_DISK_SIZE_WITHOUT_HEADER_INDEX = 8;
281 static int UNCOMPRESSED_SIZE_WITHOUT_HEADER_INDEX = 12;
282 static int PREV_BLOCK_OFFSET_INDEX = 16;
283 static int CHECKSUM_TYPE_INDEX = 24;
284 static int BYTES_PER_CHECKSUM_INDEX = 25;
285 static int ON_DISK_DATA_SIZE_WITH_HEADER_INDEX = 29;
286 }
287
288
289
290
291 private HFileBlock(HFileBlock that) {
292 this.blockType = that.blockType;
293 this.onDiskSizeWithoutHeader = that.onDiskSizeWithoutHeader;
294 this.uncompressedSizeWithoutHeader = that.uncompressedSizeWithoutHeader;
295 this.prevBlockOffset = that.prevBlockOffset;
296 this.buf = that.buf.duplicate();
297 this.offset = that.offset;
298 this.onDiskDataSizeWithHeader = that.onDiskDataSizeWithHeader;
299 this.fileContext = that.fileContext;
300 this.nextBlockOnDiskSize = that.nextBlockOnDiskSize;
301 }
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323 @VisibleForTesting
324 public HFileBlock(BlockType blockType, int onDiskSizeWithoutHeader,
325 int uncompressedSizeWithoutHeader, long prevBlockOffset, ByteBuffer b, boolean fillHeader,
326 long offset, final int nextBlockOnDiskSize, int onDiskDataSizeWithHeader,
327 HFileContext fileContext) {
328 init(blockType, onDiskSizeWithoutHeader, uncompressedSizeWithoutHeader,
329 prevBlockOffset, offset, onDiskDataSizeWithHeader, nextBlockOnDiskSize, fileContext);
330 this.buf = b;
331 if (fillHeader) {
332 overwriteHeader();
333 }
334 this.buf.rewind();
335 }
336
337
338
339
340
341
342
343
344 HFileBlock(ByteBuffer buf, boolean usesHBaseChecksum, final long offset,
345 final int nextBlockOnDiskSize, HFileContext fileContext) throws IOException {
346 buf.rewind();
347 final BlockType blockType = BlockType.read(buf);
348 final int onDiskSizeWithoutHeader = buf.getInt(Header.ON_DISK_SIZE_WITHOUT_HEADER_INDEX);
349 final int uncompressedSizeWithoutHeader =
350 buf.getInt(Header.UNCOMPRESSED_SIZE_WITHOUT_HEADER_INDEX);
351 final long prevBlockOffset = buf.getLong(Header.PREV_BLOCK_OFFSET_INDEX);
352
353
354 HFileContextBuilder fileContextBuilder = fileContext != null?
355 new HFileContextBuilder(fileContext): new HFileContextBuilder();
356 fileContextBuilder.withHBaseCheckSum(usesHBaseChecksum);
357 int onDiskDataSizeWithHeader;
358 if (usesHBaseChecksum) {
359 byte checksumType = buf.get(Header.CHECKSUM_TYPE_INDEX);
360 int bytesPerChecksum = buf.getInt(Header.BYTES_PER_CHECKSUM_INDEX);
361 onDiskDataSizeWithHeader = buf.getInt(Header.ON_DISK_DATA_SIZE_WITH_HEADER_INDEX);
362
363 fileContextBuilder.withChecksumType(ChecksumType.codeToType(checksumType));
364 fileContextBuilder.withBytesPerCheckSum(bytesPerChecksum);
365 } else {
366 fileContextBuilder.withChecksumType(ChecksumType.NULL);
367 fileContextBuilder.withBytesPerCheckSum(0);
368
369 onDiskDataSizeWithHeader = onDiskSizeWithoutHeader + headerSize(usesHBaseChecksum);
370 }
371 fileContext = fileContextBuilder.build();
372 assert usesHBaseChecksum == fileContext.isUseHBaseChecksum();
373 init(blockType, onDiskSizeWithoutHeader, uncompressedSizeWithoutHeader,
374 prevBlockOffset, offset, onDiskDataSizeWithHeader, nextBlockOnDiskSize, fileContext);
375 this.offset = offset;
376 this.buf = buf;
377 this.buf.rewind();
378 }
379
380
381
382
383 private void init(BlockType blockType, int onDiskSizeWithoutHeader,
384 int uncompressedSizeWithoutHeader, long prevBlockOffset,
385 long offset, int onDiskDataSizeWithHeader, final int nextBlockOnDiskSize,
386 HFileContext fileContext) {
387 this.blockType = blockType;
388 this.onDiskSizeWithoutHeader = onDiskSizeWithoutHeader;
389 this.uncompressedSizeWithoutHeader = uncompressedSizeWithoutHeader;
390 this.prevBlockOffset = prevBlockOffset;
391 this.offset = offset;
392 this.onDiskDataSizeWithHeader = onDiskDataSizeWithHeader;
393 this.nextBlockOnDiskSize = nextBlockOnDiskSize;
394 this.fileContext = fileContext;
395 }
396
397
398
399
400
401
402
403 private static int getOnDiskSizeWithHeader(final ByteBuffer headerBuf, boolean verifyChecksum) {
404 return headerBuf.getInt(Header.ON_DISK_SIZE_WITHOUT_HEADER_INDEX) +
405 headerSize(verifyChecksum);
406 }
407
408
409
410
411
412
413 public int getNextBlockOnDiskSize() {
414 return nextBlockOnDiskSize;
415 }
416
417 @Override
418 public BlockType getBlockType() {
419 return blockType;
420 }
421
422
423 public short getDataBlockEncodingId() {
424 if (blockType != BlockType.ENCODED_DATA) {
425 throw new IllegalArgumentException("Querying encoder ID of a block " +
426 "of type other than " + BlockType.ENCODED_DATA + ": " + blockType);
427 }
428 return buf.getShort(headerSize());
429 }
430
431
432
433
434 public int getOnDiskSizeWithHeader() {
435 return onDiskSizeWithoutHeader + headerSize();
436 }
437
438
439
440
441 int getOnDiskSizeWithoutHeader() {
442 return onDiskSizeWithoutHeader;
443 }
444
445
446
447
448 public int getUncompressedSizeWithoutHeader() {
449 return uncompressedSizeWithoutHeader;
450 }
451
452
453
454
455
456 long getPrevBlockOffset() {
457 return prevBlockOffset;
458 }
459
460
461
462
463
464 private void overwriteHeader() {
465 buf.rewind();
466 blockType.write(buf);
467 buf.putInt(onDiskSizeWithoutHeader);
468 buf.putInt(uncompressedSizeWithoutHeader);
469 buf.putLong(prevBlockOffset);
470 if (this.fileContext.isUseHBaseChecksum()) {
471 buf.put(fileContext.getChecksumType().getCode());
472 buf.putInt(fileContext.getBytesPerChecksum());
473 buf.putInt(onDiskDataSizeWithHeader);
474 }
475 }
476
477
478
479
480
481
482 public ByteBuffer getBufferWithoutHeader() {
483 ByteBuffer dup = getBufferReadOnly();
484
485 dup.position(headerSize()).limit(buf.limit() - totalChecksumBytes());
486 return dup.slice();
487 }
488
489
490
491
492
493
494
495
496
497
498
499
500 public ByteBuffer getBufferReadOnly() {
501 ByteBuffer dup = this.buf.duplicate();
502 return dup;
503 }
504
505 private void sanityCheckAssertion(long valueFromBuf, long valueFromField,
506 String fieldName) throws IOException {
507 if (valueFromBuf != valueFromField) {
508 throw new AssertionError(fieldName + " in the buffer (" + valueFromBuf
509 + ") is different from that in the field (" + valueFromField + ")");
510 }
511 }
512
513 private void sanityCheckAssertion(BlockType valueFromBuf, BlockType valueFromField)
514 throws IOException {
515 if (valueFromBuf != valueFromField) {
516 throw new IOException("Block type stored in the buffer: " +
517 valueFromBuf + ", block type field: " + valueFromField);
518 }
519 }
520
521
522
523
524
525
526
527
528
529 @VisibleForTesting
530 void sanityCheck() throws IOException {
531
532 ByteBuffer dup = this.buf.duplicate();
533 dup.rewind();
534 sanityCheckAssertion(BlockType.read(dup), blockType);
535 sanityCheckAssertion(dup.getInt(), onDiskSizeWithoutHeader, "onDiskSizeWithoutHeader");
536 sanityCheckAssertion(dup.getInt(), uncompressedSizeWithoutHeader,
537 "uncompressedSizeWithoutHeader");
538 sanityCheckAssertion(dup.getLong(), prevBlockOffset, "prevBlockOffset");
539 if (this.fileContext.isUseHBaseChecksum()) {
540 sanityCheckAssertion(dup.get(), this.fileContext.getChecksumType().getCode(), "checksumType");
541 sanityCheckAssertion(dup.getInt(), this.fileContext.getBytesPerChecksum(),
542 "bytesPerChecksum");
543 sanityCheckAssertion(dup.getInt(), onDiskDataSizeWithHeader, "onDiskDataSizeWithHeader");
544 }
545
546 int cksumBytes = totalChecksumBytes();
547 int expectedBufLimit = onDiskDataSizeWithHeader + cksumBytes;
548 if (dup.limit() != expectedBufLimit) {
549 throw new AssertionError("Expected limit " + expectedBufLimit + ", got " + dup.limit());
550 }
551
552
553
554 int hdrSize = headerSize();
555 if (dup.capacity() != expectedBufLimit && dup.capacity() != expectedBufLimit + hdrSize) {
556 throw new AssertionError("Invalid buffer capacity: " + dup.capacity() +
557 ", expected " + expectedBufLimit + " or " + (expectedBufLimit + hdrSize));
558 }
559 }
560
561 @Override
562 public String toString() {
563 StringBuilder sb = new StringBuilder()
564 .append("[")
565 .append("blockType=").append(blockType)
566 .append(", fileOffset=").append(offset)
567 .append(", headerSize=").append(headerSize())
568 .append(", onDiskSizeWithoutHeader=").append(onDiskSizeWithoutHeader)
569 .append(", uncompressedSizeWithoutHeader=").append(uncompressedSizeWithoutHeader)
570 .append(", prevBlockOffset=").append(prevBlockOffset)
571 .append(", isUseHBaseChecksum=").append(fileContext.isUseHBaseChecksum());
572 if (fileContext.isUseHBaseChecksum()) {
573 sb.append(", checksumType=").append(ChecksumType.codeToType(this.buf.get(24)))
574 .append(", bytesPerChecksum=").append(this.buf.getInt(24 + 1))
575 .append(", onDiskDataSizeWithHeader=").append(onDiskDataSizeWithHeader);
576 } else {
577 sb.append(", onDiskDataSizeWithHeader=").append(onDiskDataSizeWithHeader)
578 .append("(").append(onDiskSizeWithoutHeader)
579 .append("+").append(HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM).append(")");
580 }
581 String dataBegin = null;
582 if (buf.hasArray()) {
583 dataBegin = Bytes.toStringBinary(buf.array(), buf.arrayOffset() + headerSize(),
584 Math.min(32, buf.limit() - buf.arrayOffset() - headerSize()));
585 } else {
586 ByteBuffer bufWithoutHeader = getBufferWithoutHeader();
587 byte[] dataBeginBytes = new byte[Math.min(32,
588 bufWithoutHeader.limit() - bufWithoutHeader.position())];
589 bufWithoutHeader.get(dataBeginBytes);
590 dataBegin = Bytes.toStringBinary(dataBeginBytes);
591 }
592 sb.append(", getOnDiskSizeWithHeader=").append(getOnDiskSizeWithHeader())
593 .append(", totalChecksumBytes=").append(totalChecksumBytes())
594 .append(", isUnpacked=").append(isUnpacked())
595 .append(", buf=[").append(buf).append("]")
596 .append(", dataBeginsWith=").append(dataBegin)
597 .append(", fileContext=").append(fileContext)
598 .append(", nextBlockOnDiskSize=").append(nextBlockOnDiskSize)
599 .append("]");
600 return sb.toString();
601 }
602
603
604
605
606
607 HFileBlock unpack(HFileContext fileContext, FSReader reader) throws IOException {
608 if (!fileContext.isCompressedOrEncrypted()) {
609
610
611
612 return this;
613 }
614
615 HFileBlock unpacked = new HFileBlock(this);
616 unpacked.allocateBuffer();
617
618 HFileBlockDecodingContext ctx = blockType == BlockType.ENCODED_DATA ?
619 reader.getBlockDecodingContext() : reader.getDefaultBlockDecodingContext();
620
621 ByteBuffer dup = this.buf.duplicate();
622 dup.position(this.headerSize());
623 dup = dup.slice();
624 ctx.prepareDecoding(unpacked.getOnDiskSizeWithoutHeader(),
625 unpacked.getUncompressedSizeWithoutHeader(), unpacked.getBufferWithoutHeader(),
626 dup);
627 return unpacked;
628 }
629
630
631
632
633
634
635 private void allocateBuffer() {
636 int cksumBytes = totalChecksumBytes();
637 int headerSize = headerSize();
638 int capacityNeeded = headerSize + uncompressedSizeWithoutHeader + cksumBytes;
639
640
641 ByteBuffer newBuf = ByteBuffer.allocate(capacityNeeded);
642
643
644
645 ByteBuffer dup = buf.duplicate();
646 dup.position(0);
647 dup.get(newBuf.array(), newBuf.arrayOffset(), headerSize);
648
649 buf = newBuf;
650
651 buf.limit(headerSize + uncompressedSizeWithoutHeader + cksumBytes);
652 }
653
654
655
656
657
658 public boolean isUnpacked() {
659 final int cksumBytes = totalChecksumBytes();
660 final int headerSize = headerSize();
661 final int expectedCapacity = headerSize + uncompressedSizeWithoutHeader + cksumBytes;
662 final int bufCapacity = buf.capacity();
663 return bufCapacity == expectedCapacity || bufCapacity == expectedCapacity + headerSize;
664 }
665
666
667 public void sanityCheckUncompressedSize() throws IOException {
668 if (onDiskSizeWithoutHeader != uncompressedSizeWithoutHeader + totalChecksumBytes()) {
669 throw new IOException("Using no compression but "
670 + "onDiskSizeWithoutHeader=" + onDiskSizeWithoutHeader + ", "
671 + "uncompressedSizeWithoutHeader=" + uncompressedSizeWithoutHeader
672 + ", numChecksumbytes=" + totalChecksumBytes());
673 }
674 }
675
676
677
678
679
680
681 long getOffset() {
682 if (offset < 0) {
683 throw new IllegalStateException("HFile block offset not initialized properly");
684 }
685 return offset;
686 }
687
688
689
690
691 public DataInputStream getByteStream() {
692 ByteBuffer dup = this.buf.duplicate();
693 dup.position(this.headerSize());
694 return new DataInputStream(new ByteBufferInputStream(dup));
695 }
696
697 @Override
698 public long heapSize() {
699 long size = ClassSize.align(
700 ClassSize.OBJECT +
701
702 3 * ClassSize.REFERENCE +
703
704
705 4 * Bytes.SIZEOF_INT +
706
707 2 * Bytes.SIZEOF_LONG +
708
709 fileContext.heapSize()
710 );
711
712 if (buf != null) {
713
714 size += ClassSize.align(buf.capacity() + BYTE_BUFFER_HEAP_SIZE);
715 }
716
717 return ClassSize.align(size);
718 }
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734 static boolean readWithExtra(InputStream in, byte[] buf,
735 int bufOffset, int necessaryLen, int extraLen) throws IOException {
736 int bytesRemaining = necessaryLen + extraLen;
737 while (bytesRemaining > 0) {
738 int ret = in.read(buf, bufOffset, bytesRemaining);
739 if (ret == -1 && bytesRemaining <= extraLen) {
740
741 break;
742 }
743 if (ret < 0) {
744 throw new IOException("Premature EOF from inputStream (read "
745 + "returned " + ret + ", was trying to read " + necessaryLen
746 + " necessary bytes and " + extraLen + " extra bytes, "
747 + "successfully read "
748 + (necessaryLen + extraLen - bytesRemaining));
749 }
750 bufOffset += ret;
751 bytesRemaining -= ret;
752 }
753 return bytesRemaining <= 0;
754 }
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774 @VisibleForTesting
775 static boolean positionalReadWithExtra(FSDataInputStream in,
776 long position, byte[] buf, int bufOffset, int necessaryLen, int extraLen)
777 throws IOException {
778 int bytesRemaining = necessaryLen + extraLen;
779 int bytesRead = 0;
780 while (bytesRead < necessaryLen) {
781 int ret = in.read(position, buf, bufOffset, bytesRemaining);
782 if (ret < 0) {
783 throw new IOException("Premature EOF from inputStream (positional read "
784 + "returned " + ret + ", was trying to read " + necessaryLen
785 + " necessary bytes and " + extraLen + " extra bytes, "
786 + "successfully read " + bytesRead);
787 }
788 position += ret;
789 bufOffset += ret;
790 bytesRemaining -= ret;
791 bytesRead += ret;
792 }
793 return bytesRead != necessaryLen && bytesRemaining <= 0;
794 }
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809 public static class Writer {
810 private enum State {
811 INIT,
812 WRITING,
813 BLOCK_READY
814 };
815
816
817 private State state = State.INIT;
818
819
820 private final HFileDataBlockEncoder dataBlockEncoder;
821
822 private HFileBlockEncodingContext dataBlockEncodingCtx;
823
824
825 private HFileBlockDefaultEncodingContext defaultBlockEncodingCtx;
826
827
828
829
830
831
832
833 private ByteArrayOutputStream baosInMemory;
834
835
836
837
838
839
840 private BlockType blockType;
841
842
843
844
845
846 private DataOutputStream userDataStream;
847
848
849
850 private int unencodedDataSizeWritten;
851
852
853
854 private int encodedDataSizeWritten;
855
856
857
858
859
860
861 private ByteArrayOutputStream onDiskBlockBytesWithHeader;
862
863
864
865
866
867
868
869 private byte[] onDiskChecksum = HConstants.EMPTY_BYTE_ARRAY;
870
871
872
873
874
875 private long startOffset;
876
877
878
879
880
881 private long[] prevOffsetByType;
882
883
884 private long prevOffset;
885
886 private HFileContext fileContext;
887
888
889
890
891 public Writer(HFileDataBlockEncoder dataBlockEncoder, HFileContext fileContext) {
892 if (fileContext.getBytesPerChecksum() < HConstants.HFILEBLOCK_HEADER_SIZE) {
893 throw new RuntimeException("Unsupported value of bytesPerChecksum. " +
894 " Minimum is " + HConstants.HFILEBLOCK_HEADER_SIZE + " but the configured value is " +
895 fileContext.getBytesPerChecksum());
896 }
897 this.dataBlockEncoder = dataBlockEncoder != null?
898 dataBlockEncoder: NoOpDataBlockEncoder.INSTANCE;
899 this.dataBlockEncodingCtx = this.dataBlockEncoder.
900 newDataBlockEncodingContext(HConstants.HFILEBLOCK_DUMMY_HEADER, fileContext);
901
902 this.defaultBlockEncodingCtx = new HFileBlockDefaultEncodingContext(null,
903 HConstants.HFILEBLOCK_DUMMY_HEADER, fileContext);
904
905 baosInMemory = new ByteArrayOutputStream();
906 prevOffsetByType = new long[BlockType.values().length];
907 for (int i = 0; i < prevOffsetByType.length; ++i) {
908 prevOffsetByType[i] = UNSET;
909 }
910
911
912 this.fileContext = fileContext;
913 }
914
915
916
917
918
919
920
921 DataOutputStream startWriting(BlockType newBlockType)
922 throws IOException {
923 if (state == State.BLOCK_READY && startOffset != -1) {
924
925
926 prevOffsetByType[blockType.getId()] = startOffset;
927 }
928
929 startOffset = -1;
930 blockType = newBlockType;
931
932 baosInMemory.reset();
933 baosInMemory.write(HConstants.HFILEBLOCK_DUMMY_HEADER);
934
935 state = State.WRITING;
936
937
938 userDataStream = new DataOutputStream(baosInMemory);
939 if (newBlockType == BlockType.DATA) {
940 this.dataBlockEncoder.startBlockEncoding(dataBlockEncodingCtx, userDataStream);
941 }
942 this.unencodedDataSizeWritten = 0;
943 this.encodedDataSizeWritten = 0;
944 return userDataStream;
945 }
946
947
948
949
950
951
952 void write(Cell cell) throws IOException{
953 expectState(State.WRITING);
954 int posBeforeEncode = this.userDataStream.size();
955 this.unencodedDataSizeWritten +=
956 this.dataBlockEncoder.encode(cell, dataBlockEncodingCtx, this.userDataStream);
957 this.encodedDataSizeWritten += this.userDataStream.size() - posBeforeEncode;
958 }
959
960
961
962
963
964
965
966
967 DataOutputStream getUserDataStream() {
968 expectState(State.WRITING);
969 return userDataStream;
970 }
971
972
973
974
975
976 void ensureBlockReady() throws IOException {
977 Preconditions.checkState(state != State.INIT,
978 "Unexpected state: " + state);
979
980 if (state == State.BLOCK_READY) {
981 return;
982 }
983
984
985 finishBlock();
986 }
987
988
989
990
991
992
993
994 private void finishBlock() throws IOException {
995 if (blockType == BlockType.DATA) {
996 this.dataBlockEncoder.endBlockEncoding(dataBlockEncodingCtx, userDataStream,
997 baosInMemory.getBuffer(), blockType);
998 blockType = dataBlockEncodingCtx.getBlockType();
999 }
1000 userDataStream.flush();
1001 prevOffset = prevOffsetByType[blockType.getId()];
1002
1003
1004
1005 state = State.BLOCK_READY;
1006 Bytes compressAndEncryptDat;
1007 if (blockType == BlockType.DATA || blockType == BlockType.ENCODED_DATA) {
1008 compressAndEncryptDat = dataBlockEncodingCtx.
1009 compressAndEncrypt(baosInMemory.getBuffer(),
1010 0, baosInMemory.size());
1011 } else {
1012 compressAndEncryptDat = defaultBlockEncodingCtx.
1013 compressAndEncrypt(baosInMemory.getBuffer(),
1014 0, baosInMemory.size());
1015 }
1016 if (compressAndEncryptDat == null) {
1017 compressAndEncryptDat = new Bytes(baosInMemory.getBuffer(),
1018 0, baosInMemory.size());
1019 }
1020 if (onDiskBlockBytesWithHeader == null) {
1021 onDiskBlockBytesWithHeader = new ByteArrayOutputStream(compressAndEncryptDat.getLength());
1022 }
1023 onDiskBlockBytesWithHeader.reset();
1024 onDiskBlockBytesWithHeader.write(compressAndEncryptDat.get(),
1025 compressAndEncryptDat.getOffset(), compressAndEncryptDat.getLength());
1026
1027 int numBytes = (int) ChecksumUtil.numBytes(
1028 onDiskBlockBytesWithHeader.size(),
1029 fileContext.getBytesPerChecksum());
1030
1031
1032 putHeader(onDiskBlockBytesWithHeader,
1033 onDiskBlockBytesWithHeader.size() + numBytes,
1034 baosInMemory.size(), onDiskBlockBytesWithHeader.size());
1035
1036 if (onDiskChecksum.length != numBytes) {
1037 onDiskChecksum = new byte[numBytes];
1038 }
1039 ChecksumUtil.generateChecksums(
1040 onDiskBlockBytesWithHeader.getBuffer(), 0,onDiskBlockBytesWithHeader.size(),
1041 onDiskChecksum, 0, fileContext.getChecksumType(), fileContext.getBytesPerChecksum());
1042 }
1043 private void putHeader(ByteArrayOutputStream dest, int onDiskSize,
1044 int uncompressedSize, int onDiskDataSize) {
1045 putHeader(dest.getBuffer(),0, onDiskSize, uncompressedSize, onDiskDataSize);
1046 }
1047
1048
1049
1050
1051
1052
1053
1054
1055 private void putHeader(byte[] dest, int offset, int onDiskSize,
1056 int uncompressedSize, int onDiskDataSize) {
1057 offset = blockType.put(dest, offset);
1058 offset = Bytes.putInt(dest, offset, onDiskSize - HConstants.HFILEBLOCK_HEADER_SIZE);
1059 offset = Bytes.putInt(dest, offset, uncompressedSize - HConstants.HFILEBLOCK_HEADER_SIZE);
1060 offset = Bytes.putLong(dest, offset, prevOffset);
1061 offset = Bytes.putByte(dest, offset, fileContext.getChecksumType().getCode());
1062 offset = Bytes.putInt(dest, offset, fileContext.getBytesPerChecksum());
1063 Bytes.putInt(dest, offset, onDiskDataSize);
1064 }
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074 void writeHeaderAndData(FSDataOutputStream out) throws IOException {
1075 long offset = out.getPos();
1076 if (startOffset != UNSET && offset != startOffset) {
1077 throw new IOException("A " + blockType + " block written to a "
1078 + "stream twice, first at offset " + startOffset + ", then at "
1079 + offset);
1080 }
1081 startOffset = offset;
1082
1083 finishBlockAndWriteHeaderAndData((DataOutputStream) out);
1084 }
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095 protected void finishBlockAndWriteHeaderAndData(DataOutputStream out)
1096 throws IOException {
1097 ensureBlockReady();
1098 long startTime = System.currentTimeMillis();
1099 out.write(onDiskBlockBytesWithHeader.getBuffer(), 0, onDiskBlockBytesWithHeader.size());
1100 out.write(onDiskChecksum);
1101 HFile.updateWriteLatency(System.currentTimeMillis() - startTime);
1102 }
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114 byte[] getHeaderAndDataForTest() throws IOException {
1115 ensureBlockReady();
1116
1117
1118 byte[] output =
1119 new byte[onDiskBlockBytesWithHeader.size()
1120 + onDiskChecksum.length];
1121 System.arraycopy(onDiskBlockBytesWithHeader.getBuffer(), 0, output, 0,
1122 onDiskBlockBytesWithHeader.size());
1123 System.arraycopy(onDiskChecksum, 0, output,
1124 onDiskBlockBytesWithHeader.size(), onDiskChecksum.length);
1125 return output;
1126 }
1127
1128
1129
1130
1131 void release() {
1132 if (dataBlockEncodingCtx != null) {
1133 dataBlockEncodingCtx.close();
1134 dataBlockEncodingCtx = null;
1135 }
1136 if (defaultBlockEncodingCtx != null) {
1137 defaultBlockEncodingCtx.close();
1138 defaultBlockEncodingCtx = null;
1139 }
1140 }
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150 int getOnDiskSizeWithoutHeader() {
1151 expectState(State.BLOCK_READY);
1152 return onDiskBlockBytesWithHeader.size() +
1153 onDiskChecksum.length - HConstants.HFILEBLOCK_HEADER_SIZE;
1154 }
1155
1156
1157
1158
1159
1160
1161
1162
1163 int getOnDiskSizeWithHeader() {
1164 expectState(State.BLOCK_READY);
1165 return onDiskBlockBytesWithHeader.size() + onDiskChecksum.length;
1166 }
1167
1168
1169
1170
1171 int getUncompressedSizeWithoutHeader() {
1172 expectState(State.BLOCK_READY);
1173 return baosInMemory.size() - HConstants.HFILEBLOCK_HEADER_SIZE;
1174 }
1175
1176
1177
1178
1179 int getUncompressedSizeWithHeader() {
1180 expectState(State.BLOCK_READY);
1181 return baosInMemory.size();
1182 }
1183
1184
1185 boolean isWriting() {
1186 return state == State.WRITING;
1187 }
1188
1189
1190
1191
1192
1193
1194
1195
1196 public int encodedBlockSizeWritten() {
1197 if (state != State.WRITING)
1198 return 0;
1199 return this.encodedDataSizeWritten;
1200 }
1201
1202
1203
1204
1205
1206
1207
1208
1209 int blockSizeWritten() {
1210 if (state != State.WRITING) return 0;
1211 return this.unencodedDataSizeWritten;
1212 }
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222 @VisibleForTesting
1223 ByteBuffer cloneUncompressedBufferWithHeader() {
1224 expectState(State.BLOCK_READY);
1225 byte[] uncompressedBlockBytesWithHeader = baosInMemory.toByteArray();
1226 int numBytes = (int) ChecksumUtil.numBytes(
1227 onDiskBlockBytesWithHeader.size(),
1228 fileContext.getBytesPerChecksum());
1229 putHeader(uncompressedBlockBytesWithHeader, 0,
1230 onDiskBlockBytesWithHeader.size() + numBytes,
1231 uncompressedBlockBytesWithHeader.length, onDiskBlockBytesWithHeader.size());
1232 return ByteBuffer.wrap(uncompressedBlockBytesWithHeader);
1233 }
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243 private ByteBuffer cloneOnDiskBufferWithHeader() {
1244 expectState(State.BLOCK_READY);
1245 return ByteBuffer.wrap(onDiskBlockBytesWithHeader.toByteArray());
1246 }
1247
1248 private void expectState(State expectedState) {
1249 if (state != expectedState) {
1250 throw new IllegalStateException("Expected state: " + expectedState +
1251 ", actual state: " + state);
1252 }
1253 }
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265 void writeBlock(BlockWritable bw, FSDataOutputStream out)
1266 throws IOException {
1267 bw.writeToBlock(startWriting(bw.getBlockType()));
1268 writeHeaderAndData(out);
1269 }
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284 HFileBlock getBlockForCaching(CacheConfig cacheConf) {
1285 HFileContext newContext = new HFileContextBuilder()
1286 .withBlockSize(fileContext.getBlocksize())
1287 .withBytesPerCheckSum(0)
1288 .withChecksumType(ChecksumType.NULL)
1289 .withCompression(fileContext.getCompression())
1290 .withDataBlockEncoding(fileContext.getDataBlockEncoding())
1291 .withHBaseCheckSum(fileContext.isUseHBaseChecksum())
1292 .withCompressTags(fileContext.isCompressTags())
1293 .withIncludesMvcc(fileContext.isIncludesMvcc())
1294 .withIncludesTags(fileContext.isIncludesTags())
1295 .build();
1296 return new HFileBlock(blockType, getOnDiskSizeWithoutHeader(),
1297 getUncompressedSizeWithoutHeader(), prevOffset,
1298 cacheConf.shouldCacheCompressed(blockType.getCategory())?
1299 cloneOnDiskBufferWithHeader() :
1300 cloneUncompressedBufferWithHeader(),
1301 FILL_HEADER, startOffset, UNSET,
1302 onDiskBlockBytesWithHeader.size() + onDiskChecksum.length, newContext);
1303 }
1304 }
1305
1306
1307 interface BlockWritable {
1308
1309
1310 BlockType getBlockType();
1311
1312
1313
1314
1315
1316
1317
1318 void writeToBlock(DataOutput out) throws IOException;
1319 }
1320
1321
1322
1323
1324 interface BlockIterator {
1325
1326
1327
1328
1329 HFileBlock nextBlock() throws IOException;
1330
1331
1332
1333
1334
1335 HFileBlock nextBlockWithBlockType(BlockType blockType) throws IOException;
1336 }
1337
1338
1339 interface FSReader {
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350 HFileBlock readBlockData(long offset, long onDiskSize, boolean pread, boolean updateMetrics)
1351 throws IOException;
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362 BlockIterator blockRange(long startOffset, long endOffset);
1363
1364
1365 void closeStreams() throws IOException;
1366
1367
1368 HFileBlockDecodingContext getBlockDecodingContext();
1369
1370
1371 HFileBlockDecodingContext getDefaultBlockDecodingContext();
1372
1373
1374
1375
1376
1377 void unbufferStream();
1378 }
1379
1380
1381
1382
1383
1384 private abstract static class AbstractFSReader implements FSReader {
1385
1386
1387
1388 protected long fileSize;
1389
1390
1391 protected final int hdrSize;
1392
1393
1394 protected HFileSystem hfs;
1395
1396 protected final Lock streamLock = new ReentrantLock();
1397
1398
1399 public static final int DEFAULT_BUFFER_SIZE = 1 << 20;
1400
1401 protected HFileContext fileContext;
1402
1403 protected String pathName;
1404
1405 public AbstractFSReader(long fileSize, HFileSystem hfs, Path path, HFileContext fileContext)
1406 throws IOException {
1407 this.fileSize = fileSize;
1408 this.hfs = hfs;
1409 if (path != null) {
1410 this.pathName = path.toString();
1411 }
1412 this.fileContext = fileContext;
1413 this.hdrSize = headerSize(fileContext.isUseHBaseChecksum());
1414 }
1415
1416 @Override
1417 public BlockIterator blockRange(final long startOffset,
1418 final long endOffset) {
1419 final FSReader owner = this;
1420 return new BlockIterator() {
1421 private long offset = startOffset;
1422
1423 private long length = -1;
1424
1425 @Override
1426 public HFileBlock nextBlock() throws IOException {
1427 if (offset >= endOffset) {
1428 return null;
1429 }
1430 HFileBlock b = readBlockData(offset, length, false, false);
1431 offset += b.getOnDiskSizeWithHeader();
1432 length = b.getNextBlockOnDiskSize();
1433 return b.unpack(fileContext, owner);
1434 }
1435
1436 @Override
1437 public HFileBlock nextBlockWithBlockType(BlockType blockType)
1438 throws IOException {
1439 HFileBlock blk = nextBlock();
1440 if (blk.getBlockType() != blockType) {
1441 throw new IOException("Expected block of type " + blockType
1442 + " but found " + blk.getBlockType());
1443 }
1444 return blk;
1445 }
1446 };
1447 }
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465 protected int readAtOffset(FSDataInputStream istream, byte [] dest, int destOffset, int size,
1466 boolean peekIntoNextBlock, long fileOffset, boolean pread) throws IOException {
1467 if (peekIntoNextBlock && destOffset + size + hdrSize > dest.length) {
1468
1469
1470 throw new IOException("Attempted to read " + size + " bytes and " +
1471 hdrSize + " bytes of next header into a " + dest.length +
1472 "-byte array at offset " + destOffset);
1473 }
1474
1475 if (!pread && streamLock.tryLock()) {
1476
1477 try {
1478 HFileUtil.seekOnMultipleSources(istream, fileOffset);
1479
1480 long realOffset = istream.getPos();
1481 if (realOffset != fileOffset) {
1482 throw new IOException("Tried to seek to " + fileOffset + " to "
1483 + "read " + size + " bytes, but pos=" + realOffset
1484 + " after seek");
1485 }
1486
1487 if (!peekIntoNextBlock) {
1488 IOUtils.readFully(istream, dest, destOffset, size);
1489 return -1;
1490 }
1491
1492
1493 if (!readWithExtra(istream, dest, destOffset, size, hdrSize)) {
1494 return -1;
1495 }
1496 } finally {
1497 streamLock.unlock();
1498 }
1499 } else {
1500
1501 int extraSize = peekIntoNextBlock ? hdrSize : 0;
1502 if (!positionalReadWithExtra(istream, fileOffset, dest, destOffset, size, extraSize)) {
1503 return -1;
1504 }
1505 }
1506
1507 assert peekIntoNextBlock;
1508 return Bytes.toInt(dest, destOffset + size + BlockType.MAGIC_LENGTH) + hdrSize;
1509 }
1510
1511 }
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523 private static class PrefetchedHeader {
1524 long offset = -1;
1525 byte [] header = new byte[HConstants.HFILEBLOCK_HEADER_SIZE];
1526 final ByteBuffer buf = ByteBuffer.wrap(header, 0, HConstants.HFILEBLOCK_HEADER_SIZE);
1527
1528 @Override
1529 public String toString() {
1530 return "offset=" + this.offset + ", header=" + Bytes.toStringBinary(header);
1531 }
1532 }
1533
1534
1535
1536
1537 static class FSReaderImpl extends AbstractFSReader {
1538
1539
1540 protected FSDataInputStreamWrapper streamWrapper;
1541
1542 private HFileBlockDecodingContext encodedBlockDecodingCtx;
1543
1544
1545 private final HFileBlockDefaultDecodingContext defaultDecodingCtx;
1546
1547
1548
1549
1550
1551
1552
1553 private AtomicReference<PrefetchedHeader> prefetchedHeader =
1554 new AtomicReference<PrefetchedHeader>(new PrefetchedHeader());
1555
1556 public FSReaderImpl(FSDataInputStreamWrapper stream, long fileSize, HFileSystem hfs, Path path,
1557 HFileContext fileContext) throws IOException {
1558 super(fileSize, hfs, path, fileContext);
1559 this.streamWrapper = stream;
1560
1561 this.streamWrapper.prepareForBlockReader(!fileContext.isUseHBaseChecksum());
1562 defaultDecodingCtx = new HFileBlockDefaultDecodingContext(fileContext);
1563 encodedBlockDecodingCtx = defaultDecodingCtx;
1564 }
1565
1566
1567
1568
1569
1570 FSReaderImpl(FSDataInputStream istream, long fileSize, HFileContext fileContext)
1571 throws IOException {
1572 this(new FSDataInputStreamWrapper(istream), fileSize, null, null, fileContext);
1573 }
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585 @Override
1586 public HFileBlock readBlockData(long offset, long onDiskSizeWithHeaderL, boolean pread,
1587 boolean updateMetrics) throws IOException {
1588
1589
1590
1591
1592
1593 boolean doVerificationThruHBaseChecksum = streamWrapper.shouldUseHBaseChecksum();
1594 FSDataInputStream is = streamWrapper.getStream(doVerificationThruHBaseChecksum);
1595
1596 HFileBlock blk = readBlockDataInternal(is, offset,
1597 onDiskSizeWithHeaderL, pread,
1598 doVerificationThruHBaseChecksum, updateMetrics);
1599 if (blk == null) {
1600 HFile.LOG.warn("HBase checksum verification failed for file " +
1601 pathName + " at offset " +
1602 offset + " filesize " + fileSize +
1603 ". Retrying read with HDFS checksums turned on...");
1604
1605 if (!doVerificationThruHBaseChecksum) {
1606 String msg = "HBase checksum verification failed for file " +
1607 pathName + " at offset " +
1608 offset + " filesize " + fileSize +
1609 " but this cannot happen because doVerify is " +
1610 doVerificationThruHBaseChecksum;
1611 HFile.LOG.warn(msg);
1612 throw new IOException(msg);
1613 }
1614 HFile.CHECKSUM_FAILURES.increment();
1615
1616
1617
1618
1619
1620
1621
1622 is = this.streamWrapper.fallbackToFsChecksum(CHECKSUM_VERIFICATION_NUM_IO_THRESHOLD);
1623 doVerificationThruHBaseChecksum = false;
1624 blk = readBlockDataInternal(is, offset, onDiskSizeWithHeaderL, pread,
1625 doVerificationThruHBaseChecksum, updateMetrics);
1626 if (blk != null) {
1627 HFile.LOG.warn("HDFS checksum verification succeeded for file " +
1628 pathName + " at offset " +
1629 offset + " filesize " + fileSize);
1630 }
1631 }
1632 if (blk == null && !doVerificationThruHBaseChecksum) {
1633 String msg = "readBlockData failed, possibly due to " +
1634 "checksum verification failed for file " + pathName +
1635 " at offset " + offset + " filesize " + fileSize;
1636 HFile.LOG.warn(msg);
1637 throw new IOException(msg);
1638 }
1639
1640
1641
1642
1643
1644
1645
1646
1647 streamWrapper.checksumOk();
1648 return blk;
1649 }
1650
1651
1652
1653
1654
1655 private static int checkAndGetSizeAsInt(final long onDiskSizeWithHeaderL, final int hdrSize)
1656 throws IOException {
1657 if ((onDiskSizeWithHeaderL < hdrSize && onDiskSizeWithHeaderL != -1)
1658 || onDiskSizeWithHeaderL >= Integer.MAX_VALUE) {
1659 throw new IOException("Invalid onDisksize=" + onDiskSizeWithHeaderL
1660 + ": expected to be at least " + hdrSize
1661 + " and at most " + Integer.MAX_VALUE + ", or -1");
1662 }
1663 return (int)onDiskSizeWithHeaderL;
1664 }
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676 private ByteBuffer getCachedHeader(final long offset) {
1677 PrefetchedHeader ph = this.prefetchedHeader.get();
1678 return ph != null && ph.offset == offset? ph.buf: null;
1679 }
1680
1681
1682
1683
1684
1685
1686 private void cacheNextBlockHeader(final long offset,
1687 final byte [] header, final int headerOffset, final int headerLength) {
1688 PrefetchedHeader ph = new PrefetchedHeader();
1689 ph.offset = offset;
1690 System.arraycopy(header, headerOffset, ph.header, 0, headerLength);
1691 this.prefetchedHeader.set(ph);
1692 }
1693
1694
1695
1696
1697
1698
1699 private void verifyOnDiskSizeMatchesHeader(final int passedIn, final ByteBuffer headerBuf,
1700 final long offset, boolean verifyChecksum)
1701 throws IOException {
1702
1703 int fromHeader = getOnDiskSizeWithHeader(headerBuf, verifyChecksum);
1704 if (passedIn != fromHeader) {
1705 throw new IOException("Passed in onDiskSizeWithHeader=" + passedIn + " != " + fromHeader +
1706 ", offset=" + offset + ", fileContext=" + this.fileContext);
1707 }
1708 }
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724 protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
1725 long onDiskSizeWithHeaderL, boolean pread, boolean verifyChecksum, boolean updateMetrics)
1726 throws IOException {
1727 if (offset < 0) {
1728 throw new IOException("Invalid offset=" + offset + " trying to read "
1729 + "block (onDiskSize=" + onDiskSizeWithHeaderL + ")");
1730 }
1731 int onDiskSizeWithHeader = checkAndGetSizeAsInt(onDiskSizeWithHeaderL, hdrSize);
1732
1733
1734
1735 ByteBuffer headerBuf = getCachedHeader(offset);
1736 if (LOG.isTraceEnabled()) {
1737 LOG.trace("Reading " + this.fileContext.getHFileName() + " at offset=" + offset +
1738 ", pread=" + pread + ", verifyChecksum=" + verifyChecksum + ", cachedHeader=" +
1739 headerBuf + ", onDiskSizeWithHeader=" + onDiskSizeWithHeader);
1740 }
1741 long startTime = System.currentTimeMillis();
1742 if (onDiskSizeWithHeader <= 0) {
1743
1744
1745
1746
1747
1748
1749 if (headerBuf == null) {
1750 if (LOG.isTraceEnabled()) {
1751 LOG.trace("Extra see to get block size!", new RuntimeException());
1752 }
1753 headerBuf = ByteBuffer.allocate(hdrSize);
1754 readAtOffset(is, headerBuf.array(), headerBuf.arrayOffset(), hdrSize, false,
1755 offset, pread);
1756 }
1757 onDiskSizeWithHeader = getOnDiskSizeWithHeader(headerBuf,
1758 this.fileContext.isUseHBaseChecksum());
1759 }
1760
1761 int preReadHeaderSize = headerBuf == null? 0 : hdrSize;
1762
1763
1764
1765
1766
1767
1768
1769 byte [] onDiskBlock = new byte[onDiskSizeWithHeader + hdrSize];
1770 int nextBlockOnDiskSize = readAtOffset(is, onDiskBlock, preReadHeaderSize,
1771 onDiskSizeWithHeader - preReadHeaderSize, true, offset + preReadHeaderSize, pread);
1772 if (headerBuf != null) {
1773
1774
1775 System.arraycopy(headerBuf.array(), headerBuf.arrayOffset(), onDiskBlock, 0, hdrSize);
1776 } else {
1777 headerBuf = ByteBuffer.wrap(onDiskBlock, 0, hdrSize);
1778 }
1779
1780
1781 assert onDiskSizeWithHeader > this.hdrSize;
1782 verifyOnDiskSizeMatchesHeader(onDiskSizeWithHeader, headerBuf, offset,
1783 this.fileContext.isUseHBaseChecksum());
1784 ByteBuffer onDiskBlockByteBuffer = ByteBuffer.wrap(onDiskBlock, 0, onDiskSizeWithHeader);
1785
1786 if (verifyChecksum &&
1787 !validateChecksum(offset, onDiskBlockByteBuffer, hdrSize)) {
1788 return null;
1789 }
1790 long duration = System.currentTimeMillis() - startTime;
1791 if (updateMetrics) {
1792 HFile.updateReadLatency(duration, pread);
1793 }
1794
1795
1796
1797 HFileBlock hFileBlock =
1798 new HFileBlock(onDiskBlockByteBuffer, this.fileContext.isUseHBaseChecksum(), offset,
1799 nextBlockOnDiskSize, fileContext);
1800
1801 if (!fileContext.isCompressedOrEncrypted()) {
1802 hFileBlock.sanityCheckUncompressed();
1803 }
1804 if (LOG.isTraceEnabled()) {
1805 LOG.trace("Read " + hFileBlock + " in " + duration + " ns");
1806 }
1807
1808 if (nextBlockOnDiskSize != -1) {
1809 cacheNextBlockHeader(offset + hFileBlock.getOnDiskSizeWithHeader(),
1810 onDiskBlock, onDiskSizeWithHeader, hdrSize);
1811 }
1812 return hFileBlock;
1813 }
1814
1815 void setIncludesMemstoreTS(boolean includesMemstoreTS) {
1816 this.fileContext.setIncludesMvcc(includesMemstoreTS);
1817 }
1818
1819 void setDataBlockEncoder(HFileDataBlockEncoder encoder) {
1820 encodedBlockDecodingCtx = encoder.newDataBlockDecodingContext(this.fileContext);
1821 }
1822
1823 @Override
1824 public HFileBlockDecodingContext getBlockDecodingContext() {
1825 return this.encodedBlockDecodingCtx;
1826 }
1827
1828 @Override
1829 public HFileBlockDecodingContext getDefaultBlockDecodingContext() {
1830 return this.defaultDecodingCtx;
1831 }
1832
1833
1834
1835
1836
1837
1838 protected boolean validateChecksum(long offset, ByteBuffer data, int hdrSize)
1839 throws IOException {
1840
1841
1842
1843
1844
1845 if (!fileContext.isUseHBaseChecksum()) {
1846 return false;
1847 }
1848 return ChecksumUtil.validateChecksum(data, pathName, offset, hdrSize);
1849 }
1850
1851 @Override
1852 public void closeStreams() throws IOException {
1853 streamWrapper.close();
1854 }
1855
1856 @Override
1857 public void unbufferStream() {
1858
1859
1860 if (streamLock.tryLock()) {
1861 try {
1862 this.streamWrapper.unbuffer();
1863 } finally {
1864 streamLock.unlock();
1865 }
1866 }
1867 }
1868
1869 @Override
1870 public String toString() {
1871 return "hfs=" + hfs + ", path=" + pathName + ", fileContext=" + fileContext;
1872 }
1873 }
1874
1875
1876 void sanityCheckUncompressed() throws IOException {
1877 if (onDiskSizeWithoutHeader != uncompressedSizeWithoutHeader +
1878 totalChecksumBytes()) {
1879 throw new IOException("Using no compression but "
1880 + "onDiskSizeWithoutHeader=" + onDiskSizeWithoutHeader + ", "
1881 + "uncompressedSizeWithoutHeader=" + uncompressedSizeWithoutHeader
1882 + ", numChecksumbytes=" + totalChecksumBytes());
1883 }
1884 }
1885
1886
1887 @Override
1888 public int getSerializedLength() {
1889 if (buf != null) {
1890
1891 return this.buf.limit() + BLOCK_METADATA_SPACE;
1892 }
1893 return 0;
1894 }
1895
1896
1897 @Override
1898 public void serialize(ByteBuffer destination, boolean includeNextBlockOnDiskSize) {
1899 ByteBufferUtils.copyFromBufferToBuffer(destination, this.buf, 0,
1900 getSerializedLength() - BLOCK_METADATA_SPACE);
1901 destination = addMetaData(destination, includeNextBlockOnDiskSize);
1902
1903
1904
1905
1906 destination.flip();
1907 }
1908
1909
1910
1911
1912 public ByteBuffer getMetaData() {
1913 ByteBuffer bb = ByteBuffer.allocate(BLOCK_METADATA_SPACE);
1914 bb = addMetaData(bb, true);
1915 bb.flip();
1916 return bb;
1917 }
1918
1919
1920
1921
1922
1923 private ByteBuffer addMetaData(final ByteBuffer destination, boolean includeNextBlockMetadata) {
1924 destination.put(this.fileContext.isUseHBaseChecksum() ? (byte) 1 : (byte) 0);
1925 destination.putLong(this.offset);
1926 if (includeNextBlockMetadata) {
1927 destination.putInt(this.nextBlockOnDiskSize);
1928 }
1929 return destination;
1930 }
1931
1932
1933 @Override
1934 public CacheableDeserializer<Cacheable> getDeserializer() {
1935 return HFileBlock.BLOCK_DESERIALIZER;
1936 }
1937
1938 @Override
1939 public int hashCode() {
1940 final int prime = 31;
1941 int result = 1;
1942 result = prime * result + ((blockType == null) ? 0 : blockType.hashCode());
1943 result = prime * result + ((buf == null) ? 0 : buf.hashCode());
1944 result = prime * result + ((fileContext == null) ? 0 : fileContext.hashCode());
1945 result = prime * result + nextBlockOnDiskSize;
1946 result = prime * result + (int) (offset ^ (offset >>> 32));
1947 result = prime * result + onDiskDataSizeWithHeader;
1948 result = prime * result + onDiskSizeWithoutHeader;
1949 result = prime * result + (int) (prevBlockOffset ^ (prevBlockOffset >>> 32));
1950 result = prime * result + uncompressedSizeWithoutHeader;
1951 return result;
1952 }
1953
1954 @Override
1955 public boolean equals(Object comparison) {
1956 if (this == comparison) {
1957 return true;
1958 }
1959 if (comparison == null) {
1960 return false;
1961 }
1962 if (comparison.getClass() != this.getClass()) {
1963 return false;
1964 }
1965
1966 HFileBlock castedComparison = (HFileBlock) comparison;
1967
1968 if (castedComparison.blockType != this.blockType) {
1969 return false;
1970 }
1971 if (castedComparison.nextBlockOnDiskSize != this.nextBlockOnDiskSize) {
1972 return false;
1973 }
1974
1975 if (castedComparison.offset != this.offset) {
1976 return false;
1977 }
1978 if (castedComparison.onDiskSizeWithoutHeader != this.onDiskSizeWithoutHeader) {
1979 return false;
1980 }
1981 if (castedComparison.prevBlockOffset != this.prevBlockOffset) {
1982 return false;
1983 }
1984 if (castedComparison.uncompressedSizeWithoutHeader != this.uncompressedSizeWithoutHeader) {
1985 return false;
1986 }
1987 if (ByteBufferUtils.compareTo(this.buf, 0, this.buf.limit(), castedComparison.buf, 0,
1988 castedComparison.buf.limit()) != 0) {
1989 return false;
1990 }
1991 return true;
1992 }
1993
1994 public DataBlockEncoding getDataBlockEncoding() {
1995 if (blockType == BlockType.ENCODED_DATA) {
1996 return DataBlockEncoding.getEncodingById(getDataBlockEncodingId());
1997 }
1998 return DataBlockEncoding.NONE;
1999 }
2000
2001 byte getChecksumType() {
2002 return this.fileContext.getChecksumType().getCode();
2003 }
2004
2005 int getBytesPerChecksum() {
2006 return this.fileContext.getBytesPerChecksum();
2007 }
2008
2009
2010 int getOnDiskDataSizeWithHeader() {
2011 return this.onDiskDataSizeWithHeader;
2012 }
2013
2014
2015
2016
2017
2018 int totalChecksumBytes() {
2019
2020
2021
2022
2023 if (!fileContext.isUseHBaseChecksum() || this.fileContext.getBytesPerChecksum() == 0) {
2024 return 0;
2025 }
2026 return (int) ChecksumUtil.numBytes(onDiskDataSizeWithHeader,
2027 this.fileContext.getBytesPerChecksum());
2028 }
2029
2030
2031
2032
2033 public int headerSize() {
2034 return headerSize(this.fileContext.isUseHBaseChecksum());
2035 }
2036
2037
2038
2039
2040 public static int headerSize(boolean usesHBaseChecksum) {
2041 return usesHBaseChecksum?
2042 HConstants.HFILEBLOCK_HEADER_SIZE: HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM;
2043 }
2044
2045
2046
2047
2048 byte[] getDummyHeaderForVersion() {
2049 return getDummyHeaderForVersion(this.fileContext.isUseHBaseChecksum());
2050 }
2051
2052
2053
2054
2055 static private byte[] getDummyHeaderForVersion(boolean usesHBaseChecksum) {
2056 return usesHBaseChecksum? HConstants.HFILEBLOCK_DUMMY_HEADER: DUMMY_HEADER_NO_CHECKSUM;
2057 }
2058
2059
2060
2061
2062
2063 HFileContext getHFileContext() {
2064 return this.fileContext;
2065 }
2066
2067
2068
2069
2070
2071
2072 @VisibleForTesting
2073 static String toStringHeader(ByteBuffer buf) throws IOException {
2074 byte[] magicBuf = new byte[Math.min(buf.limit() - buf.position(), BlockType.MAGIC_LENGTH)];
2075 buf.get(magicBuf);
2076 BlockType bt = BlockType.parse(magicBuf, 0, BlockType.MAGIC_LENGTH);
2077 int compressedBlockSizeNoHeader = buf.getInt();
2078 int uncompressedBlockSizeNoHeader = buf.getInt();
2079 long prevBlockOffset = buf.getLong();
2080 byte cksumtype = buf.get();
2081 long bytesPerChecksum = buf.getInt();
2082 long onDiskDataSizeWithHeader = buf.getInt();
2083 return " Header dump: magic: " + Bytes.toString(magicBuf) +
2084 " blockType " + bt +
2085 " compressedBlockSizeNoHeader " +
2086 compressedBlockSizeNoHeader +
2087 " uncompressedBlockSizeNoHeader " +
2088 uncompressedBlockSizeNoHeader +
2089 " prevBlockOffset " + prevBlockOffset +
2090 " checksumType " + ChecksumType.codeToType(cksumtype) +
2091 " bytesPerChecksum " + bytesPerChecksum +
2092 " onDiskDataSizeWithHeader " + onDiskDataSizeWithHeader;
2093 }
2094 }