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