/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.uci.ics.hyracks.dataflow.std.group;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.logging.Logger;

import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksStageletContext;
import edu.uci.ics.hyracks.api.dataflow.IActivityGraphBuilder;
import edu.uci.ics.hyracks.api.dataflow.IOperatorDescriptor;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ITuplePartitionComputerFactory;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.io.FileReference;
import edu.uci.ics.hyracks.api.job.IOperatorEnvironment;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.io.RunFileReader;
import edu.uci.ics.hyracks.dataflow.common.io.RunFileWriter;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractActivityNode;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.util.ReferenceEntry;
import edu.uci.ics.hyracks.dataflow.std.util.ReferencedPriorityQueue;

/**
 * This is an implementation of the external hash group operator.
 * The motivation of this operator is that when tuples are processed in
 * parallel, distinguished aggregating keys partitioned on one node may exceed
 * the main memory, so aggregation results should be output onto the disk to
 * make space for aggregating more input tuples.
 */
public class ExternalHashGroupOperatorDescriptor extends AbstractOperatorDescriptor {

    /**
     * The input frame identifier (in the job environment)
     */
    private static final String GROUPTABLES = "gtables";

    /**
     * The runs files identifier (in the job environment)
     */
    private static final String RUNS = "runs";

    /**
     * The fields used for grouping (grouping keys).
     */
    private final int[] keyFields;

    /**
     * The comparator for checking the grouping conditions, corresponding to the {@link #keyFields}.
     */
    private final IBinaryComparatorFactory[] comparatorFactories;

    /**
     * The aggregator factory for the aggregating field, corresponding to the {@link #aggregateFields}.
     */
    private IAccumulatingAggregatorFactory aggregatorFactory;

    /**
     * The maximum number of frames in the main memory.
     */
    private final int framesLimit;

    /**
     * Indicate whether the final output will be sorted or not.
     */
    private final boolean sortOutput;

    /**
     * Partition computer factory
     */
    private final ITuplePartitionComputerFactory tpcf;

    /**
     * The size of the in-memory table, which should be specified now by the
     * creator of this operator descriptor.
     */
    private final int tableSize;

    /**
     * XXX Logger for debug information
     */
    private static Logger LOGGER = Logger.getLogger(ExternalHashGroupOperatorDescriptor.class.getName());

    /**
     * Constructor of the external hash group operator descriptor.
     * 
     * @param spec
     * @param keyFields
     *            The fields as keys of grouping.
     * @param framesLimit
     *            The maximum number of frames to be used in memory.
     * @param sortOutput
     *            Whether the output should be sorted or not. Note that if the
     *            input data is large enough for external grouping, the output
     *            will be sorted surely. The only case that when the output is
     *            not sorted is when the size of the input data can be grouped
     *            in memory and this parameter is false.
     * @param tpcf
     *            The partitioner.
     * @param comparatorFactories
     *            The comparators.
     * @param aggregatorFactory
     *            The aggregators.
     * @param recordDescriptor
     *            The record descriptor for the input data.
     * @param tableSize
     *            The maximum size of the in memory table usable to this
     *            operator.
     */
    public ExternalHashGroupOperatorDescriptor(JobSpecification spec, int[] keyFields, int framesLimit,
            boolean sortOutput, ITuplePartitionComputerFactory tpcf, IBinaryComparatorFactory[] comparatorFactories,
            IAccumulatingAggregatorFactory aggregatorFactory, RecordDescriptor recordDescriptor, int tableSize) {
        super(spec, 1, 1);
        this.framesLimit = framesLimit;
        if (framesLimit <= 1) {
            // Minimum of 2 frames: 1 for input records, and 1 for output
            // aggregation results.
            throw new IllegalStateException();
        }
        this.aggregatorFactory = aggregatorFactory;
        this.keyFields = keyFields;
        this.comparatorFactories = comparatorFactories;

        this.sortOutput = sortOutput;

        this.tpcf = tpcf;

        this.tableSize = tableSize;

        // Set the record descriptor. Note that since this operator is a unary
        // operator,
        // only the first record descritpor is used here.
        recordDescriptors[0] = recordDescriptor;
    }

    /**
     * 
     */
    private static final long serialVersionUID = 1L;

    /*
     * (non-Javadoc)
     * 
     * @see
     * edu.uci.ics.hyracks.api.dataflow.IOperatorDescriptor#contributeTaskGraph
     * (edu.uci.ics.hyracks.api.dataflow.IActivityGraphBuilder)
     */
    @Override
    public void contributeTaskGraph(IActivityGraphBuilder builder) {
        PartialAggregateActivity partialAggAct = new PartialAggregateActivity();
        MergeActivity mergeAct = new MergeActivity();

        builder.addTask(partialAggAct);
        builder.addSourceEdge(0, partialAggAct, 0);

        builder.addTask(mergeAct);
        builder.addTargetEdge(0, mergeAct, 0);

        // FIXME Block or not?
        builder.addBlockingEdge(partialAggAct, mergeAct);

    }

    private class PartialAggregateActivity extends AbstractActivityNode {

        /**
         * 
         */
        private static final long serialVersionUID = 1L;

        @Override
        public IOperatorNodePushable createPushRuntime(final IHyracksStageletContext ctx,
                final IOperatorEnvironment env, final IRecordDescriptorProvider recordDescProvider, int partition,
                int nPartitions) {
            // Create the in-memory hash table
            final SpillableGroupingHashTable gTable = new SpillableGroupingHashTable(ctx, keyFields,
                    comparatorFactories, tpcf, aggregatorFactory, recordDescProvider.getInputRecordDescriptor(
                            getOperatorId(), 0), recordDescriptors[0],
                            // Always take one frame for the input records
                            framesLimit - 1, tableSize);
            // Create the tuple accessor
            final FrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(),
                    recordDescProvider.getInputRecordDescriptor(getOperatorId(), 0));
            // Create the partial aggregate activity node
            IOperatorNodePushable op = new AbstractUnaryInputSinkOperatorNodePushable() {

                /**
                 * Run files
                 */
                private LinkedList<RunFileReader> runs;

                @Override
                public void close() throws HyracksDataException {
                    if (gTable.getFrameCount() >= 0) {
                        if (runs.size() <= 0) {
                            // All in memory
                            env.set(GROUPTABLES, gTable);
                        } else {
                            // flush the memory into the run file.
                            flushFramesToRun();
                        }
                    }
                    env.set(RUNS, runs);
                }

                @Override
                public void flush() throws HyracksDataException {

                }

                /**
                 * Process the next input buffer.
                 * The actual insertion is processed in {@link #gTable}. It will
                 * check whether it is possible to contain the data into the
                 * main memory or not. If not, it will indicate the operator to
                 * flush the content of the table into a run file.
                 */
                @Override
                public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
                    accessor.reset(buffer);
                    int tupleCount = accessor.getTupleCount();
                    for (int i = 0; i < tupleCount; i++) {
                        // If the group table is too large, flush the table into
                        // a run file.
                        if (!gTable.insert(accessor, i)) {
                            flushFramesToRun();
                            if (!gTable.insert(accessor, i))
                                throw new HyracksDataException(
                                "Failed to insert a new buffer into the aggregate operator!");
                        }
                    }

                }

                @Override
                public void open() throws HyracksDataException {
                    runs = new LinkedList<RunFileReader>();
                    gTable.reset();
                }

                /**
                 * Flush the content of the group table into a run file.
                 * During the flushing, the hash table will be sorted as first.
                 * After that, a run file handler is initialized and the hash
                 * table is flushed into the run file.
                 * 
                 * @throws HyracksDataException
                 */
                private void flushFramesToRun() throws HyracksDataException {
                    // Sort the contents of the hash table.
                    gTable.sortFrames();
                    FileReference runFile;
                    try {
                        runFile = ctx.getJobletContext().createWorkspaceFile(
                                ExternalHashGroupOperatorDescriptor.class.getSimpleName());
                    } catch (IOException e) {
                        throw new HyracksDataException(e);
                    }
                    RunFileWriter writer = new RunFileWriter(runFile, ctx.getIOManager());
                    writer.open();
                    try {
                        gTable.flushFrames(writer, true);
                    } catch (Exception ex) {
                        throw new HyracksDataException(ex);
                    } finally {
                        writer.close();
                    }
                    gTable.reset();
                    runs.add(((RunFileWriter) writer).createReader());
                    LOGGER.warning("Created run file: " + runFile.getFile().getAbsolutePath());
                }

            };

            return op;
        }

        @Override
        public IOperatorDescriptor getOwner() {
            return ExternalHashGroupOperatorDescriptor.this;
        }

    }

    private class MergeActivity extends AbstractActivityNode {

        /**
         * 
         */
        private static final long serialVersionUID = 1L;

        @Override
        public IOperatorNodePushable createPushRuntime(final IHyracksStageletContext ctx,
                final IOperatorEnvironment env, IRecordDescriptorProvider recordDescProvider, int partition,
                int nPartitions) {
            final IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
            for (int i = 0; i < comparatorFactories.length; ++i) {
                comparators[i] = comparatorFactories[i].createBinaryComparator();
            }
            IOperatorNodePushable op = new AbstractUnaryOutputSourceOperatorNodePushable() {
                /**
                 * Input frames, one for each run file.
                 */
                private List<ByteBuffer> inFrames;

                /**
                 * Output frame.
                 */
                private ByteBuffer outFrame;

                /**
                 * List of the run files to be merged
                 */
                LinkedList<RunFileReader> runs;

                /**
                 * Tuple appender for the output frame {@link #outFrame}.
                 */
                private FrameTupleAppender outFrameAppender;

                private ISpillableAccumulatingAggregator visitingAggregator;
                private ArrayTupleBuilder visitingKeyTuple;

                @SuppressWarnings("unchecked")
                @Override
                public void initialize() throws HyracksDataException {
                    runs = (LinkedList<RunFileReader>) env.get(RUNS);
                    writer.open();

                    try {
                        if (runs.size() <= 0) {
                            // If the aggregate results can be fit into
                            // memory...
                            SpillableGroupingHashTable gTable = (SpillableGroupingHashTable) env.get(GROUPTABLES);
                            if (gTable != null) {
                                gTable.flushFrames(writer, sortOutput);
                            }
                            env.set(GROUPTABLES, null);
                        } else {
                            // Otherwise, merge the run files into a single file
                            inFrames = new ArrayList<ByteBuffer>();
                            outFrame = ctx.allocateFrame();
                            outFrameAppender = new FrameTupleAppender(ctx.getFrameSize());
                            outFrameAppender.reset(outFrame, true);
                            for (int i = 0; i < framesLimit - 1; ++i) {
                                inFrames.add(ctx.allocateFrame());
                            }
                            int passCount = 0;
                            while (runs.size() > 0) {
                                passCount++;
                                try {
                                    doPass(runs, passCount);
                                } catch (Exception e) {
                                    throw new HyracksDataException(e);
                                }
                            }
                        }

                    } finally {
                        writer.close();
                    }
                    env.set(RUNS, null);
                }

                /**
                 * Merge the run files once.
                 * 
                 * @param runs
                 * @param passCount
                 * @throws HyracksDataException
                 * @throws IOException
                 */
                private void doPass(LinkedList<RunFileReader> runs, int passCount) throws HyracksDataException,
                IOException {
                    FileReference newRun = null;
                    IFrameWriter writer = this.writer;
                    boolean finalPass = false;

                    int[] storedKeys = new int[keyFields.length];
                    // Get the list of the fields in the stored records.
                    for (int i = 0; i < keyFields.length; ++i) {
                        storedKeys[i] = i;
                    }

                    // Release the space not used
                    if (runs.size() + 1 <= framesLimit) {
                        // If there are run files no more than the available
                        // frame slots...
                        // No run file to be generated, since the result can be
                        // directly
                        // outputted into the output frame for write.
                        finalPass = true;
                        for (int i = inFrames.size() - 1; i >= runs.size(); i--) {
                            inFrames.remove(i);
                        }
                    } else {
                        // Otherwise, a new run file will be created
                        newRun = ctx.getJobletContext().createWorkspaceFile(
                                ExternalHashGroupOperatorDescriptor.class.getSimpleName());
                        writer = new RunFileWriter(newRun, ctx.getIOManager());
                        writer.open();
                    }
                    try {
                        // Create run file read handler for each input frame
                        RunFileReader[] runFileReaders = new RunFileReader[inFrames.size()];
                        // Create input frame accessor
                        FrameTupleAccessor[] tupleAccessors = new FrameTupleAccessor[inFrames.size()];
                        Comparator<ReferenceEntry> comparator = createEntryComparator(comparators);
                        ReferencedPriorityQueue topTuples = new ReferencedPriorityQueue(ctx.getFrameSize(),
                                recordDescriptors[0], inFrames.size(), comparator);
                        // For the index of tuples visited in each frame.
                        int[] tupleIndexes = new int[inFrames.size()];
                        for (int i = 0; i < inFrames.size(); i++) {
                            tupleIndexes[i] = 0;
                            int runIndex = topTuples.peek().getRunid();
                            runFileReaders[runIndex] = runs.get(runIndex);
                            runFileReaders[runIndex].open();
                            // Load the first frame of the file into the main
                            // memory
                            if (runFileReaders[runIndex].nextFrame(inFrames.get(runIndex))) {
                                // initialize the tuple accessor for the frame
                                tupleAccessors[runIndex] = new FrameTupleAccessor(ctx.getFrameSize(),
                                        recordDescriptors[0]);
                                tupleAccessors[runIndex].reset(inFrames.get(runIndex));
                                setNextTopTuple(runIndex, tupleIndexes, runFileReaders, tupleAccessors, topTuples);
                            } else {
                                closeRun(runIndex, runFileReaders, tupleAccessors);
                            }
                        }
                        // Merge
                        // Get a key holder for the current working
                        // aggregator keys
                        visitingAggregator = null;
                        visitingKeyTuple = null;
                        // Loop on all run files, and update the key
                        // holder.
                        while (!topTuples.areRunsExhausted()) {
                            // Get the top record
                            ReferenceEntry top = topTuples.peek();
                            int tupleIndex = top.getTupleIndex();
                            int runIndex = topTuples.peek().getRunid();
                            FrameTupleAccessor fta = top.getAccessor();
                            if (visitingAggregator == null) {
                                // Initialize the aggregator
                                visitingAggregator = aggregatorFactory.createSpillableAggregator(ctx,
                                        recordDescriptors[0], recordDescriptors[0]);
                                // Initialize the partial aggregation result
                                visitingAggregator.initFromPartial(fta, tupleIndex, keyFields);
                                visitingKeyTuple = new ArrayTupleBuilder(recordDescriptors[0].getFields().length);
                                for (int i = 0; i < keyFields.length; i++) {
                                    visitingKeyTuple.addField(fta, tupleIndex, keyFields[i]);
                                }
                            } else {
                                if (compareTupleWithFrame(visitingKeyTuple, fta, tupleIndex, storedKeys, keyFields,
                                        comparators) == 0) {
                                    // If the two partial results are on the
                                    // same key
                                    visitingAggregator.accumulatePartialResult(fta, tupleIndex, keyFields);
                                } else {
                                    // Otherwise, write the partial result back
                                    // to the output frame
                                    if (!visitingAggregator.output(outFrameAppender, visitingKeyTuple)) {
                                        FrameUtils.flushFrame(outFrame, writer);
                                        outFrameAppender.reset(outFrame, true);
                                        if (!visitingAggregator.output(outFrameAppender, visitingKeyTuple)) {
                                            throw new IllegalStateException();
                                        }
                                    }
                                    // Reset the partial aggregation result
                                    visitingAggregator.initFromPartial(fta, tupleIndex, keyFields);
                                    visitingKeyTuple.reset();
                                    for (int i = 0; i < keyFields.length; i++) {
                                        visitingKeyTuple.addField(fta, tupleIndex, keyFields[i]);
                                    }
                                }
                            }
                            tupleIndexes[runIndex]++;
                            setNextTopTuple(runIndex, tupleIndexes, runFileReaders, tupleAccessors, topTuples);
                        }
                        // Output the last aggregation result in the frame
                        if (visitingAggregator != null) {
                            if (!visitingAggregator.output(outFrameAppender, visitingKeyTuple)) {
                                FrameUtils.flushFrame(outFrame, writer);
                                outFrameAppender.reset(outFrame, true);
                                if (!visitingAggregator.output(outFrameAppender, visitingKeyTuple)) {
                                    throw new IllegalStateException();
                                }
                            }
                        }
                        // Output data into run file writer after all tuples
                        // have been checked
                        if (outFrameAppender.getTupleCount() > 0) {
                            FrameUtils.flushFrame(outFrame, writer);
                            outFrameAppender.reset(outFrame, true);
                        }
                        // empty the input frames
                        runs.subList(0, inFrames.size()).clear();
                        // insert the new run file into the beginning of the run
                        // file list
                        if (!finalPass) {
                            runs.add(0, ((RunFileWriter) writer).createReader());
                        }
                    } catch (Exception ex) {
                        throw new HyracksDataException(ex);
                    } finally {
                        if (!finalPass) {
                            writer.close();
                        }
                    }
                }

                /**
                 * Insert the tuple into the priority queue.
                 * 
                 * @param runIndex
                 * @param tupleIndexes
                 * @param runCursors
                 * @param tupleAccessors
                 * @param topTuples
                 * @throws IOException
                 */
                private void setNextTopTuple(int runIndex, int[] tupleIndexes, RunFileReader[] runCursors,
                        FrameTupleAccessor[] tupleAccessors, ReferencedPriorityQueue topTuples) throws IOException {
                    boolean exists = hasNextTuple(runIndex, tupleIndexes, runCursors, tupleAccessors);
                    if (exists) {
                        topTuples.popAndReplace(tupleAccessors[runIndex], tupleIndexes[runIndex]);
                    } else {
                        topTuples.pop();
                        closeRun(runIndex, runCursors, tupleAccessors);
                    }
                }

                /**
                 * Check whether there are any more tuples to be checked for the
                 * given run file from the corresponding input frame.
                 * If the input frame for this run file is exhausted, load a new
                 * frame of the run file into the input frame.
                 * 
                 * @param runIndex
                 * @param tupleIndexes
                 * @param runCursors
                 * @param tupleAccessors
                 * @return
                 * @throws IOException
                 */
                private boolean hasNextTuple(int runIndex, int[] tupleIndexes, RunFileReader[] runCursors,
                        FrameTupleAccessor[] tupleAccessors) throws IOException {

                    if (tupleAccessors[runIndex] == null || runCursors[runIndex] == null) {
                        /*
                         * Return false if the targeting run file is not
                         * available, or the frame for the run file is not
                         * available.
                         */
                        return false;
                    } else if (tupleIndexes[runIndex] >= tupleAccessors[runIndex].getTupleCount()) {
                        /*
                         * If all tuples in the targeting frame have been
                         * checked.
                         */
                        ByteBuffer buf = tupleAccessors[runIndex].getBuffer(); // same-as-inFrames.get(runIndex)
                        // Refill the buffer with contents from the run file.
                        if (runCursors[runIndex].nextFrame(buf)) {
                            tupleIndexes[runIndex] = 0;
                            return hasNextTuple(runIndex, tupleIndexes, runCursors, tupleAccessors);
                        } else {
                            return false;
                        }
                    } else {
                        return true;
                    }
                }

                /**
                 * Close the run file, and also the corresponding readers and
                 * input frame.
                 * 
                 * @param index
                 * @param runCursors
                 * @param tupleAccessor
                 * @throws HyracksDataException
                 */
                private void closeRun(int index, RunFileReader[] runCursors, IFrameTupleAccessor[] tupleAccessor)
                throws HyracksDataException {
                    runCursors[index].close();
                    runCursors[index] = null;
                    tupleAccessor[index] = null;
                }

                /**
                 * Compare a tuple (in the format of a {@link ArrayTupleBuilder} ) with a record in a frame (in the format of a {@link FrameTupleAccessor}). Comparing keys and comparators
                 * are specified for this method as inputs.
                 * 
                 * @param tuple0
                 * @param accessor1
                 * @param tIndex1
                 * @param keys0
                 * @param keys1
                 * @param comparators
                 * @return
                 */
                private int compareTupleWithFrame(ArrayTupleBuilder tuple0, FrameTupleAccessor accessor1, int tIndex1,
                        int[] keys0, int[] keys1, IBinaryComparator[] comparators) {
                    int tStart1 = accessor1.getTupleStartOffset(tIndex1);
                    int fStartOffset1 = accessor1.getFieldSlotsLength() + tStart1;

                    for (int i = 0; i < keys0.length; ++i) {
                        int fIdx0 = keys0[i];
                        int fStart0 = (i == 0 ? 0 : tuple0.getFieldEndOffsets()[fIdx0 - 1]);
                        int fEnd0 = tuple0.getFieldEndOffsets()[fIdx0];
                        int fLen0 = fEnd0 - fStart0;

                        int fIdx1 = keys1[i];
                        int fStart1 = accessor1.getFieldStartOffset(tIndex1, fIdx1);
                        int fEnd1 = accessor1.getFieldEndOffset(tIndex1, fIdx1);
                        int fLen1 = fEnd1 - fStart1;

                        int c = comparators[i].compare(tuple0.getByteArray(), fStart0, fLen0, accessor1.getBuffer()
                                .array(), fStart1 + fStartOffset1, fLen1);
                        if (c != 0) {
                            return c;
                        }
                    }
                    return 0;
                }
            };
            return op;
        }

        @Override
        public IOperatorDescriptor getOwner() {
            return ExternalHashGroupOperatorDescriptor.this;
        }

        private Comparator<ReferenceEntry> createEntryComparator(final IBinaryComparator[] comparators) {
            return new Comparator<ReferenceEntry>() {
                public int compare(ReferenceEntry tp1, ReferenceEntry tp2) {
                    FrameTupleAccessor fta1 = (FrameTupleAccessor) tp1.getAccessor();
                    FrameTupleAccessor fta2 = (FrameTupleAccessor) tp2.getAccessor();
                    int j1 = (Integer) tp1.getTupleIndex();
                    int j2 = (Integer) tp2.getTupleIndex();
                    byte[] b1 = fta1.getBuffer().array();
                    byte[] b2 = fta2.getBuffer().array();
                    for (int f = 0; f < keyFields.length; ++f) {
                        int fIdx = keyFields[f];
                        int s1 = fta1.getTupleStartOffset(j1) + fta1.getFieldSlotsLength()
                        + fta1.getFieldStartOffset(j1, fIdx);
                        int l1 = fta1.getFieldEndOffset(j1, fIdx) - fta1.getFieldStartOffset(j1, fIdx);
                        int s2 = fta2.getTupleStartOffset(j2) + fta2.getFieldSlotsLength()
                        + fta2.getFieldStartOffset(j2, fIdx);
                        int l2 = fta2.getFieldEndOffset(j2, fIdx) - fta2.getFieldStartOffset(j2, fIdx);
                        int c = comparators[f].compare(b1, s1, l1, b2, s2, l2);
                        if (c != 0) {
                            return c;
                        }
                    }
                    return 0;
                }
            };
        }

    }
}