private void processCacheCollisions(long[] collisionMask,
      List<? extends CacheChunk> toDecompress, MemoryBuffer[] targetBuffers,
      List<MemoryBuffer> cacheBuffers) {
    if (collisionMask == null) return;
    assert collisionMask.length >= (toDecompress.size() >>> 6);
    // There are some elements that were cached in parallel, take care of them.
    long maskVal = -1;
    for (int i = 0; i < toDecompress.size(); ++i) {
      if ((i & 63) == 0) {
        maskVal = collisionMask[i >>> 6];
      }
      if ((maskVal & 1) == 1) {
        // Cache has found an old buffer for the key and put it into array instead of our new one.
        CacheChunk replacedChunk = toDecompress.get(i);
        MemoryBuffer replacementBuffer = targetBuffers[i];
        if (isTracingEnabled) {
          LOG.trace("Discarding data due to cache collision: " + replacedChunk.getBuffer()
              + " replaced with " + replacementBuffer);
        }
        trace.logCacheCollision(replacedChunk, replacementBuffer);
        assert replacedChunk.getBuffer() != replacementBuffer : i + " was not replaced in the results "
            + "even though mask is [" + Long.toBinaryString(maskVal) + "]";
        replacedChunk.handleCacheCollision(cacheWrapper, replacementBuffer, cacheBuffers);
      }
      maskVal >>= 1;
    }
  }
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