code: @Override
    public long spill(long size, MemoryConsumer trigger) throws IOException {
      if (trigger != this) {
        if (readingIterator != null) {
          return readingIterator.spill();
        }
        return 0L; // this should throw exception
      }
  
      if (inMemSorter == null || inMemSorter.numRecords() <= 0) {
        return 0L;
      }
  
      logger.info("Thread {} spilling sort data of {} to disk ({} {} so far)",
        Thread.currentThread().getId(),
        Utils.bytesToString(getMemoryUsage()),
        spillWriters.size(),
        spillWriters.size() > 1 ? " times" : " time");
  
      ShuffleWriteMetrics writeMetrics = new ShuffleWriteMetrics();
  
      final UnsafeSorterSpillWriter spillWriter =
        new UnsafeSorterSpillWriter(blockManager, fileBufferSizeBytes, writeMetrics,
          inMemSorter.numRecords());
      spillWriters.add(spillWriter);
      spillIterator(inMemSorter.getSortedIterator(), spillWriter);
  
      final long spillSize = freeMemory();
      // Note that this is more-or-less going to be a multiple of the page size, so wasted space in
      // pages will currently be counted as memory spilled even though that space isn't actually
      // written to disk. This also counts the space needed to store the sorter's pointer array.
      inMemSorter.reset();
      // Reset the in-memory sorter's pointer array only after freeing up the memory pages holding the
      // records. Otherwise, if the task is over allocated memory, then without freeing the memory
      // pages, we might not be able to get memory for the pointer array.
  
      taskContext.taskMetrics().incMemoryBytesSpilled(spillSize);
      taskContext.taskMetrics().incDiskBytesSpilled(writeMetrics.bytesWritten());
      totalSpillBytes += spillSize;
      return spillSize;
    }
docs: Sort and spill the current records in response to memory pressure.