View Javadoc

1   /*
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.io.hfile;
20  
21  import java.io.ByteArrayOutputStream;
22  import java.io.DataInput;
23  import java.io.DataInputStream;
24  import java.io.DataOutput;
25  import java.io.DataOutputStream;
26  import java.io.IOException;
27  import java.nio.ByteBuffer;
28  import java.util.ArrayList;
29  import java.util.Arrays;
30  import java.util.Collections;
31  import java.util.List;
32  import java.util.concurrent.atomic.AtomicReference;
33  
34  import org.apache.commons.logging.Log;
35  import org.apache.commons.logging.LogFactory;
36  import org.apache.hadoop.classification.InterfaceAudience;
37  import org.apache.hadoop.conf.Configuration;
38  import org.apache.hadoop.fs.FSDataOutputStream;
39  import org.apache.hadoop.hbase.HConstants;
40  import org.apache.hadoop.hbase.KeyValue;
41  import org.apache.hadoop.hbase.KeyValue.KVComparator;
42  import org.apache.hadoop.hbase.io.HeapSize;
43  import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
44  import org.apache.hadoop.hbase.io.hfile.HFile.CachingBlockReader;
45  import org.apache.hadoop.hbase.util.Bytes;
46  import org.apache.hadoop.hbase.util.ClassSize;
47  import org.apache.hadoop.hbase.util.CompoundBloomFilterWriter;
48  import org.apache.hadoop.io.WritableUtils;
49  import org.apache.hadoop.util.StringUtils;
50  
51  /**
52   * Provides functionality to write ({@link BlockIndexWriter}) and read
53   * ({@link BlockIndexReader}) single-level and multi-level block indexes.
54   *
55   * Examples of how to use the block index writer can be found in
56   * {@link CompoundBloomFilterWriter} and {@link HFileWriterV2}. Examples of how
57   * to use the reader can be found in {@link HFileReaderV2} and
58   * TestHFileBlockIndex.
59   */
60  @InterfaceAudience.Private
61  public class HFileBlockIndex {
62  
63    private static final Log LOG = LogFactory.getLog(HFileBlockIndex.class);
64  
65    static final int DEFAULT_MAX_CHUNK_SIZE = 128 * 1024;
66  
67    /**
68     * The maximum size guideline for index blocks (both leaf, intermediate, and
69     * root). If not specified, <code>DEFAULT_MAX_CHUNK_SIZE</code> is used.
70     */
71    public static final String MAX_CHUNK_SIZE_KEY = "hfile.index.block.max.size";
72  
73    /**
74     * The number of bytes stored in each "secondary index" entry in addition to
75     * key bytes in the non-root index block format. The first long is the file
76     * offset of the deeper-level block the entry points to, and the int that
77     * follows is that block's on-disk size without including header.
78     */
79    static final int SECONDARY_INDEX_ENTRY_OVERHEAD = Bytes.SIZEOF_INT
80        + Bytes.SIZEOF_LONG;
81  
82    /**
83     * Error message when trying to use inline block API in single-level mode.
84     */
85    private static final String INLINE_BLOCKS_NOT_ALLOWED =
86        "Inline blocks are not allowed in the single-level-only mode";
87  
88    /**
89     * The size of a meta-data record used for finding the mid-key in a
90     * multi-level index. Consists of the middle leaf-level index block offset
91     * (long), its on-disk size without header included (int), and the mid-key
92     * entry's zero-based index in that leaf index block.
93     */
94    private static final int MID_KEY_METADATA_SIZE = Bytes.SIZEOF_LONG +
95        2 * Bytes.SIZEOF_INT;
96  
97    /**
98     * The reader will always hold the root level index in the memory. Index
99     * blocks at all other levels will be cached in the LRU cache in practice,
100    * although this API does not enforce that.
101    *
102    * All non-root (leaf and intermediate) index blocks contain what we call a
103    * "secondary index": an array of offsets to the entries within the block.
104    * This allows us to do binary search for the entry corresponding to the
105    * given key without having to deserialize the block.
106    */
107   public static class BlockIndexReader implements HeapSize {
108     /** Needed doing lookup on blocks. */
109     private final KVComparator comparator;
110 
111     // Root-level data.
112     private byte[][] blockKeys;
113     private long[] blockOffsets;
114     private int[] blockDataSizes;
115     private int rootCount = 0;
116 
117     // Mid-key metadata.
118     private long midLeafBlockOffset = -1;
119     private int midLeafBlockOnDiskSize = -1;
120     private int midKeyEntry = -1;
121 
122     /** Pre-computed mid-key */
123     private AtomicReference<byte[]> midKey = new AtomicReference<byte[]>();
124 
125     /**
126      * The number of levels in the block index tree. One if there is only root
127      * level, two for root and leaf levels, etc.
128      */
129     private int searchTreeLevel;
130 
131     /** A way to read {@link HFile} blocks at a given offset */
132     private CachingBlockReader cachingBlockReader;
133 
134     public BlockIndexReader(final KVComparator c, final int treeLevel,
135         final CachingBlockReader cachingBlockReader) {
136       this(c, treeLevel);
137       this.cachingBlockReader = cachingBlockReader;
138     }
139 
140     public BlockIndexReader(final KVComparator c, final int treeLevel)
141     {
142       comparator = c;
143       searchTreeLevel = treeLevel;
144     }
145 
146     /**
147      * @return true if the block index is empty.
148      */
149     public boolean isEmpty() {
150       return blockKeys.length == 0;
151     }
152 
153     /**
154      * Verifies that the block index is non-empty and throws an
155      * {@link IllegalStateException} otherwise.
156      */
157     public void ensureNonEmpty() {
158       if (blockKeys.length == 0) {
159         throw new IllegalStateException("Block index is empty or not loaded");
160       }
161     }
162 
163     /**
164      * Return the data block which contains this key. This function will only
165      * be called when the HFile version is larger than 1.
166      *
167      * @param key the key we are looking for
168      * @param keyOffset the offset of the key in its byte array
169      * @param keyLength the length of the key
170      * @param currentBlock the current block, to avoid re-reading the same
171      *          block
172      * @return reader a basic way to load blocks
173      * @throws IOException
174      */
175     public HFileBlock seekToDataBlock(final byte[] key, int keyOffset,
176         int keyLength, HFileBlock currentBlock, boolean cacheBlocks,
177         boolean pread, boolean isCompaction)
178         throws IOException {
179       BlockWithScanInfo blockWithScanInfo = loadDataBlockWithScanInfo(key, keyOffset, keyLength,
180           currentBlock, cacheBlocks, pread, isCompaction);
181       if (blockWithScanInfo == null) {
182         return null;
183       } else {
184         return blockWithScanInfo.getHFileBlock();
185       }
186     }
187 
188     /**
189      * Return the BlockWithScanInfo which contains the DataBlock with other scan info
190      * such as nextIndexedKey.
191      * This function will only be called when the HFile version is larger than 1.
192      *
193      * @param key the key we are looking for
194      * @param keyOffset the offset of the key in its byte array
195      * @param keyLength the length of the key
196      * @param currentBlock the current block, to avoid re-reading the same
197      *          block
198      * @param cacheBlocks
199      * @param pread
200      * @param isCompaction
201      * @return the BlockWithScanInfo which contains the DataBlock with other scan info
202      *         such as nextIndexedKey.
203      * @throws IOException
204      */
205     public BlockWithScanInfo loadDataBlockWithScanInfo(final byte[] key, int keyOffset,
206         int keyLength, HFileBlock currentBlock, boolean cacheBlocks,
207         boolean pread, boolean isCompaction)
208         throws IOException {
209       int rootLevelIndex = rootBlockContainingKey(key, keyOffset, keyLength);
210       if (rootLevelIndex < 0 || rootLevelIndex >= blockOffsets.length) {
211         return null;
212       }
213 
214       // the next indexed key
215       byte[] nextIndexedKey = null;
216 
217       // Read the next-level (intermediate or leaf) index block.
218       long currentOffset = blockOffsets[rootLevelIndex];
219       int currentOnDiskSize = blockDataSizes[rootLevelIndex];
220 
221       if (rootLevelIndex < blockKeys.length - 1) {
222         nextIndexedKey = blockKeys[rootLevelIndex + 1];
223       } else {
224         nextIndexedKey = HConstants.NO_NEXT_INDEXED_KEY;
225       }
226 
227       int lookupLevel = 1; // How many levels deep we are in our lookup.
228       int index = -1;
229 
230       HFileBlock block;
231       while (true) {
232 
233         if (currentBlock != null && currentBlock.getOffset() == currentOffset)
234         {
235           // Avoid reading the same block again, even with caching turned off.
236           // This is crucial for compaction-type workload which might have
237           // caching turned off. This is like a one-block cache inside the
238           // scanner.
239           block = currentBlock;
240         } else {
241           // Call HFile's caching block reader API. We always cache index
242           // blocks, otherwise we might get terrible performance.
243           boolean shouldCache = cacheBlocks || (lookupLevel < searchTreeLevel);
244           BlockType expectedBlockType;
245           if (lookupLevel < searchTreeLevel - 1) {
246             expectedBlockType = BlockType.INTERMEDIATE_INDEX;
247           } else if (lookupLevel == searchTreeLevel - 1) {
248             expectedBlockType = BlockType.LEAF_INDEX;
249           } else {
250             // this also accounts for ENCODED_DATA
251             expectedBlockType = BlockType.DATA;
252           }
253           block = cachingBlockReader.readBlock(currentOffset,
254               currentOnDiskSize, shouldCache, pread, isCompaction,
255               expectedBlockType);
256         }
257 
258         if (block == null) {
259           throw new IOException("Failed to read block at offset " +
260               currentOffset + ", onDiskSize=" + currentOnDiskSize);
261         }
262 
263         // Found a data block, break the loop and check our level in the tree.
264         if (block.getBlockType().isData()) {
265           break;
266         }
267 
268         // Not a data block. This must be a leaf-level or intermediate-level
269         // index block. We don't allow going deeper than searchTreeLevel.
270         if (++lookupLevel > searchTreeLevel) {
271           throw new IOException("Search Tree Level overflow: lookupLevel="+
272               lookupLevel + ", searchTreeLevel=" + searchTreeLevel);
273         }
274 
275         // Locate the entry corresponding to the given key in the non-root
276         // (leaf or intermediate-level) index block.
277         ByteBuffer buffer = block.getBufferWithoutHeader();
278         index = locateNonRootIndexEntry(buffer, key, keyOffset, keyLength, comparator);
279         if (index == -1) {
280           throw new IOException("The key "
281               + Bytes.toStringBinary(key, keyOffset, keyLength)
282               + " is before the" + " first key of the non-root index block "
283               + block);
284         }
285 
286         currentOffset = buffer.getLong();
287         currentOnDiskSize = buffer.getInt();
288 
289         // Only update next indexed key if there is a next indexed key in the current level
290         byte[] tmpNextIndexedKey = getNonRootIndexedKey(buffer, index + 1);
291         if (tmpNextIndexedKey != null) {
292           nextIndexedKey = tmpNextIndexedKey;
293         }
294       }
295 
296       if (lookupLevel != searchTreeLevel) {
297         throw new IOException("Reached a data block at level " + lookupLevel +
298             " but the number of levels is " + searchTreeLevel);
299       }
300 
301       // set the next indexed key for the current block.
302       BlockWithScanInfo blockWithScanInfo = new BlockWithScanInfo(block, nextIndexedKey);
303       return blockWithScanInfo;
304     }
305 
306     /**
307      * An approximation to the {@link HFile}'s mid-key. Operates on block
308      * boundaries, and does not go inside blocks. In other words, returns the
309      * first key of the middle block of the file.
310      *
311      * @return the first key of the middle block
312      */
313     public byte[] midkey() throws IOException {
314       if (rootCount == 0)
315         throw new IOException("HFile empty");
316 
317       byte[] targetMidKey = this.midKey.get();
318       if (targetMidKey != null) {
319         return targetMidKey;
320       }
321 
322       if (midLeafBlockOffset >= 0) {
323         if (cachingBlockReader == null) {
324           throw new IOException("Have to read the middle leaf block but " +
325               "no block reader available");
326         }
327 
328         // Caching, using pread, assuming this is not a compaction.
329         HFileBlock midLeafBlock = cachingBlockReader.readBlock(
330             midLeafBlockOffset, midLeafBlockOnDiskSize, true, true, false,
331             BlockType.LEAF_INDEX);
332 
333         ByteBuffer b = midLeafBlock.getBufferWithoutHeader();
334         int numDataBlocks = b.getInt();
335         int keyRelOffset = b.getInt(Bytes.SIZEOF_INT * (midKeyEntry + 1));
336         int keyLen = b.getInt(Bytes.SIZEOF_INT * (midKeyEntry + 2)) -
337             keyRelOffset;
338         int keyOffset = b.arrayOffset() +
339             Bytes.SIZEOF_INT * (numDataBlocks + 2) + keyRelOffset +
340             SECONDARY_INDEX_ENTRY_OVERHEAD;
341         targetMidKey = Arrays.copyOfRange(b.array(), keyOffset, keyOffset + keyLen);
342       } else {
343         // The middle of the root-level index.
344         targetMidKey = blockKeys[rootCount / 2];
345       }
346 
347       this.midKey.set(targetMidKey);
348       return targetMidKey;
349     }
350 
351     /**
352      * @param i from 0 to {@link #getRootBlockCount() - 1}
353      */
354     public byte[] getRootBlockKey(int i) {
355       return blockKeys[i];
356     }
357 
358     /**
359      * @param i from 0 to {@link #getRootBlockCount() - 1}
360      */
361     public long getRootBlockOffset(int i) {
362       return blockOffsets[i];
363     }
364 
365     /**
366      * @param i zero-based index of a root-level block
367      * @return the on-disk size of the root-level block for version 2, or the
368      *         uncompressed size for version 1
369      */
370     public int getRootBlockDataSize(int i) {
371       return blockDataSizes[i];
372     }
373 
374     /**
375      * @return the number of root-level blocks in this block index
376      */
377     public int getRootBlockCount() {
378       return rootCount;
379     }
380 
381     /**
382      * Finds the root-level index block containing the given key.
383      *
384      * @param key
385      *          Key to find
386      * @return Offset of block containing <code>key</code> (between 0 and the
387      *         number of blocks - 1) or -1 if this file does not contain the
388      *         request.
389      */
390     public int rootBlockContainingKey(final byte[] key, int offset,
391         int length) {
392       int pos = Bytes.binarySearch(blockKeys, key, offset, length,
393           comparator);
394       // pos is between -(blockKeys.length + 1) to blockKeys.length - 1, see
395       // binarySearch's javadoc.
396 
397       if (pos >= 0) {
398         // This means this is an exact match with an element of blockKeys.
399         assert pos < blockKeys.length;
400         return pos;
401       }
402 
403       // Otherwise, pos = -(i + 1), where blockKeys[i - 1] < key < blockKeys[i],
404       // and i is in [0, blockKeys.length]. We are returning j = i - 1 such that
405       // blockKeys[j] <= key < blockKeys[j + 1]. In particular, j = -1 if
406       // key < blockKeys[0], meaning the file does not contain the given key.
407 
408       int i = -pos - 1;
409       assert 0 <= i && i <= blockKeys.length;
410       return i - 1;
411     }
412 
413     /**
414      * Adds a new entry in the root block index. Only used when reading.
415      *
416      * @param key Last key in the block
417      * @param offset file offset where the block is stored
418      * @param dataSize the uncompressed data size
419      */
420     private void add(final byte[] key, final long offset, final int dataSize) {
421       blockOffsets[rootCount] = offset;
422       blockKeys[rootCount] = key;
423       blockDataSizes[rootCount] = dataSize;
424       rootCount++;
425     }
426 
427     /**
428      * The indexed key at the ith position in the nonRootIndex. The position starts at 0.
429      * @param nonRootIndex
430      * @param i the ith position
431      * @return The indexed key at the ith position in the nonRootIndex.
432      */
433     private byte[] getNonRootIndexedKey(ByteBuffer nonRootIndex, int i) {
434       int numEntries = nonRootIndex.getInt(0);
435       if (i < 0 || i >= numEntries) {
436         return null;
437       }
438 
439       // Entries start after the number of entries and the secondary index.
440       // The secondary index takes numEntries + 1 ints.
441       int entriesOffset = Bytes.SIZEOF_INT * (numEntries + 2);
442       // Targetkey's offset relative to the end of secondary index
443       int targetKeyRelOffset = nonRootIndex.getInt(
444           Bytes.SIZEOF_INT * (i + 1));
445 
446       // The offset of the target key in the blockIndex buffer
447       int targetKeyOffset = entriesOffset     // Skip secondary index
448           + targetKeyRelOffset               // Skip all entries until mid
449           + SECONDARY_INDEX_ENTRY_OVERHEAD;  // Skip offset and on-disk-size
450 
451       // We subtract the two consecutive secondary index elements, which
452       // gives us the size of the whole (offset, onDiskSize, key) tuple. We
453       // then need to subtract the overhead of offset and onDiskSize.
454       int targetKeyLength = nonRootIndex.getInt(Bytes.SIZEOF_INT * (i + 2)) -
455         targetKeyRelOffset - SECONDARY_INDEX_ENTRY_OVERHEAD;
456 
457       int from = nonRootIndex.arrayOffset() + targetKeyOffset;
458       int to = from + targetKeyLength;
459       return Arrays.copyOfRange(nonRootIndex.array(), from, to);
460     }
461 
462     /**
463      * Performs a binary search over a non-root level index block. Utilizes the
464      * secondary index, which records the offsets of (offset, onDiskSize,
465      * firstKey) tuples of all entries.
466      *
467      * @param key the key we are searching for offsets to individual entries in
468      *          the blockIndex buffer
469      * @param keyOffset the offset of the key in its byte array
470      * @param keyLength the length of the key
471      * @param nonRootIndex the non-root index block buffer, starting with the
472      *          secondary index. The position is ignored.
473      * @return the index i in [0, numEntries - 1] such that keys[i] <= key <
474      *         keys[i + 1], if keys is the array of all keys being searched, or
475      *         -1 otherwise
476      * @throws IOException
477      */
478     static int binarySearchNonRootIndex(byte[] key, int keyOffset,
479         int keyLength, ByteBuffer nonRootIndex,
480         KVComparator comparator) {
481 
482       int numEntries = nonRootIndex.getInt(0);
483       int low = 0;
484       int high = numEntries - 1;
485       int mid = 0;
486 
487       // Entries start after the number of entries and the secondary index.
488       // The secondary index takes numEntries + 1 ints.
489       int entriesOffset = Bytes.SIZEOF_INT * (numEntries + 2);
490 
491       // If we imagine that keys[-1] = -Infinity and
492       // keys[numEntries] = Infinity, then we are maintaining an invariant that
493       // keys[low - 1] < key < keys[high + 1] while narrowing down the range.
494 
495       while (low <= high) {
496         mid = (low + high) >>> 1;
497 
498         // Midkey's offset relative to the end of secondary index
499         int midKeyRelOffset = nonRootIndex.getInt(
500             Bytes.SIZEOF_INT * (mid + 1));
501 
502         // The offset of the middle key in the blockIndex buffer
503         int midKeyOffset = entriesOffset       // Skip secondary index
504             + midKeyRelOffset                  // Skip all entries until mid
505             + SECONDARY_INDEX_ENTRY_OVERHEAD;  // Skip offset and on-disk-size
506 
507         // We subtract the two consecutive secondary index elements, which
508         // gives us the size of the whole (offset, onDiskSize, key) tuple. We
509         // then need to subtract the overhead of offset and onDiskSize.
510         int midLength = nonRootIndex.getInt(Bytes.SIZEOF_INT * (mid + 2)) -
511             midKeyRelOffset - SECONDARY_INDEX_ENTRY_OVERHEAD;
512 
513         // we have to compare in this order, because the comparator order
514         // has special logic when the 'left side' is a special key.
515         int cmp = comparator.compareFlatKey(key, keyOffset, keyLength,
516             nonRootIndex.array(), nonRootIndex.arrayOffset() + midKeyOffset,
517             midLength);
518 
519         // key lives above the midpoint
520         if (cmp > 0)
521           low = mid + 1; // Maintain the invariant that keys[low - 1] < key
522         // key lives below the midpoint
523         else if (cmp < 0)
524           high = mid - 1; // Maintain the invariant that key < keys[high + 1]
525         else
526           return mid; // exact match
527       }
528 
529       // As per our invariant, keys[low - 1] < key < keys[high + 1], meaning
530       // that low - 1 < high + 1 and (low - high) <= 1. As per the loop break
531       // condition, low >= high + 1. Therefore, low = high + 1.
532 
533       if (low != high + 1) {
534         throw new IllegalStateException("Binary search broken: low=" + low
535             + " " + "instead of " + (high + 1));
536       }
537 
538       // OK, our invariant says that keys[low - 1] < key < keys[low]. We need to
539       // return i such that keys[i] <= key < keys[i + 1]. Therefore i = low - 1.
540       int i = low - 1;
541 
542       // Some extra validation on the result.
543       if (i < -1 || i >= numEntries) {
544         throw new IllegalStateException("Binary search broken: result is " +
545             i + " but expected to be between -1 and (numEntries - 1) = " +
546             (numEntries - 1));
547       }
548 
549       return i;
550     }
551 
552     /**
553      * Search for one key using the secondary index in a non-root block. In case
554      * of success, positions the provided buffer at the entry of interest, where
555      * the file offset and the on-disk-size can be read.
556      *
557      * @param nonRootBlock a non-root block without header. Initial position
558      *          does not matter.
559      * @param key the byte array containing the key
560      * @param keyOffset the offset of the key in its byte array
561      * @param keyLength the length of the key
562      * @return the index position where the given key was found,
563      *         otherwise return -1 in the case the given key is before the first key.
564      *
565      */
566     static int locateNonRootIndexEntry(ByteBuffer nonRootBlock, byte[] key,
567         int keyOffset, int keyLength, KVComparator comparator) {
568       int entryIndex = binarySearchNonRootIndex(key, keyOffset, keyLength,
569           nonRootBlock, comparator);
570 
571       if (entryIndex != -1) {
572         int numEntries = nonRootBlock.getInt(0);
573 
574         // The end of secondary index and the beginning of entries themselves.
575         int entriesOffset = Bytes.SIZEOF_INT * (numEntries + 2);
576 
577         // The offset of the entry we are interested in relative to the end of
578         // the secondary index.
579         int entryRelOffset = nonRootBlock.getInt(Bytes.SIZEOF_INT
580             * (1 + entryIndex));
581 
582         nonRootBlock.position(entriesOffset + entryRelOffset);
583       }
584 
585       return entryIndex;
586     }
587 
588     /**
589      * Read in the root-level index from the given input stream. Must match
590      * what was written into the root level by
591      * {@link BlockIndexWriter#writeIndexBlocks(FSDataOutputStream)} at the
592      * offset that function returned.
593      *
594      * @param in the buffered input stream or wrapped byte input stream
595      * @param numEntries the number of root-level index entries
596      * @throws IOException
597      */
598     public void readRootIndex(DataInput in, final int numEntries)
599         throws IOException {
600       blockOffsets = new long[numEntries];
601       blockKeys = new byte[numEntries][];
602       blockDataSizes = new int[numEntries];
603 
604       // If index size is zero, no index was written.
605       if (numEntries > 0) {
606         for (int i = 0; i < numEntries; ++i) {
607           long offset = in.readLong();
608           int dataSize = in.readInt();
609           byte[] key = Bytes.readByteArray(in);
610           add(key, offset, dataSize);
611         }
612       }
613     }
614     
615     /**
616      * Read in the root-level index from the given input stream. Must match
617      * what was written into the root level by
618      * {@link BlockIndexWriter#writeIndexBlocks(FSDataOutputStream)} at the
619      * offset that function returned.
620      *
621      * @param blk the HFile block
622      * @param numEntries the number of root-level index entries
623      * @return the buffered input stream or wrapped byte input stream
624      * @throws IOException
625      */
626     public DataInputStream readRootIndex(HFileBlock blk, final int numEntries) throws IOException {
627       DataInputStream in = blk.getByteStream();
628       readRootIndex(in, numEntries);
629       return in;
630     }
631 
632     /**
633      * Read the root-level metadata of a multi-level block index. Based on
634      * {@link #readRootIndex(DataInput, int)}, but also reads metadata
635      * necessary to compute the mid-key in a multi-level index.
636      *
637      * @param blk the HFile block
638      * @param numEntries the number of root-level index entries
639      * @throws IOException
640      */
641     public void readMultiLevelIndexRoot(HFileBlock blk,
642         final int numEntries) throws IOException {
643       DataInputStream in = readRootIndex(blk, numEntries);
644       // after reading the root index the checksum bytes have to
645       // be subtracted to know if the mid key exists.
646       int checkSumBytes = blk.totalChecksumBytes();
647       if ((in.available() - checkSumBytes) < MID_KEY_METADATA_SIZE) {
648         // No mid-key metadata available.
649         return;
650       }
651       midLeafBlockOffset = in.readLong();
652       midLeafBlockOnDiskSize = in.readInt();
653       midKeyEntry = in.readInt();
654     }
655 
656     @Override
657     public String toString() {
658       StringBuilder sb = new StringBuilder();
659       sb.append("size=" + rootCount).append("\n");
660       for (int i = 0; i < rootCount; i++) {
661         sb.append("key=").append(KeyValue.keyToString(blockKeys[i]))
662             .append("\n  offset=").append(blockOffsets[i])
663             .append(", dataSize=" + blockDataSizes[i]).append("\n");
664       }
665       return sb.toString();
666     }
667 
668     @Override
669     public long heapSize() {
670       long heapSize = ClassSize.align(6 * ClassSize.REFERENCE +
671           2 * Bytes.SIZEOF_INT + ClassSize.OBJECT);
672 
673       // Mid-key metadata.
674       heapSize += MID_KEY_METADATA_SIZE;
675 
676       // Calculating the size of blockKeys
677       if (blockKeys != null) {
678         // Adding array + references overhead
679         heapSize += ClassSize.align(ClassSize.ARRAY + blockKeys.length
680             * ClassSize.REFERENCE);
681 
682         // Adding bytes
683         for (byte[] key : blockKeys) {
684           heapSize += ClassSize.align(ClassSize.ARRAY + key.length);
685         }
686       }
687 
688       if (blockOffsets != null) {
689         heapSize += ClassSize.align(ClassSize.ARRAY + blockOffsets.length
690             * Bytes.SIZEOF_LONG);
691       }
692 
693       if (blockDataSizes != null) {
694         heapSize += ClassSize.align(ClassSize.ARRAY + blockDataSizes.length
695             * Bytes.SIZEOF_INT);
696       }
697 
698       return ClassSize.align(heapSize);
699     }
700 
701   }
702 
703   /**
704    * Writes the block index into the output stream. Generate the tree from
705    * bottom up. The leaf level is written to disk as a sequence of inline
706    * blocks, if it is larger than a certain number of bytes. If the leaf level
707    * is not large enough, we write all entries to the root level instead.
708    *
709    * After all leaf blocks have been written, we end up with an index
710    * referencing the resulting leaf index blocks. If that index is larger than
711    * the allowed root index size, the writer will break it up into
712    * reasonable-size intermediate-level index block chunks write those chunks
713    * out, and create another index referencing those chunks. This will be
714    * repeated until the remaining index is small enough to become the root
715    * index. However, in most practical cases we will only have leaf-level
716    * blocks and the root index, or just the root index.
717    */
718   public static class BlockIndexWriter implements InlineBlockWriter {
719     /**
720      * While the index is being written, this represents the current block
721      * index referencing all leaf blocks, with one exception. If the file is
722      * being closed and there are not enough blocks to complete even a single
723      * leaf block, no leaf blocks get written and this contains the entire
724      * block index. After all levels of the index were written by
725      * {@link #writeIndexBlocks(FSDataOutputStream)}, this contains the final
726      * root-level index.
727      */
728     private BlockIndexChunk rootChunk = new BlockIndexChunk();
729 
730     /**
731      * Current leaf-level chunk. New entries referencing data blocks get added
732      * to this chunk until it grows large enough to be written to disk.
733      */
734     private BlockIndexChunk curInlineChunk = new BlockIndexChunk();
735 
736     /**
737      * The number of block index levels. This is one if there is only root
738      * level (even empty), two if there a leaf level and root level, and is
739      * higher if there are intermediate levels. This is only final after
740      * {@link #writeIndexBlocks(FSDataOutputStream)} has been called. The
741      * initial value accounts for the root level, and will be increased to two
742      * as soon as we find out there is a leaf-level in
743      * {@link #blockWritten(long, int)}.
744      */
745     private int numLevels = 1;
746 
747     private HFileBlock.Writer blockWriter;
748     private byte[] firstKey = null;
749 
750     /**
751      * The total number of leaf-level entries, i.e. entries referenced by
752      * leaf-level blocks. For the data block index this is equal to the number
753      * of data blocks.
754      */
755     private long totalNumEntries;
756 
757     /** Total compressed size of all index blocks. */
758     private long totalBlockOnDiskSize;
759 
760     /** Total uncompressed size of all index blocks. */
761     private long totalBlockUncompressedSize;
762 
763     /** The maximum size guideline of all multi-level index blocks. */
764     private int maxChunkSize;
765 
766     /** Whether we require this block index to always be single-level. */
767     private boolean singleLevelOnly;
768 
769     /** Block cache, or null if cache-on-write is disabled */
770     private BlockCache blockCache;
771 
772     /** Name to use for computing cache keys */
773     private String nameForCaching;
774 
775     /** Creates a single-level block index writer */
776     public BlockIndexWriter() {
777       this(null, null, null);
778       singleLevelOnly = true;
779     }
780 
781     /**
782      * Creates a multi-level block index writer.
783      *
784      * @param blockWriter the block writer to use to write index blocks
785      * @param blockCache if this is not null, index blocks will be cached
786      *    on write into this block cache.
787      */
788     public BlockIndexWriter(HFileBlock.Writer blockWriter,
789         BlockCache blockCache, String nameForCaching) {
790       if ((blockCache == null) != (nameForCaching == null)) {
791         throw new IllegalArgumentException("Block cache and file name for " +
792             "caching must be both specified or both null");
793       }
794 
795       this.blockWriter = blockWriter;
796       this.blockCache = blockCache;
797       this.nameForCaching = nameForCaching;
798       this.maxChunkSize = HFileBlockIndex.DEFAULT_MAX_CHUNK_SIZE;
799     }
800 
801     public void setMaxChunkSize(int maxChunkSize) {
802       if (maxChunkSize <= 0) {
803         throw new IllegalArgumentException("Invald maximum index block size");
804       }
805       this.maxChunkSize = maxChunkSize;
806     }
807 
808     /**
809      * Writes the root level and intermediate levels of the block index into
810      * the output stream, generating the tree from bottom up. Assumes that the
811      * leaf level has been inline-written to the disk if there is enough data
812      * for more than one leaf block. We iterate by breaking the current level
813      * of the block index, starting with the index of all leaf-level blocks,
814      * into chunks small enough to be written to disk, and generate its parent
815      * level, until we end up with a level small enough to become the root
816      * level.
817      *
818      * If the leaf level is not large enough, there is no inline block index
819      * anymore, so we only write that level of block index to disk as the root
820      * level.
821      *
822      * @param out FSDataOutputStream
823      * @return position at which we entered the root-level index.
824      * @throws IOException
825      */
826     public long writeIndexBlocks(FSDataOutputStream out) throws IOException {
827       if (curInlineChunk != null && curInlineChunk.getNumEntries() != 0) {
828         throw new IOException("Trying to write a multi-level block index, " +
829             "but are " + curInlineChunk.getNumEntries() + " entries in the " +
830             "last inline chunk.");
831       }
832 
833       // We need to get mid-key metadata before we create intermediate
834       // indexes and overwrite the root chunk.
835       byte[] midKeyMetadata = numLevels > 1 ? rootChunk.getMidKeyMetadata()
836           : null;
837 
838       if (curInlineChunk != null) {
839         while (rootChunk.getRootSize() > maxChunkSize) {
840           rootChunk = writeIntermediateLevel(out, rootChunk);
841           numLevels += 1;
842         }
843       }
844 
845       // write the root level
846       long rootLevelIndexPos = out.getPos();
847 
848       {
849         DataOutput blockStream =
850             blockWriter.startWriting(BlockType.ROOT_INDEX);
851         rootChunk.writeRoot(blockStream);
852         if (midKeyMetadata != null)
853           blockStream.write(midKeyMetadata);
854         blockWriter.writeHeaderAndData(out);
855       }
856 
857       // Add root index block size
858       totalBlockOnDiskSize += blockWriter.getOnDiskSizeWithoutHeader();
859       totalBlockUncompressedSize +=
860           blockWriter.getUncompressedSizeWithoutHeader();
861 
862       if (LOG.isTraceEnabled()) {
863         LOG.trace("Wrote a " + numLevels + "-level index with root level at pos "
864           + rootLevelIndexPos + ", " + rootChunk.getNumEntries()
865           + " root-level entries, " + totalNumEntries + " total entries, "
866           + StringUtils.humanReadableInt(this.totalBlockOnDiskSize) +
867           " on-disk size, "
868           + StringUtils.humanReadableInt(totalBlockUncompressedSize) +
869           " total uncompressed size.");
870       }
871       return rootLevelIndexPos;
872     }
873 
874     /**
875      * Writes the block index data as a single level only. Does not do any
876      * block framing.
877      *
878      * @param out the buffered output stream to write the index to. Typically a
879      *          stream writing into an {@link HFile} block.
880      * @param description a short description of the index being written. Used
881      *          in a log message.
882      * @throws IOException
883      */
884     public void writeSingleLevelIndex(DataOutput out, String description)
885         throws IOException {
886       expectNumLevels(1);
887 
888       if (!singleLevelOnly)
889         throw new IOException("Single-level mode is turned off");
890 
891       if (rootChunk.getNumEntries() > 0)
892         throw new IOException("Root-level entries already added in " +
893             "single-level mode");
894 
895       rootChunk = curInlineChunk;
896       curInlineChunk = new BlockIndexChunk();
897 
898       if (LOG.isTraceEnabled()) {
899         LOG.trace("Wrote a single-level " + description + " index with "
900           + rootChunk.getNumEntries() + " entries, " + rootChunk.getRootSize()
901           + " bytes");
902       }
903       rootChunk.writeRoot(out);
904     }
905 
906     /**
907      * Split the current level of the block index into intermediate index
908      * blocks of permitted size and write those blocks to disk. Return the next
909      * level of the block index referencing those intermediate-level blocks.
910      *
911      * @param out
912      * @param currentLevel the current level of the block index, such as the a
913      *          chunk referencing all leaf-level index blocks
914      * @return the parent level block index, which becomes the root index after
915      *         a few (usually zero) iterations
916      * @throws IOException
917      */
918     private BlockIndexChunk writeIntermediateLevel(FSDataOutputStream out,
919         BlockIndexChunk currentLevel) throws IOException {
920       // Entries referencing intermediate-level blocks we are about to create.
921       BlockIndexChunk parent = new BlockIndexChunk();
922 
923       // The current intermediate-level block index chunk.
924       BlockIndexChunk curChunk = new BlockIndexChunk();
925 
926       for (int i = 0; i < currentLevel.getNumEntries(); ++i) {
927         curChunk.add(currentLevel.getBlockKey(i),
928             currentLevel.getBlockOffset(i), currentLevel.getOnDiskDataSize(i));
929 
930         if (curChunk.getRootSize() >= maxChunkSize)
931           writeIntermediateBlock(out, parent, curChunk);
932       }
933 
934       if (curChunk.getNumEntries() > 0) {
935         writeIntermediateBlock(out, parent, curChunk);
936       }
937 
938       return parent;
939     }
940 
941     private void writeIntermediateBlock(FSDataOutputStream out,
942         BlockIndexChunk parent, BlockIndexChunk curChunk) throws IOException {
943       long beginOffset = out.getPos();
944       DataOutputStream dos = blockWriter.startWriting(
945           BlockType.INTERMEDIATE_INDEX);
946       curChunk.writeNonRoot(dos);
947       byte[] curFirstKey = curChunk.getBlockKey(0);
948       blockWriter.writeHeaderAndData(out);
949 
950       if (blockCache != null) {
951         HFileBlock blockForCaching = blockWriter.getBlockForCaching();
952         blockCache.cacheBlock(new BlockCacheKey(nameForCaching,
953             beginOffset, DataBlockEncoding.NONE, 
954             blockForCaching.getBlockType()), blockForCaching);
955       }
956 
957       // Add intermediate index block size
958       totalBlockOnDiskSize += blockWriter.getOnDiskSizeWithoutHeader();
959       totalBlockUncompressedSize +=
960           blockWriter.getUncompressedSizeWithoutHeader();
961 
962       // OFFSET is the beginning offset the chunk of block index entries.
963       // SIZE is the total byte size of the chunk of block index entries
964       // + the secondary index size
965       // FIRST_KEY is the first key in the chunk of block index
966       // entries.
967       parent.add(curFirstKey, beginOffset,
968           blockWriter.getOnDiskSizeWithHeader());
969 
970       // clear current block index chunk
971       curChunk.clear();
972       curFirstKey = null;
973     }
974 
975     /**
976      * @return how many block index entries there are in the root level
977      */
978     public final int getNumRootEntries() {
979       return rootChunk.getNumEntries();
980     }
981 
982     /**
983      * @return the number of levels in this block index.
984      */
985     public int getNumLevels() {
986       return numLevels;
987     }
988 
989     private void expectNumLevels(int expectedNumLevels) {
990       if (numLevels != expectedNumLevels) {
991         throw new IllegalStateException("Number of block index levels is "
992             + numLevels + "but is expected to be " + expectedNumLevels);
993       }
994     }
995 
996     /**
997      * Whether there is an inline block ready to be written. In general, we
998      * write an leaf-level index block as an inline block as soon as its size
999      * as serialized in the non-root format reaches a certain threshold.
1000      */
1001     @Override
1002     public boolean shouldWriteBlock(boolean closing) {
1003       if (singleLevelOnly) {
1004         throw new UnsupportedOperationException(INLINE_BLOCKS_NOT_ALLOWED);
1005       }
1006 
1007       if (curInlineChunk == null) {
1008         throw new IllegalStateException("curInlineChunk is null; has shouldWriteBlock been " +
1009             "called with closing=true and then called again?");
1010       }
1011 
1012       if (curInlineChunk.getNumEntries() == 0) {
1013         return false;
1014       }
1015 
1016       // We do have some entries in the current inline chunk.
1017       if (closing) {
1018         if (rootChunk.getNumEntries() == 0) {
1019           // We did not add any leaf-level blocks yet. Instead of creating a
1020           // leaf level with one block, move these entries to the root level.
1021 
1022           expectNumLevels(1);
1023           rootChunk = curInlineChunk;
1024           curInlineChunk = null;  // Disallow adding any more index entries.
1025           return false;
1026         }
1027 
1028         return true;
1029       } else {
1030         return curInlineChunk.getNonRootSize() >= maxChunkSize;
1031       }
1032     }
1033 
1034     /**
1035      * Write out the current inline index block. Inline blocks are non-root
1036      * blocks, so the non-root index format is used.
1037      *
1038      * @param out
1039      */
1040     @Override
1041     public void writeInlineBlock(DataOutput out) throws IOException {
1042       if (singleLevelOnly)
1043         throw new UnsupportedOperationException(INLINE_BLOCKS_NOT_ALLOWED);
1044 
1045       // Write the inline block index to the output stream in the non-root
1046       // index block format.
1047       curInlineChunk.writeNonRoot(out);
1048 
1049       // Save the first key of the inline block so that we can add it to the
1050       // parent-level index.
1051       firstKey = curInlineChunk.getBlockKey(0);
1052 
1053       // Start a new inline index block
1054       curInlineChunk.clear();
1055     }
1056 
1057     /**
1058      * Called after an inline block has been written so that we can add an
1059      * entry referring to that block to the parent-level index.
1060      */
1061     @Override
1062     public void blockWritten(long offset, int onDiskSize, int uncompressedSize)
1063     {
1064       // Add leaf index block size
1065       totalBlockOnDiskSize += onDiskSize;
1066       totalBlockUncompressedSize += uncompressedSize;
1067 
1068       if (singleLevelOnly)
1069         throw new UnsupportedOperationException(INLINE_BLOCKS_NOT_ALLOWED);
1070 
1071       if (firstKey == null) {
1072         throw new IllegalStateException("Trying to add second-level index " +
1073             "entry with offset=" + offset + " and onDiskSize=" + onDiskSize +
1074             "but the first key was not set in writeInlineBlock");
1075       }
1076 
1077       if (rootChunk.getNumEntries() == 0) {
1078         // We are writing the first leaf block, so increase index level.
1079         expectNumLevels(1);
1080         numLevels = 2;
1081       }
1082 
1083       // Add another entry to the second-level index. Include the number of
1084       // entries in all previous leaf-level chunks for mid-key calculation.
1085       rootChunk.add(firstKey, offset, onDiskSize, totalNumEntries);
1086       firstKey = null;
1087     }
1088 
1089     @Override
1090     public BlockType getInlineBlockType() {
1091       return BlockType.LEAF_INDEX;
1092     }
1093 
1094     /**
1095      * Add one index entry to the current leaf-level block. When the leaf-level
1096      * block gets large enough, it will be flushed to disk as an inline block.
1097      *
1098      * @param firstKey the first key of the data block
1099      * @param blockOffset the offset of the data block
1100      * @param blockDataSize the on-disk size of the data block ({@link HFile}
1101      *          format version 2), or the uncompressed size of the data block (
1102      *          {@link HFile} format version 1).
1103      */
1104     public void addEntry(byte[] firstKey, long blockOffset, int blockDataSize)
1105     {
1106       curInlineChunk.add(firstKey, blockOffset, blockDataSize);
1107       ++totalNumEntries;
1108     }
1109 
1110     /**
1111      * @throws IOException if we happened to write a multi-level index.
1112      */
1113     public void ensureSingleLevel() throws IOException {
1114       if (numLevels > 1) {
1115         throw new IOException ("Wrote a " + numLevels + "-level index with " +
1116             rootChunk.getNumEntries() + " root-level entries, but " +
1117             "this is expected to be a single-level block index.");
1118       }
1119     }
1120 
1121     /**
1122      * @return true if we are using cache-on-write. This is configured by the
1123      *         caller of the constructor by either passing a valid block cache
1124      *         or null.
1125      */
1126     @Override
1127     public boolean getCacheOnWrite() {
1128       return blockCache != null;
1129     }
1130 
1131     /**
1132      * The total uncompressed size of the root index block, intermediate-level
1133      * index blocks, and leaf-level index blocks.
1134      *
1135      * @return the total uncompressed size of all index blocks
1136      */
1137     public long getTotalUncompressedSize() {
1138       return totalBlockUncompressedSize;
1139     }
1140 
1141   }
1142 
1143   /**
1144    * A single chunk of the block index in the process of writing. The data in
1145    * this chunk can become a leaf-level, intermediate-level, or root index
1146    * block.
1147    */
1148   static class BlockIndexChunk {
1149 
1150     /** First keys of the key range corresponding to each index entry. */
1151     private final List<byte[]> blockKeys = new ArrayList<byte[]>();
1152 
1153     /** Block offset in backing stream. */
1154     private final List<Long> blockOffsets = new ArrayList<Long>();
1155 
1156     /** On-disk data sizes of lower-level data or index blocks. */
1157     private final List<Integer> onDiskDataSizes = new ArrayList<Integer>();
1158 
1159     /**
1160      * The cumulative number of sub-entries, i.e. entries on deeper-level block
1161      * index entries. numSubEntriesAt[i] is the number of sub-entries in the
1162      * blocks corresponding to this chunk's entries #0 through #i inclusively.
1163      */
1164     private final List<Long> numSubEntriesAt = new ArrayList<Long>();
1165 
1166     /**
1167      * The offset of the next entry to be added, relative to the end of the
1168      * "secondary index" in the "non-root" format representation of this index
1169      * chunk. This is the next value to be added to the secondary index.
1170      */
1171     private int curTotalNonRootEntrySize = 0;
1172 
1173     /**
1174      * The accumulated size of this chunk if stored in the root index format.
1175      */
1176     private int curTotalRootSize = 0;
1177 
1178     /**
1179      * The "secondary index" used for binary search over variable-length
1180      * records in a "non-root" format block. These offsets are relative to the
1181      * end of this secondary index.
1182      */
1183     private final List<Integer> secondaryIndexOffsetMarks =
1184         new ArrayList<Integer>();
1185 
1186     /**
1187      * Adds a new entry to this block index chunk.
1188      *
1189      * @param firstKey the first key in the block pointed to by this entry
1190      * @param blockOffset the offset of the next-level block pointed to by this
1191      *          entry
1192      * @param onDiskDataSize the on-disk data of the block pointed to by this
1193      *          entry, including header size
1194      * @param curTotalNumSubEntries if this chunk is the root index chunk under
1195      *          construction, this specifies the current total number of
1196      *          sub-entries in all leaf-level chunks, including the one
1197      *          corresponding to the second-level entry being added.
1198      */
1199     void add(byte[] firstKey, long blockOffset, int onDiskDataSize,
1200         long curTotalNumSubEntries) {
1201       // Record the offset for the secondary index
1202       secondaryIndexOffsetMarks.add(curTotalNonRootEntrySize);
1203       curTotalNonRootEntrySize += SECONDARY_INDEX_ENTRY_OVERHEAD
1204           + firstKey.length;
1205 
1206       curTotalRootSize += Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT
1207           + WritableUtils.getVIntSize(firstKey.length) + firstKey.length;
1208 
1209       blockKeys.add(firstKey);
1210       blockOffsets.add(blockOffset);
1211       onDiskDataSizes.add(onDiskDataSize);
1212 
1213       if (curTotalNumSubEntries != -1) {
1214         numSubEntriesAt.add(curTotalNumSubEntries);
1215 
1216         // Make sure the parallel arrays are in sync.
1217         if (numSubEntriesAt.size() != blockKeys.size()) {
1218           throw new IllegalStateException("Only have key/value count " +
1219               "stats for " + numSubEntriesAt.size() + " block index " +
1220               "entries out of " + blockKeys.size());
1221         }
1222       }
1223     }
1224 
1225     /**
1226      * The same as {@link #add(byte[], long, int, long)} but does not take the
1227      * key/value into account. Used for single-level indexes.
1228      *
1229      * @see {@link #add(byte[], long, int, long)}
1230      */
1231     public void add(byte[] firstKey, long blockOffset, int onDiskDataSize) {
1232       add(firstKey, blockOffset, onDiskDataSize, -1);
1233     }
1234 
1235     public void clear() {
1236       blockKeys.clear();
1237       blockOffsets.clear();
1238       onDiskDataSizes.clear();
1239       secondaryIndexOffsetMarks.clear();
1240       numSubEntriesAt.clear();
1241       curTotalNonRootEntrySize = 0;
1242       curTotalRootSize = 0;
1243     }
1244 
1245     /**
1246      * Finds the entry corresponding to the deeper-level index block containing
1247      * the given deeper-level entry (a "sub-entry"), assuming a global 0-based
1248      * ordering of sub-entries.
1249      *
1250      * <p>
1251      * <i> Implementation note. </i> We are looking for i such that
1252      * numSubEntriesAt[i - 1] <= k < numSubEntriesAt[i], because a deeper-level
1253      * block #i (0-based) contains sub-entries # numSubEntriesAt[i - 1]'th
1254      * through numSubEntriesAt[i] - 1, assuming a global 0-based ordering of
1255      * sub-entries. i is by definition the insertion point of k in
1256      * numSubEntriesAt.
1257      *
1258      * @param k sub-entry index, from 0 to the total number sub-entries - 1
1259      * @return the 0-based index of the entry corresponding to the given
1260      *         sub-entry
1261      */
1262     public int getEntryBySubEntry(long k) {
1263       // We define mid-key as the key corresponding to k'th sub-entry
1264       // (0-based).
1265 
1266       int i = Collections.binarySearch(numSubEntriesAt, k);
1267 
1268       // Exact match: cumulativeWeight[i] = k. This means chunks #0 through
1269       // #i contain exactly k sub-entries, and the sub-entry #k (0-based)
1270       // is in the (i + 1)'th chunk.
1271       if (i >= 0)
1272         return i + 1;
1273 
1274       // Inexact match. Return the insertion point.
1275       return -i - 1;
1276     }
1277 
1278     /**
1279      * Used when writing the root block index of a multi-level block index.
1280      * Serializes additional information allowing to efficiently identify the
1281      * mid-key.
1282      *
1283      * @return a few serialized fields for finding the mid-key
1284      * @throws IOException if could not create metadata for computing mid-key
1285      */
1286     public byte[] getMidKeyMetadata() throws IOException {
1287       ByteArrayOutputStream baos = new ByteArrayOutputStream(
1288           MID_KEY_METADATA_SIZE);
1289       DataOutputStream baosDos = new DataOutputStream(baos);
1290       long totalNumSubEntries = numSubEntriesAt.get(blockKeys.size() - 1);
1291       if (totalNumSubEntries == 0) {
1292         throw new IOException("No leaf-level entries, mid-key unavailable");
1293       }
1294       long midKeySubEntry = (totalNumSubEntries - 1) / 2;
1295       int midKeyEntry = getEntryBySubEntry(midKeySubEntry);
1296 
1297       baosDos.writeLong(blockOffsets.get(midKeyEntry));
1298       baosDos.writeInt(onDiskDataSizes.get(midKeyEntry));
1299 
1300       long numSubEntriesBefore = midKeyEntry > 0
1301           ? numSubEntriesAt.get(midKeyEntry - 1) : 0;
1302       long subEntryWithinEntry = midKeySubEntry - numSubEntriesBefore;
1303       if (subEntryWithinEntry < 0 || subEntryWithinEntry > Integer.MAX_VALUE)
1304       {
1305         throw new IOException("Could not identify mid-key index within the "
1306             + "leaf-level block containing mid-key: out of range ("
1307             + subEntryWithinEntry + ", numSubEntriesBefore="
1308             + numSubEntriesBefore + ", midKeySubEntry=" + midKeySubEntry
1309             + ")");
1310       }
1311 
1312       baosDos.writeInt((int) subEntryWithinEntry);
1313 
1314       if (baosDos.size() != MID_KEY_METADATA_SIZE) {
1315         throw new IOException("Could not write mid-key metadata: size=" +
1316             baosDos.size() + ", correct size: " + MID_KEY_METADATA_SIZE);
1317       }
1318 
1319       // Close just to be good citizens, although this has no effect.
1320       baos.close();
1321 
1322       return baos.toByteArray();
1323     }
1324 
1325     /**
1326      * Writes the block index chunk in the non-root index block format. This
1327      * format contains the number of entries, an index of integer offsets
1328      * for quick binary search on variable-length records, and tuples of
1329      * block offset, on-disk block size, and the first key for each entry.
1330      *
1331      * @param out
1332      * @throws IOException
1333      */
1334     void writeNonRoot(DataOutput out) throws IOException {
1335       // The number of entries in the block.
1336       out.writeInt(blockKeys.size());
1337 
1338       if (secondaryIndexOffsetMarks.size() != blockKeys.size()) {
1339         throw new IOException("Corrupted block index chunk writer: " +
1340             blockKeys.size() + " entries but " +
1341             secondaryIndexOffsetMarks.size() + " secondary index items");
1342       }
1343 
1344       // For each entry, write a "secondary index" of relative offsets to the
1345       // entries from the end of the secondary index. This works, because at
1346       // read time we read the number of entries and know where the secondary
1347       // index ends.
1348       for (int currentSecondaryIndex : secondaryIndexOffsetMarks)
1349         out.writeInt(currentSecondaryIndex);
1350 
1351       // We include one other element in the secondary index to calculate the
1352       // size of each entry more easily by subtracting secondary index elements.
1353       out.writeInt(curTotalNonRootEntrySize);
1354 
1355       for (int i = 0; i < blockKeys.size(); ++i) {
1356         out.writeLong(blockOffsets.get(i));
1357         out.writeInt(onDiskDataSizes.get(i));
1358         out.write(blockKeys.get(i));
1359       }
1360     }
1361 
1362     /**
1363      * @return the size of this chunk if stored in the non-root index block
1364      *         format
1365      */
1366     int getNonRootSize() {
1367       return Bytes.SIZEOF_INT                          // Number of entries
1368           + Bytes.SIZEOF_INT * (blockKeys.size() + 1)  // Secondary index
1369           + curTotalNonRootEntrySize;                  // All entries
1370     }
1371 
1372     /**
1373      * Writes this chunk into the given output stream in the root block index
1374      * format. This format is similar to the {@link HFile} version 1 block
1375      * index format, except that we store on-disk size of the block instead of
1376      * its uncompressed size.
1377      *
1378      * @param out the data output stream to write the block index to. Typically
1379      *          a stream writing into an {@link HFile} block.
1380      * @throws IOException
1381      */
1382     void writeRoot(DataOutput out) throws IOException {
1383       for (int i = 0; i < blockKeys.size(); ++i) {
1384         out.writeLong(blockOffsets.get(i));
1385         out.writeInt(onDiskDataSizes.get(i));
1386         Bytes.writeByteArray(out, blockKeys.get(i));
1387       }
1388     }
1389 
1390     /**
1391      * @return the size of this chunk if stored in the root index block format
1392      */
1393     int getRootSize() {
1394       return curTotalRootSize;
1395     }
1396 
1397     /**
1398      * @return the number of entries in this block index chunk
1399      */
1400     public int getNumEntries() {
1401       return blockKeys.size();
1402     }
1403 
1404     public byte[] getBlockKey(int i) {
1405       return blockKeys.get(i);
1406     }
1407 
1408     public long getBlockOffset(int i) {
1409       return blockOffsets.get(i);
1410     }
1411 
1412     public int getOnDiskDataSize(int i) {
1413       return onDiskDataSizes.get(i);
1414     }
1415 
1416     public long getCumulativeNumKV(int i) {
1417       if (i < 0)
1418         return 0;
1419       return numSubEntriesAt.get(i);
1420     }
1421 
1422   }
1423 
1424   public static int getMaxChunkSize(Configuration conf) {
1425     return conf.getInt(MAX_CHUNK_SIZE_KEY, DEFAULT_MAX_CHUNK_SIZE);
1426   }
1427 }