Merge lp:~tjoneslo/akiban-server/file-merge-sort into lp:~akiban-technologies/akiban-server/trunk

[Thomas Jones-Low]

Add the FileMergeSort to the akiban trunk.

This relies upon a library called java-merge-sort, done by the same people that did the Jackson JSON libaray already in use.

All of the code is in the new MergeJoinSorter and consists of class to translate between the AkibanServer way of doing things and what the java merge sort expected.

SortKey -> the single class that holds the KeyState for comparison and the single Key for the whole row.
KeyReadCursor -> Reads rows from the input cursor, converts them to SortKeys, and passes them to the java merge sort for processing. This is the external input.
KeyFinalCursor -> reads from the final sorted output and return rows through the cursor interface for next operator in line. This is the external output.

KeyReader/KeyWriter -> used by the java merge sort library to read/write the intermediate blocks of SortKeys.

KeySortCompare -> used by the java merge sort library to compare the SortKeys to put them into the correct order.

Internally the java merge sort library reads input (KeyReadCursor) and sorts it (KeySortCompare) into a buffer (40MB). If size of the input exceeds the buffer size, the (full) buffer is written to disk (KeyWriter) and the process repeated with a new buffer. When all of the input has been read, the java merge sort library does a 16 way merge (KeyReader) to create the final output. This final output is written (KeyFinalCursor) into the row stream.

Also two complete new tests, one for the KeyReader/Writer pair, the other for the KeyReadCursor and KeyFinalCursor to make sure the process doesn't magle rows on the way through. And Extend the SortIT test to verify the sorting works as expected.

[Thomas Jones-Low (tjoneslo)]

This isn't currently hooked up to the any of the Sort operators.

[Nathan Williams (nwilliams)]

Nice! This looks pretty good overall. A handful of small issues below.

The exceptions on lines 166 and 192 probably needs bubbled up.

Line 291, 306 and 319 look suspicious. InputStream returns -1 at EOF, whereas this is checking read < size. Maybe an assert instead?

Line 491, a new config instead of persistit.tmpvoldir might be nice. Or perhaps an akserver.tmp_dir, which TreeService can then assign to tmpvoldir if otherwise unset.

Did you investigate if the fasterxml.Sorter closes all of the temp files from the provider on finish? On error? If not, I wonder if we need to hang onto them all and ensure they are in close().

Line 496, could SimplQueryContext look at adapter.getSession().sessionId() instead of throwing? Or is there no Session either?

Line 550, is that TODO relevant or just stale?

Line 576, I think we just have to eat that one. Maybe log for... good measure?

TestKeyReaderWriter.random being static and unseeded makes me a little nervous. A non-deterministic IT is asking for headaches.

Does surefire actually pick up TestKeyReaderWriter? I think our pattern only looks for *Test.java.

Maybe a new configuration that get shimmed into SortConfig? I can imagine wanting to set that pretty quickly.

review: Needs Fixing

[lp:~tjoneslo/akiban-server/file-merge-sort]

2707. By tjoneslo on 2013-07-27

Address merge comments.
Add new error code for returning problems on IO errors
Clean up other error handling.

[Thomas Jones-Low (tjoneslo)]

> Nice! This looks pretty good overall. A handful of small issues below.
>
> The exceptions on lines 166 and 192 probably needs bubbled up.
>
Done

> Line 291, 306 and 319 look suspicious. InputStream returns -1 at EOF, whereas
> this is checking read < size. Maybe an assert instead?
>
Converted to an assert.

> Line 491, a new config instead of persistit.tmpvoldir might be nice. Or
> perhaps an akserver.tmp_dir, which TreeService can then assign to tmpvoldir if
> otherwise unset.
>
tmpvoldir is a persistit specific configuration item, the only way to alter the manner in which it gets set is to make changes to Persistit.

Added the akserver.tmp_dir variable and use that instead.

> Did you investigate if the fasterxml.Sorter closes all of the temp files from
> the provider on finish? On error? If not, I wonder if we need to hang onto
> them all and ensure they are in close().
>
   The initial read into the temp files does so correctly. The merge stage does not, and I've filed a bug in the Merge-sort github. https://github.com/cowtowncoder/java-merge-sort/issues/8

> Line 496, could SimplQueryContext look at adapter.getSession().sessionId()
> instead of throwing? Or is there no Session either?
>

Fixed, removed the try/catch in MergeJoinSorter too. The adapter session sessionIDs and the context sessionIDs are different creatures (the former is from SessionService, the latter are from the MontiorService) but for these purposes they can be the same.

> Line 550, is that TODO relevant or just stale?
>
It's relevant. I still don't have an answer for that.

> Line 576, I think we just have to eat that one. Maybe log for... good measure?
>

> TestKeyReaderWriter.random being static and unseeded makes me a little
> nervous. A non-deterministic IT is asking for headaches.
>

Fixed.

> Does surefire actually pick up TestKeyReaderWriter? I think our pattern only
> looks for *Test.java.
>

Renamed to KeyReaderWriterTest

> Maybe a new configuration that get shimmed into SortConfig? I can imagine
> wanting to set that pretty quickly.

The only other configuration item is the size of the sort buffer. Added a akserver.sort.memory configuration item and set the default to 64M.

review: Needs Resubmitting

[Nathan Williams (nwilliams)]

> tmpvoldir is a persistit specific configuration item,
> the only way to alter the manner in which it gets set
> is to make changes to Persistit.

Sorry, I wasn't clear. TreeServiceImpl#setupPersistitProperties() does some remapping already. It can take the new tmp_dir config and map it to tmpvoldir. That way we only need one in the config file.

>> Line 550, is that TODO relevant or just stale?
> It's relevant. I still don't have an answer for that.

Now 576. I think it needs to throw an exception instead of returning null. Null indicates end of stream and query would succeed, whereas an IOException may mean one of the sort files couldn't be read (for example).

The Random in the test is still static, which makes it sensitive to run order. Could we make it not static for the unlikely-but-frustrating scenario if something ever goes wrong?

review: Needs Fixing

[lp:~tjoneslo/akiban-server/file-merge-sort]

2708. By tjoneslo on 2013-07-29

Update TreeServiceImpl to use new tmp_dir for persistit tmpvoldir to have only one config item
Fix missing exception in MergeJoinSorter
Fix the reused static random in the Key tests.

[Thomas Jones-Low (tjoneslo)]

address review comments

- Update TreeServiceImpl processing to use akserver.tmp_dir into persisitit.tmpvoldir
- Throw exception from MergeJoinSort
- Remove/fix the static random() used in two places that may cause problems.

review: Needs Resubmitting

[Nathan Williams (nwilliams)]

Excellent. Thanks for the tweaks.

review: Approve

=== modified file 'pom.xml'
--- pom.xml	2013-06-14 11:23:11 +0000
+++ pom.xml	2013-07-29 16:50:42 +0000
@@ -206,6 +206,11 @@
             <version>4.0.0</version>
         </dependency>
         <dependency>
+           <groupId>com.fasterxml.util</groupId>
+           <artifactId>java-merge-sort</artifactId>
+           <version>0.7.1</version>
+        </dependency>
+        <dependency>
           <groupId>org.apache.httpcomponents</groupId>
           <artifactId>httpclient</artifactId>
           <version>4.2.3</version>

=== modified file 'src/main/java/com/akiban/qp/operator/SimpleQueryContext.java'
--- src/main/java/com/akiban/qp/operator/SimpleQueryContext.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/qp/operator/SimpleQueryContext.java	2013-07-29 16:50:42 +0000
@@ -77,6 +77,10 @@
 
     @Override
     public int getSessionId() {
+        if (adapter != null) {
+            return (int)adapter.getSession().sessionId();
+        }
+
         throw new UnsupportedOperationException();
     }
 

=== modified file 'src/main/java/com/akiban/qp/persistitadapter/PersistitAdapter.java'
--- src/main/java/com/akiban/qp/persistitadapter/PersistitAdapter.java	2013-07-18 18:28:04 +0000
+++ src/main/java/com/akiban/qp/persistitadapter/PersistitAdapter.java	2013-07-29 16:50:42 +0000
@@ -93,6 +93,7 @@
                                InOutTap loadTap)
     {
         return new PersistitSorter(context, bindings, input, rowType, ordering, sortOption, loadTap);
+        //return new MergeJoinSorter(context, bindings, input, rowType, ordering, sortOption, loadTap);
     }
 
     @Override

=== added file 'src/main/java/com/akiban/qp/persistitadapter/indexcursor/MergeJoinSorter.java'
--- src/main/java/com/akiban/qp/persistitadapter/indexcursor/MergeJoinSorter.java	1970-01-01 00:00:00 +0000
+++ src/main/java/com/akiban/qp/persistitadapter/indexcursor/MergeJoinSorter.java	2013-07-29 16:50:42 +0000
@@ -0,0 +1,621 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package com.akiban.qp.persistitadapter.indexcursor;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import com.akiban.qp.operator.API;
+import com.akiban.qp.operator.API.Ordering;
+import com.akiban.qp.operator.CursorLifecycle;
+import com.akiban.qp.operator.QueryBindings;
+import com.akiban.qp.operator.QueryContext;
+import com.akiban.qp.operator.RowCursor;
+import com.akiban.qp.persistitadapter.Sorter;
+import com.akiban.qp.row.Row;
+import com.akiban.qp.row.ValuesHolderRow;
+import com.akiban.qp.rowtype.RowType;
+import com.akiban.server.PersistitKeyPValueSource;
+import com.akiban.server.PersistitKeyPValueTarget;
+import com.akiban.server.api.dml.ColumnSelector;
+import com.akiban.server.error.MergeSortIOException;
+import com.akiban.server.types3.TInstance;
+import com.akiban.server.types3.pvalue.PValueSource;
+import com.akiban.server.types3.pvalue.PValueTargets;
+import com.akiban.util.tap.InOutTap;
+import com.persistit.Key;
+import com.persistit.KeyState;
+import com.persistit.Persistit;
+import com.persistit.exception.KeyTooLongException;
+
+import com.fasterxml.sort.DataReader;
+import com.fasterxml.sort.DataReaderFactory;
+import com.fasterxml.sort.DataWriter;
+import com.fasterxml.sort.DataWriterFactory;
+import com.fasterxml.sort.SortConfig;
+import com.fasterxml.sort.TempFileProvider;
+
+/**
+ * <h1>Overview</h1>
+ *
+ * Sort rows by inserting them in sorted order into a memory buffer, then into on-disk files, and performing 
+ * a multiway merge sort on the resulting files. 
+ *
+ * <h1>Behavior</h1>
+ *
+ * The rows of the input stream are written into a memory pool (40MB as a default). When (if) the memory pool is filled,
+ * the pool is written to disk, then emptied and rows are written again to the memory pool. When the final row is written
+ * to the disk files are merged in an n-way merge sort. The default is 16 way sort. When all of the temp files are merged
+ * into one file, the rows are read from the file in sorted order. 
+ * 
+ * If the initial input stream does not produce enough data to overflow the memory pool, no disk files will be produced. 
+ *
+ * <h1>Performance</h1>
+ *
+ * The MergeJoinSorter generates IO dependent upon the size of the input stream. 
+ * If the input stream generates less than the memory pool size (40MB default), there is no IO generated. 
+ * If the input stream generates more than the memory pool size, but less than 16x the pool size, it should
+ * generate two read and two writes for each row. One write to, one read from into the initial temporary file, 
+ * one write to, one read from the final sorted temporary file. For each 16x larger the input set gets it adds
+ * one more write to/read from temporary file cycle.  
+ *
+ * <h1>Memory Requirements</h1>
+ *
+ * The MergeJoinSorter allocates a single memory buffer for each instance to perform an initial sort,
+ * defaulting to 40MB in size. 
+*/
+
+public class MergeJoinSorter implements Sorter {
+
+    private QueryContext context;
+    private QueryBindings bindings;
+    private RowCursor input;
+    private RowType rowType;
+    private Ordering ordering;
+    private InOutTap loadTap;
+
+    private File finalFile;
+
+    private final SorterAdapter<?, ?, ?> sorterAdapter;
+    private final List<Integer> orderChanges;
+    private Key sortKey;
+    private Comparator<SortKey> compare;
+    
+    public MergeJoinSorter (QueryContext context,
+            QueryBindings bindings,
+            RowCursor input,
+            RowType rowType,
+            API.Ordering ordering,
+            API.SortOption sortOption,
+            InOutTap loadTap)
+    {
+        this.context = context;
+        this.bindings = bindings;
+        this.input = input;
+        this.rowType = rowType;
+        this.ordering = ordering.copy();
+        this.loadTap = loadTap;
+        
+        this.sortKey = new Key ((Persistit)null);
+        this.sorterAdapter = new PValueSorterAdapter();
+        // Note: init may change this.ordering
+        sorterAdapter.init(rowType, this.ordering, this.sortKey, null, this.context, this.bindings, sortOption);
+        // Explicitly use input ordering to avoid appended field
+        this.orderChanges = new ArrayList<>();
+        List<Comparator<KeyState>> comparators = new ArrayList<>();
+        for(int i = 0; i < ordering.sortColumns(); ++i) {
+            Comparator<KeyState> c = ordering.ascending(i) ? ASC_COMPARATOR : DESC_COMPARATOR;
+            if(i == 0 || ordering.ascending(i-1) != ordering.ascending(i)) {
+                orderChanges.add(i);
+                comparators.add(c);
+            }
+        }
+        this.orderChanges.add(ordering.sortColumns());
+        this.compare = new KeySortCompare(comparators);
+        
+    }
+
+    @Override
+    public RowCursor sort() {
+        try {
+            loadTree();
+        } catch (IOException e) {
+            throw new MergeSortIOException(e);
+        }
+        return cursor();
+    }
+
+    @Override
+    public void close() {
+    }
+    
+    private void loadTree() throws FileNotFoundException, IOException {
+        MergeTempFileProvider tmpFileProvider = new MergeTempFileProvider(context);
+        finalFile  = tmpFileProvider.provide();
+        
+        com.fasterxml.sort.Sorter<SortKey> s = new com.fasterxml.sort.Sorter<SortKey> (
+                getSortConfig(tmpFileProvider),
+                new KeyReaderFactory(), 
+                new KeyWriterFactory(), 
+                compare);
+        s.sort(new KeyReadCursor(), new KeyWriter(new FileOutputStream(finalFile)));
+    }
+    
+    private RowCursor cursor() {
+        KeyFinalCursor cursor = null;
+        try {
+            cursor = new KeyFinalCursor (finalFile, rowType);
+        } catch (FileNotFoundException e) {
+            throw new MergeSortIOException(e);
+        }
+        return cursor;
+    }
+
+    public KeyReadCursor readCursor() { 
+        return new KeyReadCursor(); 
+    }
+    
+    private SortConfig getSortConfig (MergeTempFileProvider tmpFileProvider) {
+        long maxMemory = Long.parseLong(context.getServiceManager().getConfigurationService().getProperty("akserver.sort.memory"));
+        return new SortConfig().withTempFileProvider(tmpFileProvider).withMaxMemoryUsage(maxMemory);
+    }
+    /*
+     * Base class for reading/writing bytes - 
+     * KeyState[] is list of key segments broken by ASC/DESC ordering
+     * rowKey is the whole, unaltered row of data. 
+     */
+    public static class SortKey {
+        public List<KeyState> sortKeys;
+        public Key rowKey;
+     
+        public SortKey () {
+            this.sortKeys = new ArrayList<>();
+            this.rowKey = new Key ((Persistit)null);
+            rowKey.clear();
+        }
+        
+        public SortKey (List<KeyState> sortKeys, Key rowKey) {
+            this.sortKeys = sortKeys;
+            this.rowKey = rowKey;
+        }
+        
+        // Sorter uses size of elements to determine when the 
+        // presort buffer is full. 
+        public int getSize() {
+            int size = 0;
+            for (KeyState state : sortKeys) {
+                size += state.getBytes().length + 4;
+                size += 4;
+            }
+            size += rowKey.getEncodedSize() + 4;
+            return size;
+        }
+    }
+
+    private class KeyReaderFactory extends DataReaderFactory<SortKey> {
+
+        @Override
+        public DataReader<SortKey> constructReader(InputStream arg0)
+                throws IOException {
+            return new KeyReader(arg0);
+        }
+        
+    }
+    
+    public static class KeyReader extends DataReader<SortKey> {
+
+        private InputStream is;
+        private ByteBuffer length;
+        public KeyReader (InputStream is) {
+            this.is = is;
+            length = ByteBuffer.allocate(4);
+        }
+        
+        @Override
+        public void close() throws IOException {
+            is.close();
+        }
+
+        @Override
+        public int estimateSizeInBytes(SortKey arg0) {
+            return arg0.getSize();
+        }
+
+        @Override
+        public SortKey readNext() throws IOException {
+            
+            SortKey key = new SortKey();
+            int states = readLength();
+            if (states < 0) { 
+                return null;
+            }
+            for (int i = 0; i < states; i++) {
+                KeyState state = readKeyState();
+                if (state == null) {
+                    return null;
+                }
+                key.sortKeys.add(state);
+            }
+            
+            key.rowKey = readKey();
+            
+            return key;
+        }
+
+        private KeyState readKeyState() throws IOException {
+            int size = readLength();
+            if (size < 1) {
+                return null;
+            }
+            byte[] bytes = new byte[size];
+            int bytesRead = is.read(bytes);
+            
+            assert bytesRead == size: "Invalid byte count on key state read";
+            
+            return new KeyState(bytes);
+        }
+        
+        private Key readKey() throws IOException{
+            int size = readLength();
+            if (size < 1) {
+                return null;
+            }
+            Key key = new Key ((Persistit)null);
+            key.setMaximumSize(size);
+            int bytesRead = is.read(key.getEncodedBytes(), 0, size);
+            
+            assert bytesRead == size : "Invalid byte count on key read";
+
+            key.setEncodedSize(size);
+            return key;
+        }
+        
+        private int readLength() throws IOException {
+            length.clear();
+            int bytesRead = is.read(length.array());
+            if (bytesRead == -1) { // EOF marker
+                return -1;
+            } 
+            assert bytesRead == 4 : "Invalid byte count on length read";
+            return length.getInt();
+        }
+    }
+
+    /*
+     * Class to read rows from the input cursor to the Sort, 
+     * converting them to SortKey elements for the Sorter. 
+     */
+    public class KeyReadCursor extends DataReader<SortKey> {
+        
+        private int rowCount = 0;
+        private Key convertKey;
+        private int rowFields;
+        private TInstance tFieldTypes[];
+        private PersistitKeyPValueTarget valueTarget;
+        
+        public KeyReadCursor () {
+            this.rowFields = rowType.nFields();
+            this.convertKey = new Key ((Persistit)null);
+            this.tFieldTypes = new TInstance[rowFields];
+            for (int i = 0; i < rowFields; i++) {
+                tFieldTypes[i] = rowType.typeInstanceAt(i);
+            }
+            valueTarget = new PersistitKeyPValueTarget();
+            valueTarget.attach(convertKey);
+        }
+        
+        @Override
+        public void close() throws IOException {
+            // Do Nothing;
+        }
+
+        @Override
+        public int estimateSizeInBytes(SortKey arg0) {
+            return arg0.getSize();
+        }
+
+        @Override
+        public SortKey readNext() throws IOException {
+            SortKey sortKey = null;
+            loadTap.in();
+            try {
+                Row row = input.next();
+                context.checkQueryCancelation();
+    
+                if (row != null) {
+                    ++rowCount;
+                    sortKey = new SortKey (createKey(row, rowCount), createValue(row));
+                }
+            } finally {
+                loadTap.out();
+            }
+            return sortKey;
+        }
+        
+        private List<KeyState> createKey(Row row, int rowCount) {
+            KeyState[] states = new KeyState[orderChanges.size() - 1];
+            for(int i = 0; i < states.length; ++i) {
+                int startOffset = orderChanges.get(i);
+                int endOffset = orderChanges.get(i + 1);
+                boolean isLast = i == states.length - 1;
+                // Loop for key growth
+                while(true) {
+                    try {
+                        sortKey.clear();
+                        for(int j = startOffset; j < endOffset; ++j) {
+                            sorterAdapter.evaluateToKey(row, j);
+                        }
+                        if(isLast && sorterAdapter.preserveDuplicates()) {
+                            sortKey.append(rowCount);
+                        }
+                        break;
+                    } catch (KeyTooLongException e) {
+                        enlargeKey(sortKey);
+                    }
+                }
+                states[i] = new KeyState(sortKey);
+            }
+            return Arrays.asList(states);
+        }
+
+        private Key createValue(Row row)
+        {
+            while(true) {
+                try {
+                    convertKey.clear();
+                    for (int i = 0; i < rowFields; i++) {
+                        //sorterAdapter.evaluateToTarget(row, i);
+                        PValueSource field = row.pvalue(i);
+                        //putFieldToTarget(field, i, oFieldTypes, tFieldTypes);
+                        tFieldTypes[i].writeCanonical(field, valueTarget);
+                    }
+                    break;
+                } catch (KeyTooLongException e) {
+                    enlargeKey(convertKey);
+                }
+            }
+            return new Key(convertKey);
+        }
+        
+        private void enlargeKey (Key key) {
+            if (key.getMaximumSize() == Key.MAX_KEY_LENGTH_UPPER_BOUND) {
+                throw new KeyTooLongException("Maximum size exceeded=" + Key.MAX_KEY_LENGTH_UPPER_BOUND);
+            }
+            key.setMaximumSize(Math.min((key.getMaximumSize() * 2), Key.MAX_KEY_LENGTH_UPPER_BOUND));
+        }
+        
+        public int rowCount() {
+            return rowCount;
+        }
+    }
+    
+    private class KeyWriterFactory extends DataWriterFactory<SortKey> {
+
+        @Override
+        public DataWriter<SortKey> constructWriter(OutputStream arg0)
+                throws IOException {
+            return new KeyWriter(arg0);
+        }
+    }
+    
+    public static class KeyWriter extends DataWriter<SortKey> {
+        private OutputStream os;
+        private ByteBuffer length;
+
+        public KeyWriter(OutputStream os) {
+            this.os = os;
+            length = ByteBuffer.allocate(4);
+        }
+        @Override
+        public void close() throws IOException {
+            os.close();
+            
+        }
+
+        @Override
+        public void writeEntry(SortKey arg0) throws IOException {
+            writeInt(arg0.sortKeys.size());
+            for (KeyState state : arg0.sortKeys) {
+                writeKeyState (state);
+            }
+            writeKey (arg0.rowKey);
+        }
+        
+        private void writeKeyState (KeyState state) throws IOException {
+            writeInt (state.getBytes().length);
+            os.write(state.getBytes());
+        }
+        
+        private void writeKey (Key key) throws IOException {
+            writeInt(key.getEncodedSize());
+            os.write(key.getEncodedBytes(), 0, key.getEncodedSize());
+        }
+        
+        private void writeInt (int size) throws IOException {
+            length.clear();
+            length.putInt(size);
+            os.write(length.array());
+        }
+    }
+
+    /*
+     * Class to provide temporary file names for inserting the 
+     * overflow buffers to disk. Implemented to the MergeJoin sort interface
+     */
+    public class MergeTempFileProvider implements TempFileProvider {
+        
+        private final File directory;
+        private final String prefix;
+        private final String suffix;
+        public MergeTempFileProvider (QueryContext context) {
+            directory = new File (context.getServiceManager().getConfigurationService().getProperty("akserver.tmp_dir"));
+            suffix = ".tmp";
+            String tmpPrefix;
+            tmpPrefix = "sort-" +  context.getSessionId() + "-";
+            prefix = tmpPrefix;
+        }
+
+        @Override
+        public File provide() throws IOException {
+            File f = File.createTempFile(prefix, suffix, directory);
+            f.deleteOnExit();
+            return f;
+        }
+    }
+    
+    /*
+     * Class to create a cursor which reads the final sorted output
+     * from the file, returning each sorted item as a Row. 
+     */
+    public static class KeyFinalCursor implements RowCursor {
+        private boolean isIdle = true;
+        private boolean isDestroyed = false;
+
+        private final KeyReader read; 
+        private final RowType rowType;
+        private PersistitKeyPValueSource valueSources[]; 
+        
+        public KeyFinalCursor (File inputFile, RowType rowType) throws FileNotFoundException {
+            this (new FileInputStream(inputFile), rowType);
+        }
+        
+        public KeyFinalCursor(InputStream stream, RowType rowType) {
+            read = new KeyReader(stream);
+            this.rowType = rowType;
+            valueSources = new PersistitKeyPValueSource[rowType.nFields()];
+            for (int i = 0; i < rowType.nFields(); i++) {
+                valueSources[i] = new PersistitKeyPValueSource (rowType.typeInstanceAt(i));
+            }
+        }
+        
+        @Override
+        public void open() {
+            CursorLifecycle.checkIdle(this);
+            isIdle = false;
+        }
+
+        @Override
+        public Row next() {
+            CursorLifecycle.checkIdleOrActive(this);
+            SortKey key;
+            Row row = null;
+            try {
+                key = read.readNext();
+            } catch (IOException e) {
+                throw new MergeSortIOException (e);
+            }
+            if (key != null) {
+                row = createRow (key);
+                return row;
+            }
+            return null;
+        }
+        
+        private Row createRow (SortKey key) {
+            ValuesHolderRow rowCopy = new ValuesHolderRow(rowType, true);
+            for(int i = 0 ; i < rowType.nFields(); ++i) {
+                valueSources[i].attach(key.rowKey, i, valueSources[i].tInstance());
+                PValueTargets.copyFrom(valueSources[i], rowCopy.pvalueAt(i));
+            }
+            return rowCopy;
+        }
+        
+        @Override
+        public void close() {
+            CursorLifecycle.checkIdleOrActive(this);
+            if(!isIdle) {
+                try {
+                    read.close();
+                } catch (IOException e) {
+                    // TODO: manage this exception? 
+                }
+                isIdle = true;
+            }
+        }
+
+        @Override
+        public void jump(Row row, ColumnSelector columnSelector) {
+            throw new UnsupportedOperationException();            
+        }
+
+        @Override
+        public void destroy() {
+            isDestroyed = true;
+        }
+
+        @Override
+        public boolean isIdle() {
+            return !isDestroyed && isIdle;
+        }
+
+        @Override
+        public boolean isActive() {
+            return !isDestroyed && !isIdle;
+        }
+
+        @Override
+        public boolean isDestroyed() {
+            return isDestroyed;
+        }
+    }
+    
+    /*
+     * Comparison function, implemented for MergeSort to compare
+     * the KeyState lists generated by the KeyReadCursor
+     */
+    public static class KeySortCompare implements Comparator<SortKey> {
+        private final Comparator<KeyState>[] comparators;
+
+        @SuppressWarnings("unchecked")
+        private KeySortCompare (List<Comparator<KeyState>> comparators) {
+            this.comparators = comparators.toArray(new Comparator[comparators.size()]);
+        }
+
+        @Override
+        public int compare(SortKey o1, SortKey o2) {
+            int val = 0;
+            for (int i = 0; (i < comparators.length) && (val == 0); ++i) {
+                val = comparators[i].compare(o1.sortKeys.get(i), o2.sortKeys.get(i));
+            }
+            return val;
+        }
+    }
+
+    private static final Comparator<KeyState> ASC_COMPARATOR = new Comparator<KeyState>() {
+        @Override
+        public int compare(KeyState k1, KeyState k2) {
+            return k1.compareTo(k2);
+        }
+    };
+
+    private static final Comparator<KeyState> DESC_COMPARATOR = new Comparator<KeyState>() {
+        @Override
+        public int compare(KeyState k1, KeyState k2) {
+            return k2.compareTo(k1);
+        }
+    };
+}
\ No newline at end of file

=== modified file 'src/main/java/com/akiban/server/error/ErrorCode.java'
--- src/main/java/com/akiban/server/error/ErrorCode.java	2013-07-19 19:40:21 +0000
+++ src/main/java/com/akiban/server/error/ErrorCode.java	2013-07-29 16:50:42 +0000
@@ -422,6 +422,7 @@
     PROTOBUF_READ           ("53", "00A", Importance.ERROR, ProtobufReadException.class),
     PROTOBUF_WRITE          ("53", "00B", Importance.ERROR, ProtobufWriteException.class),
     INVALID_ALTER           ("53", "00C", Importance.ERROR, InvalidAlterException.class),
+    MERGE_SORT_IO           ("53", "00D", Importance.ERROR, MergeSortIOException.class),
     
     // Class 55 - Type conversion errors
     UNKNOWN_TYPE            ("55", "001", Importance.DEBUG, UnknownDataTypeException.class),    

=== added file 'src/main/java/com/akiban/server/error/MergeSortIOException.java'
--- src/main/java/com/akiban/server/error/MergeSortIOException.java	1970-01-01 00:00:00 +0000
+++ src/main/java/com/akiban/server/error/MergeSortIOException.java	2013-07-29 16:50:42 +0000
@@ -0,0 +1,27 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package com.akiban.server.error;
+
+import java.io.IOException;
+
+public class MergeSortIOException extends InvalidOperationException {
+
+    public MergeSortIOException(IOException ex) {
+        super(ErrorCode.MERGE_SORT_IO, ex.getMessage());
+    }
+
+}

=== modified file 'src/main/java/com/akiban/server/service/tree/TreeServiceImpl.java'
--- src/main/java/com/akiban/server/service/tree/TreeServiceImpl.java	2013-05-03 00:59:49 +0000
+++ src/main/java/com/akiban/server/service/tree/TreeServiceImpl.java	2013-07-29 16:50:42 +0000
@@ -42,6 +42,7 @@
 import com.akiban.server.service.session.Session;
 import com.akiban.server.service.session.SessionService;
 import com.google.inject.Inject;
+import com.persistit.Configuration;
 import com.persistit.Exchange;
 import com.persistit.Key;
 import com.persistit.Persistit;
@@ -63,6 +64,8 @@
     private static final String PERSISTIT_MODULE_NAME = "persistit.";
 
     private static final String DATAPATH_PROP_NAME = "datapath";
+    
+    private static final String TEMPDIR_NAME = "akserver.tmp_dir";
 
     private static final String BUFFER_SIZE_PROP_NAME = "buffersize";
     
@@ -189,6 +192,13 @@
         properties.setProperty(DATAPATH_PROP_NAME, datapath);
         ensureDirectoryExists(datapath, false);
 
+        //
+        // Copied the akserver.tmp_dir property to the Persistit tmpvoldir 
+        // The latter is used for temporary Persistit volumes used for sorting.
+        final String tmpPath = configService.getProperty(TEMPDIR_NAME);
+        properties.setProperty(Configuration.TEMPORARY_VOLUME_DIR_PROPERTY_NAME, tmpPath);
+        ensureDirectoryExists(tmpPath, false);
+        
         // Get the configured buffer size:
         // Default is 16K. Can be overridden with
         //

=== modified file 'src/main/resources/com/akiban/server/error/error_code.properties'
--- src/main/resources/com/akiban/server/error/error_code.properties	2013-07-19 19:40:21 +0000
+++ src/main/resources/com/akiban/server/error/error_code.properties	2013-07-29 16:50:42 +0000
@@ -302,6 +302,8 @@
 PROTOBUF_READ           = Error while deserializing protobuf message type {0}: {1}
 PROTOBUF_WRITE          = Error while serializing protobuf message type {0}: {1}
 INVALID_ALTER           = Invalid alter request on `{0}`.`{1}`: {2}
+MERGE_SORT_IO           = Merge Sort had an unexpected IOException: {0}
+ 
 #
 # Class 54 - Program limit exceeded
 #

=== modified file 'src/main/resources/com/akiban/server/service/config/configuration-defaults.properties'
--- src/main/resources/com/akiban/server/service/config/configuration-defaults.properties	2013-07-26 23:42:34 +0000
+++ src/main/resources/com/akiban/server/service/config/configuration-defaults.properties	2013-07-29 16:50:42 +0000
@@ -33,6 +33,7 @@
 akserver.routines.script_class_path=
 akserver.index_statistics.bucket_count=32
 akserver.restrict_user_schema=false
+akserver.tmp_dir=/tmp
 akserver.text.indexpath=/tmp/aktext
 akserver.text.backgroundInterval=3000
 # EXPERIMENTAL
@@ -44,6 +45,8 @@
 akserver.write_lock_enabled=true
 akserver.lookaheadQuantum.indexScan=1
 akserver.lookaheadQuantum.groupLookup=1
+## 64M per sort instance
+akserver.sort.memory=67108864
 
 persistit.buffersize=16384
 

=== added directory 'src/test/java/com/akiban/qp/persistitadapter'
=== added directory 'src/test/java/com/akiban/qp/persistitadapter/indexcursor'
=== added file 'src/test/java/com/akiban/qp/persistitadapter/indexcursor/KeyFinalCursorIT.java'
--- src/test/java/com/akiban/qp/persistitadapter/indexcursor/KeyFinalCursorIT.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/qp/persistitadapter/indexcursor/KeyFinalCursorIT.java	2013-07-29 16:50:42 +0000
@@ -0,0 +1,170 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.akiban.qp.persistitadapter.indexcursor;
+
+import static com.akiban.qp.operator.API.cursor;
+import static com.akiban.qp.operator.API.valuesScan_Default;
+import static org.junit.Assert.*;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import com.akiban.qp.operator.API;
+import com.akiban.qp.operator.Cursor;
+import com.akiban.qp.operator.Operator;
+import com.akiban.qp.operator.RowCursor;
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter.KeyReadCursor;
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter.KeyReader;
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter.KeyWriter;
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter.KeyFinalCursor;
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter.SortKey;
+import com.akiban.qp.row.BindableRow;
+import com.akiban.qp.row.RowBase;
+import com.akiban.qp.rowtype.RowType;
+import com.akiban.qp.rowtype.Schema;
+import com.akiban.server.test.it.qp.OperatorITBase;
+import com.akiban.server.test.it.qp.TestRow;
+import com.akiban.server.types3.mcompat.mtypes.MNumeric;
+import com.akiban.server.types3.mcompat.mtypes.MString;
+import com.akiban.server.types3.texpressions.TPreparedField;
+import com.akiban.util.tap.Tap;
+import com.persistit.Key;
+
+public class KeyFinalCursorIT extends OperatorITBase {
+
+
+    private Schema schema;
+    private ByteArrayOutputStream os;
+    private ByteArrayInputStream is;
+    private SortKey startKey; 
+    private KeyWriter writer;
+    private List<BindableRow> bindRows;
+    private final Random random = new Random (100);
+    
+    @Before
+    public void createFileBuffers() {
+        schema = new Schema(ais());
+        os = new ByteArrayOutputStream();
+        startKey = new SortKey();
+        writer = new KeyWriter(os);
+        bindRows = new ArrayList<>();
+
+    }
+
+    @Test
+    public void cycleComplete() throws IOException {
+        RowType rowType = schema.newValuesType(MNumeric.INT.instance(true));
+        
+        TestRow[] rows = new TestRow[] {
+                row(rowType, 1L),
+        };
+        
+        bindRows.add(BindableRow.of(rows[0], true));
+    
+        RowCursor cursor = cycleRows(rowType);
+        compareRows(rows, cursor);
+    }
+
+    @Test
+    public void cycleNValues() throws IOException {
+        RowType rowType = schema.newValuesType(MNumeric.INT.instance(false),MNumeric.INT.instance(true), MString.varchar());
+        TestRow[] rows = new TestRow[] {
+                row(rowType, 1L, 100L, "A"),
+        };
+        
+        bindRows.add(BindableRow.of(rows[0], true));
+        RowCursor cursor = cycleRows (rowType);
+        compareRows(rows, cursor);
+    }
+    
+    @Test
+    public void cycleNRows() throws IOException {
+        RowType rowType = schema.newValuesType(MNumeric.INT.instance(true));
+        
+        List<TestRow> rows = new ArrayList<>();
+        for (long i = 0; i < 100; i++) {
+            TestRow row = row (rowType, i);
+            rows.add(row);
+            bindRows.add(BindableRow.of(row, true));
+        }
+        RowCursor cursor = cycleRows (rowType);
+        
+        TestRow[] rowArray = new TestRow[rows.size()];
+        compareRows (rows.toArray(rowArray), cursor); 
+    }
+    
+    @Test
+    public void cycleManyRows() throws IOException {
+        RowType rowType = schema.newValuesType(MNumeric.INT.instance(false), MNumeric.INT.instance(true), MString.varchar());
+        List<TestRow> rows = new ArrayList<>();
+        for (long i = 0; i < 100; i++) {
+            TestRow row = row (rowType, random.nextInt(), i, 
+                    characters(5+random.nextInt(1000)));
+            rows.add(row);
+            bindRows.add(BindableRow.of(row, true));
+        }
+        RowCursor cursor = cycleRows(rowType);
+        TestRow[] rowArray = new TestRow[rows.size()];
+        compareRows (rows.toArray(rowArray), cursor);
+        
+    }
+    
+    private RowCursor cycleRows(RowType rowType) throws IOException {
+        KeyReadCursor keyCursor = getKeyCursor(rowType , bindRows);
+
+        startKey  = keyCursor.readNext(); 
+        while (startKey != null) {
+            writer.writeEntry(startKey);
+            startKey = keyCursor.readNext();
+        }
+        
+        is = new ByteArrayInputStream (os.toByteArray());
+        RowCursor cursor = new KeyFinalCursor(is, rowType);
+        return cursor;
+        
+    }
+    
+    private KeyReadCursor getKeyCursor (RowType rowType, List<BindableRow> rows) {
+
+        Operator op = valuesScan_Default(rows, rowType);
+        Cursor cursor = cursor(op, queryContext, queryBindings);
+        API.Ordering ordering = API.ordering();
+        ordering.append(new TPreparedField (rowType.typeInstanceAt(0), 0), true);
+        
+        MergeJoinSorter mergeSorter = new MergeJoinSorter(queryContext, queryBindings, cursor, 
+                rowType, ordering, API.SortOption.PRESERVE_DUPLICATES, Tap.createTimer("Test Tap"));
+        
+        cursor.open();
+        return mergeSorter.readCursor();
+    }
+    static final String ALPHA = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
+    public String characters(final int length) {
+        StringBuilder sb = new StringBuilder(length);
+        for( int i = 0; i < length; i++ ) 
+           sb.append(ALPHA.charAt(random.nextInt(ALPHA.length())));
+        return sb.toString();
+     }
+
+}

=== added file 'src/test/java/com/akiban/qp/persistitadapter/indexcursor/KeyReaderWriterTest.java'
--- src/test/java/com/akiban/qp/persistitadapter/indexcursor/KeyReaderWriterTest.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/qp/persistitadapter/indexcursor/KeyReaderWriterTest.java	2013-07-29 16:50:42 +0000
@@ -0,0 +1,182 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package com.akiban.qp.persistitadapter.indexcursor;
+
+import static org.junit.Assert.*;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter.KeyReader;
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter.KeyWriter;
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter.SortKey;
+import com.persistit.Key;
+import com.persistit.KeyState;
+import com.persistit.Persistit;
+
+public class KeyReaderWriterTest {
+
+    private ByteArrayOutputStream os;
+    private ByteArrayInputStream is;
+    private SortKey startKey;
+    private Key testKey;
+    private KeyWriter writer;
+    
+    @Before
+    public void createFileBuffers() {
+        os = new ByteArrayOutputStream();
+        startKey = new SortKey();
+        testKey = new Key ((Persistit)null);
+        writer = new KeyWriter(os);
+    }
+    @Test
+    public void cycleSimple() throws IOException {
+        testKey.append(1);
+        startKey.sortKeys.add(new KeyState(testKey));
+        startKey.rowKey.append(1);
+        writer.writeEntry(startKey);
+        verifyInput();
+    }
+    
+    @Test
+    public void cycleString() throws IOException {
+        testKey.append("abcd");
+        startKey.sortKeys.add(new KeyState(testKey));
+        startKey.rowKey.append("abcd");
+        writer.writeEntry(startKey);
+        verifyInput();
+    }
+
+    @Test
+    public void cycleNIntegers() throws IOException {
+        for (int i = 0; i < 400; i++) {
+            testKey.append(i);
+            startKey.rowKey.append(i);
+        }
+        startKey.sortKeys.add(new KeyState(testKey));
+        writer.writeEntry(startKey);
+        verifyInput();
+    }
+    
+    @Test
+    public void cycle2Keys() throws IOException {
+        
+        testKey.append(1);
+        startKey.sortKeys.add(new KeyState(testKey));
+        startKey.rowKey.append(1);
+        writer.writeEntry(startKey);
+        writer.writeEntry(startKey);
+        
+        is = new ByteArrayInputStream (os.toByteArray());
+        KeyReader reader = new KeyReader (is);
+
+        SortKey endKey = reader.readNext();
+        assertTrue (startKey.rowKey.compareTo(endKey.rowKey) == 0);
+        assertTrue (startKey.sortKeys.get(0).compareTo(endKey.sortKeys.get(0)) == 0);
+        endKey = reader.readNext();
+        assertTrue (startKey.rowKey.compareTo(endKey.rowKey) == 0);
+        assertTrue (startKey.sortKeys.get(0).compareTo(endKey.sortKeys.get(0)) == 0);
+        endKey = reader.readNext();
+        assertNull (endKey);
+    }
+    
+    @Test
+    public void cycleNKeys() throws IOException{
+        List<SortKey> keys = new ArrayList<>(100);
+        for (int i = 0; i < 100; i++) {
+            SortKey newKey = new SortKey();
+            newKey.rowKey.append(i);
+            testKey.clear();
+            testKey.append(i);
+            newKey.sortKeys.add(new KeyState (testKey));
+            keys.add(newKey);
+        }
+        verifyNKeys (keys);
+    }
+    
+    @Test
+    public void cycleNStrings() throws IOException {
+        List<SortKey> keys = new ArrayList<>(100);
+        for (int i = 0; i < 100; i++) {
+            SortKey newKey = new SortKey();
+            String value = characters(5+random.nextInt(1000));
+            newKey.rowKey.append(value);
+            testKey.clear();
+            testKey.append(value);
+            newKey.sortKeys.add(new KeyState (testKey));
+            keys.add(newKey);
+        }
+        verifyNKeys (keys);
+    }
+    
+    @Test
+    public void cycleNMultiKeys () throws IOException {
+        List<SortKey> keys = new ArrayList<>(100);
+        for (int i = 0; i < 100; i++) {
+            SortKey newKey = new SortKey();
+            newKey.rowKey.append(random.nextInt());
+            newKey.rowKey.append(null);
+            newKey.rowKey.append(characters(3+random.nextInt(25)));
+            newKey.rowKey.append(characters(3+random.nextInt(25)));
+            testKey.clear();
+            testKey.append(i);
+            newKey.sortKeys.add(new KeyState (testKey));
+            keys.add(newKey);
+        }
+        verifyNKeys(keys);
+    }
+    
+    private void verifyInput() throws IOException {
+        is = new ByteArrayInputStream (os.toByteArray());
+        KeyReader reader = new KeyReader (is);
+        SortKey endKey = reader.readNext();
+        assertTrue (startKey.rowKey.compareTo(endKey.rowKey) == 0);
+        assertTrue (startKey.sortKeys.get(0).compareTo(endKey.sortKeys.get(0)) == 0);
+        
+    }
+    
+    private void verifyNKeys(List<SortKey> keys) throws IOException  {
+        for (SortKey key : keys) {
+            writer.writeEntry(key);
+        }
+        is = new ByteArrayInputStream (os.toByteArray());
+        KeyReader reader = new KeyReader (is);
+
+        SortKey endKey;
+        for (SortKey startKey : keys) {
+            endKey = reader.readNext();
+            assertTrue (startKey.rowKey.compareTo(endKey.rowKey) == 0);
+            assertTrue (startKey.sortKeys.get(0).compareTo(endKey.sortKeys.get(0)) == 0);
+        }
+    }
+
+    static final String ALPHA = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
+    final Random random = new Random(100);
+    public String characters(final int length) {
+        StringBuilder sb = new StringBuilder(length);
+        for( int i = 0; i < length; i++ ) 
+           sb.append(ALPHA.charAt(random.nextInt(ALPHA.length())));
+        return sb.toString();
+     }
+}

=== modified file 'src/test/java/com/akiban/server/test/it/sort/MemorySorterIT.java'
--- src/test/java/com/akiban/server/test/it/sort/MemorySorterIT.java	2013-07-08 18:16:50 +0000
+++ src/test/java/com/akiban/server/test/it/sort/MemorySorterIT.java	2013-07-29 16:50:42 +0000
@@ -23,7 +23,6 @@
 import com.akiban.qp.operator.QueryContext;
 import com.akiban.qp.persistitadapter.Sorter;
 import com.akiban.qp.persistitadapter.indexcursor.MemorySorter;
-import com.akiban.qp.persistitadapter.indexcursor.PersistitSorter;
 import com.akiban.qp.rowtype.RowType;
 import com.akiban.util.tap.InOutTap;
 

=== added file 'src/test/java/com/akiban/server/test/it/sort/MergeJoinSorterIT.java'
--- src/test/java/com/akiban/server/test/it/sort/MergeJoinSorterIT.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/server/test/it/sort/MergeJoinSorterIT.java	2013-07-29 16:50:42 +0000
@@ -0,0 +1,49 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package com.akiban.server.test.it.sort;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.akiban.qp.operator.API.Ordering;
+import com.akiban.qp.operator.API.SortOption;
+import com.akiban.qp.operator.Cursor;
+import com.akiban.qp.operator.QueryBindings;
+import com.akiban.qp.operator.QueryContext;
+import com.akiban.qp.persistitadapter.Sorter;
+import com.akiban.qp.persistitadapter.indexcursor.MergeJoinSorter;
+import com.akiban.qp.rowtype.RowType;
+import com.akiban.util.tap.InOutTap;
+
+public class MergeJoinSorterIT extends SorterITBase {
+
+    @Override
+    public Map<String,String> startupConfigProperties() {
+        Map<String,String> props = new HashMap<>();
+        props.putAll(super.startupConfigProperties());
+        props.put("akserver.tmp_dir","/tmp/akserver-junit/");
+        return props;
+    }
+
+    @Override
+    public Sorter createSorter(QueryContext context, QueryBindings bindings,
+            Cursor input, RowType rowType, Ordering ordering,
+            SortOption sortOption, InOutTap loadTap) {
+        return new MergeJoinSorter(context, bindings, input, rowType, ordering, sortOption, loadTap);
+    }
+
+}