@Override
  public long[] putFileData(Object fileKey, DiskRange[] ranges, MemoryBuffer[] buffers,
      long baseOffset, Priority priority, LowLevelCacheCounters qfCounters, CacheTag tag) {
    long[] result = null;
    assert buffers.length == ranges.length;
    FileCache<ConcurrentSkipListMap<Long, LlapDataBuffer>> subCache =
        FileCache.getOrAddFileSubCache(cache, fileKey, CACHE_CTOR, tag);
    try {
      for (int i = 0; i < ranges.length; ++i) {
        LlapDataBuffer buffer = (LlapDataBuffer)buffers[i];
        if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
          LlapIoImpl.LOCKING_LOGGER.trace("Locking {} at put time", buffer);
        }
        boolean canLock = lockBuffer(buffer, false);
        assert canLock;
        long offset = ranges[i].getOffset() + baseOffset;
        assert buffer.declaredCachedLength == LlapDataBuffer.UNKNOWN_CACHED_LENGTH;
        buffer.declaredCachedLength = ranges[i].getLength();
        while (true) { // Overwhelmingly executes once, or maybe twice (replacing stale value).
          LlapDataBuffer oldVal = subCache.getCache().putIfAbsent(offset, buffer);
          if (oldVal == null) {
            buffer.setStart(offset);
            buffer.setFileCache(subCache);
            // Cached successfully, add to policy.
            cachePolicy.cache(buffer, priority);
            if (qfCounters != null) {
              qfCounters.recordAllocBytes(buffer.byteBuffer.remaining(), buffer.allocSize);
            }
            break;
          }
          if (LlapIoImpl.CACHE_LOGGER.isTraceEnabled()) {
            LlapIoImpl.CACHE_LOGGER.trace("Trying to cache when the chunk is already cached for" +
                " {}@{} (base {}); old {}, new {}", fileKey, offset, baseOffset, oldVal, buffer);
          }

          if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
            LlapIoImpl.LOCKING_LOGGER.trace("Locking {} due to cache collision", oldVal);
          }
          if (lockBuffer(oldVal, true)) {
            // We don't do proper overlap checking because it would cost cycles and we
            // think it will never happen. We do perform the most basic check here.
            if (oldVal.declaredCachedLength != buffer.declaredCachedLength) {
              throw new RuntimeException("Found a block with different length at the same offset: "
                  + oldVal.declaredCachedLength + " vs " + buffer.declaredCachedLength + " @" + offset
                  + " (base " + baseOffset + ")");
            }
            // We found an old, valid block for this key in the cache.
            if (LlapIoImpl.LOCKING_LOGGER.isTraceEnabled()) {
              LlapIoImpl.LOCKING_LOGGER.trace("Unlocking {} due to cache collision with {}",
                  buffer, oldVal);
            }

            // This is always set to ranges[i].getLength() prior inserting into the map to avoid inconsistency with the
            // check above. However once we decided that this new buffer will not be cached, we should unset
            // declaredCachedLength, so that it can be instantly deallocated at unlockBuffer()'s else branch.
            buffer.declaredCachedLength = LlapDataBuffer.UNKNOWN_CACHED_LENGTH;
            unlockBuffer(buffer, false);
            buffers[i] = oldVal;
            if (result == null) {
              result = new long[align64(buffers.length) >>> 6];
            }
            result[i >>> 6] |= (1L << (i & 63)); // indicate that we've replaced the value
            break;
          }
          // We found some old value but couldn't incRef it; remove it.
          subCache.getCache().remove(offset, oldVal);
        }
      }
    } finally {
      subCache.decRef();
    }
    return result;
  }
Adds cached buffer to buffer list.
@param currentNotCached Pointer to the list node where we are inserting.
@param currentCached The cached buffer found for this node, to insert.
@return The new currentNotCached pointer, following the cached buffer insertion.
llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java