long compactValues(@NotNull List<CompactionRecordInfo> infos, @NotNull PersistentHashMapValueStorage storage) throws IOException {
    if (myOptions.myHasNoChunks) {
      return compactValuesWithoutChunks(infos, storage);
    }

    PriorityQueue<CompactionRecordInfo> records = new PriorityQueue<>(
      infos.size(), (info, info2) -> Long.compare(info2.valueAddress, info.valueAddress));

    records.addAll(infos);

    final int fileBufferLength = 256 * 1024;
    final int maxRecordHeader = 5 /* max length - variable int */ + 10 /* max long offset*/;
    final byte[] buffer = new byte[fileBufferLength + maxRecordHeader];
    byte[] reusedAccumulatedChunksBuffer = {};

    long lastReadOffset = mySize;
    long lastConsumedOffset = lastReadOffset;
    long allRecordsStart = 0;
    int fragments = 0;
    int newFragments = 0;

    byte[] stuffFromPreviousRecord = null;
    int bytesRead = (int)(mySize - (mySize / fileBufferLength) * fileBufferLength);
    long retained = 0;

    while (lastReadOffset != 0) {
      final long readStartOffset = lastReadOffset - bytesRead;
      myCompactionModeReader.get(readStartOffset, buffer, 0, bytesRead); // buffer contains [readStartOffset, readStartOffset + bytesRead)

      while (!records.isEmpty()) {
        final CompactionRecordInfo info = records.peek();
        if (info.valueAddress >= readStartOffset) {
          if (info.valueAddress >= lastReadOffset) {
            throw new IOException("Value storage is corrupted: value file size:" + mySize + ", readStartOffset:" + readStartOffset +
                                  ", record address:" + info.valueAddress + "; file: " + myPath);
          }
          // record start is inside our buffer

          final int recordStartInBuffer = (int)(info.valueAddress - readStartOffset);
          DataInputStream inputStream = toDataInputStream(buffer, recordStartInBuffer, buffer.length);

          if (stuffFromPreviousRecord != null && fileBufferLength - recordStartInBuffer < maxRecordHeader) {
            // add additional bytes to read offset / size
            if (allRecordsStart != 0) {
              myCompactionModeReader.get(allRecordsStart, buffer, bytesRead, maxRecordHeader);
            }
            else {
              final int maxAdditionalBytes = Math.min(stuffFromPreviousRecord.length, maxRecordHeader);
              System.arraycopy(stuffFromPreviousRecord, 0, buffer, bytesRead, maxAdditionalBytes);
            }
          }

          int available = inputStream.available();
          int chunkSize = readChunkSize(inputStream);
          final long prevChunkAddress = readPrevChunkAddress(info.valueAddress, inputStream);
          final int dataOffset = available - inputStream.available();

          byte[] accumulatedChunksBuffer;
          if (info.value != null) {
            int defragmentedChunkSize = info.value.length + chunkSize;
            if (prevChunkAddress == 0) {
              if (defragmentedChunkSize >= reusedAccumulatedChunksBuffer.length) {
                reusedAccumulatedChunksBuffer = new byte[defragmentedChunkSize];
              }
              accumulatedChunksBuffer = reusedAccumulatedChunksBuffer;
            }
            else {
              accumulatedChunksBuffer = new byte[defragmentedChunkSize];
              retained += defragmentedChunkSize;
            }
            System.arraycopy(info.value, 0, accumulatedChunksBuffer, chunkSize, info.value.length);
          }
          else {
            if (prevChunkAddress == 0) {
              if (chunkSize >= reusedAccumulatedChunksBuffer.length) reusedAccumulatedChunksBuffer = new byte[chunkSize];
              accumulatedChunksBuffer = reusedAccumulatedChunksBuffer;
            }
            else {
              accumulatedChunksBuffer = new byte[chunkSize];
              retained += chunkSize;
            }
          }

          final int chunkSizeOutOfBuffer = MathUtil.clamp((int)(info.valueAddress + dataOffset + chunkSize - lastReadOffset), 0, chunkSize);
          if (chunkSizeOutOfBuffer > 0) {
            if (allRecordsStart != 0) {
              myCompactionModeReader.get(allRecordsStart, accumulatedChunksBuffer, chunkSize - chunkSizeOutOfBuffer, chunkSizeOutOfBuffer);
            }
            else {
              int offsetInStuffFromPreviousRecord = Math.max((int)(info.valueAddress + dataOffset - lastReadOffset), 0);
              // stuffFromPreviousRecord starts from lastReadOffset
              System.arraycopy(stuffFromPreviousRecord, offsetInStuffFromPreviousRecord, accumulatedChunksBuffer,
                               chunkSize - chunkSizeOutOfBuffer, chunkSizeOutOfBuffer);
            }
          }

          stuffFromPreviousRecord = null;
          allRecordsStart = 0;

          lastConsumedOffset = info.valueAddress;
          checkPreconditions(accumulatedChunksBuffer, chunkSize);

          System.arraycopy(buffer, recordStartInBuffer + dataOffset, accumulatedChunksBuffer, 0, chunkSize - chunkSizeOutOfBuffer);

          ++fragments;
          records.remove(info);
          if (info.value != null) {
            chunkSize += info.value.length;
            retained -= info.value.length;
            info.value = null;
          }

          if (prevChunkAddress == 0) {
            info.newValueAddress = storage.appendBytes(accumulatedChunksBuffer, 0, chunkSize, info.newValueAddress);
            ++newFragments;
          }
          else {
            if (retained > SOFT_MAX_RETAINED_LIMIT &&
                accumulatedChunksBuffer.length > BLOCK_SIZE_TO_WRITE_WHEN_SOFT_MAX_RETAINED_LIMIT_IS_HIT ||
                retained > MAX_RETAINED_LIMIT_WHEN_COMPACTING) {
              // to avoid OOME we need to save bytes in accumulatedChunksBuffer
              newFragments +=
                saveAccumulatedDataOnDiskPreservingWriteOrder(storage, info, prevChunkAddress, accumulatedChunksBuffer, chunkSize);
              retained -= accumulatedChunksBuffer.length;
              continue;
            }
            info.value = accumulatedChunksBuffer;
            info.valueAddress = prevChunkAddress;
            records.add(info);
          }
        }
        else {
          // [readStartOffset,lastConsumedOffset) is from previous segment
          if (stuffFromPreviousRecord == null) {
            stuffFromPreviousRecord = new byte[(int)(lastConsumedOffset - readStartOffset)];
            System.arraycopy(buffer, 0, stuffFromPreviousRecord, 0, stuffFromPreviousRecord.length);
          }
          else {
            allRecordsStart = readStartOffset;
          }
          break; // request next read
        }
      }

      lastReadOffset -= bytesRead;
      bytesRead = fileBufferLength;
    }

    return fragments | ((long)newFragments << 32);
  }
long compactValues(@NotNull List<CompactionRecordInfo> infos, @NotNull PersistentHashMapValueStorage storage)