| /* |
| * 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; |
| } |
| }; |
| } |
| |
| } |
| } |
