public DiskRangeList readEncodedStream(long baseOffset, DiskRangeList start, long cOffset,
      long endCOffset, ColumnStreamData csd, long unlockUntilCOffset, long streamOffset,
      IdentityHashMap<ByteBuffer, Boolean> toRelease) throws IOException {
    if (csd.getCacheBuffers() == null) {
      csd.setCacheBuffers(new ArrayList<MemoryBuffer>());
    } else {
      csd.getCacheBuffers().clear();
    }
    if (cOffset == endCOffset) return null;
    List<ProcCacheChunk> toDecompress = null;
    List<IncompleteCb> badEstimates = null;
    List<ByteBuffer> toReleaseCopies = null;
    if (isCompressed) {
      toReleaseCopies = new ArrayList<>();
      toDecompress = new ArrayList<>();
      badEstimates = new ArrayList<>();
    }

    // 1. Find our bearings in the stream. Normally, iter will already point either to where we
    // want to be, or just before. However, RGs can overlap due to encoding, so we may have
    // to return to a previous block.
    DiskRangeList current = findExactPosition(start, cOffset);
    if (isTracingEnabled) {
      LOG.trace("Starting read for [" + cOffset + "," + endCOffset + ") at " + current);
    }
    trace.logStartRead(current);

    CacheChunk lastUncompressed = null;

    // 2. Go thru the blocks; add stuff to results and prepare the decompression work (see below).
    try {
      lastUncompressed = isCompressed ?
          prepareRangesForCompressedRead(cOffset, endCOffset, streamOffset, unlockUntilCOffset,
              current, csd, toRelease, toReleaseCopies, toDecompress, badEstimates)
        : prepareRangesForUncompressedRead(
            cOffset, endCOffset, streamOffset, unlockUntilCOffset, current, csd);
    } catch (Exception ex) {
      LOG.error("Failed " + (isCompressed ? "" : "un") + "compressed read; cOffset " + cOffset
          + ", endCOffset " + endCOffset + ", streamOffset " + streamOffset
          + ", unlockUntilCOffset " + unlockUntilCOffset + "; ranges passed in "
          + RecordReaderUtils.stringifyDiskRanges(start) + "; ranges passed to prepare "
          + RecordReaderUtils.stringifyDiskRanges(current)); // Don't log exception here.
      throw (ex instanceof IOException) ? (IOException)ex : new IOException(ex);
    }

    // 2.5. Remember the bad estimates for future reference.
    if (badEstimates != null && !badEstimates.isEmpty()) {
      // Relies on the fact that cache does not actually store these.
      DiskRange[] cacheKeys = badEstimates.toArray(new DiskRange[badEstimates.size()]);
      long[] result = cacheWrapper.putFileData(fileKey, cacheKeys, null, baseOffset, tag);
      assert result == null; // We don't expect conflicts from bad estimates.
    }

    if (toDecompress == null || toDecompress.isEmpty()) {
      releaseBuffers(toReleaseCopies, false);
      return lastUncompressed; // Nothing to do.
    }

    // 3. Allocate the buffers, prepare cache keys.
    // At this point, we have read all the CBs we need to read. cacheBuffers contains some cache
    // data and some unallocated membufs for decompression. toDecompress contains all the work we
    // need to do, and each item points to one of the membufs in cacheBuffers as target. The iter
    // has also been adjusted to point to these buffers instead of compressed data for the ranges.
    MemoryBuffer[] targetBuffers = new MemoryBuffer[toDecompress.size()];
    DiskRange[] cacheKeys = new DiskRange[toDecompress.size()];
    int ix = 0;
    for (ProcCacheChunk chunk : toDecompress) {
      cacheKeys[ix] = chunk; // Relies on the fact that cache does not actually store these.
      targetBuffers[ix] = chunk.getBuffer();
      ++ix;
    }
    boolean isAllocated = false;
    try {
      allocateMultiple(targetBuffers, bufferSize);
      isAllocated = true;
    } finally {
      // toDecompress/targetBuffers contents are actually already added to some structures that
      // will be cleaned up on error. Remove the unallocated buffers; keep the cached buffers in.
      if (!isAllocated) {
        // Inefficient - this only happens during cleanup on errors.
        for (MemoryBuffer buf : targetBuffers) {
          csd.getCacheBuffers().remove(buf);
        }
        for (ProcCacheChunk chunk : toDecompress) {
          chunk.buffer = null;
        }
      }
    }

    // 4. Now decompress (or copy) the data into cache buffers.
    int decompressedIx = 0;
    try {
      while (decompressedIx < toDecompress.size()) {
        ProcCacheChunk chunk = toDecompress.get(decompressedIx);
        ByteBuffer dest = chunk.getBuffer().getByteBufferRaw();
        if (chunk.isOriginalDataCompressed) {
          boolean isOk = false;
          try {
            decompressChunk(chunk.originalData, codec, dest);
            isOk = true;
          } finally {
            if (!isOk) {
              isCodecFailure = true;
            }
          }
        } else {
          copyUncompressedChunk(chunk.originalData, dest);
        }

        if (isTracingEnabled) {
          LOG.trace("Locking " + chunk.getBuffer() + " due to reuse (after decompression)");
        }
        // After we set originalData to null, we incref the buffer and the cleanup would decref it.
        // Note that this assumes the failure during incref means incref didn't occur.
        try {
          cacheWrapper.reuseBuffer(chunk.getBuffer());
        } finally {
          chunk.originalData = null;
        }
        ++decompressedIx;
      }
    } finally {
      // This will only execute on error. Deallocate the remaining allocated buffers explicitly.
      // The ones that were already incref-ed will be cleaned up with the regular cache buffers.
      while (decompressedIx < toDecompress.size()) {
        ProcCacheChunk chunk = toDecompress.get(decompressedIx);
        csd.getCacheBuffers().remove(chunk.getBuffer());
        try {
          cacheWrapper.getAllocator().deallocate(chunk.getBuffer());
        } catch (Throwable t) {
          LOG.error("Ignoring the cleanup error after another error", t);
        }
        chunk.setBuffer(null);
        ++decompressedIx;
      }
    }

    // 5. Release the copies we made directly to the cleaner.
    releaseBuffers(toReleaseCopies, false);

    // 6. Finally, put uncompressed data to cache.
    if (fileKey != null) {
      long[] collisionMask = cacheWrapper.putFileData(
          fileKey, cacheKeys, targetBuffers, baseOffset, tag);
      processCacheCollisions(collisionMask, toDecompress, targetBuffers, csd.getCacheBuffers());
    }

    // 7. It may happen that we know we won't use some cache buffers anymore (the alternative
    //    is that we will use the same buffers for other streams in separate calls).
    //    Release initial refcounts.
    for (ProcCacheChunk chunk : toDecompress) {
      ponderReleaseInitialRefcount(unlockUntilCOffset, streamOffset, chunk);
    }

    return lastUncompressed;
  }
Uncompresses part of the stream. RGs can overlap, so we cannot just go and decompress
and remove what we have returned. We will keep iterator as a "hint" point.
@param baseOffset Absolute offset of boundaries and ranges relative to file, for cache keys.
@param start Ordered ranges containing file data. Helpful if they point close to cOffset.
@param cOffset Start offset to decompress.
@param endCOffset End offset to decompress; estimate, partial CBs will be ignored.
@param csd Stream data, to add the results.
@param unlockUntilCOffset The offset until which the buffers can be unlocked in cache, as
                          they will not be used in future calls (see the class comment in
                          EncodedReaderImpl about refcounts).
@return Last buffer cached during decompression. Cache buffers are never removed from
        the master list, so they are safe to keep as iterators for various streams.
ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java