@Override
    public int read(byte[] array, final int arrayOffset, final int len) throws IOException {
      long readStartPos = position;
      DiskRangeList drl = new DiskRangeList(readStartPos, readStartPos + len);
      DataCache.BooleanRef gotAllData = new DataCache.BooleanRef();
      drl = cache.getFileData(fileKey, drl, 0, new DataCache.DiskRangeListFactory() {
        @Override
        public DiskRangeList createCacheChunk(
            MemoryBuffer buffer, long startOffset, long endOffset) {
          return new CacheChunk(buffer, startOffset, endOffset);
        }
      }, gotAllData);
      if (LOG.isDebugEnabled()) {
        LOG.debug("Buffers after cache " + RecordReaderUtils.stringifyDiskRanges(drl));
      }
      if (gotAllData.value) {
        long sizeRead = 0;
        while (drl != null) {
          assert drl.hasData();
          long from = drl.getOffset(), to = drl.getEnd();
          int offsetFromReadStart = (int)(from - readStartPos), candidateSize = (int)(to - from);
          ByteBuffer data = drl.getData().duplicate();
          data.get(array, arrayOffset + offsetFromReadStart, candidateSize);
          cache.releaseBuffer(((CacheChunk)drl).getBuffer());
          sizeRead += candidateSize;
          drl = drl.next;
        }
        validateAndUpdatePosition(len, sizeRead);
        return len;
      }
      int maxAlloc = cache.getAllocator().getMaxAllocation();
      // We have some disk data. Separate it by column chunk and put into cache.

      // We started with a single DRL, so we assume there will be no consecutive missing blocks
      // after the cache has inserted cache data. We also assume all the missing parts will
      // represent one or several column chunks, since we always cache on column chunk boundaries.
      DiskRangeList current = drl;
      FileSystem fs = path.getFileSystem(conf);
      try (FSDataInputStream is = fs.open(path, bufferSize)) {
        Allocator allocator = cache.getAllocator();
        long sizeRead = 0;
        while (current != null) {
          DiskRangeList candidate = current;
          current = current.next;
          long from = candidate.getOffset(), to = candidate.getEnd();
          // The offset in the destination array for the beginning of this missing range.
          int offsetFromReadStart = (int)(from - readStartPos), candidateSize = (int)(to - from);
          if (candidate.hasData()) {
            ByteBuffer data = candidate.getData().duplicate();
            data.get(array, arrayOffset + offsetFromReadStart, candidateSize);
            cache.releaseBuffer(((CacheChunk)candidate).getBuffer());
            sizeRead += candidateSize;
            continue;
          }
          // The data is not in cache.

          // Account for potential partial chunks.
          SortedMap<Long, Long> chunksInThisRead = getAndValidateMissingChunks(maxAlloc, from, to);

          is.seek(from);
          is.readFully(array, arrayOffset + offsetFromReadStart, candidateSize);
          sizeRead += candidateSize;
          // Now copy missing chunks (and parts of chunks) into cache buffers.
          if (fileKey == null || cache == null) continue;
          int extraDiskDataOffset = 0;
          // TODO: should we try to make a giant array for one cache call to avoid overhead?
          for (Map.Entry<Long, Long> missingChunk : chunksInThisRead.entrySet()) {
            long chunkFrom = Math.max(from, missingChunk.getKey()),
                chunkTo = Math.min(to, missingChunk.getValue()),
                chunkLength = chunkTo - chunkFrom;
            // TODO: if we allow partial reads (right now we disable this), we'd have to handle it here.
            //       chunksInThisRead should probably be changed to be a struct array indicating both
            //       partial and full sizes for each chunk; then the partial ones could be merged
            //       with the previous partial ones, and any newly-full chunks put in the cache.
            MemoryBuffer[] largeBuffers = null, smallBuffer = null, newCacheData = null;
            try {
              int largeBufCount = (int) (chunkLength / maxAlloc);
              int smallSize = (int) (chunkLength % maxAlloc);
              int chunkPartCount = largeBufCount + ((smallSize > 0) ? 1 : 0);
              DiskRange[] cacheRanges = new DiskRange[chunkPartCount];
              int extraOffsetInChunk = 0;
              if (maxAlloc < chunkLength) {
                largeBuffers = new MemoryBuffer[largeBufCount];
                // Note: we don't use StoppableAllocator here - this is not on an IO thread.
                allocator.allocateMultiple(largeBuffers, maxAlloc, cache.getDataBufferFactory());
                for (int i = 0; i < largeBuffers.length; ++i) {
                  // By definition here we copy up to the limit of the buffer.
                  ByteBuffer bb = largeBuffers[i].getByteBufferRaw();
                  int remaining = bb.remaining();
                  assert remaining == maxAlloc;
                  copyDiskDataToCacheBuffer(array,
                      arrayOffset + offsetFromReadStart + extraDiskDataOffset,
                      remaining, bb, cacheRanges, i, chunkFrom + extraOffsetInChunk);
                  extraDiskDataOffset += remaining;
                  extraOffsetInChunk += remaining;
                }
              }
              newCacheData = largeBuffers;
              largeBuffers = null;
              if (smallSize > 0) {
                smallBuffer = new MemoryBuffer[1];
                // Note: we don't use StoppableAllocator here - this is not on an IO thread.
                allocator.allocateMultiple(smallBuffer, smallSize, cache.getDataBufferFactory());
                ByteBuffer bb = smallBuffer[0].getByteBufferRaw();
                copyDiskDataToCacheBuffer(array,
                    arrayOffset + offsetFromReadStart + extraDiskDataOffset,
                    smallSize, bb, cacheRanges, largeBufCount, chunkFrom + extraOffsetInChunk);
                extraDiskDataOffset += smallSize;
                extraOffsetInChunk += smallSize; // Not strictly necessary, no one will look at it.
                if (newCacheData == null) {
                  newCacheData = smallBuffer;
                } else {
                  // TODO: add allocate overload with an offset and length
                  MemoryBuffer[] combinedCacheData = new MemoryBuffer[largeBufCount + 1];
                  System.arraycopy(newCacheData, 0, combinedCacheData, 0, largeBufCount);
                  newCacheData = combinedCacheData;
                  newCacheData[largeBufCount] = smallBuffer[0];
                }
                smallBuffer = null;
              }
              cache.putFileData(fileKey, cacheRanges, newCacheData, 0, tag);
            } finally {
              // We do not use the new cache buffers for the actual read, given the way read() API is.
              // Therefore, we don't need to handle cache collisions - just decref all the buffers.
              if (newCacheData != null) {
                for (MemoryBuffer buffer : newCacheData) {
                  if (buffer == null) continue;
                  cache.releaseBuffer(buffer);
                }
              }
              // If we have failed before building newCacheData, deallocate other the allocated.
              if (largeBuffers != null) {
                for (MemoryBuffer buffer : largeBuffers) {
                  if (buffer == null) continue;
                  allocator.deallocate(buffer);
                }
              }
              if (smallBuffer != null && smallBuffer[0] != null) {
                allocator.deallocate(smallBuffer[0]);
              }
            }
          }
        }
        validateAndUpdatePosition(len, sizeRead);
      }
      return len;
    }
This is currently only used by Parquet; however, a generally applicable approach is used -
you pass in a set of offset pairs for a file, and the file is cached with these boundaries.
Don't add anything format specific here.
ql/src/java/org/apache/hadoop/hive/llap/LlapCacheAwareFs.java