@VisibleForTesting
    Iterator<InternalRow> sort() throws IOException {
        try {
            final UnsafeSorterIterator sortedIterator = sorter.getSortedIterator();
            if (!sortedIterator.hasNext()) {
                // Since we won't ever call next() on an empty iterator, we need to clean up resources
                // here in order to prevent memory leaks.
                cleanupResources();
            }
            return new Iterator<InternalRow>() {

                private final int numFields = schema.size();
                private UnsafeRow row = new UnsafeRow(numFields);

                @Override
                public boolean hasNext() {
                    return sortedIterator.hasNext();
                }

                @Override
                public UnsafeRow next() {
                    try {
                        sortedIterator.loadNext();
                        row.pointTo(
                                sortedIterator.getBaseObject(),
                                sortedIterator.getBaseOffset(),
                                sortedIterator.getRecordLength());
                        if (!hasNext()) {
                            UnsafeRow copy = row.copy(); // so that we don't have dangling pointers to freed page
                            row = null; // so that we don't keep references to the base object
                            cleanupResources();
                            return copy;
                        } else {
                            return row;
                        }
                    } catch (IOException e) {
                        cleanupResources();
                        // Scala iterators don't declare any checked exceptions, so we need to use this hack
                        // to re-throw the exception:
                        Platform.throwException(e);
                    }
                    throw new RuntimeException("Exception should have been re-thrown in next()");
                }

                ;
            };
        } catch (IOException e) {
            cleanupResources();
            throw e;
        }
    }
Return the peak memory used so far, in bytes.
indexr-query-opt/src/main/java/org/apache/spark/sql/execution/UnsafeExternalRowSorter.java