private void ponderReleaseInitialRefcount(
      long unlockUntilCOffset, long streamStartOffset, CacheChunk cc) {
    // Don't release if the buffer contains any data beyond the acceptable boundary.
    if (cc.getEnd() > unlockUntilCOffset) return;
    assert cc.getBuffer() != null;
    try {
      releaseInitialRefcount(cc, false);
    } catch (AssertionError e) {
      LOG.error("BUG: releasing initial refcount; stream start " + streamStartOffset + ", "
          + "unlocking until " + unlockUntilCOffset + " from [" + cc + "]: " + e.getMessage());
      throw e;
    }
    // Release all the previous buffers that we may not have been able to release due to reuse,
    // as long as they are still in the same stream and are not already released.
    DiskRangeList prev = cc.prev;
    while (true) {
      // Do not release beyond current stream (we don't know which RGs that buffer is for).
      if ((prev == null) || (prev.getEnd() <= streamStartOffset)) break;
      // Only release cache chunks; do not release ProcCacheChunks - they may not yet have data.
      if (prev.getClass() != CacheChunk.class) break;

      CacheChunk prevCc = (CacheChunk)prev;
      if (prevCc.buffer == null) break;
      try {
        releaseInitialRefcount(prevCc, true);
      } catch (AssertionError e) {
        LOG.error("BUG: releasing initial refcount; stream start " + streamStartOffset + ", "
            + "unlocking until " + unlockUntilCOffset + " from [" + cc + "] and backtracked to ["
            + prevCc + "]: " + e.getMessage());
        throw e;
      }
      prev = prev.prev;
    }
  }
To achieve some sort of consistent cache boundaries, we will cache streams deterministically;
in segments starting w/stream start, and going for either stream size or some fixed size.
If we are not reading the entire segment's worth of data, then we will not cache the partial
RGs; the breakage of cache assumptions (no interleaving blocks, etc.) is way too much PITA
to handle just for this case.
We could avoid copy in non-zcr case and manage the buffer that was not allocated by our
allocator. Uncompressed case is not mainline though so let's not complicate it.
@param kind
ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java