You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by ji...@apache.org on 2016/02/26 06:54:03 UTC
[06/11] incubator-asterixdb-hyracks git commit: Implemented the
memory-bounded HashGroupby and HashJoin for BigObject
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractExternalSortRunMerger.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractExternalSortRunMerger.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractExternalSortRunMerger.java
new file mode 100644
index 0000000..24154f7
--- /dev/null
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractExternalSortRunMerger.java
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 at
+ *
+ * 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 org.apache.hyracks.dataflow.std.sort;
+
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.hyracks.api.comm.FrameConstants;
+import org.apache.hyracks.api.comm.IFrameWriter;
+import org.apache.hyracks.api.comm.VSizeFrame;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputer;
+import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
+import org.apache.hyracks.dataflow.common.io.RunFileWriter;
+import org.apache.hyracks.dataflow.std.sort.util.GroupVSizeFrame;
+
+public abstract class AbstractExternalSortRunMerger {
+
+ protected final IHyracksTaskContext ctx;
+ protected final IFrameWriter writer;
+
+ private final List<GeneratedRunFileReader> runs;
+ private final BitSet currentGenerationRunAvailable;
+ private final IBinaryComparator[] comparators;
+ private final INormalizedKeyComputer nmkComputer;
+ private final RecordDescriptor recordDesc;
+ private final int framesLimit;
+ private final int MAX_FRAME_SIZE;
+ private final int topK;
+ private List<GroupVSizeFrame> inFrames;
+ private VSizeFrame outputFrame;
+ private ISorter sorter;
+
+ private static final Logger LOGGER = Logger.getLogger(AbstractExternalSortRunMerger.class.getName());
+
+ public AbstractExternalSortRunMerger(IHyracksTaskContext ctx, ISorter sorter, List<GeneratedRunFileReader> runs,
+ IBinaryComparator[] comparators, INormalizedKeyComputer nmkComputer, RecordDescriptor recordDesc,
+ int framesLimit, IFrameWriter writer) {
+ this(ctx, sorter, runs, comparators, nmkComputer, recordDesc, framesLimit, Integer.MAX_VALUE, writer);
+ }
+
+ public AbstractExternalSortRunMerger(IHyracksTaskContext ctx, ISorter sorter, List<GeneratedRunFileReader> runs,
+ IBinaryComparator[] comparators, INormalizedKeyComputer nmkComputer, RecordDescriptor recordDesc,
+ int framesLimit, int topK, IFrameWriter writer) {
+ this.ctx = ctx;
+ this.sorter = sorter;
+ this.runs = new LinkedList<>(runs);
+ this.currentGenerationRunAvailable = new BitSet(runs.size());
+ this.comparators = comparators;
+ this.nmkComputer = nmkComputer;
+ this.recordDesc = recordDesc;
+ this.framesLimit = framesLimit;
+ this.writer = writer;
+ this.MAX_FRAME_SIZE = FrameConstants.MAX_NUM_MINFRAME * ctx.getInitialFrameSize();
+ this.topK = topK;
+ }
+
+ public void process() throws HyracksDataException {
+ IFrameWriter finalWriter = null;
+ try {
+ if (runs.size() <= 0) {
+ finalWriter = prepareSkipMergingFinalResultWriter(writer);
+ finalWriter.open();
+ if (sorter != null) {
+ if (sorter.hasRemaining()) {
+ sorter.flush(finalWriter);
+ }
+ sorter.close();
+ }
+ } else {
+ /** recycle sort buffer */
+ if (sorter != null) {
+ sorter.close();
+ }
+
+ finalWriter = prepareFinalMergeResultWriter(writer);
+ finalWriter.open();
+
+ int maxMergeWidth = framesLimit - 1;
+
+ inFrames = new ArrayList<>(maxMergeWidth);
+ outputFrame = new VSizeFrame(ctx);
+ List<GeneratedRunFileReader> partialRuns = new ArrayList<>(maxMergeWidth);
+
+ int stop = runs.size();
+ currentGenerationRunAvailable.set(0, stop);
+
+ while (true) {
+
+ int unUsed = selectPartialRuns(maxMergeWidth * ctx.getInitialFrameSize(), runs, partialRuns,
+ currentGenerationRunAvailable, stop);
+ prepareFrames(unUsed, inFrames, partialRuns);
+
+ if (!currentGenerationRunAvailable.isEmpty() || stop < runs.size()) {
+ GeneratedRunFileReader reader;
+ if (partialRuns.size() == 1) {
+ if (!currentGenerationRunAvailable.isEmpty()) {
+ throw new HyracksDataException(
+ "The record is too big to put into the merging frame, please"
+ + " allocate more sorting memory");
+ } else {
+ reader = partialRuns.get(0);
+ }
+
+ } else {
+ RunFileWriter mergeFileWriter = prepareIntermediateMergeRunFile();
+ IFrameWriter mergeResultWriter = prepareIntermediateMergeResultWriter(mergeFileWriter);
+
+ mergeResultWriter.open();
+ merge(mergeResultWriter, partialRuns);
+ mergeResultWriter.close();
+
+ reader = mergeFileWriter.createReader();
+ }
+ runs.add(reader);
+
+ if (currentGenerationRunAvailable.isEmpty()) {
+
+ if (LOGGER.isLoggable(Level.FINE)) {
+ LOGGER.fine("generated runs:" + stop);
+ }
+ runs.subList(0, stop).clear();
+ currentGenerationRunAvailable.clear();
+ currentGenerationRunAvailable.set(0, runs.size());
+ stop = runs.size();
+ }
+ } else {
+ if (LOGGER.isLoggable(Level.FINE)) {
+ LOGGER.fine("final runs:" + stop);
+ }
+ merge(finalWriter, partialRuns);
+ break;
+ }
+ }
+ }
+ } catch (Exception e) {
+ finalWriter.fail();
+ throw new HyracksDataException(e);
+ } finally {
+ finalWriter.close();
+ }
+ }
+
+ private static int selectPartialRuns(int budget, List<GeneratedRunFileReader> runs,
+ List<GeneratedRunFileReader> partialRuns, BitSet runAvailable, int stop) {
+ partialRuns.clear();
+ int maxFrameSizeOfGenRun = 0;
+ int nextRunId = runAvailable.nextSetBit(0);
+ while (budget > 0 && nextRunId >= 0 && nextRunId < stop) {
+ int runFrameSize = runs.get(nextRunId).getMaxFrameSize();
+ if (budget - runFrameSize >= 0) {
+ partialRuns.add(runs.get(nextRunId));
+ budget -= runFrameSize;
+ runAvailable.clear(nextRunId);
+ maxFrameSizeOfGenRun = Math.max(runFrameSize, maxFrameSizeOfGenRun);
+ }
+ nextRunId = runAvailable.nextSetBit(nextRunId + 1);
+ }
+ return budget;
+ }
+
+ private void prepareFrames(int extraFreeMem, List<GroupVSizeFrame> inFrames,
+ List<GeneratedRunFileReader> partialRuns) throws HyracksDataException {
+ if (extraFreeMem > 0 && partialRuns.size() > 1) {
+ int extraFrames = extraFreeMem / ctx.getInitialFrameSize();
+ int avg = (extraFrames / partialRuns.size()) * ctx.getInitialFrameSize();
+ int residue = (extraFrames % partialRuns.size());
+ for (int i = 0; i < residue; i++) {
+ partialRuns.get(i).updateSize(Math.min(MAX_FRAME_SIZE,
+ partialRuns.get(i).getMaxFrameSize() + avg + ctx.getInitialFrameSize()));
+ }
+ for (int i = residue; i < partialRuns.size() && avg > 0; i++) {
+ partialRuns.get(i).updateSize(Math.min(MAX_FRAME_SIZE, partialRuns.get(i).getMaxFrameSize() + avg));
+ }
+ }
+
+ if (inFrames.size() > partialRuns.size()) {
+ inFrames.subList(partialRuns.size(), inFrames.size()).clear();
+ }
+ int i;
+ for (i = 0; i < inFrames.size(); i++) {
+ inFrames.get(i).resize(partialRuns.get(i).getMaxFrameSize());
+ }
+ for (; i < partialRuns.size(); i++) {
+ inFrames.add(new GroupVSizeFrame(ctx, partialRuns.get(i).getMaxFrameSize()));
+ }
+ }
+
+ abstract protected IFrameWriter prepareSkipMergingFinalResultWriter(IFrameWriter nextWriter)
+ throws HyracksDataException;
+
+ abstract protected RunFileWriter prepareIntermediateMergeRunFile() throws HyracksDataException;
+
+ abstract protected IFrameWriter prepareIntermediateMergeResultWriter(RunFileWriter mergeFileWriter)
+ throws HyracksDataException;
+
+ abstract protected IFrameWriter prepareFinalMergeResultWriter(IFrameWriter nextWriter) throws HyracksDataException;
+
+ abstract protected int[] getSortFields();
+
+ private void merge(IFrameWriter writer, List<GeneratedRunFileReader> partialRuns) throws HyracksDataException {
+ RunMergingFrameReader merger = new RunMergingFrameReader(ctx, partialRuns, inFrames, getSortFields(),
+ comparators, nmkComputer, recordDesc, topK);
+ int io = 0;
+ merger.open();
+ try {
+ while (merger.nextFrame(outputFrame)) {
+ FrameUtils.flushFrame(outputFrame.getBuffer(), writer);
+ io++;
+ }
+ } finally {
+ merger.close();
+ if (LOGGER.isLoggable(Level.FINE)) {
+ LOGGER.fine("Output " + io + " frames");
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java
index e56d51d..f213b97 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java
@@ -36,8 +36,10 @@ import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.util.IntSerDeUtils;
+import org.apache.hyracks.dataflow.std.buffermanager.BufferInfo;
+import org.apache.hyracks.dataflow.std.buffermanager.IFrameBufferManager;
import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.IFrameBufferManager;
public abstract class AbstractFrameSorter implements IFrameSorter {
@@ -60,6 +62,9 @@ public abstract class AbstractFrameSorter implements IFrameSorter {
protected int[] tPointers;
protected int tupleCount;
+ private FrameTupleAccessor fta2;
+ private BufferInfo info = new BufferInfo(null, -1, -1);
+
public AbstractFrameSorter(IHyracksTaskContext ctx, IFrameBufferManager bufferManager, int[] sortFields,
INormalizedKeyComputerFactory firstKeyNormalizerFactory, IBinaryComparatorFactory[] comparatorFactories,
RecordDescriptor recordDescriptor) throws HyracksDataException {
@@ -82,6 +87,7 @@ public abstract class AbstractFrameSorter implements IFrameSorter {
this.outputAppender = new FrameTupleAppender();
this.outputFrame = new VSizeFrame(ctx);
this.outputLimit = outputLimit;
+ this.fta2 = new FrameTupleAccessor(recordDescriptor);
}
@Override
@@ -106,9 +112,8 @@ public abstract class AbstractFrameSorter implements IFrameSorter {
public void sort() throws HyracksDataException {
tupleCount = 0;
for (int i = 0; i < bufferManager.getNumFrames(); ++i) {
- inputTupleAccessor
- .reset(bufferManager.getFrame(i), bufferManager.getFrameStartOffset(i),
- bufferManager.getFrameSize(i));
+ bufferManager.getFrame(i, info);
+ inputTupleAccessor.reset(info.getBuffer(), info.getStartOffset(), info.getLength());
tupleCount += inputTupleAccessor.getTupleCount();
}
if (tPointers == null || tPointers.length < tupleCount * PTR_SIZE) {
@@ -117,9 +122,8 @@ public abstract class AbstractFrameSorter implements IFrameSorter {
int ptr = 0;
int sfIdx = sortFields[0];
for (int i = 0; i < bufferManager.getNumFrames(); ++i) {
- inputTupleAccessor
- .reset(bufferManager.getFrame(i), bufferManager.getFrameStartOffset(i),
- bufferManager.getFrameSize(i));
+ bufferManager.getFrame(i, info);
+ inputTupleAccessor.reset(info.getBuffer(), info.getStartOffset(), info.getLength());
int tCount = inputTupleAccessor.getTupleCount();
byte[] array = inputTupleAccessor.getBuffer().array();
for (int j = 0; j < tCount; ++j) {
@@ -163,9 +167,8 @@ public abstract class AbstractFrameSorter implements IFrameSorter {
int i = tPointers[ptr * PTR_SIZE + ID_FRAMEID];
int tStart = tPointers[ptr * PTR_SIZE + ID_TUPLE_START];
int tEnd = tPointers[ptr * PTR_SIZE + ID_TUPLE_END];
- ByteBuffer buffer = bufferManager.getFrame(i);
- inputTupleAccessor.reset(buffer, bufferManager.getFrameStartOffset(i), bufferManager.getFrameSize(i));
-
+ bufferManager.getFrame(i, info);
+ inputTupleAccessor.reset(info.getBuffer(), info.getStartOffset(), info.getLength());
int flushed = FrameUtils.appendToWriter(writer, outputAppender, inputTupleAccessor, tStart, tEnd);
if (flushed > 0) {
maxFrameSize = Math.max(maxFrameSize, flushed);
@@ -181,6 +184,45 @@ public abstract class AbstractFrameSorter implements IFrameSorter {
return maxFrameSize;
}
+ protected final int compare(int tp1, int tp2) throws HyracksDataException {
+ int i1 = tPointers[tp1 * 4];
+ int j1 = tPointers[tp1 * 4 + 1];
+ int v1 = tPointers[tp1 * 4 + 3];
+
+ int tp2i = tPointers[tp2 * 4];
+ int tp2j = tPointers[tp2 * 4 + 1];
+ int tp2v = tPointers[tp2 * 4 + 3];
+
+ if (v1 != tp2v) {
+ return ((((long) v1) & 0xffffffffL) < (((long) tp2v) & 0xffffffffL)) ? -1 : 1;
+ }
+ int i2 = tp2i;
+ int j2 = tp2j;
+ bufferManager.getFrame(i1, info);
+ byte[] b1 = info.getBuffer().array();
+ inputTupleAccessor.reset(info.getBuffer(), info.getStartOffset(), info.getLength());
+
+ bufferManager.getFrame(i2, info);
+ byte[] b2 = info.getBuffer().array();
+ fta2.reset(info.getBuffer(), info.getStartOffset(), info.getLength());
+ for (int f = 0; f < comparators.length; ++f) {
+ int fIdx = sortFields[f];
+ int f1Start = fIdx == 0 ? 0 : IntSerDeUtils.getInt(b1, j1 + (fIdx - 1) * 4);
+ int f1End = IntSerDeUtils.getInt(b1, j1 + fIdx * 4);
+ int s1 = j1 + inputTupleAccessor.getFieldSlotsLength() + f1Start;
+ int l1 = f1End - f1Start;
+ int f2Start = fIdx == 0 ? 0 : IntSerDeUtils.getInt(b2, j2 + (fIdx - 1) * 4);
+ int f2End = IntSerDeUtils.getInt(b2, j2 + fIdx * 4);
+ int s2 = j2 + fta2.getFieldSlotsLength() + f2Start;
+ int l2 = f2End - f2Start;
+ int c = comparators[f].compare(b1, s1, l1, b2, s2, l2);
+ if (c != 0) {
+ return c;
+ }
+ }
+ return 0;
+ }
+
@Override
public void close() {
tupleCount = 0;
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
index b5ba099..bcebb7d 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
@@ -24,26 +24,27 @@ import java.util.List;
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
import org.apache.hyracks.dataflow.common.io.RunFileWriter;
public abstract class AbstractSortRunGenerator implements IRunGenerator {
- protected final List<RunAndMaxFrameSizePair> runAndMaxSizes;
+ protected final List<GeneratedRunFileReader> generatedRunFileReaders;
public AbstractSortRunGenerator() {
- runAndMaxSizes = new LinkedList<>();
+ generatedRunFileReaders = new LinkedList<>();
}
abstract public ISorter getSorter() throws HyracksDataException;
@Override
public void open() throws HyracksDataException {
- runAndMaxSizes.clear();
+ generatedRunFileReaders.clear();
}
@Override
public void close() throws HyracksDataException {
if (getSorter().hasRemaining()) {
- if (runAndMaxSizes.size() <= 0) {
+ if (generatedRunFileReaders.size() <= 0) {
getSorter().sort();
} else {
flushFramesToRun();
@@ -60,13 +61,12 @@ public abstract class AbstractSortRunGenerator implements IRunGenerator {
RunFileWriter runWriter = getRunFileWriter();
IFrameWriter flushWriter = getFlushableFrameWriter(runWriter);
flushWriter.open();
- int maxFlushedFrameSize;
try {
- maxFlushedFrameSize = getSorter().flush(flushWriter);
+ getSorter().flush(flushWriter);
} finally {
flushWriter.close();
}
- runAndMaxSizes.add(new RunAndMaxFrameSizePair(runWriter.createDeleteOnCloseReader(), maxFlushedFrameSize));
+ generatedRunFileReaders.add(runWriter.createDeleteOnCloseReader());
getSorter().reset();
}
@@ -75,7 +75,7 @@ public abstract class AbstractSortRunGenerator implements IRunGenerator {
}
@Override
- public List<RunAndMaxFrameSizePair> getRuns() {
- return runAndMaxSizes;
+ public List<GeneratedRunFileReader> getRuns() {
+ return generatedRunFileReaders;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
index 8856c41..665a90b 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
@@ -42,6 +42,7 @@ import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
import org.apache.hyracks.dataflow.std.base.AbstractActivityNode;
import org.apache.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
import org.apache.hyracks.dataflow.std.base.AbstractStateObject;
@@ -92,7 +93,7 @@ public abstract class AbstractSorterOperatorDescriptor extends AbstractOperatorD
}
public static class SortTaskState extends AbstractStateObject {
- public List<RunAndMaxFrameSizePair> runAndMaxFrameSizePairs;
+ public List<GeneratedRunFileReader> generatedRunFileReaders;
public ISorter sorter;
public SortTaskState(JobId jobId, TaskId taskId) {
@@ -142,7 +143,7 @@ public abstract class AbstractSorterOperatorDescriptor extends AbstractOperatorD
SortTaskState state = new SortTaskState(ctx.getJobletContext().getJobId(),
new TaskId(getActivityId(), partition));
runGen.close();
- state.runAndMaxFrameSizePairs = runGen.getRuns();
+ state.generatedRunFileReaders = runGen.getRuns();
state.sorter = runGen.getSorter();
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("InitialNumberOfRuns:" + runGen.getRuns().size());
@@ -166,9 +167,9 @@ public abstract class AbstractSorterOperatorDescriptor extends AbstractOperatorD
super(id);
}
- protected abstract ExternalSortRunMerger getSortRunMerger(IHyracksTaskContext ctx,
+ protected abstract AbstractExternalSortRunMerger getSortRunMerger(IHyracksTaskContext ctx,
IRecordDescriptorProvider recordDescProvider, IFrameWriter writer, ISorter sorter,
- List<RunAndMaxFrameSizePair> runs, IBinaryComparator[] comparators, INormalizedKeyComputer nmkComputer,
+ List<GeneratedRunFileReader> runs, IBinaryComparator[] comparators, INormalizedKeyComputer nmkComputer,
int necessaryFrames);
@Override
@@ -180,17 +181,16 @@ public abstract class AbstractSorterOperatorDescriptor extends AbstractOperatorD
public void initialize() throws HyracksDataException {
SortTaskState state = (SortTaskState) ctx
.getStateObject(new TaskId(new ActivityId(getOperatorId(), SORT_ACTIVITY_ID), partition));
- List<RunAndMaxFrameSizePair> runs = state.runAndMaxFrameSizePairs;
+ List<GeneratedRunFileReader> runs = state.generatedRunFileReaders;
ISorter sorter = state.sorter;
IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
for (int i = 0; i < comparatorFactories.length; ++i) {
comparators[i] = comparatorFactories[i].createBinaryComparator();
}
- INormalizedKeyComputer nmkComputer = firstKeyNormalizerFactory == null ?
- null :
- firstKeyNormalizerFactory.createNormalizedKeyComputer();
- ExternalSortRunMerger merger = getSortRunMerger(ctx, recordDescProvider, writer, sorter, runs,
- comparators, nmkComputer, framesLimit);
+ INormalizedKeyComputer nmkComputer = firstKeyNormalizerFactory == null ? null
+ : firstKeyNormalizerFactory.createNormalizedKeyComputer();
+ AbstractExternalSortRunMerger merger = getSortRunMerger(ctx, recordDescProvider, writer, sorter,
+ runs, comparators, nmkComputer, framesLimit);
merger.process();
}
};
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortOperatorDescriptor.java
index fd9187d..36d0c1b 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortOperatorDescriptor.java
@@ -18,15 +18,21 @@
*/
package org.apache.hyracks.dataflow.std.sort;
+import java.util.List;
+
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.ActivityId;
-import org.apache.hyracks.api.dataflow.value.*;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputer;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
+import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.EnumFreeSlotPolicy;
-
-import java.util.List;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
+import org.apache.hyracks.dataflow.std.buffermanager.EnumFreeSlotPolicy;
public class ExternalSortOperatorDescriptor extends AbstractSorterOperatorDescriptor {
@@ -61,8 +67,8 @@ public class ExternalSortOperatorDescriptor extends AbstractSorterOperatorDescri
@Override
protected AbstractSortRunGenerator getRunGenerator(IHyracksTaskContext ctx,
IRecordDescriptorProvider recordDescProvider) throws HyracksDataException {
- return new ExternalSortRunGenerator(ctx, sortFields, firstKeyNormalizerFactory,
- comparatorFactories, recordDescriptors[0], alg, policy, framesLimit, outputLimit);
+ return new ExternalSortRunGenerator(ctx, sortFields, firstKeyNormalizerFactory, comparatorFactories,
+ recordDescriptors[0], alg, policy, framesLimit, outputLimit);
}
};
}
@@ -72,10 +78,11 @@ public class ExternalSortOperatorDescriptor extends AbstractSorterOperatorDescri
return new AbstractSorterOperatorDescriptor.MergeActivity(id) {
@Override
protected ExternalSortRunMerger getSortRunMerger(IHyracksTaskContext ctx,
- IRecordDescriptorProvider recordDescProvider, IFrameWriter writer, ISorter sorter, List<RunAndMaxFrameSizePair> runs, IBinaryComparator[] comparators,
+ IRecordDescriptorProvider recordDescProvider, IFrameWriter writer, ISorter sorter,
+ List<GeneratedRunFileReader> runs, IBinaryComparator[] comparators,
INormalizedKeyComputer nmkComputer, int necessaryFrames) {
- return new ExternalSortRunMerger(ctx, sorter, runs, sortFields, comparators,
- nmkComputer, recordDescriptors[0], necessaryFrames, outputLimit, writer);
+ return new ExternalSortRunMerger(ctx, sorter, runs, sortFields, comparators, nmkComputer,
+ recordDescriptors[0], necessaryFrames, outputLimit, writer);
}
};
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java
index a8ab898..3075719 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java
@@ -18,8 +18,6 @@
*/
package org.apache.hyracks.dataflow.std.sort;
-import java.nio.ByteBuffer;
-
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
@@ -28,20 +26,9 @@ import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.dataflow.common.io.RunFileWriter;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.EnumFreeSlotPolicy;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotBiggestFirst;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotLastFit;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotSmallestFit;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.IFrameBufferManager;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.IFrameFreeSlotPolicy;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableFrameMemoryManager;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableFramePool;
-
-public class ExternalSortRunGenerator extends AbstractSortRunGenerator {
+import org.apache.hyracks.dataflow.std.buffermanager.EnumFreeSlotPolicy;
- protected final IHyracksTaskContext ctx;
- protected final IFrameSorter frameSorter;
- protected final int maxSortFrames;
+public class ExternalSortRunGenerator extends AbstractExternalSortRunGenerator {
public ExternalSortRunGenerator(IHyracksTaskContext ctx, int[] sortFields,
INormalizedKeyComputerFactory firstKeyNormalizerFactory, IBinaryComparatorFactory[] comparatorFactories,
@@ -62,40 +49,8 @@ public class ExternalSortRunGenerator extends AbstractSortRunGenerator {
INormalizedKeyComputerFactory firstKeyNormalizerFactory, IBinaryComparatorFactory[] comparatorFactories,
RecordDescriptor recordDesc, Algorithm alg, EnumFreeSlotPolicy policy, int framesLimit, int outputLimit)
throws HyracksDataException {
- this.ctx = ctx;
- maxSortFrames = framesLimit - 1;
-
- IFrameFreeSlotPolicy freeSlotPolicy = null;
- switch (policy) {
- case SMALLEST_FIT:
- freeSlotPolicy = new FrameFreeSlotSmallestFit();
- break;
- case LAST_FIT:
- freeSlotPolicy = new FrameFreeSlotLastFit(maxSortFrames);
- break;
- case BIGGEST_FIT:
- freeSlotPolicy = new FrameFreeSlotBiggestFirst(maxSortFrames);
- break;
- }
- IFrameBufferManager bufferManager = new VariableFrameMemoryManager(
- new VariableFramePool(ctx, maxSortFrames * ctx.getInitialFrameSize()), freeSlotPolicy);
- if (alg == Algorithm.MERGE_SORT) {
- frameSorter = new FrameSorterMergeSort(ctx, bufferManager, sortFields, firstKeyNormalizerFactory,
- comparatorFactories, recordDesc, outputLimit);
- } else {
- frameSorter = new FrameSorterQuickSort(ctx, bufferManager, sortFields, firstKeyNormalizerFactory,
- comparatorFactories, recordDesc, outputLimit);
- }
- }
-
- @Override
- public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- if (!frameSorter.insertFrame(buffer)) {
- flushFramesToRun();
- if (!frameSorter.insertFrame(buffer)) {
- throw new HyracksDataException("The given frame is too big to insert into the sorting memory.");
- }
- }
+ super(ctx, sortFields, firstKeyNormalizerFactory, comparatorFactories, recordDesc, alg, policy, framesLimit,
+ outputLimit);
}
@Override
@@ -110,8 +65,4 @@ public class ExternalSortRunGenerator extends AbstractSortRunGenerator {
return writer;
}
- @Override
- public ISorter getSorter() {
- return frameSorter;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunMerger.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunMerger.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunMerger.java
index d44826e..e773a3b 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunMerger.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunMerger.java
@@ -18,255 +18,54 @@
*/
package org.apache.hyracks.dataflow.std.sort;
-import java.util.ArrayList;
-import java.util.BitSet;
-import java.util.LinkedList;
import java.util.List;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-import org.apache.hyracks.api.comm.FrameConstants;
-import org.apache.hyracks.api.comm.IFrameReader;
import org.apache.hyracks.api.comm.IFrameWriter;
-import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputer;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
-import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
import org.apache.hyracks.dataflow.common.io.RunFileWriter;
-import org.apache.hyracks.dataflow.std.sort.util.GroupVSizeFrame;
-public class ExternalSortRunMerger {
+public class ExternalSortRunMerger extends AbstractExternalSortRunMerger {
- protected final IHyracksTaskContext ctx;
- protected final IFrameWriter writer;
-
- private final List<RunAndMaxFrameSizePair> runs;
- private final BitSet currentGenerationRunAvailable;
private final int[] sortFields;
- private final IBinaryComparator[] comparators;
- private final INormalizedKeyComputer nmkComputer;
- private final RecordDescriptor recordDesc;
- private final int framesLimit;
- private final int MAX_FRAME_SIZE;
- private final ArrayList<IFrameReader> tempRuns;
- private final int topK;
- private List<GroupVSizeFrame> inFrames;
- private VSizeFrame outputFrame;
- private ISorter sorter;
-
- private static final Logger LOGGER = Logger.getLogger(ExternalSortRunMerger.class.getName());
-
- public ExternalSortRunMerger(IHyracksTaskContext ctx, ISorter sorter, List<RunAndMaxFrameSizePair> runs,
- int[] sortFields, IBinaryComparator[] comparators, INormalizedKeyComputer nmkComputer,
- RecordDescriptor recordDesc, int framesLimit, IFrameWriter writer) {
- this(ctx, sorter, runs, sortFields, comparators, nmkComputer, recordDesc, framesLimit, Integer.MAX_VALUE,
- writer);
- }
- public ExternalSortRunMerger(IHyracksTaskContext ctx, ISorter sorter, List<RunAndMaxFrameSizePair> runs,
+ public ExternalSortRunMerger(IHyracksTaskContext ctx, ISorter sorter, List<GeneratedRunFileReader> runs,
int[] sortFields, IBinaryComparator[] comparators, INormalizedKeyComputer nmkComputer,
RecordDescriptor recordDesc, int framesLimit, int topK, IFrameWriter writer) {
- this.ctx = ctx;
- this.sorter = sorter;
- this.runs = new LinkedList<>(runs);
- this.currentGenerationRunAvailable = new BitSet(runs.size());
+ super(ctx, sorter, runs, comparators, nmkComputer, recordDesc, framesLimit, topK, writer);
this.sortFields = sortFields;
- this.comparators = comparators;
- this.nmkComputer = nmkComputer;
- this.recordDesc = recordDesc;
- this.framesLimit = framesLimit;
- this.writer = writer;
- this.MAX_FRAME_SIZE = FrameConstants.MAX_NUM_MINFRAME * ctx.getInitialFrameSize();
- this.topK = topK;
- this.tempRuns = new ArrayList<>(runs.size());
- }
-
- public void process() throws HyracksDataException {
- IFrameWriter finalWriter = null;
- try {
- if (runs.size() <= 0) {
- finalWriter = prepareSkipMergingFinalResultWriter(writer);
- finalWriter.open();
- if (sorter != null) {
- if (sorter.hasRemaining()) {
- sorter.flush(finalWriter);
- }
- sorter.close();
- }
- } else {
- /** recycle sort buffer */
- if (sorter != null) {
- sorter.close();
- }
-
- finalWriter = prepareFinalMergeResultWriter(writer);
- finalWriter.open();
-
- int maxMergeWidth = framesLimit - 1;
-
- inFrames = new ArrayList<>(maxMergeWidth);
- outputFrame = new VSizeFrame(ctx);
- List<RunAndMaxFrameSizePair> partialRuns = new ArrayList<>(maxMergeWidth);
-
- int stop = runs.size();
- currentGenerationRunAvailable.set(0, stop);
-
- while (true) {
-
- int unUsed = selectPartialRuns(maxMergeWidth * ctx.getInitialFrameSize(), runs, partialRuns,
- currentGenerationRunAvailable, stop);
- prepareFrames(unUsed, inFrames, partialRuns);
-
- if (!currentGenerationRunAvailable.isEmpty() || stop < runs.size()) {
- IFrameReader reader;
- int mergedMaxFrameSize;
- if (partialRuns.size() == 1) {
- if (!currentGenerationRunAvailable.isEmpty()) {
- throw new HyracksDataException(
- "The record is too big to put into the merging frame, please"
- + " allocate more sorting memory");
- } else {
- reader = partialRuns.get(0).run;
- mergedMaxFrameSize = partialRuns.get(0).maxFrameSize;
- }
-
- } else {
- RunFileWriter mergeFileWriter = prepareIntermediateMergeRunFile();
- IFrameWriter mergeResultWriter = prepareIntermediateMergeResultWriter(mergeFileWriter);
-
- mergeResultWriter.open();
- mergedMaxFrameSize = merge(mergeResultWriter, partialRuns);
- mergeResultWriter.close();
-
- reader = mergeFileWriter.createDeleteOnCloseReader();
- }
-
- appendNewRuns(reader, mergedMaxFrameSize);
- if (currentGenerationRunAvailable.isEmpty()) {
-
- if (LOGGER.isLoggable(Level.FINE)) {
- LOGGER.fine("generated runs:" + stop);
- }
- runs.subList(0, stop).clear();
- currentGenerationRunAvailable.clear();
- currentGenerationRunAvailable.set(0, runs.size());
- stop = runs.size();
- }
- } else {
- if (LOGGER.isLoggable(Level.FINE)) {
- LOGGER.fine("final runs:" + stop);
- }
- merge(finalWriter, partialRuns);
- break;
- }
- }
- }
- } catch (Exception e) {
- finalWriter.fail();
- throw new HyracksDataException(e);
- } finally {
- finalWriter.close();
- }
- }
-
- private void appendNewRuns(IFrameReader reader, int mergedPartialMaxSize) {
- runs.add(new RunAndMaxFrameSizePair(reader, mergedPartialMaxSize));
- }
-
- private static int selectPartialRuns(int budget, List<RunAndMaxFrameSizePair> runs,
- List<RunAndMaxFrameSizePair> partialRuns, BitSet runAvailable, int stop) {
- partialRuns.clear();
- int maxFrameSizeOfGenRun = 0;
- int nextRunId = runAvailable.nextSetBit(0);
- while (budget > 0 && nextRunId >= 0 && nextRunId < stop) {
- int runFrameSize = runs.get(nextRunId).maxFrameSize;
- if (budget - runFrameSize >= 0) {
- partialRuns.add(runs.get(nextRunId));
- budget -= runFrameSize;
- runAvailable.clear(nextRunId);
- maxFrameSizeOfGenRun = runFrameSize > maxFrameSizeOfGenRun ? runFrameSize : maxFrameSizeOfGenRun;
- }
- nextRunId = runAvailable.nextSetBit(nextRunId + 1);
- }
- return budget;
- }
-
- private void prepareFrames(int extraFreeMem, List<GroupVSizeFrame> inFrames, List<RunAndMaxFrameSizePair> patialRuns)
- throws HyracksDataException {
- if (extraFreeMem > 0 && patialRuns.size() > 1) {
- int extraFrames = extraFreeMem / ctx.getInitialFrameSize();
- int avg = (extraFrames / patialRuns.size()) * ctx.getInitialFrameSize();
- int residue = (extraFrames % patialRuns.size());
- for (int i = 0; i < residue; i++) {
- patialRuns.get(i).updateSize(
- Math.min(MAX_FRAME_SIZE, patialRuns.get(i).maxFrameSize + avg + ctx.getInitialFrameSize()));
- }
- for (int i = residue; i < patialRuns.size() && avg > 0; i++) {
- patialRuns.get(i).updateSize(Math.min(MAX_FRAME_SIZE, patialRuns.get(i).maxFrameSize + avg));
- }
- }
-
- if (inFrames.size() > patialRuns.size()) {
- inFrames.subList(patialRuns.size(), inFrames.size()).clear();
- }
- int i;
- for (i = 0; i < inFrames.size(); i++) {
- inFrames.get(i).resize(patialRuns.get(i).maxFrameSize);
- }
- for (; i < patialRuns.size(); i++) {
- inFrames.add(new GroupVSizeFrame(ctx, patialRuns.get(i).maxFrameSize));
- }
}
+ @Override
protected IFrameWriter prepareSkipMergingFinalResultWriter(IFrameWriter nextWriter) throws HyracksDataException {
return nextWriter;
}
+ @Override
protected RunFileWriter prepareIntermediateMergeRunFile() throws HyracksDataException {
FileReference newRun = ctx.createManagedWorkspaceFile(ExternalSortRunMerger.class.getSimpleName());
return new RunFileWriter(newRun, ctx.getIOManager());
}
+ @Override
protected IFrameWriter prepareIntermediateMergeResultWriter(RunFileWriter mergeFileWriter)
throws HyracksDataException {
return mergeFileWriter;
}
+ @Override
protected IFrameWriter prepareFinalMergeResultWriter(IFrameWriter nextWriter) throws HyracksDataException {
return nextWriter;
}
+ @Override
protected int[] getSortFields() {
return sortFields;
}
- private int merge(IFrameWriter writer, List<RunAndMaxFrameSizePair> partialRuns) throws HyracksDataException {
- tempRuns.clear();
- for (int i = 0; i < partialRuns.size(); i++) {
- tempRuns.add(partialRuns.get(i).run);
- }
- RunMergingFrameReader merger = new RunMergingFrameReader(ctx, tempRuns, inFrames, getSortFields(), comparators,
- nmkComputer, recordDesc, topK);
- int maxFrameSize = 0;
- int io = 0;
- merger.open();
- try {
- while (merger.nextFrame(outputFrame)) {
- FrameUtils.flushFrame(outputFrame.getBuffer(), writer);
- maxFrameSize = maxFrameSize < outputFrame.getFrameSize() ? outputFrame.getFrameSize() : maxFrameSize;
- io++;
- }
- } finally {
- merger.close();
- if (LOGGER.isLoggable(Level.FINE)) {
- LOGGER.fine("Output " + io + " frames");
- }
- }
- return maxFrameSize;
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterMergeSort.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterMergeSort.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterMergeSort.java
index 5ed11b9..f1f1b1b 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterMergeSort.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterMergeSort.java
@@ -18,21 +18,16 @@
*/
package org.apache.hyracks.dataflow.std.sort;
-import java.nio.ByteBuffer;
-
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-import org.apache.hyracks.dataflow.common.util.IntSerDeUtils;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.IFrameBufferManager;
+import org.apache.hyracks.dataflow.std.buffermanager.IFrameBufferManager;
public class FrameSorterMergeSort extends AbstractFrameSorter {
private int[] tPointersTemp;
- private FrameTupleAccessor fta2;
public FrameSorterMergeSort(IHyracksTaskContext ctx, IFrameBufferManager bufferManager, int[] sortFields,
INormalizedKeyComputerFactory firstKeyNormalizerFactory, IBinaryComparatorFactory[] comparatorFactories,
@@ -46,7 +41,6 @@ public class FrameSorterMergeSort extends AbstractFrameSorter {
RecordDescriptor recordDescriptor, int outputLimit) throws HyracksDataException {
super(ctx, bufferManager, sortFields, firstKeyNormalizerFactory, comparatorFactories, recordDescriptor,
outputLimit);
- fta2 = new FrameTupleAccessor(recordDescriptor);
}
@Override
@@ -122,42 +116,4 @@ public class FrameSorterMergeSort extends AbstractFrameSorter {
tPointersTemp[dest * 4 + 3] = tPointers[src * 4 + 3];
}
- private int compare(int tp1, int tp2) throws HyracksDataException {
- int i1 = tPointers[tp1 * 4];
- int j1 = tPointers[tp1 * 4 + 1];
- int v1 = tPointers[tp1 * 4 + 3];
-
- int tp2i = tPointers[tp2 * 4];
- int tp2j = tPointers[tp2 * 4 + 1];
- int tp2v = tPointers[tp2 * 4 + 3];
-
- if (v1 != tp2v) {
- return ((((long) v1) & 0xffffffffL) < (((long) tp2v) & 0xffffffffL)) ? -1 : 1;
- }
- int i2 = tp2i;
- int j2 = tp2j;
- ByteBuffer buf1 = super.bufferManager.getFrame(i1);
- ByteBuffer buf2 = super.bufferManager.getFrame(i2);
- byte[] b1 = buf1.array();
- byte[] b2 = buf2.array();
- inputTupleAccessor.reset(buf1);
- fta2.reset(buf2);
- for (int f = 0; f < comparators.length; ++f) {
- int fIdx = sortFields[f];
- int f1Start = fIdx == 0 ? 0 : IntSerDeUtils.getInt(b1, j1 + (fIdx - 1) * 4);
- int f1End = IntSerDeUtils.getInt(b1, j1 + fIdx * 4);
- int s1 = j1 + inputTupleAccessor.getFieldSlotsLength() + f1Start;
- int l1 = f1End - f1Start;
- int f2Start = fIdx == 0 ? 0 : IntSerDeUtils.getInt(b2, j2 + (fIdx - 1) * 4);
- int f2End = IntSerDeUtils.getInt(b2, j2 + fIdx * 4);
- int s2 = j2 + fta2.getFieldSlotsLength() + f2Start;
- int l2 = f2End - f2Start;
- int c = comparators[f].compare(b1, s1, l1, b2, s2, l2);
- if (c != 0) {
- return c;
- }
- }
- return 0;
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterQuickSort.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterQuickSort.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterQuickSort.java
index a970a0e..cf864f6 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterQuickSort.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/FrameSorterQuickSort.java
@@ -18,20 +18,15 @@
*/
package org.apache.hyracks.dataflow.std.sort;
-import java.nio.ByteBuffer;
-
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.IFrameBufferManager;
+import org.apache.hyracks.dataflow.std.buffermanager.IFrameBufferManager;
public class FrameSorterQuickSort extends AbstractFrameSorter {
- private FrameTupleAccessor fta2;
-
public FrameSorterQuickSort(IHyracksTaskContext ctx, IFrameBufferManager bufferManager, int[] sortFields,
INormalizedKeyComputerFactory firstKeyNormalizerFactory, IBinaryComparatorFactory[] comparatorFactories,
RecordDescriptor recordDescriptor) throws HyracksDataException {
@@ -44,19 +39,15 @@ public class FrameSorterQuickSort extends AbstractFrameSorter {
RecordDescriptor recordDescriptor, int outputLimit) throws HyracksDataException {
super(ctx, bufferManager, sortFields, firstKeyNormalizerFactory, comparatorFactories, recordDescriptor,
outputLimit);
- fta2 = new FrameTupleAccessor(recordDescriptor);
}
@Override
void sortTupleReferences() throws HyracksDataException {
- sort(tPointers, 0, tupleCount);
+ sort(0, tupleCount);
}
- void sort(int[] tPointers, int offset, int length) throws HyracksDataException {
+ void sort(int offset, int length) throws HyracksDataException {
int m = offset + (length >> 1);
- int mi = tPointers[m * 4];
- int mj = tPointers[m * 4 + 1];
- int mv = tPointers[m * 4 + 3];
int a = offset;
int b = a;
@@ -64,7 +55,7 @@ public class FrameSorterQuickSort extends AbstractFrameSorter {
int d = c;
while (true) {
while (b <= c) {
- int cmp = compare(tPointers, b, mi, mj, mv);
+ int cmp = compare(b, m);
if (cmp > 0) {
break;
}
@@ -74,7 +65,7 @@ public class FrameSorterQuickSort extends AbstractFrameSorter {
++b;
}
while (c >= b) {
- int cmp = compare(tPointers, c, mi, mj, mv);
+ int cmp = compare(c, m);
if (cmp < 0) {
break;
}
@@ -96,10 +87,10 @@ public class FrameSorterQuickSort extends AbstractFrameSorter {
vecswap(tPointers, b, n - s, s);
if ((s = b - a) > 1) {
- sort(tPointers, offset, s);
+ sort(offset, s);
}
if ((s = d - c) > 1) {
- sort(tPointers, n - s, s);
+ sort(n - s, s);
}
}
@@ -117,37 +108,4 @@ public class FrameSorterQuickSort extends AbstractFrameSorter {
}
}
- private int compare(int[] tPointers, int tp1, int tp2i, int tp2j, int tp2v) throws HyracksDataException {
- int i1 = tPointers[tp1 * 4];
- int j1 = tPointers[tp1 * 4 + 1];
- int v1 = tPointers[tp1 * 4 + 3];
- if (v1 != tp2v) {
- return ((((long) v1) & 0xffffffffL) < (((long) tp2v) & 0xffffffffL)) ? -1 : 1;
- }
- int i2 = tp2i;
- int j2 = tp2j;
- ByteBuffer buf1 = super.bufferManager.getFrame(i1);
- ByteBuffer buf2 = super.bufferManager.getFrame(i2);
- byte[] b1 = buf1.array();
- byte[] b2 = buf2.array();
- inputTupleAccessor.reset(buf1);
- fta2.reset(buf2);
- for (int f = 0; f < comparators.length; ++f) {
- int fIdx = sortFields[f];
- int f1Start = fIdx == 0 ? 0 : buf1.getInt(j1 + (fIdx - 1) * 4);
- int f1End = buf1.getInt(j1 + fIdx * 4);
- int s1 = j1 + inputTupleAccessor.getFieldSlotsLength() + f1Start;
- int l1 = f1End - f1Start;
- int f2Start = fIdx == 0 ? 0 : buf2.getInt(j2 + (fIdx - 1) * 4);
- int f2End = buf2.getInt(j2 + fIdx * 4);
- int s2 = j2 + fta2.getFieldSlotsLength() + f2Start;
- int l2 = f2End - f2Start;
- int c = comparators[f].compare(b1, s1, l1, b2, s2, l2);
- if (c != 0) {
- return c;
- }
- }
- return 0;
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java
index 7f2847c..648be02 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java
@@ -31,10 +31,10 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.io.RunFileWriter;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.IFramePool;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.ITupleBufferManager;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableFramePool;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableTupleMemoryManager;
+import org.apache.hyracks.dataflow.std.buffermanager.IFramePool;
+import org.apache.hyracks.dataflow.std.buffermanager.IDeletableTupleBufferManager;
+import org.apache.hyracks.dataflow.std.buffermanager.VariableFramePool;
+import org.apache.hyracks.dataflow.std.buffermanager.VariableDeletableTupleMemoryManager;
public class HeapSortRunGenerator extends AbstractSortRunGenerator {
protected final IHyracksTaskContext ctx;
@@ -64,7 +64,7 @@ public class HeapSortRunGenerator extends AbstractSortRunGenerator {
@Override
public void open() throws HyracksDataException {
IFramePool framePool = new VariableFramePool(ctx, (frameLimit - 1) * ctx.getInitialFrameSize());
- ITupleBufferManager bufferManager = new VariableTupleMemoryManager(framePool, recordDescriptor);
+ IDeletableTupleBufferManager bufferManager = new VariableDeletableTupleMemoryManager(framePool, recordDescriptor);
tupleSorter = new TupleSorterHeapSort(ctx, bufferManager, topK, sortFields, nmkFactory, comparatorFactories);
super.open();
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HybridTopKSortRunGenerator.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HybridTopKSortRunGenerator.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HybridTopKSortRunGenerator.java
index af271ff..9be4bc6 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HybridTopKSortRunGenerator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HybridTopKSortRunGenerator.java
@@ -30,9 +30,10 @@ import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.dataflow.common.io.RunFileWriter;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotBiggestFirst;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableFrameMemoryManager;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableFramePool;
+import org.apache.hyracks.dataflow.std.buffermanager.EnumFreeSlotPolicy;
+import org.apache.hyracks.dataflow.std.buffermanager.FrameFreeSlotPolicyFactory;
+import org.apache.hyracks.dataflow.std.buffermanager.VariableFrameMemoryManager;
+import org.apache.hyracks.dataflow.std.buffermanager.VariableFramePool;
public class HybridTopKSortRunGenerator extends HeapSortRunGenerator {
private static final Logger LOG = Logger.getLogger(HybridTopKSortRunGenerator.class.getName());
@@ -59,8 +60,8 @@ public class HybridTopKSortRunGenerator extends HeapSortRunGenerator {
@Override
protected RunFileWriter getRunFileWriter() throws HyracksDataException {
- FileReference file = ctx.getJobletContext().createManagedWorkspaceFile(
- HybridTopKSortRunGenerator.class.getSimpleName());
+ FileReference file = ctx.getJobletContext()
+ .createManagedWorkspaceFile(HybridTopKSortRunGenerator.class.getSimpleName());
return new RunFileWriter(file, ctx.getIOManager());
}
@@ -95,7 +96,8 @@ public class HybridTopKSortRunGenerator extends HeapSortRunGenerator {
if (frameSorter == null) {
VariableFrameMemoryManager bufferManager = new VariableFrameMemoryManager(
new VariableFramePool(ctx, (frameLimit - 1) * ctx.getInitialFrameSize()),
- new FrameFreeSlotBiggestFirst(frameLimit - 1));
+ FrameFreeSlotPolicyFactory.createFreeSlotPolicy(EnumFreeSlotPolicy.BIGGEST_FIT,
+ frameLimit - 1));
frameSorter = new FrameSorterMergeSort(ctx, bufferManager, sortFields, nmkFactory, comparatorFactories,
recordDescriptor, topK);
if (LOG.isLoggable(Level.FINE)) {
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
index 70075b1..02880dd 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
@@ -21,6 +21,7 @@ package org.apache.hyracks.dataflow.std.sort;
import java.util.List;
import org.apache.hyracks.api.comm.IFrameWriter;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
/**
* @author pouria
@@ -31,5 +32,5 @@ public interface IRunGenerator extends IFrameWriter {
/**
* @return the list of generated (sorted) runs
*/
- List<RunAndMaxFrameSizePair> getRuns();
+ List<GeneratedRunFileReader> getRuns();
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ISorter.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ISorter.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ISorter.java
index 5f83a79..3fb2b54 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ISorter.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ISorter.java
@@ -30,7 +30,7 @@ public interface ISorter {
void sort() throws HyracksDataException;
- void close();
+ void close() throws HyracksDataException;
int flush(IFrameWriter writer) throws HyracksDataException;
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/InMemorySortOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/InMemorySortOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/InMemorySortOperatorDescriptor.java
index 2ffe437..521c2ae 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/InMemorySortOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/InMemorySortOperatorDescriptor.java
@@ -40,10 +40,11 @@ import org.apache.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
import org.apache.hyracks.dataflow.std.base.AbstractStateObject;
import org.apache.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable;
import org.apache.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotLastFit;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.IFrameBufferManager;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableFrameMemoryManager;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableFramePool;
+import org.apache.hyracks.dataflow.std.buffermanager.EnumFreeSlotPolicy;
+import org.apache.hyracks.dataflow.std.buffermanager.FrameFreeSlotPolicyFactory;
+import org.apache.hyracks.dataflow.std.buffermanager.IFrameBufferManager;
+import org.apache.hyracks.dataflow.std.buffermanager.VariableFrameMemoryManager;
+import org.apache.hyracks.dataflow.std.buffermanager.VariableFramePool;
public class InMemorySortOperatorDescriptor extends AbstractOperatorDescriptor {
private static final long serialVersionUID = 1L;
@@ -119,7 +120,8 @@ public class InMemorySortOperatorDescriptor extends AbstractOperatorDescriptor {
new TaskId(getActivityId(), partition));
IFrameBufferManager frameBufferManager = new VariableFrameMemoryManager(
- new VariableFramePool(ctx, VariableFramePool.UNLIMITED_MEMORY), new FrameFreeSlotLastFit());
+ new VariableFramePool(ctx, VariableFramePool.UNLIMITED_MEMORY),
+ FrameFreeSlotPolicyFactory.createFreeSlotPolicy(EnumFreeSlotPolicy.LAST_FIT));
state.frameSorter = new FrameSorterMergeSort(ctx, frameBufferManager, sortFields,
firstKeyNormalizerFactory, comparatorFactories, recordDescriptors[0]);
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/RunAndMaxFrameSizePair.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/RunAndMaxFrameSizePair.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/RunAndMaxFrameSizePair.java
deleted file mode 100644
index 1dfccfd..0000000
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/RunAndMaxFrameSizePair.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 at
- *
- * 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 org.apache.hyracks.dataflow.std.sort;
-
-import org.apache.hyracks.api.comm.IFrameReader;
-
-public class RunAndMaxFrameSizePair {
- public IFrameReader run;
- public int maxFrameSize;
-
- public RunAndMaxFrameSizePair(IFrameReader run, int maxFrameSize) {
- this.run = run;
- this.maxFrameSize = maxFrameSize;
- }
-
- void updateSize(int newMaxSize){
- this.maxFrameSize = newMaxSize;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TopKSorterOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TopKSorterOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TopKSorterOperatorDescriptor.java
index cd0008f..5df09c6 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TopKSorterOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TopKSorterOperatorDescriptor.java
@@ -19,13 +19,19 @@
package org.apache.hyracks.dataflow.std.sort;
+import java.util.List;
+
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.ActivityId;
-import org.apache.hyracks.api.dataflow.value.*;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputer;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
+import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
-
-import java.util.List;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
public class TopKSorterOperatorDescriptor extends AbstractSorterOperatorDescriptor {
@@ -56,10 +62,11 @@ public class TopKSorterOperatorDescriptor extends AbstractSorterOperatorDescript
return new MergeActivity(id) {
@Override
protected ExternalSortRunMerger getSortRunMerger(IHyracksTaskContext ctx,
- IRecordDescriptorProvider recordDescProvider, IFrameWriter writer, ISorter sorter, List<RunAndMaxFrameSizePair> runs, IBinaryComparator[] comparators,
+ IRecordDescriptorProvider recordDescProvider, IFrameWriter writer, ISorter sorter,
+ List<GeneratedRunFileReader> runs, IBinaryComparator[] comparators,
INormalizedKeyComputer nmkComputer, int necessaryFrames) {
- return new ExternalSortRunMerger(ctx, sorter, runs, sortFields, comparators,
- nmkComputer, recordDescriptors[0], necessaryFrames, topK, writer);
+ return new ExternalSortRunMerger(ctx, sorter, runs, sortFields, comparators, nmkComputer,
+ recordDescriptors[0], necessaryFrames, topK, writer);
}
};
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java
index fc15bbc..a5ec88a 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java
@@ -36,8 +36,8 @@ import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.ITupleBufferAccessor;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.ITupleBufferManager;
+import org.apache.hyracks.dataflow.std.buffermanager.IDeletableTupleBufferManager;
+import org.apache.hyracks.dataflow.std.buffermanager.ITuplePointerAccessor;
import org.apache.hyracks.dataflow.std.structures.IResetableComparable;
import org.apache.hyracks.dataflow.std.structures.IResetableComparableFactory;
import org.apache.hyracks.dataflow.std.structures.MaxHeap;
@@ -70,8 +70,8 @@ public class TupleSorterHeapSort implements ITupleSorter {
}
bufferAccessor1.reset(tuplePointer);
bufferAccessor2.reset(o.tuplePointer);
- byte[] b1 = bufferAccessor1.getTupleBuffer().array();
- byte[] b2 = bufferAccessor2.getTupleBuffer().array();
+ byte[] b1 = bufferAccessor1.getBuffer().array();
+ byte[] b2 = bufferAccessor2.getBuffer().array();
for (int f = 0; f < comparators.length; ++f) {
int fIdx = sortFields[f];
@@ -104,9 +104,9 @@ public class TupleSorterHeapSort implements ITupleSorter {
}
}
- private final ITupleBufferManager bufferManager;
- private final ITupleBufferAccessor bufferAccessor1;
- private final ITupleBufferAccessor bufferAccessor2;
+ private final IDeletableTupleBufferManager bufferManager;
+ private final ITuplePointerAccessor bufferAccessor1;
+ private final ITuplePointerAccessor bufferAccessor2;
private final int topK;
private final FrameTupleAppender outputAppender;
private final IFrame outputFrame;
@@ -120,12 +120,13 @@ public class TupleSorterHeapSort implements ITupleSorter {
private MaxHeap heap;
private boolean isSorted;
- public TupleSorterHeapSort(IHyracksTaskContext ctx, ITupleBufferManager bufferManager, int topK, int[] sortFields,
+ public TupleSorterHeapSort(IHyracksTaskContext ctx, IDeletableTupleBufferManager bufferManager, int topK,
+ int[] sortFields,
INormalizedKeyComputerFactory firstKeyNormalizerFactory, IBinaryComparatorFactory[] comparatorFactories)
throws HyracksDataException {
this.bufferManager = bufferManager;
- this.bufferAccessor1 = bufferManager.getTupleAccessor();
- this.bufferAccessor2 = bufferManager.getTupleAccessor();
+ this.bufferAccessor1 = bufferManager.createTupleAccessor();
+ this.bufferAccessor2 = bufferManager.createTupleAccessor();
this.topK = topK;
this.outputFrame = new VSizeFrame(ctx);
this.outputAppender = new FrameTupleAppender();
@@ -190,7 +191,7 @@ public class TupleSorterHeapSort implements ITupleSorter {
}
bufferAccessor2.reset(maxEntry.tuplePointer);
byte[] b1 = frameTupleAccessor.getBuffer().array();
- byte[] b2 = bufferAccessor2.getTupleBuffer().array();
+ byte[] b2 = bufferAccessor2.getBuffer().array();
for (int f = 0; f < comparators.length; ++f) {
int fIdx = sortFields[f];
@@ -236,7 +237,7 @@ public class TupleSorterHeapSort implements ITupleSorter {
};
@Override
- public void close() {
+ public void close() throws HyracksDataException {
heap = null;
bufferManager.close();
isSorted = false;
@@ -254,7 +255,7 @@ public class TupleSorterHeapSort implements ITupleSorter {
HeapEntry minEntry = (HeapEntry) entries[i];
bufferAccessor1.reset(minEntry.tuplePointer);
int flushed = FrameUtils
- .appendToWriter(writer, outputAppender, bufferAccessor1.getTupleBuffer().array(),
+ .appendToWriter(writer, outputAppender, bufferAccessor1.getBuffer().array(),
bufferAccessor1.getTupleStartOffset(), bufferAccessor1.getTupleLength());
if (flushed > 0) {
maxFrameSize = Math.max(maxFrameSize, flushed);
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/EnumFreeSlotPolicy.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/EnumFreeSlotPolicy.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/EnumFreeSlotPolicy.java
deleted file mode 100644
index 21ed3e2..0000000
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/EnumFreeSlotPolicy.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 at
- *
- * 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 org.apache.hyracks.dataflow.std.sort.buffermanager;
-
-public enum EnumFreeSlotPolicy {
- SMALLEST_FIT,
- LAST_FIT,
- BIGGEST_FIT,
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotBiggestFirst.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotBiggestFirst.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotBiggestFirst.java
deleted file mode 100644
index 200f27d..0000000
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotBiggestFirst.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 at
- *
- * 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 org.apache.hyracks.dataflow.std.sort.buffermanager;
-
-import org.apache.hyracks.dataflow.std.structures.IResetableComparable;
-import org.apache.hyracks.dataflow.std.structures.IResetableComparableFactory;
-import org.apache.hyracks.dataflow.std.structures.MaxHeap;
-
-public class FrameFreeSlotBiggestFirst implements IFrameFreeSlotPolicy {
- private static final int INVALID = -1;
-
- class SpaceEntryFactory implements IResetableComparableFactory {
- @Override
- public IResetableComparable createResetableComparable() {
- return new SpaceEntry();
- }
- }
-
- class SpaceEntry implements IResetableComparable<SpaceEntry> {
- int space;
- int id;
-
- SpaceEntry() {
- space = INVALID;
- id = INVALID;
- }
-
- @Override
- public int compareTo(SpaceEntry o) {
- if (o.space != space) {
- if (o.space == INVALID) {
- return 1;
- }
- if (space == INVALID) {
- return -1;
- }
- return space < o.space ? -1 : 1;
- }
- return 0;
- }
-
- @Override
- public void reset(SpaceEntry other) {
- space = other.space;
- id = other.id;
- }
-
- void reset(int space, int id) {
- this.space = space;
- this.id = id;
- }
- }
-
- private MaxHeap heap;
- private SpaceEntry tempEntry;
-
- public FrameFreeSlotBiggestFirst(int initialCapacity) {
- heap = new MaxHeap(new SpaceEntryFactory(), initialCapacity);
- tempEntry = new SpaceEntry();
- }
-
- @Override
- public int popBestFit(int tobeInsertedSize) {
- if (!heap.isEmpty()) {
- heap.peekMax(tempEntry);
- if (tempEntry.space >= tobeInsertedSize) {
- heap.getMax(tempEntry);
- return tempEntry.id;
- }
- }
- return -1;
- }
-
- @Override
- public void pushNewFrame(int frameID, int freeSpace) {
- tempEntry.reset(freeSpace, frameID);
- heap.insert(tempEntry);
- }
-
- @Override
- public void reset() {
- heap.reset();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotLastFit.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotLastFit.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotLastFit.java
deleted file mode 100644
index 0235c6f..0000000
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotLastFit.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 at
- *
- * 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 org.apache.hyracks.dataflow.std.sort.buffermanager;
-
-import java.util.Arrays;
-
-public class FrameFreeSlotLastFit implements IFrameFreeSlotPolicy {
- private static int INITIAL_CAPACITY = 10;
-
- private class FrameSpace {
- int frameId;
- int freeSpace;
-
- FrameSpace(int frameId, int freeSpace) {
- reset(frameId, freeSpace);
- }
-
- void reset(int frameId, int freeSpace) {
- this.frameId = frameId;
- this.freeSpace = freeSpace;
- }
- }
-
- private FrameSpace[] frameSpaces;
- private int size;
-
- public FrameFreeSlotLastFit(int maxFrames) {
- frameSpaces = new FrameSpace[maxFrames];
- size = 0;
- }
-
- public FrameFreeSlotLastFit() {
- this(INITIAL_CAPACITY);
- }
-
- @Override
- public int popBestFit(int tobeInsertedSize) {
- for (int i = size - 1; i >= 0; i--) {
- if (frameSpaces[i].freeSpace >= tobeInsertedSize) {
- FrameSpace ret = frameSpaces[i];
- System.arraycopy(frameSpaces, i + 1, frameSpaces, i, size - i - 1);
- frameSpaces[--size] = ret;
- return ret.frameId;
- }
- }
- return -1;
- }
-
- @Override
- public void pushNewFrame(int frameID, int freeSpace) {
- if (size >= frameSpaces.length) {
- frameSpaces = Arrays.copyOf(frameSpaces, size * 2);
- }
- if (frameSpaces[size] == null) {
- frameSpaces[size++] = new FrameSpace(frameID, freeSpace);
- } else {
- frameSpaces[size++].reset(frameID, freeSpace);
- }
- }
-
- @Override
- public void reset() {
- size = 0;
- for (int i = frameSpaces.length - 1; i >= 0; i--) {
- frameSpaces[i] = null;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/6abc63e2/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotSmallestFit.java
----------------------------------------------------------------------
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotSmallestFit.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotSmallestFit.java
deleted file mode 100644
index 8331a93..0000000
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/buffermanager/FrameFreeSlotSmallestFit.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 at
- *
- * 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 org.apache.hyracks.dataflow.std.sort.buffermanager;
-
-import java.util.LinkedList;
-import java.util.Map;
-import java.util.TreeMap;
-
-public class FrameFreeSlotSmallestFit implements IFrameFreeSlotPolicy {
-
- private TreeMap<Integer, LinkedList<Integer>> freeSpaceIndex;
-
- public FrameFreeSlotSmallestFit() {
- freeSpaceIndex = new TreeMap<>();
- }
-
- @Override
- public int popBestFit(int tobeInsertedSize) {
- Map.Entry<Integer, LinkedList<Integer>> entry = freeSpaceIndex.ceilingEntry(tobeInsertedSize);
- if (entry == null) {
- return -1;
- }
- int id = entry.getValue().removeFirst();
- if (entry.getValue().isEmpty()) {
- freeSpaceIndex.remove(entry.getKey());
- }
- return id;
- }
-
- @Override
- public void pushNewFrame(int frameID, int freeSpace) {
- Map.Entry<Integer, LinkedList<Integer>> entry = freeSpaceIndex.ceilingEntry(freeSpace);
- if (entry == null || entry.getKey() != freeSpace) {
- LinkedList<Integer> linkedList = new LinkedList<>();
- linkedList.add(frameID);
- freeSpaceIndex.put(freeSpace, linkedList);
- } else {
- entry.getValue().add(frameID);
- }
- }
-
- @Override
- public void reset() {
- freeSpaceIndex.clear();
- }
-}