001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.io.hfile;
019
020import static org.apache.hadoop.hbase.io.ByteBuffAllocator.HEAP;
021import java.io.DataInputStream;
022import java.io.DataOutput;
023import java.io.DataOutputStream;
024import java.io.IOException;
025import java.nio.ByteBuffer;
026import java.util.ArrayList;
027import java.util.List;
028import java.util.concurrent.atomic.AtomicReference;
029import java.util.concurrent.locks.Lock;
030import java.util.concurrent.locks.ReentrantLock;
031import org.apache.hadoop.fs.FSDataInputStream;
032import org.apache.hadoop.fs.FSDataOutputStream;
033import org.apache.hadoop.hbase.Cell;
034import org.apache.hadoop.hbase.HConstants;
035import org.apache.hadoop.hbase.fs.HFileSystem;
036import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
037import org.apache.hadoop.hbase.io.ByteBuffAllocator;
038import org.apache.hadoop.hbase.io.ByteBuffInputStream;
039import org.apache.hadoop.hbase.io.ByteBufferWriterDataOutputStream;
040import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper;
041import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
042import org.apache.hadoop.hbase.io.encoding.EncodingState;
043import org.apache.hadoop.hbase.io.encoding.HFileBlockDecodingContext;
044import org.apache.hadoop.hbase.io.encoding.HFileBlockDefaultDecodingContext;
045import org.apache.hadoop.hbase.io.encoding.HFileBlockDefaultEncodingContext;
046import org.apache.hadoop.hbase.io.encoding.HFileBlockEncodingContext;
047import org.apache.hadoop.hbase.io.util.BlockIOUtils;
048import org.apache.hadoop.hbase.nio.ByteBuff;
049import org.apache.hadoop.hbase.nio.MultiByteBuff;
050import org.apache.hadoop.hbase.nio.SingleByteBuff;
051import org.apache.hadoop.hbase.regionserver.ShipperListener;
052import org.apache.hadoop.hbase.util.Bytes;
053import org.apache.hadoop.hbase.util.ChecksumType;
054import org.apache.hadoop.hbase.util.ClassSize;
055import org.apache.yetus.audience.InterfaceAudience;
056import org.slf4j.Logger;
057import org.slf4j.LoggerFactory;
058import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
059import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
060
061/**
062 * Cacheable Blocks of an {@link HFile} version 2 file.
063 * Version 2 was introduced in hbase-0.92.0.
064 *
065 * <p>Version 1 was the original file block. Version 2 was introduced when we changed the hbase file
066 * format to support multi-level block indexes and compound bloom filters (HBASE-3857). Support
067 * for Version 1 was removed in hbase-1.3.0.
068 *
069 * <h3>HFileBlock: Version 2</h3>
070 * In version 2, a block is structured as follows:
071 * <ul>
072 * <li><b>Header:</b> See Writer#putHeader() for where header is written; header total size is
073 * HFILEBLOCK_HEADER_SIZE
074 * <ul>
075 * <li>0. blockType: Magic record identifying the {@link BlockType} (8 bytes):
076 * e.g. <code>DATABLK*</code>
077 * <li>1. onDiskSizeWithoutHeader: Compressed -- a.k.a 'on disk' -- block size, excluding header,
078 * but including tailing checksum bytes (4 bytes)
079 * <li>2. uncompressedSizeWithoutHeader: Uncompressed block size, excluding header, and excluding
080 * checksum bytes (4 bytes)
081 * <li>3. prevBlockOffset: The offset of the previous block of the same type (8 bytes). This is
082 * used to navigate to the previous block without having to go to the block index
083 * <li>4: For minorVersions &gt;=1, the ordinal describing checksum type (1 byte)
084 * <li>5: For minorVersions &gt;=1, the number of data bytes/checksum chunk (4 bytes)
085 * <li>6: onDiskDataSizeWithHeader: For minorVersions &gt;=1, the size of data 'on disk', including
086 * header, excluding checksums (4 bytes)
087 * </ul>
088 * </li>
089 * <li><b>Raw/Compressed/Encrypted/Encoded data:</b> The compression
090 * algorithm is the same for all the blocks in an {@link HFile}. If compression is NONE, this is
091 * just raw, serialized Cells.
092 * <li><b>Tail:</b> For minorVersions &gt;=1, a series of 4 byte checksums, one each for
093 * the number of bytes specified by bytesPerChecksum.
094 * </ul>
095 *
096 * <h3>Caching</h3>
097 * Caches cache whole blocks with trailing checksums if any. We then tag on some metadata, the
098 * content of BLOCK_METADATA_SPACE which will be flag on if we are doing 'hbase'
099 * checksums and then the offset into the file which is needed when we re-make a cache key
100 * when we return the block to the cache as 'done'.
101 * See {@link Cacheable#serialize(ByteBuffer, boolean)} and {@link Cacheable#getDeserializer()}.
102 *
103 * <p>TODO: Should we cache the checksums? Down in Writer#getBlockForCaching(CacheConfig) where
104 * we make a block to cache-on-write, there is an attempt at turning off checksums. This is not the
105 * only place we get blocks to cache. We also will cache the raw return from an hdfs read. In this
106 * case, the checksums may be present. If the cache is backed by something that doesn't do ECC,
107 * say an SSD, we might want to preserve checksums. For now this is open question.
108 * <p>TODO: Over in BucketCache, we save a block allocation by doing a custom serialization.
109 * Be sure to change it if serialization changes in here. Could we add a method here that takes an
110 * IOEngine and that then serializes to it rather than expose our internals over in BucketCache?
111 * IOEngine is in the bucket subpackage. Pull it up? Then this class knows about bucketcache. Ugh.
112 */
113@InterfaceAudience.Private
114public class HFileBlock implements Cacheable {
115  private static final Logger LOG = LoggerFactory.getLogger(HFileBlock.class);
116  public static final int FIXED_OVERHEAD = ClassSize.align(ClassSize.OBJECT +
117     // BlockType, ByteBuff, MemoryType, HFileContext, ByteBuffAllocator
118      5 * ClassSize.REFERENCE +
119      // On-disk size, uncompressed size, and next block's on-disk size
120      // bytePerChecksum and onDiskDataSize
121      4 * Bytes.SIZEOF_INT +
122      // This and previous block offset
123      2 * Bytes.SIZEOF_LONG);
124
125  // Block Header fields.
126
127  // TODO: encapsulate Header related logic in this inner class.
128  static class Header {
129    // Format of header is:
130    // 8 bytes - block magic
131    // 4 bytes int - onDiskSizeWithoutHeader
132    // 4 bytes int - uncompressedSizeWithoutHeader
133    // 8 bytes long - prevBlockOffset
134    // The following 3 are only present if header contains checksum information
135    // 1 byte - checksum type
136    // 4 byte int - bytes per checksum
137    // 4 byte int - onDiskDataSizeWithHeader
138    static int BLOCK_MAGIC_INDEX = 0;
139    static int ON_DISK_SIZE_WITHOUT_HEADER_INDEX = 8;
140    static int UNCOMPRESSED_SIZE_WITHOUT_HEADER_INDEX = 12;
141    static int PREV_BLOCK_OFFSET_INDEX = 16;
142    static int CHECKSUM_TYPE_INDEX = 24;
143    static int BYTES_PER_CHECKSUM_INDEX = 25;
144    static int ON_DISK_DATA_SIZE_WITH_HEADER_INDEX = 29;
145  }
146
147  /** Type of block. Header field 0. */
148  private BlockType blockType;
149
150  /**
151   * Size on disk excluding header, including checksum. Header field 1.
152   * @see Writer#putHeader(byte[], int, int, int, int)
153   */
154  private int onDiskSizeWithoutHeader;
155
156  /**
157   * Size of pure data. Does not include header or checksums. Header field 2.
158   * @see Writer#putHeader(byte[], int, int, int, int)
159   */
160  private int uncompressedSizeWithoutHeader;
161
162  /**
163   * The offset of the previous block on disk. Header field 3.
164   * @see Writer#putHeader(byte[], int, int, int, int)
165   */
166  private long prevBlockOffset;
167
168  /**
169   * Size on disk of header + data. Excludes checksum. Header field 6,
170   * OR calculated from {@link #onDiskSizeWithoutHeader} when using HDFS checksum.
171   * @see Writer#putHeader(byte[], int, int, int, int)
172   */
173  private int onDiskDataSizeWithHeader;
174  // End of Block Header fields.
175
176  /**
177   * The in-memory representation of the hfile block. Can be on or offheap. Can be backed by
178   * a single ByteBuffer or by many. Make no assumptions.
179   *
180   * <p>Be careful reading from this <code>buf</code>. Duplicate and work on the duplicate or if
181   * not, be sure to reset position and limit else trouble down the road.
182   *
183   * <p>TODO: Make this read-only once made.
184   *
185   * <p>We are using the ByteBuff type. ByteBuffer is not extensible yet we need to be able to have
186   * a ByteBuffer-like API across multiple ByteBuffers reading from a cache such as BucketCache.
187   * So, we have this ByteBuff type. Unfortunately, it is spread all about HFileBlock. Would be
188   * good if could be confined to cache-use only but hard-to-do.
189   */
190  private ByteBuff buf;
191
192  /** Meta data that holds meta information on the hfileblock.
193   */
194  private HFileContext fileContext;
195
196  /**
197   * The offset of this block in the file. Populated by the reader for
198   * convenience of access. This offset is not part of the block header.
199   */
200  private long offset = UNSET;
201
202  /**
203   * The on-disk size of the next block, including the header and checksums if present.
204   * UNSET if unknown.
205   *
206   * Blocks try to carry the size of the next block to read in this data member. Usually
207   * we get block sizes from the hfile index but sometimes the index is not available:
208   * e.g. when we read the indexes themselves (indexes are stored in blocks, we do not
209   * have an index for the indexes). Saves seeks especially around file open when
210   * there is a flurry of reading in hfile metadata.
211   */
212  private int nextBlockOnDiskSize = UNSET;
213
214  private ByteBuffAllocator allocator;
215
216  /**
217   * On a checksum failure, do these many succeeding read requests using hdfs checksums before
218   * auto-reenabling hbase checksum verification.
219   */
220  static final int CHECKSUM_VERIFICATION_NUM_IO_THRESHOLD = 3;
221
222  private static int UNSET = -1;
223  public static final boolean FILL_HEADER = true;
224  public static final boolean DONT_FILL_HEADER = false;
225
226  // How to get the estimate correctly? if it is a singleBB?
227  public static final int MULTI_BYTE_BUFFER_HEAP_SIZE =
228      (int)ClassSize.estimateBase(MultiByteBuff.class, false);
229
230  /**
231   * Space for metadata on a block that gets stored along with the block when we cache it.
232   * There are a few bytes stuck on the end of the HFileBlock that we pull in from HDFS.
233   * 8 bytes are for the offset of this block (long) in the file. Offset is important because is is
234   * used when we remake the CacheKey when we return block to the cache when done. There is also
235   * a flag on whether checksumming is being done by hbase or not. See class comment for note on
236   * uncertain state of checksumming of blocks that come out of cache (should we or should we not?).
237   * Finally there are 4 bytes to hold the length of the next block which can save a seek on
238   * occasion if available.
239   * (This EXTRA info came in with original commit of the bucketcache, HBASE-7404. It was
240   * formerly known as EXTRA_SERIALIZATION_SPACE).
241   */
242  static final int BLOCK_METADATA_SPACE = Bytes.SIZEOF_BYTE + Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT;
243
244  /**
245   * Each checksum value is an integer that can be stored in 4 bytes.
246   */
247  static final int CHECKSUM_SIZE = Bytes.SIZEOF_INT;
248
249  static final byte[] DUMMY_HEADER_NO_CHECKSUM =
250      new byte[HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM];
251
252  /**
253   * Used deserializing blocks from Cache.
254   *
255   * <code>
256   * ++++++++++++++
257   * + HFileBlock +
258   * ++++++++++++++
259   * + Checksums  + <= Optional
260   * ++++++++++++++
261   * + Metadata!  + <= See note on BLOCK_METADATA_SPACE above.
262   * ++++++++++++++
263   * </code>
264   * @see #serialize(ByteBuffer, boolean)
265   */
266  public static final CacheableDeserializer<Cacheable> BLOCK_DESERIALIZER = new BlockDeserializer();
267
268  public static final class BlockDeserializer implements CacheableDeserializer<Cacheable> {
269    private BlockDeserializer() {
270    }
271
272    @Override
273    public HFileBlock deserialize(ByteBuff buf, ByteBuffAllocator alloc)
274        throws IOException {
275      // The buf has the file block followed by block metadata.
276      // Set limit to just before the BLOCK_METADATA_SPACE then rewind.
277      buf.limit(buf.limit() - BLOCK_METADATA_SPACE).rewind();
278      // Get a new buffer to pass the HFileBlock for it to 'own'.
279      ByteBuff newByteBuff = buf.slice();
280      // Read out the BLOCK_METADATA_SPACE content and shove into our HFileBlock.
281      buf.position(buf.limit());
282      buf.limit(buf.limit() + HFileBlock.BLOCK_METADATA_SPACE);
283      boolean usesChecksum = buf.get() == (byte) 1;
284      long offset = buf.getLong();
285      int nextBlockOnDiskSize = buf.getInt();
286      return createFromBuff(newByteBuff, usesChecksum, offset, nextBlockOnDiskSize, null, alloc);
287    }
288
289    @Override
290    public int getDeserializerIdentifier() {
291      return DESERIALIZER_IDENTIFIER;
292    }
293  }
294
295  private static final int DESERIALIZER_IDENTIFIER;
296  static {
297    DESERIALIZER_IDENTIFIER =
298        CacheableDeserializerIdManager.registerDeserializer(BLOCK_DESERIALIZER);
299  }
300
301  /**
302   * Creates a new {@link HFile} block from the given fields. This constructor
303   * is used only while writing blocks and caching,
304   * and is sitting in a byte buffer and we want to stuff the block into cache.
305   *
306   * <p>TODO: The caller presumes no checksumming
307   * <p>TODO: HFile block writer can also off-heap ? </p>
308   * required of this block instance since going into cache; checksum already verified on
309   * underlying block data pulled in from filesystem. Is that correct? What if cache is SSD?
310   *
311   * @param blockType the type of this block, see {@link BlockType}
312   * @param onDiskSizeWithoutHeader see {@link #onDiskSizeWithoutHeader}
313   * @param uncompressedSizeWithoutHeader see {@link #uncompressedSizeWithoutHeader}
314   * @param prevBlockOffset see {@link #prevBlockOffset}
315   * @param buf block buffer with header ({@link HConstants#HFILEBLOCK_HEADER_SIZE} bytes)
316   * @param fillHeader when true, write the first 4 header fields into passed buffer.
317   * @param offset the file offset the block was read from
318   * @param onDiskDataSizeWithHeader see {@link #onDiskDataSizeWithHeader}
319   * @param fileContext HFile meta data
320   */
321  @VisibleForTesting
322  public HFileBlock(BlockType blockType, int onDiskSizeWithoutHeader,
323      int uncompressedSizeWithoutHeader, long prevBlockOffset, ByteBuff buf, boolean fillHeader,
324      long offset, int nextBlockOnDiskSize, int onDiskDataSizeWithHeader, HFileContext fileContext,
325      ByteBuffAllocator allocator) {
326    this.blockType = blockType;
327    this.onDiskSizeWithoutHeader = onDiskSizeWithoutHeader;
328    this.uncompressedSizeWithoutHeader = uncompressedSizeWithoutHeader;
329    this.prevBlockOffset = prevBlockOffset;
330    this.offset = offset;
331    this.onDiskDataSizeWithHeader = onDiskDataSizeWithHeader;
332    this.nextBlockOnDiskSize = nextBlockOnDiskSize;
333    this.fileContext = fileContext;
334    this.allocator = allocator;
335    this.buf = buf;
336    if (fillHeader) {
337      overwriteHeader();
338    }
339    this.buf.rewind();
340  }
341
342  /**
343   * Creates a block from an existing buffer starting with a header. Rewinds
344   * and takes ownership of the buffer. By definition of rewind, ignores the
345   * buffer position, but if you slice the buffer beforehand, it will rewind
346   * to that point.
347   * @param buf Has header, content, and trailing checksums if present.
348   */
349  static HFileBlock createFromBuff(ByteBuff buf, boolean usesHBaseChecksum, final long offset,
350      final int nextBlockOnDiskSize, HFileContext fileContext, ByteBuffAllocator allocator)
351      throws IOException {
352    buf.rewind();
353    final BlockType blockType = BlockType.read(buf);
354    final int onDiskSizeWithoutHeader = buf.getInt(Header.ON_DISK_SIZE_WITHOUT_HEADER_INDEX);
355    final int uncompressedSizeWithoutHeader =
356        buf.getInt(Header.UNCOMPRESSED_SIZE_WITHOUT_HEADER_INDEX);
357    final long prevBlockOffset = buf.getLong(Header.PREV_BLOCK_OFFSET_INDEX);
358    // This constructor is called when we deserialize a block from cache and when we read a block in
359    // from the fs. fileCache is null when deserialized from cache so need to make up one.
360    HFileContextBuilder fileContextBuilder = fileContext != null ?
361      new HFileContextBuilder(fileContext) : new HFileContextBuilder();
362    fileContextBuilder.withHBaseCheckSum(usesHBaseChecksum);
363    int onDiskDataSizeWithHeader;
364    if (usesHBaseChecksum) {
365      byte checksumType = buf.get(Header.CHECKSUM_TYPE_INDEX);
366      int bytesPerChecksum = buf.getInt(Header.BYTES_PER_CHECKSUM_INDEX);
367      onDiskDataSizeWithHeader = buf.getInt(Header.ON_DISK_DATA_SIZE_WITH_HEADER_INDEX);
368      // Use the checksum type and bytes per checksum from header, not from fileContext.
369      fileContextBuilder.withChecksumType(ChecksumType.codeToType(checksumType));
370      fileContextBuilder.withBytesPerCheckSum(bytesPerChecksum);
371    } else {
372      fileContextBuilder.withChecksumType(ChecksumType.NULL);
373      fileContextBuilder.withBytesPerCheckSum(0);
374      // Need to fix onDiskDataSizeWithHeader; there are not checksums after-block-data
375      onDiskDataSizeWithHeader = onDiskSizeWithoutHeader + headerSize(usesHBaseChecksum);
376    }
377    fileContext = fileContextBuilder.build();
378    assert usesHBaseChecksum == fileContext.isUseHBaseChecksum();
379    return new HFileBlockBuilder()
380        .withBlockType(blockType)
381        .withOnDiskSizeWithoutHeader(onDiskSizeWithoutHeader)
382        .withUncompressedSizeWithoutHeader(uncompressedSizeWithoutHeader)
383        .withPrevBlockOffset(prevBlockOffset)
384        .withOffset(offset)
385        .withOnDiskDataSizeWithHeader(onDiskDataSizeWithHeader)
386        .withNextBlockOnDiskSize(nextBlockOnDiskSize)
387        .withHFileContext(fileContext)
388        .withByteBuffAllocator(allocator)
389        .withByteBuff(buf.rewind())
390        .withShared(!buf.hasArray())
391        .build();
392  }
393
394  /**
395   * Parse total on disk size including header and checksum.
396   * @param headerBuf Header ByteBuffer. Presumed exact size of header.
397   * @param verifyChecksum true if checksum verification is in use.
398   * @return Size of the block with header included.
399   */
400  private static int getOnDiskSizeWithHeader(final ByteBuff headerBuf,
401      boolean verifyChecksum) {
402    return headerBuf.getInt(Header.ON_DISK_SIZE_WITHOUT_HEADER_INDEX) + headerSize(verifyChecksum);
403  }
404
405  /**
406   * @return the on-disk size of the next block (including the header size and any checksums if
407   *   present) read by peeking into the next block's header; use as a hint when doing
408   *   a read of the next block when scanning or running over a file.
409   */
410  int getNextBlockOnDiskSize() {
411    return nextBlockOnDiskSize;
412  }
413
414  @Override
415  public BlockType getBlockType() {
416    return blockType;
417  }
418
419  @Override
420  public int refCnt() {
421    return buf.refCnt();
422  }
423
424  @Override
425  public HFileBlock retain() {
426    buf.retain();
427    return this;
428  }
429
430  /**
431   * Call {@link ByteBuff#release()} to decrease the reference count, if no other reference, it will
432   * return back the {@link ByteBuffer} to {@link org.apache.hadoop.hbase.io.ByteBuffAllocator}
433   */
434  @Override
435  public boolean release() {
436    return buf.release();
437  }
438
439  /** @return get data block encoding id that was used to encode this block */
440  short getDataBlockEncodingId() {
441    if (blockType != BlockType.ENCODED_DATA) {
442      throw new IllegalArgumentException("Querying encoder ID of a block " +
443          "of type other than " + BlockType.ENCODED_DATA + ": " + blockType);
444    }
445    return buf.getShort(headerSize());
446  }
447
448  /**
449   * @return the on-disk size of header + data part + checksum.
450   */
451  public int getOnDiskSizeWithHeader() {
452    return onDiskSizeWithoutHeader + headerSize();
453  }
454
455  /**
456   * @return the on-disk size of the data part + checksum (header excluded).
457   */
458  int getOnDiskSizeWithoutHeader() {
459    return onDiskSizeWithoutHeader;
460  }
461
462  /**
463   * @return the uncompressed size of data part (header and checksum excluded).
464   */
465  int getUncompressedSizeWithoutHeader() {
466    return uncompressedSizeWithoutHeader;
467  }
468
469  /**
470   * @return the offset of the previous block of the same type in the file, or
471   *         -1 if unknown
472   */
473  long getPrevBlockOffset() {
474    return prevBlockOffset;
475  }
476
477  /**
478   * Rewinds {@code buf} and writes first 4 header fields. {@code buf} position
479   * is modified as side-effect.
480   */
481  private void overwriteHeader() {
482    buf.rewind();
483    blockType.write(buf);
484    buf.putInt(onDiskSizeWithoutHeader);
485    buf.putInt(uncompressedSizeWithoutHeader);
486    buf.putLong(prevBlockOffset);
487    if (this.fileContext.isUseHBaseChecksum()) {
488      buf.put(fileContext.getChecksumType().getCode());
489      buf.putInt(fileContext.getBytesPerChecksum());
490      buf.putInt(onDiskDataSizeWithHeader);
491    }
492  }
493
494  /**
495   * Returns a buffer that does not include the header and checksum.
496   * @return the buffer with header skipped and checksum omitted.
497   */
498  public ByteBuff getBufferWithoutHeader() {
499    return this.getBufferWithoutHeader(false);
500  }
501
502  /**
503   * Returns a buffer that does not include the header or checksum.
504   * @param withChecksum to indicate whether include the checksum or not.
505   * @return the buffer with header skipped and checksum omitted.
506   */
507  public ByteBuff getBufferWithoutHeader(boolean withChecksum) {
508    ByteBuff dup = getBufferReadOnly();
509    int delta = withChecksum ? 0 : totalChecksumBytes();
510    return dup.position(headerSize()).limit(buf.limit() - delta).slice();
511  }
512
513  /**
514   * Returns a read-only duplicate of the buffer this block stores internally ready to be read.
515   * Clients must not modify the buffer object though they may set position and limit on the
516   * returned buffer since we pass back a duplicate. This method has to be public because it is used
517   * in {@link CompoundBloomFilter} to avoid object creation on every Bloom
518   * filter lookup, but has to be used with caution. Buffer holds header, block content,
519   * and any follow-on checksums if present.
520   *
521   * @return the buffer of this block for read-only operations
522   */
523  public ByteBuff getBufferReadOnly() {
524    // TODO: ByteBuf does not support asReadOnlyBuffer(). Fix.
525    ByteBuff dup = this.buf.duplicate();
526    assert dup.position() == 0;
527    return dup;
528  }
529
530  public ByteBuffAllocator getByteBuffAllocator() {
531    return this.allocator;
532  }
533
534  @VisibleForTesting
535  private void sanityCheckAssertion(long valueFromBuf, long valueFromField,
536      String fieldName) throws IOException {
537    if (valueFromBuf != valueFromField) {
538      throw new AssertionError(fieldName + " in the buffer (" + valueFromBuf
539          + ") is different from that in the field (" + valueFromField + ")");
540    }
541  }
542
543  @VisibleForTesting
544  private void sanityCheckAssertion(BlockType valueFromBuf, BlockType valueFromField)
545      throws IOException {
546    if (valueFromBuf != valueFromField) {
547      throw new IOException("Block type stored in the buffer: " +
548        valueFromBuf + ", block type field: " + valueFromField);
549    }
550  }
551
552  /**
553   * Checks if the block is internally consistent, i.e. the first
554   * {@link HConstants#HFILEBLOCK_HEADER_SIZE} bytes of the buffer contain a
555   * valid header consistent with the fields. Assumes a packed block structure.
556   * This function is primary for testing and debugging, and is not
557   * thread-safe, because it alters the internal buffer pointer.
558   * Used by tests only.
559   */
560  @VisibleForTesting
561  void sanityCheck() throws IOException {
562    // Duplicate so no side-effects
563    ByteBuff dup = this.buf.duplicate().rewind();
564    sanityCheckAssertion(BlockType.read(dup), blockType);
565
566    sanityCheckAssertion(dup.getInt(), onDiskSizeWithoutHeader, "onDiskSizeWithoutHeader");
567
568    sanityCheckAssertion(dup.getInt(), uncompressedSizeWithoutHeader,
569        "uncompressedSizeWithoutHeader");
570
571    sanityCheckAssertion(dup.getLong(), prevBlockOffset, "prevBlockOffset");
572    if (this.fileContext.isUseHBaseChecksum()) {
573      sanityCheckAssertion(dup.get(), this.fileContext.getChecksumType().getCode(), "checksumType");
574      sanityCheckAssertion(dup.getInt(), this.fileContext.getBytesPerChecksum(),
575          "bytesPerChecksum");
576      sanityCheckAssertion(dup.getInt(), onDiskDataSizeWithHeader, "onDiskDataSizeWithHeader");
577    }
578
579    int cksumBytes = totalChecksumBytes();
580    int expectedBufLimit = onDiskDataSizeWithHeader + cksumBytes;
581    if (dup.limit() != expectedBufLimit) {
582      throw new AssertionError("Expected limit " + expectedBufLimit + ", got " + dup.limit());
583    }
584
585    // We might optionally allocate HFILEBLOCK_HEADER_SIZE more bytes to read the next
586    // block's header, so there are two sensible values for buffer capacity.
587    int hdrSize = headerSize();
588    dup.rewind();
589    if (dup.remaining() != expectedBufLimit && dup.remaining() != expectedBufLimit + hdrSize) {
590      throw new AssertionError("Invalid buffer capacity: " + dup.remaining() +
591          ", expected " + expectedBufLimit + " or " + (expectedBufLimit + hdrSize));
592    }
593  }
594
595  @Override
596  public String toString() {
597    StringBuilder sb = new StringBuilder()
598      .append("[")
599      .append("blockType=").append(blockType)
600      .append(", fileOffset=").append(offset)
601      .append(", headerSize=").append(headerSize())
602      .append(", onDiskSizeWithoutHeader=").append(onDiskSizeWithoutHeader)
603      .append(", uncompressedSizeWithoutHeader=").append(uncompressedSizeWithoutHeader)
604      .append(", prevBlockOffset=").append(prevBlockOffset)
605      .append(", isUseHBaseChecksum=").append(fileContext.isUseHBaseChecksum());
606    if (fileContext.isUseHBaseChecksum()) {
607      sb.append(", checksumType=").append(ChecksumType.codeToType(this.buf.get(24)))
608        .append(", bytesPerChecksum=").append(this.buf.getInt(24 + 1))
609        .append(", onDiskDataSizeWithHeader=").append(onDiskDataSizeWithHeader);
610    } else {
611      sb.append(", onDiskDataSizeWithHeader=").append(onDiskDataSizeWithHeader)
612        .append("(").append(onDiskSizeWithoutHeader)
613        .append("+").append(HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM).append(")");
614    }
615    String dataBegin;
616    if (buf.hasArray()) {
617      dataBegin = Bytes.toStringBinary(buf.array(), buf.arrayOffset() + headerSize(),
618          Math.min(32, buf.limit() - buf.arrayOffset() - headerSize()));
619    } else {
620      ByteBuff bufWithoutHeader = getBufferWithoutHeader();
621      byte[] dataBeginBytes = new byte[Math.min(32,
622          bufWithoutHeader.limit() - bufWithoutHeader.position())];
623      bufWithoutHeader.get(dataBeginBytes);
624      dataBegin = Bytes.toStringBinary(dataBeginBytes);
625    }
626    sb.append(", getOnDiskSizeWithHeader=").append(getOnDiskSizeWithHeader())
627      .append(", totalChecksumBytes=").append(totalChecksumBytes())
628      .append(", isUnpacked=").append(isUnpacked())
629      .append(", buf=[").append(buf).append("]")
630      .append(", dataBeginsWith=").append(dataBegin)
631      .append(", fileContext=").append(fileContext)
632      .append(", nextBlockOnDiskSize=").append(nextBlockOnDiskSize)
633      .append("]");
634    return sb.toString();
635  }
636
637  /**
638   * Retrieves the decompressed/decrypted view of this block. An encoded block remains in its
639   * encoded structure. Internal structures are shared between instances where applicable.
640   */
641  HFileBlock unpack(HFileContext fileContext, FSReader reader) throws IOException {
642    if (!fileContext.isCompressedOrEncrypted()) {
643      // TODO: cannot use our own fileContext here because HFileBlock(ByteBuffer, boolean),
644      // which is used for block serialization to L2 cache, does not preserve encoding and
645      // encryption details.
646      return this;
647    }
648
649    HFileBlock unpacked = shallowClone(this);
650    unpacked.allocateBuffer(); // allocates space for the decompressed block
651    boolean succ = false;
652    try {
653      HFileBlockDecodingContext ctx = blockType == BlockType.ENCODED_DATA
654          ? reader.getBlockDecodingContext() : reader.getDefaultBlockDecodingContext();
655      // Create a duplicated buffer without the header part.
656      ByteBuff dup = this.buf.duplicate();
657      dup.position(this.headerSize());
658      dup = dup.slice();
659      // Decode the dup into unpacked#buf
660      ctx.prepareDecoding(unpacked.getOnDiskSizeWithoutHeader(),
661        unpacked.getUncompressedSizeWithoutHeader(), unpacked.getBufferWithoutHeader(true), dup);
662      succ = true;
663      return unpacked;
664    } finally {
665      if (!succ) {
666        unpacked.release();
667      }
668    }
669  }
670
671  /**
672   * Always allocates a new buffer of the correct size. Copies header bytes
673   * from the existing buffer. Does not change header fields.
674   * Reserve room to keep checksum bytes too.
675   */
676  private void allocateBuffer() {
677    int cksumBytes = totalChecksumBytes();
678    int headerSize = headerSize();
679    int capacityNeeded = headerSize + uncompressedSizeWithoutHeader + cksumBytes;
680
681    ByteBuff newBuf = allocator.allocate(capacityNeeded);
682
683    // Copy header bytes into newBuf.
684    buf.position(0);
685    newBuf.put(0, buf, 0, headerSize);
686
687    buf = newBuf;
688    // set limit to exclude next block's header
689    buf.limit(capacityNeeded);
690  }
691
692  /**
693   * Return true when this block's buffer has been unpacked, false otherwise. Note this is a
694   * calculated heuristic, not tracked attribute of the block.
695   */
696  public boolean isUnpacked() {
697    final int cksumBytes = totalChecksumBytes();
698    final int headerSize = headerSize();
699    final int expectedCapacity = headerSize + uncompressedSizeWithoutHeader + cksumBytes;
700    final int bufCapacity = buf.remaining();
701    return bufCapacity == expectedCapacity || bufCapacity == expectedCapacity + headerSize;
702  }
703
704  /**
705   * Cannot be {@link #UNSET}. Must be a legitimate value. Used re-making the {@link BlockCacheKey}
706   * when block is returned to the cache.
707   * @return the offset of this block in the file it was read from
708   */
709  long getOffset() {
710    if (offset < 0) {
711      throw new IllegalStateException("HFile block offset not initialized properly");
712    }
713    return offset;
714  }
715
716  /**
717   * @return a byte stream reading the data + checksum of this block
718   */
719  DataInputStream getByteStream() {
720    ByteBuff dup = this.buf.duplicate();
721    dup.position(this.headerSize());
722    return new DataInputStream(new ByteBuffInputStream(dup));
723  }
724
725  @Override
726  public long heapSize() {
727    long size = FIXED_OVERHEAD;
728    size += fileContext.heapSize();
729    if (buf != null) {
730      // Deep overhead of the byte buffer. Needs to be aligned separately.
731      size += ClassSize.align(buf.capacity() + MULTI_BYTE_BUFFER_HEAP_SIZE);
732    }
733    return ClassSize.align(size);
734  }
735
736  /**
737   * Will be override by {@link SharedMemHFileBlock} or {@link ExclusiveMemHFileBlock}. Return true
738   * by default.
739   */
740  public boolean isSharedMem() {
741    if (this instanceof SharedMemHFileBlock) {
742      return true;
743    } else if (this instanceof ExclusiveMemHFileBlock) {
744      return false;
745    }
746    return true;
747  }
748
749  /**
750   * Unified version 2 {@link HFile} block writer. The intended usage pattern
751   * is as follows:
752   * <ol>
753   * <li>Construct an {@link HFileBlock.Writer}, providing a compression algorithm.
754   * <li>Call {@link Writer#startWriting} and get a data stream to write to.
755   * <li>Write your data into the stream.
756   * <li>Call Writer#writeHeaderAndData(FSDataOutputStream) as many times as you need to.
757   * store the serialized block into an external stream.
758   * <li>Repeat to write more blocks.
759   * </ol>
760   * <p>
761   */
762  static class Writer implements ShipperListener {
763    private enum State {
764      INIT,
765      WRITING,
766      BLOCK_READY
767    }
768
769    /** Writer state. Used to ensure the correct usage protocol. */
770    private State state = State.INIT;
771
772    /** Data block encoder used for data blocks */
773    private final HFileDataBlockEncoder dataBlockEncoder;
774
775    private HFileBlockEncodingContext dataBlockEncodingCtx;
776
777    /** block encoding context for non-data blocks*/
778    private HFileBlockDefaultEncodingContext defaultBlockEncodingCtx;
779
780    /**
781     * The stream we use to accumulate data into a block in an uncompressed format.
782     * We reset this stream at the end of each block and reuse it. The
783     * header is written as the first {@link HConstants#HFILEBLOCK_HEADER_SIZE} bytes into this
784     * stream.
785     */
786    private ByteArrayOutputStream baosInMemory;
787
788    /**
789     * Current block type. Set in {@link #startWriting(BlockType)}. Could be
790     * changed in {@link #finishBlock()} from {@link BlockType#DATA}
791     * to {@link BlockType#ENCODED_DATA}.
792     */
793    private BlockType blockType;
794
795    /**
796     * A stream that we write uncompressed bytes to, which compresses them and
797     * writes them to {@link #baosInMemory}.
798     */
799    private DataOutputStream userDataStream;
800
801    /**
802     * Bytes to be written to the file system, including the header. Compressed
803     * if compression is turned on. It also includes the checksum data that
804     * immediately follows the block data. (header + data + checksums)
805     */
806    private ByteArrayOutputStream onDiskBlockBytesWithHeader;
807
808    /**
809     * The size of the checksum data on disk. It is used only if data is
810     * not compressed. If data is compressed, then the checksums are already
811     * part of onDiskBytesWithHeader. If data is uncompressed, then this
812     * variable stores the checksum data for this block.
813     */
814    private byte[] onDiskChecksum = HConstants.EMPTY_BYTE_ARRAY;
815
816    /**
817     * Current block's start offset in the {@link HFile}. Set in
818     * {@link #writeHeaderAndData(FSDataOutputStream)}.
819     */
820    private long startOffset;
821
822    /**
823     * Offset of previous block by block type. Updated when the next block is
824     * started.
825     */
826    private long[] prevOffsetByType;
827
828    /** The offset of the previous block of the same type */
829    private long prevOffset;
830    /** Meta data that holds information about the hfileblock**/
831    private HFileContext fileContext;
832
833    private final ByteBuffAllocator allocator;
834
835    @Override
836    public void beforeShipped() {
837      if (getEncodingState() != null) {
838        getEncodingState().beforeShipped();
839      }
840    }
841
842    EncodingState getEncodingState() {
843      return dataBlockEncodingCtx.getEncodingState();
844    }
845
846    /**
847     * @param dataBlockEncoder data block encoding algorithm to use
848     */
849    @VisibleForTesting
850    public Writer(HFileDataBlockEncoder dataBlockEncoder, HFileContext fileContext) {
851      this(dataBlockEncoder, fileContext, ByteBuffAllocator.HEAP);
852    }
853
854    public Writer(HFileDataBlockEncoder dataBlockEncoder, HFileContext fileContext,
855        ByteBuffAllocator allocator) {
856      if (fileContext.getBytesPerChecksum() < HConstants.HFILEBLOCK_HEADER_SIZE) {
857        throw new RuntimeException("Unsupported value of bytesPerChecksum. " +
858            " Minimum is " + HConstants.HFILEBLOCK_HEADER_SIZE + " but the configured value is " +
859            fileContext.getBytesPerChecksum());
860      }
861      this.allocator = allocator;
862      this.dataBlockEncoder = dataBlockEncoder != null?
863          dataBlockEncoder: NoOpDataBlockEncoder.INSTANCE;
864      this.dataBlockEncodingCtx = this.dataBlockEncoder.
865          newDataBlockEncodingContext(HConstants.HFILEBLOCK_DUMMY_HEADER, fileContext);
866      // TODO: This should be lazily instantiated since we usually do NOT need this default encoder
867      this.defaultBlockEncodingCtx = new HFileBlockDefaultEncodingContext(null,
868          HConstants.HFILEBLOCK_DUMMY_HEADER, fileContext);
869      // TODO: Set BAOS initial size. Use fileContext.getBlocksize() and add for header/checksum
870      baosInMemory = new ByteArrayOutputStream();
871      prevOffsetByType = new long[BlockType.values().length];
872      for (int i = 0; i < prevOffsetByType.length; ++i) {
873        prevOffsetByType[i] = UNSET;
874      }
875      // TODO: Why fileContext saved away when we have dataBlockEncoder and/or
876      // defaultDataBlockEncoder?
877      this.fileContext = fileContext;
878    }
879
880    /**
881     * Starts writing into the block. The previous block's data is discarded.
882     *
883     * @return the stream the user can write their data into
884     */
885    DataOutputStream startWriting(BlockType newBlockType)
886        throws IOException {
887      if (state == State.BLOCK_READY && startOffset != -1) {
888        // We had a previous block that was written to a stream at a specific
889        // offset. Save that offset as the last offset of a block of that type.
890        prevOffsetByType[blockType.getId()] = startOffset;
891      }
892
893      startOffset = -1;
894      blockType = newBlockType;
895
896      baosInMemory.reset();
897      baosInMemory.write(HConstants.HFILEBLOCK_DUMMY_HEADER);
898
899      state = State.WRITING;
900
901      // We will compress it later in finishBlock()
902      userDataStream = new ByteBufferWriterDataOutputStream(baosInMemory);
903      if (newBlockType == BlockType.DATA) {
904        this.dataBlockEncoder.startBlockEncoding(dataBlockEncodingCtx, userDataStream);
905      }
906      return userDataStream;
907    }
908
909    /**
910     * Writes the Cell to this block
911     */
912    void write(Cell cell) throws IOException{
913      expectState(State.WRITING);
914      this.dataBlockEncoder.encode(cell, dataBlockEncodingCtx, this.userDataStream);
915    }
916
917    /**
918     * Transitions the block writer from the "writing" state to the "block
919     * ready" state.  Does nothing if a block is already finished.
920     */
921    void ensureBlockReady() throws IOException {
922      Preconditions.checkState(state != State.INIT,
923          "Unexpected state: " + state);
924
925      if (state == State.BLOCK_READY) {
926        return;
927      }
928
929      // This will set state to BLOCK_READY.
930      finishBlock();
931    }
932
933    /**
934     * Finish up writing of the block.
935     * Flushes the compressing stream (if using compression), fills out the header,
936     * does any compression/encryption of bytes to flush out to disk, and manages
937     * the cache on write content, if applicable. Sets block write state to "block ready".
938     */
939    private void finishBlock() throws IOException {
940      if (blockType == BlockType.DATA) {
941        this.dataBlockEncoder.endBlockEncoding(dataBlockEncodingCtx, userDataStream,
942            baosInMemory.getBuffer(), blockType);
943        blockType = dataBlockEncodingCtx.getBlockType();
944      }
945      userDataStream.flush();
946      prevOffset = prevOffsetByType[blockType.getId()];
947
948      // We need to set state before we can package the block up for cache-on-write. In a way, the
949      // block is ready, but not yet encoded or compressed.
950      state = State.BLOCK_READY;
951      Bytes compressAndEncryptDat;
952      if (blockType == BlockType.DATA || blockType == BlockType.ENCODED_DATA) {
953        compressAndEncryptDat = dataBlockEncodingCtx.
954            compressAndEncrypt(baosInMemory.getBuffer(), 0, baosInMemory.size());
955      } else {
956        compressAndEncryptDat = defaultBlockEncodingCtx.
957            compressAndEncrypt(baosInMemory.getBuffer(), 0, baosInMemory.size());
958      }
959      if (compressAndEncryptDat == null) {
960        compressAndEncryptDat = new Bytes(baosInMemory.getBuffer(), 0, baosInMemory.size());
961      }
962      if (onDiskBlockBytesWithHeader == null) {
963        onDiskBlockBytesWithHeader = new ByteArrayOutputStream(compressAndEncryptDat.getLength());
964      }
965      onDiskBlockBytesWithHeader.reset();
966      onDiskBlockBytesWithHeader.write(compressAndEncryptDat.get(),
967            compressAndEncryptDat.getOffset(), compressAndEncryptDat.getLength());
968      // Calculate how many bytes we need for checksum on the tail of the block.
969      int numBytes = (int) ChecksumUtil.numBytes(
970          onDiskBlockBytesWithHeader.size(),
971          fileContext.getBytesPerChecksum());
972
973      // Put the header for the on disk bytes; header currently is unfilled-out
974      putHeader(onDiskBlockBytesWithHeader,
975          onDiskBlockBytesWithHeader.size() + numBytes,
976          baosInMemory.size(), onDiskBlockBytesWithHeader.size());
977      if (onDiskChecksum.length != numBytes) {
978        onDiskChecksum = new byte[numBytes];
979      }
980      ChecksumUtil.generateChecksums(
981          onDiskBlockBytesWithHeader.getBuffer(), 0,onDiskBlockBytesWithHeader.size(),
982          onDiskChecksum, 0, fileContext.getChecksumType(), fileContext.getBytesPerChecksum());
983    }
984
985    /**
986     * Put the header into the given byte array at the given offset.
987     * @param onDiskSize size of the block on disk header + data + checksum
988     * @param uncompressedSize size of the block after decompression (but
989     *          before optional data block decoding) including header
990     * @param onDiskDataSize size of the block on disk with header
991     *        and data but not including the checksums
992     */
993    private void putHeader(byte[] dest, int offset, int onDiskSize,
994        int uncompressedSize, int onDiskDataSize) {
995      offset = blockType.put(dest, offset);
996      offset = Bytes.putInt(dest, offset, onDiskSize - HConstants.HFILEBLOCK_HEADER_SIZE);
997      offset = Bytes.putInt(dest, offset, uncompressedSize - HConstants.HFILEBLOCK_HEADER_SIZE);
998      offset = Bytes.putLong(dest, offset, prevOffset);
999      offset = Bytes.putByte(dest, offset, fileContext.getChecksumType().getCode());
1000      offset = Bytes.putInt(dest, offset, fileContext.getBytesPerChecksum());
1001      Bytes.putInt(dest, offset, onDiskDataSize);
1002    }
1003
1004    private void putHeader(ByteBuff buff, int onDiskSize,
1005        int uncompressedSize, int onDiskDataSize) {
1006      buff.rewind();
1007      blockType.write(buff);
1008      buff.putInt(onDiskSize - HConstants.HFILEBLOCK_HEADER_SIZE);
1009      buff.putInt(uncompressedSize - HConstants.HFILEBLOCK_HEADER_SIZE);
1010      buff.putLong(prevOffset);
1011      buff.put(fileContext.getChecksumType().getCode());
1012      buff.putInt(fileContext.getBytesPerChecksum());
1013      buff.putInt(onDiskDataSize);
1014    }
1015
1016    private void putHeader(ByteArrayOutputStream dest, int onDiskSize,
1017        int uncompressedSize, int onDiskDataSize) {
1018      putHeader(dest.getBuffer(),0, onDiskSize, uncompressedSize, onDiskDataSize);
1019    }
1020
1021    /**
1022     * Similar to {@link #writeHeaderAndData(FSDataOutputStream)}, but records
1023     * the offset of this block so that it can be referenced in the next block
1024     * of the same type.
1025     */
1026    void writeHeaderAndData(FSDataOutputStream out) throws IOException {
1027      long offset = out.getPos();
1028      if (startOffset != UNSET && offset != startOffset) {
1029        throw new IOException("A " + blockType + " block written to a "
1030            + "stream twice, first at offset " + startOffset + ", then at "
1031            + offset);
1032      }
1033      startOffset = offset;
1034      finishBlockAndWriteHeaderAndData(out);
1035    }
1036
1037    /**
1038     * Writes the header and the compressed data of this block (or uncompressed
1039     * data when not using compression) into the given stream. Can be called in
1040     * the "writing" state or in the "block ready" state. If called in the
1041     * "writing" state, transitions the writer to the "block ready" state.
1042     * @param out the output stream to write the
1043     */
1044    protected void finishBlockAndWriteHeaderAndData(DataOutputStream out)
1045      throws IOException {
1046      ensureBlockReady();
1047      long startTime = System.currentTimeMillis();
1048      out.write(onDiskBlockBytesWithHeader.getBuffer(), 0, onDiskBlockBytesWithHeader.size());
1049      out.write(onDiskChecksum);
1050      HFile.updateWriteLatency(System.currentTimeMillis() - startTime);
1051    }
1052
1053    /**
1054     * Returns the header or the compressed data (or uncompressed data when not
1055     * using compression) as a byte array. Can be called in the "writing" state
1056     * or in the "block ready" state. If called in the "writing" state,
1057     * transitions the writer to the "block ready" state. This returns
1058     * the header + data + checksums stored on disk.
1059     *
1060     * @return header and data as they would be stored on disk in a byte array
1061     */
1062    byte[] getHeaderAndDataForTest() throws IOException {
1063      ensureBlockReady();
1064      // This is not very optimal, because we are doing an extra copy.
1065      // But this method is used only by unit tests.
1066      byte[] output =
1067          new byte[onDiskBlockBytesWithHeader.size()
1068              + onDiskChecksum.length];
1069      System.arraycopy(onDiskBlockBytesWithHeader.getBuffer(), 0, output, 0,
1070          onDiskBlockBytesWithHeader.size());
1071      System.arraycopy(onDiskChecksum, 0, output,
1072          onDiskBlockBytesWithHeader.size(), onDiskChecksum.length);
1073      return output;
1074    }
1075
1076    /**
1077     * Releases resources used by this writer.
1078     */
1079    void release() {
1080      if (dataBlockEncodingCtx != null) {
1081        dataBlockEncodingCtx.close();
1082        dataBlockEncodingCtx = null;
1083      }
1084      if (defaultBlockEncodingCtx != null) {
1085        defaultBlockEncodingCtx.close();
1086        defaultBlockEncodingCtx = null;
1087      }
1088    }
1089
1090    /**
1091     * Returns the on-disk size of the data portion of the block. This is the
1092     * compressed size if compression is enabled. Can only be called in the
1093     * "block ready" state. Header is not compressed, and its size is not
1094     * included in the return value.
1095     *
1096     * @return the on-disk size of the block, not including the header.
1097     */
1098    int getOnDiskSizeWithoutHeader() {
1099      expectState(State.BLOCK_READY);
1100      return onDiskBlockBytesWithHeader.size() +
1101          onDiskChecksum.length - HConstants.HFILEBLOCK_HEADER_SIZE;
1102    }
1103
1104    /**
1105     * Returns the on-disk size of the block. Can only be called in the
1106     * "block ready" state.
1107     *
1108     * @return the on-disk size of the block ready to be written, including the
1109     *         header size, the data and the checksum data.
1110     */
1111    int getOnDiskSizeWithHeader() {
1112      expectState(State.BLOCK_READY);
1113      return onDiskBlockBytesWithHeader.size() + onDiskChecksum.length;
1114    }
1115
1116    /**
1117     * The uncompressed size of the block data. Does not include header size.
1118     */
1119    int getUncompressedSizeWithoutHeader() {
1120      expectState(State.BLOCK_READY);
1121      return baosInMemory.size() - HConstants.HFILEBLOCK_HEADER_SIZE;
1122    }
1123
1124    /**
1125     * The uncompressed size of the block data, including header size.
1126     */
1127    int getUncompressedSizeWithHeader() {
1128      expectState(State.BLOCK_READY);
1129      return baosInMemory.size();
1130    }
1131
1132    /** @return true if a block is being written  */
1133    boolean isWriting() {
1134      return state == State.WRITING;
1135    }
1136
1137    /**
1138     * Returns the number of bytes written into the current block so far, or
1139     * zero if not writing the block at the moment. Note that this will return
1140     * zero in the "block ready" state as well.
1141     *
1142     * @return the number of bytes written
1143     */
1144    public int encodedBlockSizeWritten() {
1145      return state != State.WRITING ? 0 : this.getEncodingState().getEncodedDataSizeWritten();
1146    }
1147
1148    /**
1149     * Returns the number of bytes written into the current block so far, or
1150     * zero if not writing the block at the moment. Note that this will return
1151     * zero in the "block ready" state as well.
1152     *
1153     * @return the number of bytes written
1154     */
1155    int blockSizeWritten() {
1156      return state != State.WRITING ? 0 : this.getEncodingState().getUnencodedDataSizeWritten();
1157    }
1158
1159    /**
1160     * Clones the header followed by the uncompressed data, even if using
1161     * compression. This is needed for storing uncompressed blocks in the block
1162     * cache. Can be called in the "writing" state or the "block ready" state.
1163     * Returns only the header and data, does not include checksum data.
1164     *
1165     * @return Returns an uncompressed block ByteBuff for caching on write
1166     */
1167    ByteBuff cloneUncompressedBufferWithHeader() {
1168      expectState(State.BLOCK_READY);
1169      ByteBuff bytebuff = allocator.allocate(baosInMemory.size());
1170      baosInMemory.toByteBuff(bytebuff);
1171      int numBytes = (int) ChecksumUtil.numBytes(
1172          onDiskBlockBytesWithHeader.size(),
1173          fileContext.getBytesPerChecksum());
1174      putHeader(bytebuff, onDiskBlockBytesWithHeader.size() + numBytes,
1175          baosInMemory.size(), onDiskBlockBytesWithHeader.size());
1176      bytebuff.rewind();
1177      return bytebuff;
1178    }
1179
1180    /**
1181     * Clones the header followed by the on-disk (compressed/encoded/encrypted) data. This is needed
1182     * for storing packed blocks in the block cache. Returns only the header and data, Does not
1183     * include checksum data.
1184     * @return Returns a copy of block bytes for caching on write
1185     */
1186    private ByteBuff cloneOnDiskBufferWithHeader() {
1187      expectState(State.BLOCK_READY);
1188      ByteBuff bytebuff = allocator.allocate(onDiskBlockBytesWithHeader.size());
1189      onDiskBlockBytesWithHeader.toByteBuff(bytebuff);
1190      bytebuff.rewind();
1191      return bytebuff;
1192    }
1193
1194    private void expectState(State expectedState) {
1195      if (state != expectedState) {
1196        throw new IllegalStateException("Expected state: " + expectedState +
1197            ", actual state: " + state);
1198      }
1199    }
1200
1201    /**
1202     * Takes the given {@link BlockWritable} instance, creates a new block of
1203     * its appropriate type, writes the writable into this block, and flushes
1204     * the block into the output stream. The writer is instructed not to buffer
1205     * uncompressed bytes for cache-on-write.
1206     *
1207     * @param bw the block-writable object to write as a block
1208     * @param out the file system output stream
1209     */
1210    void writeBlock(BlockWritable bw, FSDataOutputStream out)
1211        throws IOException {
1212      bw.writeToBlock(startWriting(bw.getBlockType()));
1213      writeHeaderAndData(out);
1214    }
1215
1216    /**
1217     * Creates a new HFileBlock. Checksums have already been validated, so
1218     * the byte buffer passed into the constructor of this newly created
1219     * block does not have checksum data even though the header minor
1220     * version is MINOR_VERSION_WITH_CHECKSUM. This is indicated by setting a
1221     * 0 value in bytesPerChecksum. This method copies the on-disk or
1222     * uncompressed data to build the HFileBlock which is used only
1223     * while writing blocks and caching.
1224     *
1225     * <p>TODO: Should there be an option where a cache can ask that hbase preserve block
1226     * checksums for checking after a block comes out of the cache? Otehrwise, cache is responsible
1227     * for blocks being wholesome (ECC memory or if file-backed, it does checksumming).
1228     */
1229    HFileBlock getBlockForCaching(CacheConfig cacheConf) {
1230      HFileContext newContext = new HFileContextBuilder()
1231                                .withBlockSize(fileContext.getBlocksize())
1232                                .withBytesPerCheckSum(0)
1233                                .withChecksumType(ChecksumType.NULL) // no checksums in cached data
1234                                .withCompression(fileContext.getCompression())
1235                                .withDataBlockEncoding(fileContext.getDataBlockEncoding())
1236                                .withHBaseCheckSum(fileContext.isUseHBaseChecksum())
1237                                .withCompressTags(fileContext.isCompressTags())
1238                                .withIncludesMvcc(fileContext.isIncludesMvcc())
1239                                .withIncludesTags(fileContext.isIncludesTags())
1240                                .withColumnFamily(fileContext.getColumnFamily())
1241                                .withTableName(fileContext.getTableName())
1242                                .build();
1243      // Build the HFileBlock.
1244      HFileBlockBuilder builder = new HFileBlockBuilder();
1245      ByteBuff buff;
1246      if (cacheConf.shouldCacheCompressed(blockType.getCategory())) {
1247        buff = cloneOnDiskBufferWithHeader();
1248      } else {
1249        buff = cloneUncompressedBufferWithHeader();
1250      }
1251      return builder.withBlockType(blockType)
1252          .withOnDiskSizeWithoutHeader(getOnDiskSizeWithoutHeader())
1253          .withUncompressedSizeWithoutHeader(getUncompressedSizeWithoutHeader())
1254          .withPrevBlockOffset(prevOffset)
1255          .withByteBuff(buff)
1256          .withFillHeader(FILL_HEADER)
1257          .withOffset(startOffset)
1258          .withNextBlockOnDiskSize(UNSET)
1259          .withOnDiskDataSizeWithHeader(onDiskBlockBytesWithHeader.size() + onDiskChecksum.length)
1260          .withHFileContext(newContext)
1261          .withByteBuffAllocator(cacheConf.getByteBuffAllocator())
1262          .withShared(!buff.hasArray())
1263          .build();
1264    }
1265  }
1266
1267  /** Something that can be written into a block. */
1268  interface BlockWritable {
1269    /** The type of block this data should use. */
1270    BlockType getBlockType();
1271
1272    /**
1273     * Writes the block to the provided stream. Must not write any magic
1274     * records.
1275     *
1276     * @param out a stream to write uncompressed data into
1277     */
1278    void writeToBlock(DataOutput out) throws IOException;
1279  }
1280
1281  /**
1282   * Iterator for reading {@link HFileBlock}s in load-on-open-section, such as root data index
1283   * block, meta index block, file info block etc.
1284   */
1285  interface BlockIterator {
1286    /**
1287     * Get the next block, or null if there are no more blocks to iterate.
1288     */
1289    HFileBlock nextBlock() throws IOException;
1290
1291    /**
1292     * Similar to {@link #nextBlock()} but checks block type, throws an exception if incorrect, and
1293     * returns the HFile block
1294     */
1295    HFileBlock nextBlockWithBlockType(BlockType blockType) throws IOException;
1296
1297    /**
1298     * Now we use the {@link ByteBuffAllocator} to manage the nio ByteBuffers for HFileBlocks, so we
1299     * must deallocate all of the ByteBuffers in the end life. the BlockIterator's life cycle is
1300     * starting from opening an HFileReader and stopped when the HFileReader#close, so we will keep
1301     * track all the read blocks until we call {@link BlockIterator#freeBlocks()} when closing the
1302     * HFileReader. Sum bytes of those blocks in load-on-open section should be quite small, so
1303     * tracking them should be OK.
1304     */
1305    void freeBlocks();
1306  }
1307
1308  /** An HFile block reader with iteration ability. */
1309  interface FSReader {
1310    /**
1311     * Reads the block at the given offset in the file with the given on-disk size and uncompressed
1312     * size.
1313     * @param offset of the file to read
1314     * @param onDiskSize the on-disk size of the entire block, including all applicable headers, or
1315     *          -1 if unknown
1316     * @param pread true to use pread, otherwise use the stream read.
1317     * @param updateMetrics update the metrics or not.
1318     * @param intoHeap allocate the block's ByteBuff by {@link ByteBuffAllocator} or JVM heap. For
1319     *          LRUBlockCache, we must ensure that the block to cache is an heap one, because the
1320     *          memory occupation is based on heap now, also for {@link CombinedBlockCache}, we use
1321     *          the heap LRUBlockCache as L1 cache to cache small blocks such as IndexBlock or
1322     *          MetaBlock for faster access. So introduce an flag here to decide whether allocate
1323     *          from JVM heap or not so that we can avoid an extra off-heap to heap memory copy when
1324     *          using LRUBlockCache. For most cases, we known what's the expected block type we'll
1325     *          read, while for some special case (Example: HFileReaderImpl#readNextDataBlock()), we
1326     *          cannot pre-decide what's the expected block type, then we can only allocate block's
1327     *          ByteBuff from {@link ByteBuffAllocator} firstly, and then when caching it in
1328     *          {@link LruBlockCache} we'll check whether the ByteBuff is from heap or not, if not
1329     *          then we'll clone it to an heap one and cache it.
1330     * @return the newly read block
1331     */
1332    HFileBlock readBlockData(long offset, long onDiskSize, boolean pread, boolean updateMetrics,
1333        boolean intoHeap) throws IOException;
1334
1335    /**
1336     * Creates a block iterator over the given portion of the {@link HFile}.
1337     * The iterator returns blocks starting with offset such that offset &lt;=
1338     * startOffset &lt; endOffset. Returned blocks are always unpacked.
1339     * Used when no hfile index available; e.g. reading in the hfile index
1340     * blocks themselves on file open.
1341     *
1342     * @param startOffset the offset of the block to start iteration with
1343     * @param endOffset the offset to end iteration at (exclusive)
1344     * @return an iterator of blocks between the two given offsets
1345     */
1346    BlockIterator blockRange(long startOffset, long endOffset);
1347
1348    /** Closes the backing streams */
1349    void closeStreams() throws IOException;
1350
1351    /** Get a decoder for {@link BlockType#ENCODED_DATA} blocks from this file. */
1352    HFileBlockDecodingContext getBlockDecodingContext();
1353
1354    /** Get the default decoder for blocks from this file. */
1355    HFileBlockDecodingContext getDefaultBlockDecodingContext();
1356
1357    void setIncludesMemStoreTS(boolean includesMemstoreTS);
1358    void setDataBlockEncoder(HFileDataBlockEncoder encoder);
1359
1360    /**
1361     * To close the stream's socket. Note: This can be concurrently called from multiple threads and
1362     * implementation should take care of thread safety.
1363     */
1364    void unbufferStream();
1365  }
1366
1367  /**
1368   * Data-structure to use caching the header of the NEXT block. Only works if next read
1369   * that comes in here is next in sequence in this block.
1370   *
1371   * When we read, we read current block and the next blocks' header. We do this so we have
1372   * the length of the next block to read if the hfile index is not available (rare, at
1373   * hfile open only).
1374   */
1375  private static class PrefetchedHeader {
1376    long offset = -1;
1377    byte[] header = new byte[HConstants.HFILEBLOCK_HEADER_SIZE];
1378    final ByteBuff buf = new SingleByteBuff(ByteBuffer.wrap(header, 0, header.length));
1379
1380    @Override
1381    public String toString() {
1382      return "offset=" + this.offset + ", header=" + Bytes.toStringBinary(header);
1383    }
1384  }
1385
1386  /**
1387   * Reads version 2 HFile blocks from the filesystem.
1388   */
1389  static class FSReaderImpl implements FSReader {
1390    /** The file system stream of the underlying {@link HFile} that
1391     * does or doesn't do checksum validations in the filesystem */
1392    private FSDataInputStreamWrapper streamWrapper;
1393
1394    private HFileBlockDecodingContext encodedBlockDecodingCtx;
1395
1396    /** Default context used when BlockType != {@link BlockType#ENCODED_DATA}. */
1397    private final HFileBlockDefaultDecodingContext defaultDecodingCtx;
1398
1399    /**
1400     * Cache of the NEXT header after this. Check it is indeed next blocks header
1401     * before using it. TODO: Review. This overread into next block to fetch
1402     * next blocks header seems unnecessary given we usually get the block size
1403     * from the hfile index. Review!
1404     */
1405    private AtomicReference<PrefetchedHeader> prefetchedHeader =
1406      new AtomicReference<>(new PrefetchedHeader());
1407
1408    /** The size of the file we are reading from, or -1 if unknown. */
1409    private long fileSize;
1410
1411    /** The size of the header */
1412    @VisibleForTesting
1413    protected final int hdrSize;
1414
1415    /** The filesystem used to access data */
1416    private HFileSystem hfs;
1417
1418    private HFileContext fileContext;
1419    // Cache the fileName
1420    private String pathName;
1421
1422    private final ByteBuffAllocator allocator;
1423
1424    private final Lock streamLock = new ReentrantLock();
1425
1426    FSReaderImpl(ReaderContext readerContext, HFileContext fileContext,
1427        ByteBuffAllocator allocator) throws IOException {
1428      this.fileSize = readerContext.getFileSize();
1429      this.hfs = readerContext.getFileSystem();
1430      if (readerContext.getFilePath() != null) {
1431        this.pathName = readerContext.getFilePath().toString();
1432      }
1433      this.fileContext = fileContext;
1434      this.hdrSize = headerSize(fileContext.isUseHBaseChecksum());
1435      this.allocator = allocator;
1436
1437      this.streamWrapper = readerContext.getInputStreamWrapper();
1438      // Older versions of HBase didn't support checksum.
1439      this.streamWrapper.prepareForBlockReader(!fileContext.isUseHBaseChecksum());
1440      defaultDecodingCtx = new HFileBlockDefaultDecodingContext(fileContext);
1441      encodedBlockDecodingCtx = defaultDecodingCtx;
1442    }
1443
1444    @Override
1445    public BlockIterator blockRange(final long startOffset, final long endOffset) {
1446      final FSReader owner = this; // handle for inner class
1447      return new BlockIterator() {
1448        private volatile boolean freed = false;
1449        // Tracking all read blocks until we call freeBlocks.
1450        private List<HFileBlock> blockTracker = new ArrayList<>();
1451        private long offset = startOffset;
1452        // Cache length of next block. Current block has the length of next block in it.
1453        private long length = -1;
1454
1455        @Override
1456        public HFileBlock nextBlock() throws IOException {
1457          if (offset >= endOffset) {
1458            return null;
1459          }
1460          HFileBlock b = readBlockData(offset, length, false, false, true);
1461          offset += b.getOnDiskSizeWithHeader();
1462          length = b.getNextBlockOnDiskSize();
1463          HFileBlock uncompressed = b.unpack(fileContext, owner);
1464          if (uncompressed != b) {
1465            b.release(); // Need to release the compressed Block now.
1466          }
1467          blockTracker.add(uncompressed);
1468          return uncompressed;
1469        }
1470
1471        @Override
1472        public HFileBlock nextBlockWithBlockType(BlockType blockType) throws IOException {
1473          HFileBlock blk = nextBlock();
1474          if (blk.getBlockType() != blockType) {
1475            throw new IOException(
1476                "Expected block of type " + blockType + " but found " + blk.getBlockType());
1477          }
1478          return blk;
1479        }
1480
1481        @Override
1482        public void freeBlocks() {
1483          if (freed) {
1484            return;
1485          }
1486          blockTracker.forEach(HFileBlock::release);
1487          blockTracker = null;
1488          freed = true;
1489        }
1490      };
1491    }
1492
1493    /**
1494     * Does a positional read or a seek and read into the given byte buffer. We need take care that
1495     * we will call the {@link ByteBuff#release()} for every exit to deallocate the ByteBuffers,
1496     * otherwise the memory leak may happen.
1497     * @param dest destination buffer
1498     * @param size size of read
1499     * @param peekIntoNextBlock whether to read the next block's on-disk size
1500     * @param fileOffset position in the stream to read at
1501     * @param pread whether we should do a positional read
1502     * @param istream The input source of data
1503     * @return true to indicate the destination buffer include the next block header, otherwise only
1504     *         include the current block data without the next block header.
1505     * @throws IOException if any IO error happen.
1506     */
1507    protected boolean readAtOffset(FSDataInputStream istream, ByteBuff dest, int size,
1508        boolean peekIntoNextBlock, long fileOffset, boolean pread) throws IOException {
1509      if (!pread) {
1510        // Seek + read. Better for scanning.
1511        HFileUtil.seekOnMultipleSources(istream, fileOffset);
1512        long realOffset = istream.getPos();
1513        if (realOffset != fileOffset) {
1514          throw new IOException("Tried to seek to " + fileOffset + " to read " + size
1515              + " bytes, but pos=" + realOffset + " after seek");
1516        }
1517        if (!peekIntoNextBlock) {
1518          BlockIOUtils.readFully(dest, istream, size);
1519          return false;
1520        }
1521
1522        // Try to read the next block header
1523        if (!BlockIOUtils.readWithExtra(dest, istream, size, hdrSize)) {
1524          // did not read the next block header.
1525          return false;
1526        }
1527      } else {
1528        // Positional read. Better for random reads; or when the streamLock is already locked.
1529        int extraSize = peekIntoNextBlock ? hdrSize : 0;
1530        if (!BlockIOUtils.preadWithExtra(dest, istream, fileOffset, size, extraSize)) {
1531          // did not read the next block header.
1532          return false;
1533        }
1534      }
1535      assert peekIntoNextBlock;
1536      return true;
1537    }
1538
1539    /**
1540     * Reads a version 2 block (version 1 blocks not supported and not expected). Tries to do as
1541     * little memory allocation as possible, using the provided on-disk size.
1542     * @param offset the offset in the stream to read at
1543     * @param onDiskSizeWithHeaderL the on-disk size of the block, including the header, or -1 if
1544     *          unknown; i.e. when iterating over blocks reading in the file metadata info.
1545     * @param pread whether to use a positional read
1546     * @param updateMetrics whether to update the metrics
1547     * @param intoHeap allocate ByteBuff of block from heap or off-heap.
1548     * @see FSReader#readBlockData(long, long, boolean, boolean, boolean) for more details about the
1549     *      useHeap.
1550     */
1551    @Override
1552    public HFileBlock readBlockData(long offset, long onDiskSizeWithHeaderL, boolean pread,
1553        boolean updateMetrics, boolean intoHeap) throws IOException {
1554      // Get a copy of the current state of whether to validate
1555      // hbase checksums or not for this read call. This is not
1556      // thread-safe but the one constaint is that if we decide
1557      // to skip hbase checksum verification then we are
1558      // guaranteed to use hdfs checksum verification.
1559      boolean doVerificationThruHBaseChecksum = streamWrapper.shouldUseHBaseChecksum();
1560      FSDataInputStream is = streamWrapper.getStream(doVerificationThruHBaseChecksum);
1561
1562      HFileBlock blk = readBlockDataInternal(is, offset, onDiskSizeWithHeaderL, pread,
1563        doVerificationThruHBaseChecksum, updateMetrics, intoHeap);
1564      if (blk == null) {
1565        HFile.LOG.warn("HBase checksum verification failed for file " +
1566                       pathName + " at offset " +
1567                       offset + " filesize " + fileSize +
1568                       ". Retrying read with HDFS checksums turned on...");
1569
1570        if (!doVerificationThruHBaseChecksum) {
1571          String msg = "HBase checksum verification failed for file " +
1572                       pathName + " at offset " +
1573                       offset + " filesize " + fileSize +
1574                       " but this cannot happen because doVerify is " +
1575                       doVerificationThruHBaseChecksum;
1576          HFile.LOG.warn(msg);
1577          throw new IOException(msg); // cannot happen case here
1578        }
1579        HFile.CHECKSUM_FAILURES.increment(); // update metrics
1580
1581        // If we have a checksum failure, we fall back into a mode where
1582        // the next few reads use HDFS level checksums. We aim to make the
1583        // next CHECKSUM_VERIFICATION_NUM_IO_THRESHOLD reads avoid
1584        // hbase checksum verification, but since this value is set without
1585        // holding any locks, it can so happen that we might actually do
1586        // a few more than precisely this number.
1587        is = this.streamWrapper.fallbackToFsChecksum(CHECKSUM_VERIFICATION_NUM_IO_THRESHOLD);
1588        doVerificationThruHBaseChecksum = false;
1589        blk = readBlockDataInternal(is, offset, onDiskSizeWithHeaderL, pread,
1590          doVerificationThruHBaseChecksum, updateMetrics, intoHeap);
1591        if (blk != null) {
1592          HFile.LOG.warn("HDFS checksum verification succeeded for file " +
1593                         pathName + " at offset " +
1594                         offset + " filesize " + fileSize);
1595        }
1596      }
1597      if (blk == null && !doVerificationThruHBaseChecksum) {
1598        String msg = "readBlockData failed, possibly due to " +
1599                     "checksum verification failed for file " + pathName +
1600                     " at offset " + offset + " filesize " + fileSize;
1601        HFile.LOG.warn(msg);
1602        throw new IOException(msg);
1603      }
1604
1605      // If there is a checksum mismatch earlier, then retry with
1606      // HBase checksums switched off and use HDFS checksum verification.
1607      // This triggers HDFS to detect and fix corrupt replicas. The
1608      // next checksumOffCount read requests will use HDFS checksums.
1609      // The decrementing of this.checksumOffCount is not thread-safe,
1610      // but it is harmless because eventually checksumOffCount will be
1611      // a negative number.
1612      streamWrapper.checksumOk();
1613      return blk;
1614    }
1615
1616    /**
1617     * @return Check <code>onDiskSizeWithHeaderL</code> size is healthy and then return it as an int
1618     */
1619    private static int checkAndGetSizeAsInt(final long onDiskSizeWithHeaderL, final int hdrSize)
1620        throws IOException {
1621      if ((onDiskSizeWithHeaderL < hdrSize && onDiskSizeWithHeaderL != -1)
1622          || onDiskSizeWithHeaderL >= Integer.MAX_VALUE) {
1623        throw new IOException("Invalid onDisksize=" + onDiskSizeWithHeaderL
1624            + ": expected to be at least " + hdrSize
1625            + " and at most " + Integer.MAX_VALUE + ", or -1");
1626      }
1627      return (int)onDiskSizeWithHeaderL;
1628    }
1629
1630    /**
1631     * Verify the passed in onDiskSizeWithHeader aligns with what is in the header else something
1632     * is not right.
1633     */
1634    private void verifyOnDiskSizeMatchesHeader(final int passedIn, final ByteBuff headerBuf,
1635           final long offset, boolean verifyChecksum)
1636         throws IOException {
1637      // Assert size provided aligns with what is in the header
1638      int fromHeader = getOnDiskSizeWithHeader(headerBuf, verifyChecksum);
1639      if (passedIn != fromHeader) {
1640        throw new IOException("Passed in onDiskSizeWithHeader=" + passedIn + " != " + fromHeader +
1641            ", offset=" + offset + ", fileContext=" + this.fileContext);
1642      }
1643    }
1644
1645    /**
1646     * Check atomic reference cache for this block's header. Cache only good if next
1647     * read coming through is next in sequence in the block. We read next block's
1648     * header on the tail of reading the previous block to save a seek. Otherwise,
1649     * we have to do a seek to read the header before we can pull in the block OR
1650     * we have to backup the stream because we over-read (the next block's header).
1651     * @see PrefetchedHeader
1652     * @return The cached block header or null if not found.
1653     * @see #cacheNextBlockHeader(long, ByteBuff, int, int)
1654     */
1655    private ByteBuff getCachedHeader(final long offset) {
1656      PrefetchedHeader ph = this.prefetchedHeader.get();
1657      return ph != null && ph.offset == offset ? ph.buf : null;
1658    }
1659
1660    /**
1661     * Save away the next blocks header in atomic reference.
1662     * @see #getCachedHeader(long)
1663     * @see PrefetchedHeader
1664     */
1665    private void cacheNextBlockHeader(final long offset,
1666        ByteBuff onDiskBlock, int onDiskSizeWithHeader, int headerLength) {
1667      PrefetchedHeader ph = new PrefetchedHeader();
1668      ph.offset = offset;
1669      onDiskBlock.get(onDiskSizeWithHeader, ph.header, 0, headerLength);
1670      this.prefetchedHeader.set(ph);
1671    }
1672
1673    private int getNextBlockOnDiskSize(boolean readNextHeader, ByteBuff onDiskBlock,
1674        int onDiskSizeWithHeader) {
1675      int nextBlockOnDiskSize = -1;
1676      if (readNextHeader) {
1677        nextBlockOnDiskSize =
1678            onDiskBlock.getIntAfterPosition(onDiskSizeWithHeader + BlockType.MAGIC_LENGTH)
1679                + hdrSize;
1680      }
1681      return nextBlockOnDiskSize;
1682    }
1683
1684    private ByteBuff allocate(int size, boolean intoHeap) {
1685      return intoHeap ? HEAP.allocate(size) : allocator.allocate(size);
1686    }
1687
1688    /**
1689     * Reads a version 2 block.
1690     * @param offset the offset in the stream to read at.
1691     * @param onDiskSizeWithHeaderL the on-disk size of the block, including the header and
1692     *          checksums if present or -1 if unknown (as a long). Can be -1 if we are doing raw
1693     *          iteration of blocks as when loading up file metadata; i.e. the first read of a new
1694     *          file. Usually non-null gotten from the file index.
1695     * @param pread whether to use a positional read
1696     * @param verifyChecksum Whether to use HBase checksums. If HBase checksum is switched off, then
1697     *          use HDFS checksum. Can also flip on/off reading same file if we hit a troublesome
1698     *          patch in an hfile.
1699     * @param updateMetrics whether need to update the metrics.
1700     * @param intoHeap allocate the ByteBuff of block from heap or off-heap.
1701     * @return the HFileBlock or null if there is a HBase checksum mismatch
1702     */
1703    @VisibleForTesting
1704    protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
1705        long onDiskSizeWithHeaderL, boolean pread, boolean verifyChecksum, boolean updateMetrics,
1706        boolean intoHeap) throws IOException {
1707      if (offset < 0) {
1708        throw new IOException("Invalid offset=" + offset + " trying to read "
1709            + "block (onDiskSize=" + onDiskSizeWithHeaderL + ")");
1710      }
1711      int onDiskSizeWithHeader = checkAndGetSizeAsInt(onDiskSizeWithHeaderL, hdrSize);
1712      // Try and get cached header. Will serve us in rare case where onDiskSizeWithHeaderL is -1
1713      // and will save us having to seek the stream backwards to reread the header we
1714      // read the last time through here.
1715      ByteBuff headerBuf = getCachedHeader(offset);
1716      LOG.trace("Reading {} at offset={}, pread={}, verifyChecksum={}, cachedHeader={}, " +
1717          "onDiskSizeWithHeader={}", this.fileContext.getHFileName(), offset, pread,
1718          verifyChecksum, headerBuf, onDiskSizeWithHeader);
1719      // This is NOT same as verifyChecksum. This latter is whether to do hbase
1720      // checksums. Can change with circumstances. The below flag is whether the
1721      // file has support for checksums (version 2+).
1722      boolean checksumSupport = this.fileContext.isUseHBaseChecksum();
1723      long startTime = System.currentTimeMillis();
1724      if (onDiskSizeWithHeader <= 0) {
1725        // We were not passed the block size. Need to get it from the header. If header was
1726        // not cached (see getCachedHeader above), need to seek to pull it in. This is costly
1727        // and should happen very rarely. Currently happens on open of a hfile reader where we
1728        // read the trailer blocks to pull in the indices. Otherwise, we are reading block sizes
1729        // out of the hfile index. To check, enable TRACE in this file and you'll get an exception
1730        // in a LOG every time we seek. See HBASE-17072 for more detail.
1731        if (headerBuf == null) {
1732          if (LOG.isTraceEnabled()) {
1733            LOG.trace("Extra see to get block size!", new RuntimeException());
1734          }
1735          headerBuf = HEAP.allocate(hdrSize);
1736          readAtOffset(is, headerBuf, hdrSize, false, offset, pread);
1737          headerBuf.rewind();
1738        }
1739        onDiskSizeWithHeader = getOnDiskSizeWithHeader(headerBuf, checksumSupport);
1740      }
1741      int preReadHeaderSize = headerBuf == null? 0 : hdrSize;
1742      // Allocate enough space to fit the next block's header too; saves a seek next time through.
1743      // onDiskBlock is whole block + header + checksums then extra hdrSize to read next header;
1744      // onDiskSizeWithHeader is header, body, and any checksums if present. preReadHeaderSize
1745      // says where to start reading. If we have the header cached, then we don't need to read
1746      // it again and we can likely read from last place we left off w/o need to backup and reread
1747      // the header we read last time through here.
1748      ByteBuff onDiskBlock = this.allocate(onDiskSizeWithHeader + hdrSize, intoHeap);
1749      boolean initHFileBlockSuccess = false;
1750      try {
1751        if (headerBuf != null) {
1752          onDiskBlock.put(0, headerBuf, 0, hdrSize).position(hdrSize);
1753        }
1754        boolean readNextHeader = readAtOffset(is, onDiskBlock,
1755          onDiskSizeWithHeader - preReadHeaderSize, true, offset + preReadHeaderSize, pread);
1756        onDiskBlock.rewind(); // in case of moving position when copying a cached header
1757        int nextBlockOnDiskSize =
1758            getNextBlockOnDiskSize(readNextHeader, onDiskBlock, onDiskSizeWithHeader);
1759        if (headerBuf == null) {
1760          headerBuf = onDiskBlock.duplicate().position(0).limit(hdrSize);
1761        }
1762        // Do a few checks before we go instantiate HFileBlock.
1763        assert onDiskSizeWithHeader > this.hdrSize;
1764        verifyOnDiskSizeMatchesHeader(onDiskSizeWithHeader, headerBuf, offset, checksumSupport);
1765        ByteBuff curBlock = onDiskBlock.duplicate().position(0).limit(onDiskSizeWithHeader);
1766        // Verify checksum of the data before using it for building HFileBlock.
1767        if (verifyChecksum && !validateChecksum(offset, curBlock, hdrSize)) {
1768          return null;
1769        }
1770        long duration = System.currentTimeMillis() - startTime;
1771        if (updateMetrics) {
1772          HFile.updateReadLatency(duration, pread);
1773        }
1774        // The onDiskBlock will become the headerAndDataBuffer for this block.
1775        // If nextBlockOnDiskSizeWithHeader is not zero, the onDiskBlock already
1776        // contains the header of next block, so no need to set next block's header in it.
1777        HFileBlock hFileBlock = createFromBuff(curBlock, checksumSupport, offset,
1778          nextBlockOnDiskSize, fileContext, intoHeap ? HEAP : allocator);
1779        // Run check on uncompressed sizings.
1780        if (!fileContext.isCompressedOrEncrypted()) {
1781          hFileBlock.sanityCheckUncompressed();
1782        }
1783        LOG.trace("Read {} in {} ns", hFileBlock, duration);
1784        // Cache next block header if we read it for the next time through here.
1785        if (nextBlockOnDiskSize != -1) {
1786          cacheNextBlockHeader(offset + hFileBlock.getOnDiskSizeWithHeader(), onDiskBlock,
1787            onDiskSizeWithHeader, hdrSize);
1788        }
1789        initHFileBlockSuccess = true;
1790        return hFileBlock;
1791      } finally {
1792        if (!initHFileBlockSuccess) {
1793          onDiskBlock.release();
1794        }
1795      }
1796    }
1797
1798    @Override
1799    public void setIncludesMemStoreTS(boolean includesMemstoreTS) {
1800      this.fileContext = new HFileContextBuilder(this.fileContext)
1801        .withIncludesMvcc(includesMemstoreTS).build();
1802    }
1803
1804    @Override
1805    public void setDataBlockEncoder(HFileDataBlockEncoder encoder) {
1806      encodedBlockDecodingCtx = encoder.newDataBlockDecodingContext(this.fileContext);
1807    }
1808
1809    @Override
1810    public HFileBlockDecodingContext getBlockDecodingContext() {
1811      return this.encodedBlockDecodingCtx;
1812    }
1813
1814    @Override
1815    public HFileBlockDecodingContext getDefaultBlockDecodingContext() {
1816      return this.defaultDecodingCtx;
1817    }
1818
1819    /**
1820     * Generates the checksum for the header as well as the data and then validates it.
1821     * If the block doesn't uses checksum, returns false.
1822     * @return True if checksum matches, else false.
1823     */
1824    private boolean validateChecksum(long offset, ByteBuff data, int hdrSize) {
1825      // If this is an older version of the block that does not have checksums, then return false
1826      // indicating that checksum verification did not succeed. Actually, this method should never
1827      // be called when the minorVersion is 0, thus this is a defensive check for a cannot-happen
1828      // case. Since this is a cannot-happen case, it is better to return false to indicate a
1829      // checksum validation failure.
1830      if (!fileContext.isUseHBaseChecksum()) {
1831        return false;
1832      }
1833      return ChecksumUtil.validateChecksum(data, pathName, offset, hdrSize);
1834    }
1835
1836    @Override
1837    public void closeStreams() throws IOException {
1838      streamWrapper.close();
1839    }
1840
1841    @Override
1842    public void unbufferStream() {
1843      // To handle concurrent reads, ensure that no other client is accessing the streams while we
1844      // unbuffer it.
1845      if (streamLock.tryLock()) {
1846        try {
1847          this.streamWrapper.unbuffer();
1848        } finally {
1849          streamLock.unlock();
1850        }
1851      }
1852    }
1853
1854    @Override
1855    public String toString() {
1856      return "hfs=" + hfs + ", path=" + pathName + ", fileContext=" + fileContext;
1857    }
1858  }
1859
1860  /** An additional sanity-check in case no compression or encryption is being used. */
1861  @VisibleForTesting
1862  void sanityCheckUncompressed() throws IOException {
1863    if (onDiskSizeWithoutHeader != uncompressedSizeWithoutHeader +
1864        totalChecksumBytes()) {
1865      throw new IOException("Using no compression but "
1866          + "onDiskSizeWithoutHeader=" + onDiskSizeWithoutHeader + ", "
1867          + "uncompressedSizeWithoutHeader=" + uncompressedSizeWithoutHeader
1868          + ", numChecksumbytes=" + totalChecksumBytes());
1869    }
1870  }
1871
1872  // Cacheable implementation
1873  @Override
1874  public int getSerializedLength() {
1875    if (buf != null) {
1876      // Include extra bytes for block metadata.
1877      return this.buf.limit() + BLOCK_METADATA_SPACE;
1878    }
1879    return 0;
1880  }
1881
1882  // Cacheable implementation
1883  @Override
1884  public void serialize(ByteBuffer destination, boolean includeNextBlockMetadata) {
1885    this.buf.get(destination, 0, getSerializedLength() - BLOCK_METADATA_SPACE);
1886    destination = addMetaData(destination, includeNextBlockMetadata);
1887
1888    // Make it ready for reading. flip sets position to zero and limit to current position which
1889    // is what we want if we do not want to serialize the block plus checksums if present plus
1890    // metadata.
1891    destination.flip();
1892  }
1893
1894  /**
1895   * For use by bucketcache. This exposes internals.
1896   */
1897  public ByteBuffer getMetaData() {
1898    ByteBuffer bb = ByteBuffer.allocate(BLOCK_METADATA_SPACE);
1899    bb = addMetaData(bb, true);
1900    bb.flip();
1901    return bb;
1902  }
1903
1904  /**
1905   * Adds metadata at current position (position is moved forward). Does not flip or reset.
1906   * @return The passed <code>destination</code> with metadata added.
1907   */
1908  private ByteBuffer addMetaData(final ByteBuffer destination, boolean includeNextBlockMetadata) {
1909    destination.put(this.fileContext.isUseHBaseChecksum() ? (byte) 1 : (byte) 0);
1910    destination.putLong(this.offset);
1911    if (includeNextBlockMetadata) {
1912      destination.putInt(this.nextBlockOnDiskSize);
1913    }
1914    return destination;
1915  }
1916
1917  // Cacheable implementation
1918  @Override
1919  public CacheableDeserializer<Cacheable> getDeserializer() {
1920    return HFileBlock.BLOCK_DESERIALIZER;
1921  }
1922
1923  @Override
1924  public int hashCode() {
1925    int result = 1;
1926    result = result * 31 + blockType.hashCode();
1927    result = result * 31 + nextBlockOnDiskSize;
1928    result = result * 31 + (int) (offset ^ (offset >>> 32));
1929    result = result * 31 + onDiskSizeWithoutHeader;
1930    result = result * 31 + (int) (prevBlockOffset ^ (prevBlockOffset >>> 32));
1931    result = result * 31 + uncompressedSizeWithoutHeader;
1932    result = result * 31 + buf.hashCode();
1933    return result;
1934  }
1935
1936  @Override
1937  public boolean equals(Object comparison) {
1938    if (this == comparison) {
1939      return true;
1940    }
1941    if (comparison == null) {
1942      return false;
1943    }
1944    if (!(comparison instanceof HFileBlock)) {
1945      return false;
1946    }
1947
1948    HFileBlock castedComparison = (HFileBlock) comparison;
1949
1950    if (castedComparison.blockType != this.blockType) {
1951      return false;
1952    }
1953    if (castedComparison.nextBlockOnDiskSize != this.nextBlockOnDiskSize) {
1954      return false;
1955    }
1956    // Offset is important. Needed when we have to remake cachekey when block is returned to cache.
1957    if (castedComparison.offset != this.offset) {
1958      return false;
1959    }
1960    if (castedComparison.onDiskSizeWithoutHeader != this.onDiskSizeWithoutHeader) {
1961      return false;
1962    }
1963    if (castedComparison.prevBlockOffset != this.prevBlockOffset) {
1964      return false;
1965    }
1966    if (castedComparison.uncompressedSizeWithoutHeader != this.uncompressedSizeWithoutHeader) {
1967      return false;
1968    }
1969    if (ByteBuff.compareTo(this.buf, 0, this.buf.limit(), castedComparison.buf, 0,
1970        castedComparison.buf.limit()) != 0) {
1971      return false;
1972    }
1973    return true;
1974  }
1975
1976  DataBlockEncoding getDataBlockEncoding() {
1977    if (blockType == BlockType.ENCODED_DATA) {
1978      return DataBlockEncoding.getEncodingById(getDataBlockEncodingId());
1979    }
1980    return DataBlockEncoding.NONE;
1981  }
1982
1983  @VisibleForTesting
1984  byte getChecksumType() {
1985    return this.fileContext.getChecksumType().getCode();
1986  }
1987
1988  int getBytesPerChecksum() {
1989    return this.fileContext.getBytesPerChecksum();
1990  }
1991
1992  /** @return the size of data on disk + header. Excludes checksum. */
1993  @VisibleForTesting
1994  int getOnDiskDataSizeWithHeader() {
1995    return this.onDiskDataSizeWithHeader;
1996  }
1997
1998  /**
1999   * Calculate the number of bytes required to store all the checksums
2000   * for this block. Each checksum value is a 4 byte integer.
2001   */
2002  int totalChecksumBytes() {
2003    // If the hfile block has minorVersion 0, then there are no checksum
2004    // data to validate. Similarly, a zero value in this.bytesPerChecksum
2005    // indicates that cached blocks do not have checksum data because
2006    // checksums were already validated when the block was read from disk.
2007    if (!fileContext.isUseHBaseChecksum() || this.fileContext.getBytesPerChecksum() == 0) {
2008      return 0;
2009    }
2010    return (int) ChecksumUtil.numBytes(onDiskDataSizeWithHeader,
2011        this.fileContext.getBytesPerChecksum());
2012  }
2013
2014  /**
2015   * Returns the size of this block header.
2016   */
2017  public int headerSize() {
2018    return headerSize(this.fileContext.isUseHBaseChecksum());
2019  }
2020
2021  /**
2022   * Maps a minor version to the size of the header.
2023   */
2024  public static int headerSize(boolean usesHBaseChecksum) {
2025    return usesHBaseChecksum?
2026        HConstants.HFILEBLOCK_HEADER_SIZE: HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM;
2027  }
2028
2029  /**
2030   * Return the appropriate DUMMY_HEADER for the minor version
2031   */
2032  @VisibleForTesting
2033  // TODO: Why is this in here?
2034  byte[] getDummyHeaderForVersion() {
2035    return getDummyHeaderForVersion(this.fileContext.isUseHBaseChecksum());
2036  }
2037
2038  /**
2039   * Return the appropriate DUMMY_HEADER for the minor version
2040   */
2041  static private byte[] getDummyHeaderForVersion(boolean usesHBaseChecksum) {
2042    return usesHBaseChecksum? HConstants.HFILEBLOCK_DUMMY_HEADER: DUMMY_HEADER_NO_CHECKSUM;
2043  }
2044
2045  /**
2046   * @return This HFileBlocks fileContext which will a derivative of the
2047   *   fileContext for the file from which this block's data was originally read.
2048   */
2049  HFileContext getHFileContext() {
2050    return this.fileContext;
2051  }
2052
2053  /**
2054   * Convert the contents of the block header into a human readable string.
2055   * This is mostly helpful for debugging. This assumes that the block
2056   * has minor version > 0.
2057   */
2058  @VisibleForTesting
2059  static String toStringHeader(ByteBuff buf) throws IOException {
2060    byte[] magicBuf = new byte[Math.min(buf.limit() - buf.position(), BlockType.MAGIC_LENGTH)];
2061    buf.get(magicBuf);
2062    BlockType bt = BlockType.parse(magicBuf, 0, BlockType.MAGIC_LENGTH);
2063    int compressedBlockSizeNoHeader = buf.getInt();
2064    int uncompressedBlockSizeNoHeader = buf.getInt();
2065    long prevBlockOffset = buf.getLong();
2066    byte cksumtype = buf.get();
2067    long bytesPerChecksum = buf.getInt();
2068    long onDiskDataSizeWithHeader = buf.getInt();
2069    return " Header dump: magic: " + Bytes.toString(magicBuf) +
2070                   " blockType " + bt +
2071                   " compressedBlockSizeNoHeader " +
2072                   compressedBlockSizeNoHeader +
2073                   " uncompressedBlockSizeNoHeader " +
2074                   uncompressedBlockSizeNoHeader +
2075                   " prevBlockOffset " + prevBlockOffset +
2076                   " checksumType " + ChecksumType.codeToType(cksumtype) +
2077                   " bytesPerChecksum " + bytesPerChecksum +
2078                   " onDiskDataSizeWithHeader " + onDiskDataSizeWithHeader;
2079  }
2080
2081  private static HFileBlockBuilder createBuilder(HFileBlock blk){
2082    return new HFileBlockBuilder()
2083          .withBlockType(blk.blockType)
2084          .withOnDiskSizeWithoutHeader(blk.onDiskSizeWithoutHeader)
2085          .withUncompressedSizeWithoutHeader(blk.uncompressedSizeWithoutHeader)
2086          .withPrevBlockOffset(blk.prevBlockOffset)
2087          .withByteBuff(blk.buf.duplicate()) // Duplicate the buffer.
2088          .withOffset(blk.offset)
2089          .withOnDiskDataSizeWithHeader(blk.onDiskDataSizeWithHeader)
2090          .withNextBlockOnDiskSize(blk.nextBlockOnDiskSize)
2091          .withHFileContext(blk.fileContext)
2092          .withByteBuffAllocator(blk.allocator)
2093          .withShared(blk.isSharedMem());
2094  }
2095
2096  static HFileBlock shallowClone(HFileBlock blk) {
2097    return createBuilder(blk).build();
2098  }
2099
2100  static HFileBlock deepCloneOnHeap(HFileBlock blk) {
2101    ByteBuff deepCloned = ByteBuff.wrap(ByteBuffer.wrap(blk.buf.toBytes(0, blk.buf.limit())));
2102    return createBuilder(blk).withByteBuff(deepCloned).withShared(false).build();
2103  }
2104}