private ProcCacheChunk addOneCompressionBuffer(BufferChunk current,
      List<MemoryBuffer> cacheBuffers, List<ProcCacheChunk> toDecompress,
      IdentityHashMap<ByteBuffer, Boolean> toRelease, List<ByteBuffer> toReleaseCopies,
      List<IncompleteCb> badEstimates) throws IOException {
    ByteBuffer slice = null;
    ByteBuffer compressed = current.getData();
    long cbStartOffset = current.getOffset();
    int b0 = -1, b1 = -1, b2 = -1;
    // First, read the CB header. Due to ORC estimates, ZCR, etc. this can be complex.
    if (compressed.remaining() >= 3) {
      // The overwhelming majority of cases will go here. Read 3 bytes. Tada!
      b0 = compressed.get() & 0xff;
      b1 = compressed.get() & 0xff;
      b2 = compressed.get() & 0xff;
    } else {
      // Bad luck! Handle the corner cases where 3 bytes are in multiple blocks.
      int[] bytes = new int[3];
      current = readLengthBytesFromSmallBuffers(
          current, cbStartOffset, bytes, badEstimates, isTracingEnabled, trace);
      if (current == null) return null;
      compressed = current.getData();
      b0 = bytes[0];
      b1 = bytes[1];
      b2 = bytes[2];
    }
    int chunkLength = (b2 << 15) | (b1 << 7) | (b0 >> 1);

    if (chunkLength > bufferSize) {
      throw new IllegalArgumentException("Buffer size too small. size = " +
          bufferSize + " needed = " + chunkLength);
    }
    int consumedLength = chunkLength + OutStream.HEADER_SIZE;
    long cbEndOffset = cbStartOffset + consumedLength;
    boolean isUncompressed = ((b0 & 0x01) == 1);
    if (isTracingEnabled) {
      LOG.trace("Found CB at " + cbStartOffset + ", chunk length " + chunkLength + ", total "
          + consumedLength + ", " + (isUncompressed ? "not " : "") + "compressed");
    }
    trace.logOrcCb(cbStartOffset, chunkLength, isUncompressed);
    if (compressed.remaining() >= chunkLength) {
      // Simple case - CB fits entirely in the disk range.
      slice = compressed.slice();
      slice.limit(chunkLength);
      return addOneCompressionBlockByteBuffer(slice, isUncompressed,
          cbStartOffset, cbEndOffset, chunkLength, current, toDecompress, cacheBuffers, false);
    }
    if (current.getEnd() < cbEndOffset && !current.hasContiguousNext()) {
      badEstimates.add(addIncompleteCompressionBuffer(
          cbStartOffset, current, 0, isTracingEnabled, trace));
      return null; // This is impossible to read from this chunk.
    }

    // TODO: we could remove extra copy for isUncompressed case by copying directly to cache.
    // We need to consolidate 2 or more buffers into one to decompress.
    ByteBuffer copy = allocateBuffer(chunkLength, compressed.isDirect());
    toReleaseCopies.add(copy); // We will always release copies at the end.
    int remaining = chunkLength - compressed.remaining();
    int originalPos = compressed.position();
    copy.put(compressed);
    if (isTracingEnabled) {
      LOG.trace("Removing partial CB " + current + " from ranges after copying its contents");
    }
    trace.logPartialCb(current);
    DiskRangeList next = current.next;
    current.removeSelf();
    if (originalPos == 0 && toRelease.remove(compressed)) {
      releaseBuffer(compressed, true);
    }

    int extraChunkCount = 0;
    while (true) {
      if (!(next instanceof BufferChunk)) {
        throw new IOException("Trying to extend compressed block into uncompressed block " + next);
      }
      compressed = next.getData();
      ++extraChunkCount;
      if (compressed.remaining() >= remaining) {
        // This is the last range for this compression block. Yay!
        slice = compressed.slice();
        slice.limit(remaining);
        copy.put(slice);
        ProcCacheChunk cc = addOneCompressionBlockByteBuffer(copy, isUncompressed, cbStartOffset,
            cbEndOffset, remaining, (BufferChunk)next, toDecompress, cacheBuffers, true);
        if (compressed.remaining() <= 0 && toRelease.remove(compressed)) {
          releaseBuffer(compressed, true); // We copied the entire buffer.
        } // else there's more data to process; will be handled in next call.
        return cc;
      }
      remaining -= compressed.remaining();
      copy.put(compressed); // TODO: move into the if below; account for release call
      if (toRelease.remove(compressed)) {
        releaseBuffer(compressed, true); // We copied the entire buffer.
      }
      DiskRangeList tmp = next;
      next = next.hasContiguousNext() ? next.next : null;
      if (next != null) {
        if (isTracingEnabled) {
          LOG.trace("Removing partial CB " + tmp + " from ranges after copying its contents");
        }
        trace.logPartialCb(tmp);
        tmp.removeSelf();
      } else {
        badEstimates.add(addIncompleteCompressionBuffer(
            cbStartOffset, tmp, extraChunkCount, isTracingEnabled, trace));
        return null; // This is impossible to read from this chunk.
      }
    }
  }
Reads one compression block from the source; handles compression blocks read from
multiple ranges (usually, that would only happen with zcr).
Adds stuff to cachedBuffers, toDecompress and toRelease (see below what each does).
@param current BufferChunk where compression block starts.
@param cacheBuffers The result buffer array to add pre-allocated target cache buffer.
@param toDecompress The list of work to decompress - pairs of compressed buffers and the
                    target buffers (same as the ones added to cacheBuffers).
@param toRelease The list of buffers to release to zcr because they are no longer in use.
@param badEstimates The list of bad estimates that cannot be decompressed.
@return The resulting cache chunk.
ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java