You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by jh...@apache.org on 2015/11/17 04:02:08 UTC
[1/2] tajo git commit: TAJO-1271: Improve memory usage of
Hash-shuffle.
Repository: tajo
Updated Branches:
refs/heads/master 10159c7f8 -> 411a26d5d
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
index 35a204a..49b0e11 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashShuffleFileWriteExec.java
@@ -19,37 +19,62 @@
package org.apache.tajo.engine.planner.physical;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.SessionVars;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.statistics.TableStats;
-import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.common.TajoDataTypes.DataType;
+import org.apache.tajo.exception.TajoRuntimeException;
+import org.apache.tajo.exception.UnsupportedException;
import org.apache.tajo.plan.logical.ShuffleFileWriteNode;
-import org.apache.tajo.storage.HashShuffleAppenderWrapper;
import org.apache.tajo.storage.HashShuffleAppenderManager;
import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.tuple.memory.MemoryRowBlock;
+import org.apache.tajo.tuple.memory.RowBlock;
+import org.apache.tajo.tuple.memory.RowWriter;
+import org.apache.tajo.unit.StorageUnit;
+import org.apache.tajo.util.FileUtil;
import org.apache.tajo.worker.TaskAttemptContext;
import java.io.IOException;
-import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
/**
* <code>HashShuffleFileWriteExec</code> is a physical executor to store intermediate data into a number of
* file outputs associated with shuffle keys. The file outputs are stored on local disks.
*/
public final class HashShuffleFileWriteExec extends UnaryPhysicalExec {
- private static Log LOG = LogFactory.getLog(HashShuffleFileWriteExec.class);
- private ShuffleFileWriteNode plan;
+ private static final Log LOG = LogFactory.getLog(HashShuffleFileWriteExec.class);
+ private static final int MAXIMUM_INITIAL_BUFFER_SIZE = StorageUnit.MB;
+ private static final int MINIMUM_INITIAL_BUFFER_SIZE = 4 * StorageUnit.KB;
+ // Buffer usage is greater than threshold, it will be flush to local storage
+ private static final float BUFFER_THRESHOLD_FACTOR = 0.8f;
+
+ private final ShuffleFileWriteNode plan;
private final TableMeta meta;
- private Partitioner partitioner;
- private Map<Integer, HashShuffleAppenderWrapper> appenderMap = new HashMap<>();
+ private final Partitioner partitioner;
private final int numShuffleOutputs;
- private final int [] shuffleKeyIds;
- private HashShuffleAppenderManager hashShuffleAppenderManager;
- private int numHashShuffleBufferTuples;
+ private final int[] shuffleKeyIds;
+ private final HashShuffleAppenderManager hashShuffleAppenderManager;
+ private final int maxBufferSize;
+ private final int bufferThreshold;
+ private final int initialBufferSize;
+ private final DataType[] dataTypes;
+
+ private final Map<Integer, MemoryRowBlock> partitionMemoryMap;
+ private long writtenBytes = 0;
+ private long usedBufferSize = 0;
+ private long totalBufferCapacity = 0;
public HashShuffleFileWriteExec(TaskAttemptContext context,
final ShuffleFileWriteNode plan, final PhysicalExec child) throws IOException {
@@ -71,105 +96,151 @@ public final class HashShuffleFileWriteExec extends UnaryPhysicalExec {
}
this.partitioner = new HashPartitioner(shuffleKeyIds, numShuffleOutputs);
this.hashShuffleAppenderManager = context.getHashShuffleAppenderManager();
- this.numHashShuffleBufferTuples = context.getConf().getIntVar(ConfVars.SHUFFLE_HASH_APPENDER_BUFFER_SIZE);
+ this.maxBufferSize = context.getQueryContext().getInt(SessionVars.HASH_SHUFFLE_BUFFER_SIZE) * StorageUnit.MB;
+ this.bufferThreshold = (int) (maxBufferSize * BUFFER_THRESHOLD_FACTOR);
+ this.dataTypes = SchemaUtil.toDataTypes(outSchema);
+
+ if(numShuffleOutputs > 0){
+ //calculate initial buffer by total partition. a buffer size will be 4Kb ~ 1MB
+ this.initialBufferSize = Math.min(MAXIMUM_INITIAL_BUFFER_SIZE,
+ Math.max(maxBufferSize / numShuffleOutputs, MINIMUM_INITIAL_BUFFER_SIZE));
+ } else {
+ this.initialBufferSize = MINIMUM_INITIAL_BUFFER_SIZE;
+ }
+
+ this.partitionMemoryMap = Maps.newHashMap();
}
@Override
public void init() throws IOException {
super.init();
}
-
- private HashShuffleAppenderWrapper getAppender(int partId) throws IOException {
- HashShuffleAppenderWrapper appender = appenderMap.get(partId);
- if (appender == null) {
- appender = hashShuffleAppenderManager.getAppender(context.getConf(),
- context.getTaskId().getTaskId().getExecutionBlockId(), partId, meta, outSchema);
- appenderMap.put(partId, appender);
- }
- return appender;
- }
-
- Map<Integer, TupleList> partitionTuples = new HashMap<>();
- long writtenBytes = 0L;
@Override
public Tuple next() throws IOException {
try {
Tuple tuple;
int partId;
- int tupleCount = 0;
long numRows = 0;
while (!context.isStopped() && (tuple = child.next()) != null) {
- tupleCount++;
- numRows++;
partId = partitioner.getPartition(tuple);
- TupleList partitionTupleList = partitionTuples.get(partId);
- if (partitionTupleList == null) {
- partitionTupleList = new TupleList(1000);
- partitionTuples.put(partId, partitionTupleList);
+ MemoryRowBlock rowBlock = partitionMemoryMap.get(partId);
+ if (rowBlock == null) {
+ rowBlock = new MemoryRowBlock(dataTypes, initialBufferSize, true, plan.getStorageType());
+ partitionMemoryMap.put(partId, rowBlock);
+ totalBufferCapacity += rowBlock.capacity();
}
- partitionTupleList.add(tuple);
- if (tupleCount >= numHashShuffleBufferTuples) {
- for (Map.Entry<Integer, TupleList> entry : partitionTuples.entrySet()) {
- int appendPartId = entry.getKey();
- HashShuffleAppenderWrapper appender = getAppender(appendPartId);
- int appendedSize = appender.addTuples(context.getTaskId(), entry.getValue());
- writtenBytes += appendedSize;
- entry.getValue().clear();
+
+ RowWriter writer = rowBlock.getWriter();
+ long prevUsedMem = rowBlock.usedMem();
+ totalBufferCapacity -= rowBlock.capacity();
+
+ writer.addTuple(tuple);
+ numRows++;
+
+ totalBufferCapacity += rowBlock.capacity(); // calculate resizeable buffer capacity
+ usedBufferSize += (rowBlock.usedMem() - prevUsedMem);
+
+ // if total buffer capacity are required more than maxBufferSize,
+ // all partitions are flushed and the buffers are released
+ if (totalBufferCapacity > maxBufferSize) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Too low buffer usage. threshold: %s, total capacity: %s, used: %s",
+ FileUtil.humanReadableByteCount(maxBufferSize, false),
+ FileUtil.humanReadableByteCount(totalBufferCapacity, false),
+ FileUtil.humanReadableByteCount(usedBufferSize, false)));
}
- tupleCount = 0;
+
+ //flush and release buffer
+ flushBuffer(partitionMemoryMap, true);
+ writtenBytes += usedBufferSize;
+ totalBufferCapacity = usedBufferSize = 0;
+
+ } else if (usedBufferSize > bufferThreshold) {
+ //flush and reuse buffer
+ flushBuffer(partitionMemoryMap, false);
+ writtenBytes += usedBufferSize;
+ usedBufferSize = 0;
}
}
- // processing remained tuples
- for (Map.Entry<Integer, TupleList> entry : partitionTuples.entrySet()) {
- int appendPartId = entry.getKey();
- HashShuffleAppenderWrapper appender = getAppender(appendPartId);
- int appendedSize = appender.addTuples(context.getTaskId(), entry.getValue());
- writtenBytes += appendedSize;
- entry.getValue().clear();
- }
+ // flush remaining buffers
+ flushBuffer(partitionMemoryMap, true);
+ writtenBytes += usedBufferSize;
+ usedBufferSize = totalBufferCapacity = 0;
TableStats aggregated = (TableStats) child.getInputStats().clone();
aggregated.setNumBytes(writtenBytes);
aggregated.setNumRows(numRows);
context.setResultStats(aggregated);
- partitionTuples.clear();
-
return null;
} catch (RuntimeException e) {
LOG.error(e.getMessage(), e);
throw new IOException(e);
- } catch (Exception e) {
+ } catch (Throwable e) {
LOG.error(e.getMessage(), e);
throw new IOException(e);
}
}
+ /**
+ * flush all buffer to local storage
+ */
+ private void flushBuffer(Map<Integer, MemoryRowBlock> partitionMemoryMap, boolean releaseBuffer)
+ throws IOException, ExecutionException, InterruptedException {
+ List<Future<MemoryRowBlock>> resultList = Lists.newArrayList();
+ ArrayList<Integer> unusedBuffer = Lists.newArrayList();
+
+ for (Map.Entry<Integer, MemoryRowBlock> entry : partitionMemoryMap.entrySet()) {
+ int appendPartId = entry.getKey();
+
+ MemoryRowBlock memoryRowBlock = entry.getValue();
+ if (memoryRowBlock.getMemory().isReadable()) {
+ //flush and release buffer
+ resultList.add(hashShuffleAppenderManager.
+ writePartitions(meta, outSchema, context.getTaskId(), appendPartId, memoryRowBlock, releaseBuffer));
+ } else {
+ if (releaseBuffer) {
+ memoryRowBlock.release();
+ } else {
+ unusedBuffer.add(appendPartId);
+ }
+ }
+ }
+
+ // wait for flush to storage
+ for (Future<MemoryRowBlock> future : resultList) {
+ future.get();
+ }
+
+ if (releaseBuffer) {
+ partitionMemoryMap.clear();
+ } else {
+ // release the unused partition
+ for (Integer id : unusedBuffer) {
+ MemoryRowBlock memoryRowBlock = partitionMemoryMap.remove(id);
+ memoryRowBlock.release();
+ }
+ }
+ }
+
@Override
public void rescan() throws IOException {
- // nothing to do
+ throw new TajoRuntimeException(new UnsupportedException());
}
@Override
public void close() throws IOException{
- super.close();
- if (appenderMap != null) {
- appenderMap.clear();
- appenderMap = null;
- }
-
- for (TupleList eachList : partitionTuples.values()) {
- eachList.clear();
+ if (partitionMemoryMap.size() > 0) {
+ for (RowBlock rowBlock : partitionMemoryMap.values()) {
+ rowBlock.release();
+ }
+ partitionMemoryMap.clear();
}
- partitionTuples.clear();
- partitionTuples = null;
-
- partitioner = null;
- plan = null;
progress = 1.0f;
+ super.close();
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java
index b652b0a..cabbf4c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SortExec.java
@@ -26,6 +26,7 @@ import org.apache.tajo.worker.TaskAttemptContext;
import java.io.IOException;
import java.util.Comparator;
+import java.util.List;
public abstract class SortExec extends UnaryPhysicalExec {
@@ -39,7 +40,7 @@ public abstract class SortExec extends UnaryPhysicalExec {
this.comparator = new BaseTupleComparator(inSchema, sortSpecs);
}
- protected TupleSorter getSorter(TupleList tupleSlots) {
+ protected TupleSorter getSorter(List tupleSlots) {
if (!tupleSlots.isEmpty() && ComparableVector.isVectorizable(sortSpecs)) {
return new VectorizedSorter(tupleSlots, sortSpecs, comparator.getSortKeyIds());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/TupleSorter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/TupleSorter.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/TupleSorter.java
index abf2808..39c67d6 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/TupleSorter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/TupleSorter.java
@@ -22,6 +22,7 @@ import org.apache.tajo.storage.Tuple;
import org.apache.tajo.storage.TupleComparator;
import java.util.Collections;
+import java.util.List;
public interface TupleSorter {
@@ -29,10 +30,10 @@ public interface TupleSorter {
class DefaultSorter implements TupleSorter {
- private final TupleList target;
+ private final List target;
private final TupleComparator comparator;
- public DefaultSorter(TupleList target, TupleComparator comparator) {
+ public DefaultSorter(List target, TupleComparator comparator) {
this.target = target;
this.comparator = comparator;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/VectorizedSorter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/VectorizedSorter.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/VectorizedSorter.java
index 82f7153..44649cd 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/VectorizedSorter.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/VectorizedSorter.java
@@ -26,6 +26,7 @@ import org.apache.tajo.exception.UnsupportedException;
import org.apache.tajo.storage.Tuple;
import java.util.Iterator;
+import java.util.List;
/**
* Extract raw level values (primitive or String/byte[]) from each of key columns before sorting
@@ -35,7 +36,7 @@ public class VectorizedSorter extends ComparableVector implements IndexedSortabl
private final int[] mappings; // index indirection
- public VectorizedSorter(TupleList source, SortSpec[] sortKeys, int[] keyIndex) {
+ public VectorizedSorter(List source, SortSpec[] sortKeys, int[] keyIndex) {
super(source.size(), sortKeys, keyIndex);
source.toArray(tuples); // wish it's array list
mappings = new int[tuples.length];
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
index 815c44b..e64cd51 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/Repartitioner.java
@@ -194,8 +194,11 @@ public class Repartitioner {
long maxStats = Long.MIN_VALUE;
int maxStatsScanIdx = -1;
StringBuilder nonLeafScanNamesBuilder = new StringBuilder();
+
+ String intermediateDataFormat = schedulerContext.getMasterContext().getConf().getVar(ConfVars.SHUFFLE_FILE_FORMAT);
for (int i = 0; i < scans.length; i++) {
- if (scans[i].getTableDesc().getMeta().getDataFormat().equalsIgnoreCase("RAW")) {
+
+ if (scans[i].getTableDesc().getMeta().getDataFormat().equalsIgnoreCase(intermediateDataFormat)) {
// Intermediate data scan
hasNonLeafNode = true;
largeScanIndexList.add(i);
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
index 4b0dac9..c7cac4f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
@@ -338,6 +338,10 @@ public class TajoWorker extends CompositeService {
return;
}
+ if (hashShuffleAppenderManager != null) {
+ hashShuffleAppenderManager.shutdown();
+ }
+
if(webServer != null) {
try {
webServer.stop();
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java
index be920a1..fbd1948 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java
@@ -25,10 +25,7 @@ import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.*;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.TajoProtos.TaskAttemptState;
import org.apache.tajo.TaskAttemptId;
@@ -548,6 +545,7 @@ public class TaskImpl implements Task {
private FileFragment[] localizeFetchedData(File file, String name, TableMeta meta)
throws IOException {
+
Configuration c = new Configuration(systemConf);
c.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, "file:///");
FileSystem fs = FileSystem.get(c);
@@ -561,7 +559,7 @@ public class TaskImpl implements Task {
if (f.getLen() == 0) {
continue;
}
- tablet = new FileFragment(name, f.getPath(), 0l, f.getLen());
+ tablet = new FileFragment(name, fs.makeQualified(f.getPath()), 0l, f.getLen());
listTablets.add(tablet);
}
@@ -569,7 +567,7 @@ public class TaskImpl implements Task {
synchronized (localChunks) {
for (FileChunk chunk : localChunks) {
if (name.equals(chunk.getEbId())) {
- tablet = new FileFragment(name, new Path(chunk.getFile().getPath()), chunk.startOffset(), chunk.length());
+ tablet = new FileFragment(name, fs.makeQualified(new Path(chunk.getFile().getPath())), chunk.startOffset(), chunk.length());
listTablets.add(tablet);
}
}
@@ -611,10 +609,9 @@ public class TaskImpl implements Task {
try {
FileChunk fetched = fetcher.get();
if (fetcher.getState() == TajoProtos.FetcherState.FETCH_FINISHED && fetched != null
- && fetched.getFile() != null) {
+ && fetched.getFile() != null) {
if (fetched.fromRemote() == false) {
- localChunks.add(fetched);
- LOG.info("Add a new FileChunk to local chunk list");
+ localChunks.add(fetched);
}
break;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java
index 55fa45d..b4d4f2b 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/stream/BufferPool.java
@@ -40,11 +40,6 @@ public class BufferPool {
}
static {
- /* TODO Enable thread cache
- * Create a pooled ByteBuf allocator but disables the thread-local cache.
- * Because the TaskRunner thread is newly created
- * */
-
if (TajoConstants.IS_TEST_MODE) {
/* Disable pooling buffers for memory usage */
ALLOCATOR = UnpooledByteBufAllocator.DEFAULT;
@@ -53,7 +48,7 @@ public class BufferPool {
ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
} else {
TajoConf tajoConf = new TajoConf();
- ALLOCATOR = createPooledByteBufAllocator(true, tajoConf.getBoolean(ALLOW_CACHE, false), 0);
+ ALLOCATOR = createPooledByteBufAllocator(true, tajoConf.getBoolean(ALLOW_CACHE, true), 0);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderManager.java
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderManager.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderManager.java
index 62df9a3..79203a2 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderManager.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderManager.java
@@ -18,6 +18,9 @@
package org.apache.tajo.storage;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileStatus;
@@ -30,6 +33,9 @@ import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.storage.rawfile.DirectRawFileWriter;
+import org.apache.tajo.tuple.memory.MemoryRowBlock;
+import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.Pair;
import java.io.IOException;
@@ -37,13 +43,15 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.*;
public class HashShuffleAppenderManager {
private static final Log LOG = LogFactory.getLog(HashShuffleAppenderManager.class);
- private Map<ExecutionBlockId, Map<Integer, PartitionAppenderMeta>> appenderMap =
- new ConcurrentHashMap<>();
+ private ConcurrentMap<ExecutionBlockId, Map<Integer, PartitionAppenderMeta>> appenderMap = Maps.newConcurrentMap();
+ private ConcurrentMap<Integer, ExecutorService> executors = Maps.newConcurrentMap(); // for parallel writing
+ private List<String> temporalPaths = Lists.newArrayList();
+
private TajoConf systemConf;
private FileSystem defaultFS;
private FileSystem localFS;
@@ -59,66 +67,88 @@ public class HashShuffleAppenderManager {
// initialize DFS and LocalFileSystems
defaultFS = TajoConf.getTajoRootDir(systemConf).getFileSystem(systemConf);
localFS = FileSystem.getLocal(systemConf);
- pageSize = systemConf.getIntVar(ConfVars.SHUFFLE_HASH_APPENDER_PAGE_VOLUME) * 1024 * 1024;
+ pageSize = systemConf.getIntVar(ConfVars.SHUFFLE_HASH_APPENDER_PAGE_VOLUME) * StorageUnit.MB;
+
+ Iterable<Path> allLocalPath = lDirAllocator.getAllLocalPathsToRead(".", systemConf);
+
+ //add async hash shuffle writer
+ for (Path path : allLocalPath) {
+ temporalPaths.add(localFS.makeQualified(path).toString());
+ executors.put(temporalPaths.size() - 1, Executors.newSingleThreadExecutor());
+ }
}
- public HashShuffleAppenderWrapper getAppender(TajoConf tajoConf, ExecutionBlockId ebId, int partId,
- TableMeta meta, Schema outSchema) throws IOException {
- synchronized (appenderMap) {
- Map<Integer, PartitionAppenderMeta> partitionAppenderMap = appenderMap.get(ebId);
+ protected int getVolumeId(Path path) {
+ int i = 0;
+ for (String rootPath : temporalPaths) {
+ if (path.toString().startsWith(rootPath)) {
+ break;
+ }
+ i++;
+ }
+ Preconditions.checkPositionIndex(i, temporalPaths.size() - 1);
+ return i;
+ }
+
+ public synchronized HashShuffleAppenderWrapper getAppender(MemoryRowBlock memoryRowBlock, ExecutionBlockId ebId,
+ int partId, TableMeta meta, Schema outSchema)
+ throws IOException {
+
+ Map<Integer, PartitionAppenderMeta> partitionAppenderMap = appenderMap.get(ebId);
+
+ if (partitionAppenderMap == null) {
+ partitionAppenderMap = new ConcurrentHashMap<>();
+ appenderMap.put(ebId, partitionAppenderMap);
+ }
- if (partitionAppenderMap == null) {
- partitionAppenderMap = new ConcurrentHashMap<>();
- appenderMap.put(ebId, partitionAppenderMap);
+ PartitionAppenderMeta partitionAppenderMeta = partitionAppenderMap.get(partId);
+ if (partitionAppenderMeta == null) {
+ Path dataFile = getDataFile(ebId, partId);
+ FileSystem fs = dataFile.getFileSystem(systemConf);
+ if (fs.exists(dataFile)) {
+ FileStatus status = fs.getFileStatus(dataFile);
+ LOG.info("File " + dataFile + " already exists, size=" + status.getLen());
}
- PartitionAppenderMeta partitionAppenderMeta = partitionAppenderMap.get(partId);
- if (partitionAppenderMeta == null) {
- Path dataFile = getDataFile(ebId, partId);
- FileSystem fs = dataFile.getFileSystem(systemConf);
- if (fs.exists(dataFile)) {
- FileStatus status = fs.getFileStatus(dataFile);
- LOG.info("File " + dataFile + " already exists, size=" + status.getLen());
- }
-
- if (!fs.exists(dataFile.getParent())) {
- fs.mkdirs(dataFile.getParent());
- }
-
- FileTablespace space = (FileTablespace) TablespaceManager.get(dataFile.toUri());
- FileAppender appender = (FileAppender) space.getAppender(meta, outSchema, dataFile);
- appender.enableStats();
- appender.init();
-
- partitionAppenderMeta = new PartitionAppenderMeta();
- partitionAppenderMeta.partId = partId;
- partitionAppenderMeta.dataFile = dataFile;
- partitionAppenderMeta.appender = new HashShuffleAppenderWrapper(ebId, partId, pageSize, appender);
- partitionAppenderMeta.appender.init();
- partitionAppenderMap.put(partId, partitionAppenderMeta);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Create Hash shuffle file(partId=" + partId + "): " + dataFile);
- }
+ if (!fs.exists(dataFile.getParent())) {
+ fs.mkdirs(dataFile.getParent());
}
- return partitionAppenderMeta.appender;
+ DirectRawFileWriter appender =
+ new DirectRawFileWriter(systemConf, null, outSchema, meta, dataFile, memoryRowBlock);
+ appender.enableStats();
+ appender.init();
+
+ partitionAppenderMeta = new PartitionAppenderMeta();
+ partitionAppenderMeta.partId = partId;
+ partitionAppenderMeta.dataFile = dataFile;
+ partitionAppenderMeta.appender =
+ new HashShuffleAppenderWrapper(ebId, partId, pageSize, appender, getVolumeId(dataFile));
+ partitionAppenderMeta.appender.init();
+ partitionAppenderMap.put(partId, partitionAppenderMeta);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Create Hash shuffle file(partId=" + partId + "): " + dataFile);
+ }
}
+
+ return partitionAppenderMeta.appender;
}
public static int getPartParentId(int partId, TajoConf tajoConf) {
- return partId % tajoConf.getIntVar(TajoConf.ConfVars.HASH_SHUFFLE_PARENT_DIRS);
+ return partId % tajoConf.getIntVar(ConfVars.SHUFFLE_HASH_PARENT_DIRS);
}
private Path getDataFile(ExecutionBlockId ebId, int partId) throws IOException {
try {
// the base dir for an output dir
String executionBlockBaseDir = ebId.getQueryId().toString() + "/output" + "/" + ebId.getId() + "/hash-shuffle";
- Path baseDirPath = localFS.makeQualified(lDirAllocator.getLocalPathForWrite(executionBlockBaseDir, systemConf));
+ Path baseDirPath = lDirAllocator.getLocalPathForWrite(executionBlockBaseDir, systemConf);
//LOG.info(ebId + "'s basedir is created (" + baseDirPath + ")");
// If EB has many partition, too many shuffle file are in single directory.
- return StorageUtil.concatPath(baseDirPath, "" + getPartParentId(partId, systemConf), "" + partId);
+ return localFS.makeQualified(
+ StorageUtil.concatPath(baseDirPath, "" + getPartParentId(partId, systemConf), "" + partId));
} catch (Exception e) {
LOG.error(e.getMessage(), e);
throw new IOException(e);
@@ -126,10 +156,7 @@ public class HashShuffleAppenderManager {
}
public List<HashShuffleIntermediate> close(ExecutionBlockId ebId) throws IOException {
- Map<Integer, PartitionAppenderMeta> partitionAppenderMap = null;
- synchronized (appenderMap) {
- partitionAppenderMap = appenderMap.remove(ebId);
- }
+ Map<Integer, PartitionAppenderMeta> partitionAppenderMap = appenderMap.remove(ebId);
if (partitionAppenderMap == null) {
LOG.info("Close HashShuffleAppenderWrapper:" + ebId + ", not a hash shuffle");
@@ -158,16 +185,43 @@ public class HashShuffleAppenderManager {
}
public void finalizeTask(TaskAttemptId taskId) {
- synchronized (appenderMap) {
- Map<Integer, PartitionAppenderMeta> partitionAppenderMap =
+ Map<Integer, PartitionAppenderMeta> partitionAppenderMap =
appenderMap.get(taskId.getTaskId().getExecutionBlockId());
- if (partitionAppenderMap == null) {
- return;
- }
+ if (partitionAppenderMap == null) {
+ return;
+ }
- for (PartitionAppenderMeta eachAppender: partitionAppenderMap.values()) {
- eachAppender.appender.taskFinished(taskId);
+ for (PartitionAppenderMeta eachAppender: partitionAppenderMap.values()) {
+ eachAppender.appender.taskFinished(taskId);
+ }
+ }
+
+ /**
+ * Asynchronously write partitions.
+ */
+ public Future<MemoryRowBlock> writePartitions(TableMeta meta, Schema schema, final TaskAttemptId taskId, int partId,
+ final MemoryRowBlock rowBlock,
+ final boolean release) throws IOException {
+
+ HashShuffleAppenderWrapper appender =
+ getAppender(rowBlock, taskId.getTaskId().getExecutionBlockId(), partId, meta, schema);
+ ExecutorService executor = executors.get(appender.getVolumeId());
+ return executor.submit(new Callable<MemoryRowBlock>() {
+ @Override
+ public MemoryRowBlock call() throws Exception {
+ appender.writeRowBlock(taskId, rowBlock);
+
+ if (release) rowBlock.release();
+ else rowBlock.clear();
+
+ return rowBlock;
}
+ });
+ }
+
+ public void shutdown() {
+ for (ExecutorService service : executors.values()) {
+ service.shutdownNow();
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderWrapper.java
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderWrapper.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderWrapper.java
index bcd4388..ccd528b 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderWrapper.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/HashShuffleAppenderWrapper.java
@@ -23,6 +23,8 @@ import org.apache.commons.logging.LogFactory;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.TaskAttemptId;
import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.storage.rawfile.DirectRawFileWriter;
+import org.apache.tajo.tuple.memory.MemoryRowBlock;
import org.apache.tajo.util.Pair;
import java.io.Closeable;
@@ -36,9 +38,10 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class HashShuffleAppenderWrapper implements Closeable {
private static Log LOG = LogFactory.getLog(HashShuffleAppenderWrapper.class);
- private FileAppender appender;
+ private DirectRawFileWriter appender;
private AtomicBoolean closed = new AtomicBoolean(false);
private int partId;
+ private int volumeId;
//<taskId,<page start offset,<task start, task end>>>
private Map<TaskAttemptId, List<Pair<Long, Pair<Integer, Integer>>>> taskTupleIndexes;
@@ -56,11 +59,13 @@ public class HashShuffleAppenderWrapper implements Closeable {
private ExecutionBlockId ebId;
- public HashShuffleAppenderWrapper(ExecutionBlockId ebId, int partId, int pageSize, FileAppender appender) {
+ public HashShuffleAppenderWrapper(ExecutionBlockId ebId, int partId, int pageSize,
+ DirectRawFileWriter appender, int volumeId) {
this.ebId = ebId;
this.partId = partId;
this.appender = appender;
this.pageSize = pageSize;
+ this.volumeId = volumeId;
}
public void init() throws IOException {
@@ -73,41 +78,36 @@ public class HashShuffleAppenderWrapper implements Closeable {
* Write multiple tuples. Each tuple is written by a FileAppender which is responsible specified partition.
* After writing if a current page exceeds pageSize, pageOffset will be added.
* @param taskId
- * @param tuples
+ * @param rowBlock
* @return written bytes
* @throws java.io.IOException
*/
- public int addTuples(TaskAttemptId taskId, List<Tuple> tuples) throws IOException {
- synchronized(appender) {
- if (closed.get()) {
- return 0;
- }
- long currentPos = appender.getOffset();
+ public MemoryRowBlock writeRowBlock(TaskAttemptId taskId, MemoryRowBlock rowBlock) throws IOException {
+ if (closed.get()) {
+ return rowBlock;
+ }
- for (Tuple eachTuple: tuples) {
- appender.addTuple(eachTuple);
- }
- long posAfterWritten = appender.getOffset();
+ appender.writeRowBlock(rowBlock);
+ appender.flush();
- int writtenBytes = (int)(posAfterWritten - currentPos);
+ int rows = rowBlock.rows();
+ long posAfterWritten = appender.getOffset();
- int nextRowNum = rowNumInPage + tuples.size();
- List<Pair<Long, Pair<Integer, Integer>>> taskIndexes = taskTupleIndexes.get(taskId);
- if (taskIndexes == null) {
- taskIndexes = new ArrayList<>();
- taskTupleIndexes.put(taskId, taskIndexes);
- }
- taskIndexes.add(
- new Pair<>(currentPage.getFirst(), new Pair(rowNumInPage, nextRowNum)));
- rowNumInPage = nextRowNum;
-
- if (posAfterWritten - currentPage.getFirst() > pageSize) {
- nextPage(posAfterWritten);
- rowNumInPage = 0;
- }
+ int nextRowNum = rowNumInPage + rows;
+ List<Pair<Long, Pair<Integer, Integer>>> taskIndexes = taskTupleIndexes.get(taskId);
+ if (taskIndexes == null) {
+ taskIndexes = new ArrayList<>();
+ taskTupleIndexes.put(taskId, taskIndexes);
+ }
+ taskIndexes.add(
+ new Pair<>(currentPage.getFirst(), new Pair(rowNumInPage, nextRowNum)));
+ rowNumInPage = nextRowNum;
- return writtenBytes;
+ if (posAfterWritten - currentPage.getFirst() > pageSize) {
+ nextPage(posAfterWritten);
+ rowNumInPage = 0;
}
+ return rowBlock;
}
public long getOffset() throws IOException {
@@ -129,42 +129,35 @@ public class HashShuffleAppenderWrapper implements Closeable {
}
public void flush() throws IOException {
- synchronized(appender) {
- if (closed.get()) {
- return;
- }
- appender.flush();
+ if (closed.get()) {
+ return;
}
+ appender.flush();
}
@Override
public void close() throws IOException {
- synchronized(appender) {
- if (closed.get()) {
- return;
- }
- appender.flush();
- offset = appender.getOffset();
- if (offset > currentPage.getFirst()) {
- nextPage(offset);
- }
- appender.close();
- if (LOG.isDebugEnabled()) {
- if (!pages.isEmpty()) {
- LOG.info(ebId + ",partId=" + partId + " Appender closed: fileLen=" + offset + ", pages=" + pages.size()
- + ", lastPage=" + pages.get(pages.size() - 1));
- } else {
- LOG.info(ebId + ",partId=" + partId + " Appender closed: fileLen=" + offset + ", pages=" + pages.size());
- }
+ if (closed.getAndSet(true)) {
+ return;
+ }
+ appender.flush();
+ offset = appender.getOffset();
+ if (offset > currentPage.getFirst()) {
+ nextPage(offset);
+ }
+ appender.close();
+ if (LOG.isDebugEnabled()) {
+ if (!pages.isEmpty()) {
+ LOG.info(ebId + ",partId=" + partId + " Appender closed: fileLen=" + offset + ", pages=" + pages.size()
+ + ", lastPage=" + pages.get(pages.size() - 1));
+ } else {
+ LOG.info(ebId + ",partId=" + partId + " Appender closed: fileLen=" + offset + ", pages=" + pages.size());
}
- closed.set(true);
}
}
public TableStats getStats() {
- synchronized(appender) {
- return appender.getStats();
- }
+ return appender.getStats();
}
public List<Pair<Long, Integer>> getPages() {
@@ -184,4 +177,8 @@ public class HashShuffleAppenderWrapper implements Closeable {
public void taskFinished(TaskAttemptId taskId) {
taskTupleIndexes.remove(taskId);
}
+
+ public int getVolumeId() {
+ return volumeId;
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java
index cda39f9..26bd135 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java
@@ -36,14 +36,15 @@ import org.apache.tajo.exception.TajoRuntimeException;
import org.apache.tajo.exception.UnsupportedException;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.storage.fragment.Fragment;
+import org.apache.tajo.storage.rawfile.DirectRawFileWriter;
import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.BitArray;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
-import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
import java.nio.channels.FileChannel;
public class RawFile {
@@ -99,7 +100,7 @@ public class RawFile {
}
if(buf == null) {
- buf = BufferPool.directBuffer(conf.getInt(READ_BUFFER_SIZE, DEFAULT_BUFFER_SIZE));
+ buf = BufferPool.directBuffer(conf.getInt(READ_BUFFER_SIZE, DEFAULT_BUFFER_SIZE)).order(ByteOrder.LITTLE_ENDIAN);
buffer = buf.nioBuffer(0, buf.capacity());
}
@@ -282,96 +283,94 @@ public class RawFile {
}
switch (columnTypes[i].getType()) {
- case BOOLEAN :
- outTuple.put(i, DatumFactory.createBool(buffer.get()));
- break;
-
- case BIT :
- outTuple.put(i, DatumFactory.createBit(buffer.get()));
- break;
-
- case CHAR :
- int realLen = readRawVarint32();
- byte[] buf = new byte[realLen];
- buffer.get(buf);
- outTuple.put(i, DatumFactory.createChar(buf));
- break;
-
- case INT2 :
- outTuple.put(i, DatumFactory.createInt2(buffer.getShort()));
- break;
-
- case INT4 :
- outTuple.put(i, DatumFactory.createInt4(decodeZigZag32(readRawVarint32())));
- break;
-
- case INT8 :
- outTuple.put(i, DatumFactory.createInt8(decodeZigZag64(readRawVarint64())));
- break;
-
- case FLOAT4 :
- outTuple.put(i, DatumFactory.createFloat4(buffer.getFloat()));
- break;
-
- case FLOAT8 :
- outTuple.put(i, DatumFactory.createFloat8(buffer.getDouble()));
- break;
-
- case TEXT : {
- int len = readRawVarint32();
- byte [] strBytes = new byte[len];
- buffer.get(strBytes);
- outTuple.put(i, DatumFactory.createText(strBytes));
- break;
- }
+ case BOOLEAN:
+ outTuple.put(i, DatumFactory.createBool(buffer.get()));
+ break;
- case BLOB : {
- int len = readRawVarint32();
- byte [] rawBytes = new byte[len];
- buffer.get(rawBytes);
- outTuple.put(i, DatumFactory.createBlob(rawBytes));
- break;
- }
+ case BIT:
+ outTuple.put(i, DatumFactory.createBit(buffer.get()));
+ break;
- case PROTOBUF: {
- int len = readRawVarint32();
- byte [] rawBytes = new byte[len];
- buffer.get(rawBytes);
+ case CHAR:
+ int realLen = readRawVarint32();
+ byte[] buf = new byte[realLen];
+ buffer.get(buf);
+ outTuple.put(i, DatumFactory.createChar(buf));
+ break;
- outTuple.put(i, ProtobufDatumFactory.createDatum(columnTypes[i], rawBytes));
- break;
- }
+ case INT2:
+ outTuple.put(i, DatumFactory.createInt2(buffer.getShort()));
+ break;
- case INET4 :
- byte [] ipv4Bytes = new byte[4];
- buffer.get(ipv4Bytes);
- outTuple.put(i, DatumFactory.createInet4(ipv4Bytes));
- break;
-
- case DATE: {
- int val = buffer.getInt();
- if (val < Integer.MIN_VALUE + 1) {
- outTuple.put(i, DatumFactory.createNullDatum());
- } else {
- outTuple.put(i, DatumFactory.createFromInt4(columnTypes[i], val));
- }
- break;
+ case INT4:
+ outTuple.put(i, DatumFactory.createInt4(decodeZigZag32(readRawVarint32())));
+ break;
+
+ case INT8:
+ outTuple.put(i, DatumFactory.createInt8(decodeZigZag64(readRawVarint64())));
+ break;
+
+ case FLOAT4:
+ outTuple.put(i, DatumFactory.createFloat4(buffer.getFloat()));
+ break;
+
+ case FLOAT8:
+ outTuple.put(i, DatumFactory.createFloat8(buffer.getDouble()));
+ break;
+
+ case TEXT: {
+ int len = readRawVarint32();
+ byte[] strBytes = new byte[len];
+ buffer.get(strBytes);
+ outTuple.put(i, DatumFactory.createText(strBytes));
+ break;
+ }
+
+ case BLOB: {
+ int len = readRawVarint32();
+ byte[] rawBytes = new byte[len];
+ buffer.get(rawBytes);
+ outTuple.put(i, DatumFactory.createBlob(rawBytes));
+ break;
+ }
+
+ case PROTOBUF: {
+ int len = readRawVarint32();
+ byte[] rawBytes = new byte[len];
+ buffer.get(rawBytes);
+
+ outTuple.put(i, ProtobufDatumFactory.createDatum(columnTypes[i], rawBytes));
+ break;
+ }
+
+ case INET4:
+ outTuple.put(i, DatumFactory.createInet4(buffer.getInt()));
+ break;
+
+ case DATE: {
+ int val = buffer.getInt();
+ if (val < Integer.MIN_VALUE + 1) {
+ outTuple.put(i, DatumFactory.createNullDatum());
+ } else {
+ outTuple.put(i, DatumFactory.createFromInt4(columnTypes[i], val));
}
- case TIME:
- case TIMESTAMP: {
- long val = buffer.getLong();
- if (val < Long.MIN_VALUE + 1) {
- outTuple.put(i, DatumFactory.createNullDatum());
- } else {
- outTuple.put(i, DatumFactory.createFromInt8(columnTypes[i], val));
- }
- break;
+ break;
+ }
+ case TIME:
+ case TIMESTAMP: {
+ long val = buffer.getLong();
+ if (val < Long.MIN_VALUE + 1) {
+ outTuple.put(i, DatumFactory.createNullDatum());
+ } else {
+ outTuple.put(i, DatumFactory.createFromInt8(columnTypes[i], val));
}
- case NULL_TYPE:
- outTuple.put(i, NullDatum.get());
- break;
+ break;
+ }
+ case NULL_TYPE:
+ outTuple.put(i, NullDatum.get());
+ break;
- default:
+ default:
}
}
@@ -400,6 +399,7 @@ public class RawFile {
buffer.clear();
forceFillBuffer = true;
filePosition = fragment.getStartKey();
+ recordCount = 0;
channel.position(filePosition);
eos = false;
}
@@ -461,331 +461,11 @@ public class RawFile {
}
}
- public static class RawFileAppender extends FileAppender {
- private FileChannel channel;
- private RandomAccessFile randomAccessFile;
- private DataType[] columnTypes;
-
- private ByteBuffer buffer;
- private ByteBuf buf;
- private BitArray nullFlags;
- private int headerSize = 0;
- private static final int RECORD_SIZE = 4;
- private long pos;
-
- private TableStatistics stats;
-
- public RawFileAppender(Configuration conf, TaskAttemptId taskAttemptId,
- Schema schema, TableMeta meta, Path workDir) throws IOException {
- super(conf, taskAttemptId, schema, meta, workDir);
- }
-
- public void init() throws IOException {
- File file;
- try {
- if (path.toUri().getScheme() != null) {
- file = new File(path.toUri());
- } else {
- file = new File(path.toString());
- }
- } catch (IllegalArgumentException iae) {
- throw new IOException(iae);
- }
-
- randomAccessFile = new RandomAccessFile(file, "rw");
- channel = randomAccessFile.getChannel();
- pos = 0;
-
- columnTypes = new DataType[schema.size()];
- for (int i = 0; i < schema.size(); i++) {
- columnTypes[i] = schema.getColumn(i).getDataType();
- }
-
- buf = BufferPool.directBuffer(conf.getInt(WRITE_BUFFER_SIZE, DEFAULT_BUFFER_SIZE));
- buffer = buf.nioBuffer(0, buf.capacity());
-
- // comput the number of bytes, representing the null flags
-
- nullFlags = new BitArray(schema.size());
- headerSize = RECORD_SIZE + 2 + nullFlags.bytesLength();
-
- if (tableStatsEnabled) {
- this.stats = new TableStatistics(this.schema, columnStatsEnabled);
- }
-
- super.init();
- }
-
- @Override
- public long getOffset() throws IOException {
- return pos;
- }
-
- private void flushBuffer() throws IOException {
- buffer.flip();
- channel.write(buffer);
- buffer.clear();
- }
-
- private boolean flushBufferAndReplace(int recordOffset, int sizeToBeWritten)
- throws IOException {
-
- // if the buffer reaches the limit,
- // write the bytes from 0 to the previous record.
- if (buffer.remaining() < sizeToBeWritten) {
-
- int limit = buffer.position();
- buffer.limit(recordOffset);
- buffer.flip();
- channel.write(buffer);
- buffer.position(recordOffset);
- buffer.limit(limit);
- buffer.compact();
-
- //increase the write-buffer
- if(buffer.remaining() < sizeToBeWritten) {
- buf.setIndex(buffer.position(), buffer.limit());
- buf.ensureWritable(sizeToBeWritten);
- buffer = buf.nioBuffer(0, buf.capacity());
- buffer.position(buf.readerIndex());
- }
- return true;
- } else {
- return false;
- }
- }
-
- /**
- * Encode a ZigZag-encoded 32-bit value. ZigZag encodes signed integers
- * into values that can be efficiently encoded with varint. (Otherwise,
- * negative values must be sign-extended to 64 bits to be varint encoded,
- * thus always taking 10 bytes on the wire.)
- *
- * @param n A signed 32-bit integer.
- * @return An unsigned 32-bit integer, stored in a signed int because
- * Java has no explicit unsigned support.
- */
- public static int encodeZigZag32(final int n) {
- // Note: the right-shift must be arithmetic
- return (n << 1) ^ (n >> 31);
- }
-
- /**
- * Encode a ZigZag-encoded 64-bit value. ZigZag encodes signed integers
- * into values that can be efficiently encoded with varint. (Otherwise,
- * negative values must be sign-extended to 64 bits to be varint encoded,
- * thus always taking 10 bytes on the wire.)
- *
- * @param n A signed 64-bit integer.
- * @return An unsigned 64-bit integer, stored in a signed int because
- * Java has no explicit unsigned support.
- */
- public static long encodeZigZag64(final long n) {
- // Note: the right-shift must be arithmetic
- return (n << 1) ^ (n >> 63);
- }
-
- /**
- * Encode and write a varint. {@code value} is treated as
- * unsigned, so it won't be sign-extended if negative.
- */
- public void writeRawVarint32(int value) throws IOException {
- while (true) {
- if ((value & ~0x7F) == 0) {
- buffer.put((byte) value);
- return;
- } else {
- buffer.put((byte) ((value & 0x7F) | 0x80));
- value >>>= 7;
- }
- }
- }
-
- /**
- * Compute the number of bytes that would be needed to encode a varint.
- * {@code value} is treated as unsigned, so it won't be sign-extended if
- * negative.
- */
- public static int computeRawVarint32Size(final int value) {
- if ((value & (0xffffffff << 7)) == 0) return 1;
- if ((value & (0xffffffff << 14)) == 0) return 2;
- if ((value & (0xffffffff << 21)) == 0) return 3;
- if ((value & (0xffffffff << 28)) == 0) return 4;
- return 5;
- }
-
- /** Encode and write a varint. */
- public void writeRawVarint64(long value) throws IOException {
- while (true) {
- if ((value & ~0x7FL) == 0) {
- buffer.put((byte) value);
- return;
- } else {
- buffer.put((byte) ((value & 0x7F) | 0x80));
- value >>>= 7;
- }
- }
- }
-
- @Override
- public void addTuple(Tuple t) throws IOException {
-
- if (buffer.remaining() < headerSize) {
- flushBuffer();
- }
-
- // skip the row header
- int recordOffset = buffer.position();
- buffer.position(recordOffset + headerSize);
- // reset the null flags
- nullFlags.clear();
- for (int i = 0; i < schema.size(); i++) {
- // it is to calculate min/max values, and it is only used for the intermediate file.
- if (tableStatsEnabled) {
- stats.analyzeField(i, t);
- }
-
- if (t.isBlankOrNull(i)) {
- nullFlags.set(i);
- continue;
- }
-
- // 10 is the maximum bytes size of all types
- if (flushBufferAndReplace(recordOffset, 10)) {
- recordOffset = 0;
- }
-
- switch(columnTypes[i].getType()) {
- case NULL_TYPE:
- nullFlags.set(i);
- continue;
-
- case BOOLEAN:
- case BIT:
- buffer.put(t.getByte(i));
- break;
-
- case INT2 :
- buffer.putShort(t.getInt2(i));
- break;
-
- case INT4 :
- writeRawVarint32(encodeZigZag32(t.getInt4(i)));
- break;
-
- case INT8 :
- writeRawVarint64(encodeZigZag64(t.getInt8(i)));
- break;
-
- case FLOAT4 :
- buffer.putFloat(t.getFloat4(i));
- break;
-
- case FLOAT8 :
- buffer.putDouble(t.getFloat8(i));
- break;
-
- case CHAR:
- case TEXT: {
- byte [] strBytes = t.getBytes(i);
- if (flushBufferAndReplace(recordOffset, strBytes.length + computeRawVarint32Size(strBytes.length))) {
- recordOffset = 0;
- }
- writeRawVarint32(strBytes.length);
- buffer.put(strBytes);
- break;
- }
+ public static class RawFileAppender extends DirectRawFileWriter {
- case DATE:
- buffer.putInt(t.getInt4(i));
- break;
-
- case TIME:
- case TIMESTAMP:
- buffer.putLong(t.getInt8(i));
- break;
-
- case BLOB : {
- byte [] rawBytes = t.getBytes(i);
- if (flushBufferAndReplace(recordOffset, rawBytes.length + computeRawVarint32Size(rawBytes.length))) {
- recordOffset = 0;
- }
- writeRawVarint32(rawBytes.length);
- buffer.put(rawBytes);
- break;
- }
-
- case PROTOBUF: {
- byte [] rawBytes = t.getBytes(i);
- if (flushBufferAndReplace(recordOffset, rawBytes.length + computeRawVarint32Size(rawBytes.length))) {
- recordOffset = 0;
- }
- writeRawVarint32(rawBytes.length);
- buffer.put(rawBytes);
- break;
- }
-
- case INET4 :
- buffer.put(t.getBytes(i));
- break;
-
- default:
- throw new IOException("Cannot support data type: " + columnTypes[i].getType());
- }
- }
-
- // write a record header
- int bufferPos = buffer.position();
- buffer.position(recordOffset);
- buffer.putInt(bufferPos - recordOffset);
- byte [] flags = nullFlags.toArray();
- buffer.putShort((short) flags.length);
- buffer.put(flags);
-
- pos += bufferPos - recordOffset;
- buffer.position(bufferPos);
-
- if (tableStatsEnabled) {
- stats.incrementRow();
- }
- }
-
- @Override
- public void flush() throws IOException {
- if(buffer != null){
- flushBuffer();
- }
- }
-
- @Override
- public void close() throws IOException {
- flush();
- if (tableStatsEnabled) {
- stats.setNumBytes(getOffset());
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("RawFileAppender written: " + getOffset() + " bytes, path: " + path);
- }
-
- if(buf != null){
- buffer.clear();
- buffer = null;
-
- buf.release();
- buf = null;
- }
-
- IOUtils.cleanup(LOG, channel, randomAccessFile);
- }
-
- @Override
- public TableStats getStats() {
- if (tableStatsEnabled) {
- stats.setNumBytes(pos);
- return stats.getTableStat();
- } else {
- return null;
- }
+ public RawFileAppender(Configuration conf, TaskAttemptId taskAttemptId, Schema schema,
+ TableMeta meta, Path workDir) throws IOException {
+ super(conf, taskAttemptId, schema, meta, workDir, null);
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileScanner.java
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileScanner.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileScanner.java
index 0172484..550de63 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileScanner.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileScanner.java
@@ -35,7 +35,6 @@ import org.apache.tajo.tuple.RowBlockReader;
import org.apache.tajo.tuple.memory.MemoryRowBlock;
import org.apache.tajo.tuple.memory.RowBlock;
import org.apache.tajo.tuple.memory.UnSafeTuple;
-import org.apache.tajo.tuple.memory.ZeroCopyTuple;
import org.apache.tajo.unit.StorageUnit;
import java.io.File;
@@ -45,6 +44,9 @@ import java.io.IOException;
public class DirectRawFileScanner extends FileScanner implements SeekableScanner {
private static final Log LOG = LogFactory.getLog(DirectRawFileScanner.class);
+ public static final String READ_BUFFER_SIZE = "tajo.storage.raw.io.read-buffer.bytes";
+ public static final int DEFAULT_BUFFER_SIZE = 128 * StorageUnit.KB;
+
private SeekableInputChannel channel;
private boolean eos = false;
@@ -53,7 +55,7 @@ public class DirectRawFileScanner extends FileScanner implements SeekableScanner
private long filePosition;
private long endOffset;
- private ZeroCopyTuple unSafeTuple = new UnSafeTuple();
+ private UnSafeTuple unSafeTuple = new UnSafeTuple();
private RowBlock tupleBuffer;
private RowBlockReader reader;
@@ -61,15 +63,19 @@ public class DirectRawFileScanner extends FileScanner implements SeekableScanner
super(conf, schema, meta, fragment);
}
+ @Override
public void init() throws IOException {
initChannel();
- tupleBuffer = new MemoryRowBlock(SchemaUtil.toDataTypes(schema), 64 * StorageUnit.KB, true);
-
- reader = tupleBuffer.getReader();
-
- fetchNeeded = !next(tupleBuffer);
+ if (tupleBuffer == null) {
+ tupleBuffer = new MemoryRowBlock(SchemaUtil.toDataTypes(schema),
+ conf.getInt(READ_BUFFER_SIZE, DEFAULT_BUFFER_SIZE));
+ } else {
+ tupleBuffer.clear();
+ }
+ fetchNeeded = true;
+ eos = false;
super.init();
}
@@ -121,12 +127,12 @@ public class DirectRawFileScanner extends FileScanner implements SeekableScanner
public void seek(long offset) throws IOException {
channel.seek(offset);
filePosition = channel.position();
- tupleBuffer.getMemory().clear();
+ tupleBuffer.clear();
fetchNeeded = true;
}
public boolean next(RowBlock rowblock) throws IOException {
- long reamin = reader.remainForRead();
+ long reamin = reader == null ? 0 : reader.remainForRead();
boolean ret = rowblock.copyFromChannel(channel);
reader = rowblock.getReader();
filePosition += rowblock.getMemory().writerPosition() - reamin;
@@ -136,7 +142,7 @@ public class DirectRawFileScanner extends FileScanner implements SeekableScanner
private boolean fetchNeeded = true;
@Override
- public Tuple next() throws IOException {
+ public UnSafeTuple next() throws IOException {
if(eos) {
return null;
}
@@ -164,9 +170,9 @@ public class DirectRawFileScanner extends FileScanner implements SeekableScanner
public void reset() throws IOException {
// reload initial buffer
filePosition = fragment.getStartKey();
+ recordCount = 0;
seek(filePosition);
eos = false;
- reader.reset();
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileWriter.java
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileWriter.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileWriter.java
index 23ef059..9cbb7a0 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileWriter.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rawfile/DirectRawFileWriter.java
@@ -26,11 +26,14 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
+import org.apache.tajo.BuiltinStorages;
import org.apache.tajo.TaskAttemptId;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.exception.TajoInternalError;
+import org.apache.tajo.exception.UnsupportedException;
import org.apache.tajo.plan.serder.PlanProto.ShuffleType;
import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.storage.FileAppender;
@@ -38,6 +41,9 @@ import org.apache.tajo.storage.StorageConstants;
import org.apache.tajo.storage.TableStatistics;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.tuple.memory.MemoryRowBlock;
+import org.apache.tajo.tuple.memory.OffHeapRowBlockUtils.TupleConverter;
+import org.apache.tajo.tuple.memory.RowWriter;
+import org.apache.tajo.tuple.memory.UnSafeTuple;
import org.apache.tajo.unit.StorageUnit;
import java.io.File;
@@ -46,23 +52,36 @@ import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;
public class DirectRawFileWriter extends FileAppender {
- public static final String FILE_EXTENSION = "draw";
- private static final int BUFFER_SIZE = 64 * StorageUnit.KB;
private static final Log LOG = LogFactory.getLog(DirectRawFileWriter.class);
- private FileChannel channel;
- private RandomAccessFile randomAccessFile;
- private FSDataOutputStream fos;
- private boolean isLocal;
- private long pos;
+ public static final String WRITE_BUFFER_SIZE = "tajo.storage.raw.io.write-buffer.bytes";
+ public static final int DEFAULT_BUFFER_SIZE = 128 * StorageUnit.KB;
+ private static final float OVERFLOW_RATIO = 1.1f;
+ protected FileChannel channel;
- private TableStatistics stats;
- private ShuffleType shuffleType;
- private MemoryRowBlock memoryRowBlock;
+ protected RandomAccessFile randomAccessFile;
+ protected FSDataOutputStream fos;
+ protected long pos;
+ protected TableStatistics stats;
+
+ protected TupleConverter tupleConverter;
+ protected MemoryRowBlock rowBlock;
+ protected boolean analyzeField;
+ protected boolean hasExternalBuf;
+ protected boolean isLocal;
+
+ public DirectRawFileWriter(Configuration conf, TaskAttemptId taskAttemptId,
+ final Schema schema, final TableMeta meta, final Path path)
+ throws IOException {
+ this(conf, taskAttemptId, schema, meta, path, null);
+ }
public DirectRawFileWriter(Configuration conf, TaskAttemptId taskAttemptId,
- final Schema schema, final TableMeta meta, final Path path) throws IOException {
+ final Schema schema, final TableMeta meta, final Path path,
+ MemoryRowBlock rowBlock) throws IOException {
super(conf, taskAttemptId, schema, meta, path);
+ this.rowBlock = rowBlock;
+ this.hasExternalBuf = rowBlock != null;
}
@Override
@@ -91,19 +110,89 @@ public class DirectRawFileWriter extends FileAppender {
if (tableStatsEnabled) {
this.stats = new TableStatistics(this.schema, columnStatsEnabled);
- this.shuffleType = PlannerUtil.getShuffleType(
+ if (ShuffleType.RANGE_SHUFFLE == PlannerUtil.getShuffleType(
meta.getOption(StorageConstants.SHUFFLE_TYPE,
- PlannerUtil.getShuffleType(ShuffleType.NONE_SHUFFLE)));
+ PlannerUtil.getShuffleType(ShuffleType.NONE_SHUFFLE)))) {
+ this.analyzeField = true;
+ }
+ }
+
+ if (rowBlock == null) {
+ int bufferSize = (int) (conf.getInt(WRITE_BUFFER_SIZE, DEFAULT_BUFFER_SIZE) * OVERFLOW_RATIO);
+ rowBlock = new MemoryRowBlock(SchemaUtil.toDataTypes(schema), bufferSize, true, meta.getDataFormat());
}
- memoryRowBlock = new MemoryRowBlock(SchemaUtil.toDataTypes(schema), BUFFER_SIZE, true);
+ tupleConverter = initConverter();
+
pos = 0;
super.init();
}
+ public TupleConverter initConverter() {
+ switch (meta.getDataFormat()) {
+ case BuiltinStorages.DRAW:
+ return getDrawConverter();
+ case BuiltinStorages.RAW:
+ return getRawConverter();
+ default:
+ throw new TajoInternalError(new UnsupportedException());
+ }
+ }
+
+ private TupleConverter getDrawConverter() {
+ return new TupleConverter() {
+
+ @Override
+ public void convert(Tuple tuple, RowWriter writer) {
+ if (analyzeField) {
+ if (tuple instanceof UnSafeTuple) {
+
+ for (int i = 0; i < writer.dataTypes().length; i++) {
+ // it is to calculate min/max values, and it is only used for the intermediate file.
+ stats.analyzeField(i, tuple);
+ }
+ // write direct to memory
+ writer.addTuple(tuple);
+ } else {
+ writer.startRow();
+
+ for (int i = 0; i < writer.dataTypes().length; i++) {
+ // it is to calculate min/max values, and it is only used for the intermediate file.
+ stats.analyzeField(i, tuple);
+ writeField(i, tuple, writer);
+ }
+ writer.endRow();
+ }
+ } else {
+ // write direct to memory
+ writer.addTuple(tuple);
+ }
+ }
+ };
+ }
+
+ private TupleConverter getRawConverter() {
+ return new TupleConverter() {
+
+ @Override
+ public void convert(Tuple tuple, RowWriter writer) {
+ writer.startRow();
+
+ for (int i = 0; i < writer.dataTypes().length; i++) {
+ // it is to calculate min/max values, and it is only used for the intermediate file.
+ if (analyzeField) {
+ stats.analyzeField(i, tuple);
+ }
+ writeField(i, tuple, writer);
+ }
+ writer.endRow();
+ }
+ };
+ }
+
@Override
public long getOffset() throws IOException {
- return pos + memoryRowBlock.getMemory().writerPosition();
+ return hasExternalBuf ? pos : pos + rowBlock.getMemory().writerPosition();
}
public void writeRowBlock(MemoryRowBlock rowBlock) throws IOException {
@@ -113,33 +202,27 @@ public class DirectRawFileWriter extends FileAppender {
pos += rowBlock.getMemory().writeTo(fos);
}
- rowBlock.getMemory().clear();
-
if (tableStatsEnabled) {
- stats.incrementRows(rowBlock.rows() - stats.getNumRows());
+ stats.incrementRows(rowBlock.rows());
}
}
@Override
public void addTuple(Tuple t) throws IOException {
- if (shuffleType == ShuffleType.RANGE_SHUFFLE) {
- // it is to calculate min/max values, and it is only used for the intermediate file.
- for (int i = 0; i < schema.size(); i++) {
- stats.analyzeField(i, t);
- }
- }
- memoryRowBlock.getWriter().addTuple(t);
+ tupleConverter.convert(t, rowBlock.getWriter());
- if(memoryRowBlock.getMemory().readableBytes() >= BUFFER_SIZE) {
- writeRowBlock(memoryRowBlock);
+ if(rowBlock.usedMem() > DEFAULT_BUFFER_SIZE) {
+ writeRowBlock(rowBlock);
+ rowBlock.clear();
}
}
@Override
public void flush() throws IOException {
- if(memoryRowBlock.getMemory().isReadable()) {
- writeRowBlock(memoryRowBlock);
+ if(!hasExternalBuf && rowBlock.getMemory().isReadable()) {
+ writeRowBlock(rowBlock);
+ rowBlock.clear();
}
}
@@ -155,7 +238,9 @@ public class DirectRawFileWriter extends FileAppender {
}
IOUtils.cleanup(LOG, channel, randomAccessFile, fos);
- memoryRowBlock.release();
+ if(!hasExternalBuf && rowBlock != null) {
+ rowBlock.release();
+ }
}
@Override
[2/2] tajo git commit: TAJO-1271: Improve memory usage of
Hash-shuffle.
Posted by jh...@apache.org.
TAJO-1271: Improve memory usage of Hash-shuffle.
Closes #837
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/411a26d5
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/411a26d5
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/411a26d5
Branch: refs/heads/master
Commit: 411a26d5d45fb466d4c3c22806c67aedde14b623
Parents: 10159c7
Author: Jinho Kim <jh...@apache.org>
Authored: Tue Nov 17 12:00:43 2015 +0900
Committer: Jinho Kim <jh...@apache.org>
Committed: Tue Nov 17 12:00:43 2015 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../org/apache/tajo/TajoTestingCluster.java | 5 +-
.../main/java/org/apache/tajo/SessionVars.java | 4 +-
.../java/org/apache/tajo/conf/TajoConf.java | 10 +-
.../org/apache/tajo/storage/BufferPool.java | 6 +-
.../org/apache/tajo/tuple/BaseTupleBuilder.java | 2 +-
.../tuple/memory/CompactRowBlockWriter.java | 408 +++++++++++++++
.../tajo/tuple/memory/DirectBufTuple.java | 6 +-
.../tajo/tuple/memory/HeapRowBlockReader.java | 5 +-
.../org/apache/tajo/tuple/memory/HeapTuple.java | 11 +-
.../tajo/tuple/memory/MemoryRowBlock.java | 81 ++-
.../tuple/memory/OffHeapRowBlockReader.java | 8 +-
.../tajo/tuple/memory/OffHeapRowBlockUtils.java | 133 +++--
.../tajo/tuple/memory/ResizableLimitSpec.java | 11 +-
.../tajo/tuple/memory/ResizableMemoryBlock.java | 18 +-
.../org/apache/tajo/tuple/memory/RowBlock.java | 20 +
.../apache/tajo/tuple/memory/UnSafeTuple.java | 24 +-
.../apache/tajo/tuple/memory/ZeroCopyTuple.java | 10 +-
.../java/org/apache/tajo/util/BitArray.java | 5 +-
tajo-core-tests/pom.xml | 2 +-
.../planner/physical/TestExternalSortExec.java | 2 +-
.../planner/physical/TestHashJoinExec.java | 1 +
.../planner/physical/TestPhysicalPlanner.java | 3 +
.../physical/TestProgressExternalSortExec.java | 41 +-
.../tajo/querymaster/TestTaskStatusUpdate.java | 2 +-
.../TestTajoCli/testHelpSessionVars.result | 3 +-
.../engine/planner/PhysicalPlannerImpl.java | 4 +-
.../planner/physical/ExternalSortExec.java | 117 +++--
.../physical/HashShuffleFileWriteExec.java | 201 +++++---
.../tajo/engine/planner/physical/SortExec.java | 3 +-
.../engine/planner/physical/TupleSorter.java | 5 +-
.../planner/physical/VectorizedSorter.java | 3 +-
.../apache/tajo/querymaster/Repartitioner.java | 5 +-
.../java/org/apache/tajo/worker/TajoWorker.java | 4 +
.../java/org/apache/tajo/worker/TaskImpl.java | 15 +-
.../tajo/plan/function/stream/BufferPool.java | 7 +-
.../storage/HashShuffleAppenderManager.java | 162 ++++--
.../storage/HashShuffleAppenderWrapper.java | 107 ++--
.../java/org/apache/tajo/storage/RawFile.java | 498 ++++---------------
.../storage/rawfile/DirectRawFileScanner.java | 28 +-
.../storage/rawfile/DirectRawFileWriter.java | 145 ++++--
41 files changed, 1282 insertions(+), 845 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 14d6a82..10fa37d 100644
--- a/CHANGES
+++ b/CHANGES
@@ -8,6 +8,8 @@ Release 0.12.0 - unreleased
IMPROVEMENT
+ TAJO-1271: Improve memory usage of Hash-shuffle. (jinho)
+
TAJO-1966: Decrease memory usage of TajoTestingCluster. (jinho)
TAJO-1941: PermGen elimination in JDK 8. (Dongkyu Hwangbo via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-cluster-tests/src/test/java/org/apache/tajo/TajoTestingCluster.java
----------------------------------------------------------------------
diff --git a/tajo-cluster-tests/src/test/java/org/apache/tajo/TajoTestingCluster.java b/tajo-cluster-tests/src/test/java/org/apache/tajo/TajoTestingCluster.java
index 046a224..9e0e060 100644
--- a/tajo-cluster-tests/src/test/java/org/apache/tajo/TajoTestingCluster.java
+++ b/tajo-cluster-tests/src/test/java/org/apache/tajo/TajoTestingCluster.java
@@ -157,8 +157,9 @@ public class TajoTestingCluster {
// Python function path
conf.setStrings(ConfVars.PYTHON_CODE_DIR.varname, getClass().getResource("/python").toString());
- // Query output file
- conf.setVar(ConfVars.QUERY_OUTPUT_DEFAULT_FILE_FORMAT, BuiltinStorages.DRAW);
+ // Buffer size
+ conf.setInt(ConfVars.$EXECUTOR_EXTERNAL_SORT_BUFFER_SIZE.varname, 1);
+ conf.setInt(ConfVars.$EXECUTOR_HASH_SHUFFLE_BUFFER_SIZE.varname, 1);
/** decrease Hbase thread and memory cache for testing */
//server handler
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/SessionVars.java b/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
index 46df687..08c12a0 100644
--- a/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
+++ b/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
@@ -117,7 +117,9 @@ public enum SessionVars implements ConfigKey {
// for physical Executors
EXTSORT_BUFFER_SIZE(ConfVars.$EXECUTOR_EXTERNAL_SORT_BUFFER_SIZE, "sort buffer size for external sort (mb)", DEFAULT,
- Long.class, Validators.min("0")),
+ Integer.class, Validators.min("0")),
+ HASH_SHUFFLE_BUFFER_SIZE(ConfVars.$EXECUTOR_HASH_SHUFFLE_BUFFER_SIZE, "hash-shuffle buffer size for local disk I/O (mb)"
+ , DEFAULT, Integer.class, Validators.min("1")),
HASH_JOIN_SIZE_LIMIT(ConfVars.$EXECUTOR_HASH_JOIN_SIZE_THRESHOLD, "limited size for hash join (mb)", DEFAULT,
Long.class, Validators.min("0")),
INNER_HASH_JOIN_SIZE_LIMIT(ConfVars.$EXECUTOR_INNER_HASH_JOIN_SIZE_THRESHOLD,
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index de52e53..a2c1fb8 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -213,12 +213,11 @@ public class TajoConf extends Configuration {
SHUFFLE_FETCHER_CONNECT_TIMEOUT("tajo.shuffle.fetcher.connect.timeout-sec", 60, Validators.min("1")),
SHUFFLE_FETCHER_READ_TIMEOUT("tajo.shuffle.fetcher.read.timeout-sec", 60, Validators.min("1")),
SHUFFLE_FETCHER_READ_RETRY_MAX_NUM("tajo.shuffle.fetcher.read.retry.max-num", 2, Validators.min("0")),
- SHUFFLE_HASH_APPENDER_BUFFER_SIZE("tajo.shuffle.hash.appender.buffer.size", 10000),
- SHUFFLE_HASH_APPENDER_PAGE_VOLUME("tajo.shuffle.hash.appender.page.volumn-mb", 30),
- HASH_SHUFFLE_PARENT_DIRS("tajo.hash.shuffle.parent.dirs.count", 10),
+ SHUFFLE_HASH_APPENDER_PAGE_VOLUME("tajo.shuffle.hash.appender.page.volume-mb", 30),
+ SHUFFLE_HASH_PARENT_DIRS("tajo.shuffle.hash.parent.dirs.count", 64),
// Query output Configuration --------------------------------------------------
- QUERY_OUTPUT_DEFAULT_FILE_FORMAT("tajo.query.output.file-format", BuiltinStorages.TEXT, Validators.javaString()),
+ QUERY_OUTPUT_DEFAULT_FILE_FORMAT("tajo.query.output.file-format", BuiltinStorages.DRAW, Validators.javaString()),
// Storage Configuration --------------------------------------------------
ROWFILE_SYNC_INTERVAL("rowfile.sync.interval", 100),
@@ -334,7 +333,7 @@ public class TajoConf extends Configuration {
$QUERY_EXECUTE_PARALLEL_MAX("tajo.query.execute.parallel.max", 10),
// for physical Executors
- $EXECUTOR_EXTERNAL_SORT_BUFFER_SIZE("tajo.executor.external-sort.buffer-mb", 200L),
+ $EXECUTOR_EXTERNAL_SORT_BUFFER_SIZE("tajo.executor.external-sort.buffer-mb", 200),
$EXECUTOR_HASH_JOIN_SIZE_THRESHOLD("tajo.executor.join.common.in-memory-hash-threshold-mb", 64l, Validators.min("0")),
$EXECUTOR_INNER_HASH_JOIN_SIZE_THRESHOLD("tajo.executor.join.inner.in-memory-hash-threshold-mb", 64l,
Validators.min("0")),
@@ -342,6 +341,7 @@ public class TajoConf extends Configuration {
Validators.min("0")),
$EXECUTOR_GROUPBY_INMEMORY_HASH_THRESHOLD("tajo.executor.groupby.in-memory-hash-threshold-mb", 64l,
Validators.min("0")),
+ $EXECUTOR_HASH_SHUFFLE_BUFFER_SIZE("tajo.executor.hash-shuffle.buffer-mb", 100, Validators.min("1")),
$MAX_OUTPUT_FILE_SIZE("tajo.query.max-outfile-size-mb", 0), // zero means infinite
$CODEGEN("tajo.executor.codegen.enabled", false), // Runtime code generation (todo this is broken)
$AGG_HASH_TABLE_SIZE("tajo.executor.aggregate.hash-table.size", 10000),
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/storage/BufferPool.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/storage/BufferPool.java b/tajo-common/src/main/java/org/apache/tajo/storage/BufferPool.java
index 97ac7b4..4913d3b 100644
--- a/tajo-common/src/main/java/org/apache/tajo/storage/BufferPool.java
+++ b/tajo-common/src/main/java/org/apache/tajo/storage/BufferPool.java
@@ -42,10 +42,6 @@ public class BufferPool {
}
static {
- /* TODO Enable thread cache
- * Create a pooled ByteBuf allocator but disables the thread-local cache.
- * Because the TaskRunner thread is newly created
- * */
if (TajoConstants.IS_TEST_MODE) {
/* Disable pooling buffers for memory usage */
@@ -55,7 +51,7 @@ public class BufferPool {
ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
} else {
TajoConf tajoConf = new TajoConf();
- ALLOCATOR = createPooledByteBufAllocator(true, tajoConf.getBoolean(ALLOW_CACHE, false), 0);
+ ALLOCATOR = createPooledByteBufAllocator(true, tajoConf.getBoolean(ALLOW_CACHE, true), 0);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java b/tajo-common/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java
index cb417f3..ebdcc26 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java
@@ -85,7 +85,7 @@ public class BaseTupleBuilder extends OffHeapRowWriter implements TupleBuilder,
public UnSafeTuple buildToZeroCopyTuple() {
UnSafeTuple zcTuple = new UnSafeTuple();
- zcTuple.set(memoryBlock, memoryBlock.readerPosition(), memoryBlock.readableBytes(), dataTypes());
+ zcTuple.set(memoryBlock, memoryBlock.readerPosition(), dataTypes());
return zcTuple;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/CompactRowBlockWriter.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/CompactRowBlockWriter.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/CompactRowBlockWriter.java
new file mode 100644
index 0000000..a88d2f1
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/CompactRowBlockWriter.java
@@ -0,0 +1,408 @@
+/**
+ * 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.tajo.tuple.memory;
+
+import io.netty.util.internal.PlatformDependent;
+import org.apache.tajo.common.TajoDataTypes.DataType;
+import org.apache.tajo.datum.BooleanDatum;
+import org.apache.tajo.datum.IntervalDatum;
+import org.apache.tajo.datum.ProtobufDatum;
+import org.apache.tajo.datum.TextDatum;
+import org.apache.tajo.exception.TajoInternalError;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.util.BitArray;
+import org.apache.tajo.util.SizeOf;
+import org.apache.tajo.util.UnsafeUtil;
+
+/**
+ * This class represent serialization of RawFile
+ *
+ * Row Record Structure
+ *
+ * | row length | null flags length | null flags | field 1 | field 2| ... | field N |;
+ *
+ * | (4 bytes) (2 bytes) (N bytes) | |;
+ * Header values
+ */
+public class CompactRowBlockWriter implements RowWriter {
+ private static final int RECORD_FIELD_SIZE = 4;
+ // Maximum variant int32 size is 5
+ private static final short MAXIMUM_VARIANT_INT32 = 5;
+ // Maximum variant int64 size is 10
+ private static final short MAXIMUM_VARIANT_INT64 = 10;
+
+ private final RowBlock rowBlock;
+ private final BitArray nullFlags;
+
+ /** record capacity + offset list */
+ private final int headerSize;
+
+ private final DataType[] dataTypes;
+
+ private int curFieldIdx;
+ private int curOffset;
+
+
+ public CompactRowBlockWriter(RowBlock rowBlock) {
+ this.dataTypes = rowBlock.getDataTypes();
+ this.rowBlock = rowBlock;
+
+ // compute the number of bytes, representing the null flags
+ nullFlags = new BitArray(dataTypes.length);
+ headerSize = RECORD_FIELD_SIZE + SizeOf.SIZE_OF_SHORT + nullFlags.bytesLength();
+
+ if (!rowBlock.getMemory().hasAddress()) {
+ throw new TajoInternalError(rowBlock.getMemory().getClass().getSimpleName()
+ + " does not support to direct memory access");
+ }
+ }
+
+
+ /**
+ * Encode a ZigZag-encoded 32-bit value. ZigZag encodes signed integers
+ * into values that can be efficiently encoded with varint. (Otherwise,
+ * negative values must be sign-extended to 64 bits to be varint encoded,
+ * thus always taking 10 bytes on the wire.)
+ *
+ * @param n A signed 32-bit integer.
+ * @return An unsigned 32-bit integer, stored in a signed int because
+ * Java has no explicit unsigned support.
+ */
+ public static int encodeZigZag32(final int n) {
+ // Note: the right-shift must be arithmetic
+ return (n << 1) ^ (n >> 31);
+ }
+
+ /**
+ * Encode a ZigZag-encoded 64-bit value. ZigZag encodes signed integers
+ * into values that can be efficiently encoded with varint. (Otherwise,
+ * negative values must be sign-extended to 64 bits to be varint encoded,
+ * thus always taking 10 bytes on the wire.)
+ *
+ * @param n A signed 64-bit integer.
+ * @return An unsigned 64-bit integer, stored in a signed int because
+ * Java has no explicit unsigned support.
+ */
+ public static long encodeZigZag64(final long n) {
+ // Note: the right-shift must be arithmetic
+ return (n << 1) ^ (n >> 63);
+ }
+
+ /**
+ * Encode and write a varint. {@code value} is treated as
+ * unsigned, so it won't be sign-extended if negative.
+ */
+ public static short writeRawVarint32(long address, int value) {
+ short length = 0;
+ while (true) {
+ if ((value & ~0x7F) == 0) {
+ PlatformDependent.putByte(address + length, (byte) value);
+ length++;
+ return length;
+ } else {
+ PlatformDependent.putByte(address + length, (byte) ((value & 0x7F) | 0x80));
+ value >>>= 7;
+ length++;
+ }
+ }
+ }
+
+ /**
+ * Encode and write a varint64.
+ */
+ public static short writeRawVarint64(long address, long value) {
+ short length = 0;
+ while (true) {
+ if ((value & ~0x7FL) == 0) {
+ PlatformDependent.putByte(address + length, (byte) value);
+ length++;
+ return length;
+ } else {
+ PlatformDependent.putByte(address + length, (byte) ((value & 0x7F) | 0x80));
+ value >>>= 7;
+ length++;
+ }
+ }
+ }
+
+ /**
+ * Compute the number of bytes that would be needed to encode a varint.
+ * {@code value} is treated as unsigned, so it won't be sign-extended if
+ * negative.
+ */
+ public static int computeRawVarint32Size(final int value) {
+ if ((value & (0xffffffff << 7)) == 0) return 1;
+ if ((value & (0xffffffff << 14)) == 0) return 2;
+ if ((value & (0xffffffff << 21)) == 0) return 3;
+ if ((value & (0xffffffff << 28)) == 0) return 4;
+ return 5;
+ }
+
+ /**
+ * Current memory address of the buffer
+ *
+ * @return The memory address
+ */
+ public long address() {
+ return rowBlock.getMemory().address();
+ }
+
+ /**
+ * Current position
+ *
+ * @return The position
+ */
+ public int position() {
+ return rowBlock.getMemory().writerPosition();
+ }
+
+
+ /**
+ * Forward the address;
+ *
+ * @param length Length to be forwarded
+ */
+ public void forward(int length) {
+ rowBlock.getMemory().writerPosition(rowBlock.getMemory().writerPosition() + length);
+ }
+
+ public void ensureSize(int size) {
+ rowBlock.getMemory().ensureSize(size);
+ }
+
+ @Override
+ public DataType[] dataTypes() {
+ return rowBlock.getDataTypes();
+ }
+
+ /**
+ * Current memory address of the row
+ *
+ * @return The memory address
+ */
+ public long recordStartAddr() {
+ return currentAddr() - curOffset;
+ }
+
+ /**
+ * Memory address that point to the first byte of the buffer
+ *
+ * @return The memory address
+ */
+ private long currentAddr() {
+ return address() + position();
+ }
+
+ public int offset() {
+ return position();
+ }
+
+
+ @Override
+ public void clear() {
+ curOffset = 0;
+ curFieldIdx = 0;
+ nullFlags.clear();
+ }
+
+ @Override
+ public boolean startRow() {
+ ensureSize(headerSize);
+ nullFlags.clear();
+
+ curOffset = headerSize;
+ curFieldIdx = 0;
+ forward(headerSize);
+ return true;
+ }
+
+
+ public void endRow() {
+ long rowHeaderPos = recordStartAddr();
+ // curOffset is equivalent to a byte length of this row.
+ PlatformDependent.putInt(rowHeaderPos, curOffset);
+ rowHeaderPos += SizeOf.SIZE_OF_INT;
+
+ //set null flags
+ byte [] flags = nullFlags.toArray();
+ PlatformDependent.putShort(rowHeaderPos, (short) flags.length);
+ rowHeaderPos += SizeOf.SIZE_OF_SHORT;
+ PlatformDependent.copyMemory(flags, 0, rowHeaderPos, flags.length);
+
+ rowBlock.setRows(rowBlock.rows() + 1);
+ }
+
+ @Override
+ public void skipField() {
+ // set null flag
+ nullFlags.set(curFieldIdx);
+ curFieldIdx++;
+ }
+
+ /**
+ * set current buffer position and forward field length
+ * @param fieldLength
+ */
+ private void forwardField(int fieldLength) {
+ forward(fieldLength);
+ curOffset += fieldLength;
+
+ }
+
+ @Override
+ public void putByte(byte val) {
+ ensureSize(SizeOf.SIZE_OF_BYTE);
+ long addr = currentAddr();
+
+ PlatformDependent.putByte(addr, val);
+ curFieldIdx++;
+ forwardField(SizeOf.SIZE_OF_BYTE);
+ }
+
+ @Override
+ public void putBool(boolean val) {
+ putByte(val ? BooleanDatum.TRUE_INT : BooleanDatum.FALSE_INT);
+ }
+
+ @Override
+ public void putInt2(short val) {
+ ensureSize(SizeOf.SIZE_OF_SHORT);
+ long addr = currentAddr();
+
+ PlatformDependent.putShort(addr, val);
+ curFieldIdx++;
+ forwardField(SizeOf.SIZE_OF_SHORT);
+ }
+
+ @Override
+ public void putInt4(int val) {
+ ensureSize(MAXIMUM_VARIANT_INT32);
+
+ curFieldIdx++;
+ forwardField(writeRawVarint32(currentAddr(), encodeZigZag32(val)));
+ }
+
+ @Override
+ public void putInt8(long val) {
+ ensureSize(MAXIMUM_VARIANT_INT64);
+
+ curFieldIdx++;
+ forwardField(writeRawVarint64(currentAddr(), encodeZigZag64(val)));
+ }
+
+ @Override
+ public void putFloat4(float val) {
+ ensureSize(SizeOf.SIZE_OF_FLOAT);
+ long addr = currentAddr();
+
+ UnsafeUtil.unsafe.putFloat(addr, val);
+ curFieldIdx++;
+ forwardField(SizeOf.SIZE_OF_FLOAT);
+ }
+
+ @Override
+ public void putFloat8(double val) {
+ ensureSize(SizeOf.SIZE_OF_DOUBLE);
+ long addr = currentAddr();
+
+ UnsafeUtil.unsafe.putDouble(addr, val);
+ curFieldIdx++;
+ forwardField(SizeOf.SIZE_OF_DOUBLE);
+ }
+
+ @Override
+ public void putText(String val) {
+ putText(val.getBytes(TextDatum.DEFAULT_CHARSET));
+ }
+
+ @Override
+ public void putText(byte[] val) {
+ putBlob(val);
+ }
+
+ @Override
+ public void putBlob(byte[] val) {
+ int bytesLen = val.length;
+
+ ensureSize(MAXIMUM_VARIANT_INT32 + bytesLen);
+ long addr = currentAddr();
+
+ short length = writeRawVarint32(addr, bytesLen);
+ PlatformDependent.copyMemory(val, 0, addr + length, bytesLen);
+ curFieldIdx++;
+ forwardField(length + bytesLen);
+ }
+
+ @Override
+ public void putDate(int val) {
+ ensureSize(SizeOf.SIZE_OF_INT);
+ long addr = currentAddr();
+
+ PlatformDependent.putInt(addr, val);
+ curFieldIdx++;
+ forwardField(SizeOf.SIZE_OF_INT);
+ }
+
+ @Override
+ public void putTime(long val) {
+ ensureSize(SizeOf.SIZE_OF_LONG);
+ long addr = currentAddr();
+
+ PlatformDependent.putLong(addr, val);
+ curFieldIdx++;
+ forwardField(SizeOf.SIZE_OF_LONG);
+ }
+
+ @Override
+ public void putTimestamp(long val) {
+ putTime(val);
+ }
+
+ @Override
+ public void putInterval(IntervalDatum val) {
+ ensureSize(MAXIMUM_VARIANT_INT32 + MAXIMUM_VARIANT_INT64);
+ long addr = currentAddr();
+
+ short length = writeRawVarint32(addr, encodeZigZag32(val.getMonths()));
+ length += writeRawVarint64(addr, encodeZigZag64(val.getMilliSeconds()));
+
+ curFieldIdx++;
+ forwardField(length);
+ }
+
+ @Override
+ public void putInet4(int val) {
+ ensureSize(SizeOf.SIZE_OF_INT);
+ long addr = currentAddr();
+
+ PlatformDependent.putInt(addr, val);
+ curFieldIdx++;
+ forwardField(SizeOf.SIZE_OF_INT);
+ }
+
+ @Override
+ public void putProtoDatum(ProtobufDatum val) {
+ putBlob(val.asByteArray());
+ }
+
+ @Override
+ public void addTuple(Tuple tuple) {
+ OffHeapRowBlockUtils.convert(tuple, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/DirectBufTuple.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/DirectBufTuple.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/DirectBufTuple.java
index 10e493f..1852e6d 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/DirectBufTuple.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/DirectBufTuple.java
@@ -28,11 +28,11 @@ import static org.apache.tajo.common.TajoDataTypes.DataType;
public class DirectBufTuple extends UnSafeTuple implements Deallocatable {
private MemoryBlock memoryBlock;
- public DirectBufTuple(int length, DataType[] types) {
- ByteBuffer bb = ByteBuffer.allocateDirect(length).order(ByteOrder.LITTLE_ENDIAN);
+ public DirectBufTuple(DataType[] types) {
+ ByteBuffer bb = ByteBuffer.allocateDirect(getLength()).order(ByteOrder.LITTLE_ENDIAN);
memoryBlock = new ResizableMemoryBlock(bb);
- set(memoryBlock, 0, length, types);
+ set(memoryBlock, 0, types);
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapRowBlockReader.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapRowBlockReader.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapRowBlockReader.java
index dd377cf..ec5033b 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapRowBlockReader.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapRowBlockReader.java
@@ -48,10 +48,9 @@ public class HeapRowBlockReader implements RowBlockReader<HeapTuple> {
public boolean next(HeapTuple tuple) {
if (curRowIdxForRead < rows) {
- int recordLen = memoryBlock.getInt(curPosForRead);
- tuple.set(memoryBlock, curPosForRead, recordLen, dataTypes);
+ tuple.set(memoryBlock, curPosForRead, dataTypes);
- curPosForRead += recordLen;
+ curPosForRead += tuple.getLength();
curRowIdxForRead++;
memoryBlock.readerPosition(curPosForRead);
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapTuple.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapTuple.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapTuple.java
index 9f508b3..c6c7daf 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapTuple.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/HeapTuple.java
@@ -41,16 +41,16 @@ public class HeapTuple extends ZeroCopyTuple implements Cloneable {
private DataType[] types;
@Override
- public void set(MemoryBlock memoryBlock, int relativePos, int length, DataType[] types) {
+ public void set(MemoryBlock memoryBlock, int relativePos, DataType[] types) {
this.buffer = memoryBlock.getBuffer();
this.types = types;
- super.set(relativePos, length);
+ super.set(relativePos);
}
protected void set(final byte[] bytes, final DataType[] types) {
this.buffer = Unpooled.wrappedBuffer(bytes).order(ByteOrder.LITTLE_ENDIAN);
this.types = types;
- super.set(0, bytes.length);
+ super.set(0);
}
@Override
@@ -59,6 +59,11 @@ public class HeapTuple extends ZeroCopyTuple implements Cloneable {
}
@Override
+ public int getLength() {
+ return buffer.getInt(getRelativePos());
+ }
+
+ @Override
public TajoDataTypes.Type type(int fieldId) {
return types[fieldId].getType();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/MemoryRowBlock.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/MemoryRowBlock.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/MemoryRowBlock.java
index 922fc68..3d02f9a 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/MemoryRowBlock.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/MemoryRowBlock.java
@@ -19,6 +19,8 @@
package org.apache.tajo.tuple.memory;
import io.netty.util.internal.PlatformDependent;
+import org.apache.tajo.BuiltinStorages;
+import org.apache.tajo.annotation.NotThreadSafe;
import org.apache.tajo.exception.NotImplementedException;
import org.apache.tajo.exception.TajoInternalError;
import org.apache.tajo.tuple.RowBlockReader;
@@ -32,10 +34,12 @@ import java.nio.channels.ScatteringByteChannel;
import static org.apache.tajo.common.TajoDataTypes.DataType;
+@NotThreadSafe
public class MemoryRowBlock implements RowBlock, Deallocatable {
public static final int NULL_FIELD_OFFSET = -1;
- private DataType[] dataTypes;
+ private final DataType[] dataTypes;
+ private final String dataFormat;
// Basic States
private int maxRowNum = Integer.MAX_VALUE; // optional
@@ -45,20 +49,20 @@ public class MemoryRowBlock implements RowBlock, Deallocatable {
private MemoryBlock memory;
public MemoryRowBlock(DataType[] dataTypes, ResizableLimitSpec limitSpec, boolean isDirect) {
+ this(dataTypes, limitSpec, isDirect, BuiltinStorages.DRAW);
+ }
+
+ public MemoryRowBlock(DataType[] dataTypes, ResizableLimitSpec limitSpec, boolean isDirect, String dataFormat) {
this.memory = new ResizableMemoryBlock(limitSpec, isDirect);
this.dataTypes = dataTypes;
+ this.dataFormat = dataFormat;
}
public MemoryRowBlock(MemoryRowBlock rowBlock) {
this.memory = TUtil.checkTypeAndGet(rowBlock.getMemory().duplicate(), ResizableMemoryBlock.class);
this.rowNum = rowBlock.rowNum;
this.dataTypes = rowBlock.dataTypes;
- }
-
- public MemoryRowBlock(MemoryBlock memory, DataType[] dataTypes, int rowNum) {
- this.memory = memory;
- this.rowNum = rowNum;
- this.dataTypes = dataTypes;
+ this.dataFormat = rowBlock.dataFormat;
}
public MemoryRowBlock(DataType[] dataTypes) {
@@ -69,8 +73,13 @@ public class MemoryRowBlock implements RowBlock, Deallocatable {
this(dataTypes, new ResizableLimitSpec(bytes), true);
}
- public MemoryRowBlock(DataType[] dataTypes, int bytes, boolean isDirect) {
- this(dataTypes, new ResizableLimitSpec(bytes), isDirect);
+ public MemoryRowBlock(DataType[] dataTypes, int bytes, boolean isDirect, String dataFormat) {
+ this(dataTypes, new ResizableLimitSpec(bytes), isDirect, dataFormat);
+ }
+
+ @Override
+ public String getDataFormat() {
+ return dataFormat;
}
@Override
@@ -91,6 +100,20 @@ public class MemoryRowBlock implements RowBlock, Deallocatable {
return memory.capacity();
}
+ @Override
+ public int usedMem() {
+ return memory.writerPosition();
+ }
+
+ @Override
+ public float usage() {
+ if (usedMem() > 0) {
+ return (usedMem() / (float) capacity());
+ } else {
+ return 0.0f;
+ }
+ }
+
public int maxRowNum() {
return maxRowNum;
}
@@ -112,6 +135,15 @@ public class MemoryRowBlock implements RowBlock, Deallocatable {
@Override
public boolean copyFromChannel(ScatteringByteChannel channel) throws IOException {
+ switch (dataFormat) {
+ case BuiltinStorages.DRAW:
+ return fillDrawBuffer(channel);
+ default:
+ throw new TajoInternalError(new NotImplementedException("Heap memory writer not implemented yet"));
+ }
+ }
+
+ protected boolean fillDrawBuffer(ScatteringByteChannel channel) throws IOException {
reset();
int readBytes = memory.writeBytes(channel);
@@ -143,13 +175,23 @@ public class MemoryRowBlock implements RowBlock, Deallocatable {
@Override
public RowWriter getWriter() {
+ if (!getMemory().hasAddress()) {
+ throw new TajoInternalError(new NotImplementedException("Heap memory writer not implemented yet"));
+ }
+
if (builder == null) {
- if (!getMemory().hasAddress()) {
- throw new TajoInternalError(new NotImplementedException("Heap memory writer not implemented yet"));
- } else {
+ switch (dataFormat) {
+ case BuiltinStorages.DRAW:
this.builder = new OffHeapRowBlockWriter(this);
+ break;
+ case BuiltinStorages.RAW:
+ this.builder = new CompactRowBlockWriter(this);
+ break;
+ default:
+ throw new TajoInternalError(new NotImplementedException(dataFormat + " memory writer not implemented yet"));
}
}
+
return builder;
}
@@ -165,10 +207,17 @@ public class MemoryRowBlock implements RowBlock, Deallocatable {
@Override
public RowBlockReader getReader() {
- if (!getMemory().hasAddress()) {
- return new HeapRowBlockReader(this);
- } else {
- return new OffHeapRowBlockReader(this);
+
+ switch (dataFormat) {
+ case BuiltinStorages.DRAW: {
+ if (!getMemory().hasAddress()) {
+ return new HeapRowBlockReader(this);
+ } else {
+ return new OffHeapRowBlockReader(this);
+ }
+ }
+ default:
+ throw new TajoInternalError(new NotImplementedException(dataFormat + " memory writer not implemented yet"));
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockReader.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockReader.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockReader.java
index ccaeffc..c5673e3 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockReader.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockReader.java
@@ -18,7 +18,6 @@
package org.apache.tajo.tuple.memory;
-import io.netty.util.internal.PlatformDependent;
import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.exception.TajoInternalError;
import org.apache.tajo.tuple.RowBlockReader;
@@ -54,11 +53,8 @@ public class OffHeapRowBlockReader implements RowBlockReader<ZeroCopyTuple> {
public boolean next(ZeroCopyTuple tuple) {
if (curRowIdxForRead < rows) {
- long recordStartPtr = memoryBlock.address() + curPosForRead;
- int recordLen = PlatformDependent.getInt(recordStartPtr);
- tuple.set(memoryBlock, curPosForRead, recordLen, dataTypes);
-
- curPosForRead += recordLen;
+ tuple.set(memoryBlock, curPosForRead, dataTypes);
+ curPosForRead += tuple.getLength();
curRowIdxForRead++;
memoryBlock.readerPosition(curPosForRead);
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockUtils.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockUtils.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockUtils.java
index e8f219c..1aca22f 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockUtils.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockUtils.java
@@ -32,6 +32,11 @@ import java.util.Comparator;
import java.util.List;
public class OffHeapRowBlockUtils {
+ private static TupleConverter tupleConverter;
+
+ static {
+ tupleConverter = new TupleConverter();
+ }
public static List<Tuple> sort(MemoryRowBlock rowBlock, Comparator<Tuple> comparator) {
List<Tuple> tupleList = Lists.newArrayList();
@@ -80,62 +85,86 @@ public class OffHeapRowBlockUtils {
return tuples;
}
- public static void convert(Tuple tuple, RowWriter writer) {
- writer.startRow();
+ /**
+ * This class is tuple converter to the RowBlock
+ */
+ public static class TupleConverter {
- for (int i = 0; i < writer.dataTypes().length; i++) {
- if (tuple.isBlankOrNull(i)) {
- writer.skipField();
- continue;
+ public void convert(Tuple tuple, RowWriter writer) {
+ writer.startRow();
+
+ for (int i = 0; i < writer.dataTypes().length; i++) {
+ writeField(i, tuple, writer);
}
- switch (writer.dataTypes()[i].getType()) {
- case BOOLEAN:
- writer.putBool(tuple.getBool(i));
- break;
- case BIT:
- writer.putByte(tuple.getByte(i));
- break;
- case INT1:
- case INT2:
- writer.putInt2(tuple.getInt2(i));
- break;
- case INT4:
- case DATE:
- case INET4:
- writer.putInt4(tuple.getInt4(i));
- break;
- case INT8:
- case TIMESTAMP:
- case TIME:
- writer.putInt8(tuple.getInt8(i));
- break;
- case FLOAT4:
- writer.putFloat4(tuple.getFloat4(i));
- break;
- case FLOAT8:
- writer.putFloat8(tuple.getFloat8(i));
- break;
- case CHAR:
- case TEXT:
- writer.putText(tuple.getBytes(i));
- break;
- case BLOB:
- writer.putBlob(tuple.getBytes(i));
- break;
- case INTERVAL:
- writer.putInterval((IntervalDatum) tuple.getInterval(i));
- break;
- case PROTOBUF:
- writer.putProtoDatum((ProtobufDatum) tuple.getProtobufDatum(i));
- break;
- case NULL_TYPE:
+
+ writer.endRow();
+ }
+
+ protected void writeField(int colIdx, Tuple tuple, RowWriter writer) {
+
+ if (tuple.isBlankOrNull(colIdx)) {
writer.skipField();
- break;
- default:
- throw new TajoRuntimeException(
- new UnsupportedException("unknown data type '" + writer.dataTypes()[i].getType().name() + "'"));
+ } else {
+ switch (writer.dataTypes()[colIdx].getType()) {
+ case BOOLEAN:
+ writer.putBool(tuple.getBool(colIdx));
+ break;
+ case BIT:
+ writer.putByte(tuple.getByte(colIdx));
+ break;
+ case INT1:
+ case INT2:
+ writer.putInt2(tuple.getInt2(colIdx));
+ break;
+ case INT4:
+ writer.putInt4(tuple.getInt4(colIdx));
+ break;
+ case DATE:
+ writer.putDate(tuple.getInt4(colIdx));
+ break;
+ case INT8:
+ writer.putInt8(tuple.getInt8(colIdx));
+ break;
+ case TIMESTAMP:
+ writer.putTimestamp(tuple.getInt8(colIdx));
+ break;
+ case TIME:
+ writer.putTime(tuple.getInt8(colIdx));
+ break;
+ case FLOAT4:
+ writer.putFloat4(tuple.getFloat4(colIdx));
+ break;
+ case FLOAT8:
+ writer.putFloat8(tuple.getFloat8(colIdx));
+ break;
+ case CHAR:
+ case TEXT:
+ writer.putText(tuple.getBytes(colIdx));
+ break;
+ case BLOB:
+ writer.putBlob(tuple.getBytes(colIdx));
+ break;
+ case INTERVAL:
+ writer.putInterval((IntervalDatum) tuple.getInterval(colIdx));
+ break;
+ case PROTOBUF:
+ writer.putProtoDatum((ProtobufDatum) tuple.getProtobufDatum(colIdx));
+ break;
+ case INET4:
+ writer.putInet4(tuple.getInt4(colIdx));
+ break;
+ case NULL_TYPE:
+ writer.skipField();
+ break;
+ default:
+ throw new TajoRuntimeException(
+ new UnsupportedException("unknown data type '" + writer.dataTypes()[colIdx].getType().name() + "'"));
+ }
}
}
- writer.endRow();
+ }
+
+ public static void convert(Tuple tuple, RowWriter writer) {
+ tupleConverter.convert(tuple, writer);
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableLimitSpec.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableLimitSpec.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableLimitSpec.java
index 614b3fb..ddf50ab 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableLimitSpec.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableLimitSpec.java
@@ -21,6 +21,8 @@ package org.apache.tajo.tuple.memory;
import com.google.common.base.Preconditions;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.exception.TajoRuntimeException;
+import org.apache.tajo.exception.UnsupportedException;
import org.apache.tajo.util.FileUtil;
/**
@@ -29,7 +31,7 @@ import org.apache.tajo.util.FileUtil;
* due to ByteBuffer.
*/
public class ResizableLimitSpec {
- private final Log LOG = LogFactory.getLog(ResizableLimitSpec.class);
+ private static final Log LOG = LogFactory.getLog(ResizableLimitSpec.class);
public static final int MAX_SIZE_BYTES = Integer.MAX_VALUE;
public static final ResizableLimitSpec DEFAULT_LIMIT = new ResizableLimitSpec(Integer.MAX_VALUE);
@@ -114,10 +116,11 @@ public class ResizableLimitSpec {
return (int) initSize;
}
- if (currentSize > Integer.MAX_VALUE) {
- LOG.warn("Current size already exceeds the maximum size (" + Integer.MAX_VALUE + " bytes)");
- return Integer.MAX_VALUE;
+ if (currentSize == Integer.MAX_VALUE) {
+ throw new TajoRuntimeException(new UnsupportedException(
+ "Current size already exceeds the maximum size (" + Integer.MAX_VALUE + " bytes)"));
}
+
long nextSize = (long) (currentSize + ((float) currentSize * incRatio));
if (nextSize > limitBytes) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableMemoryBlock.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableMemoryBlock.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableMemoryBlock.java
index 22c2561..09faff9 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableMemoryBlock.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ResizableMemoryBlock.java
@@ -39,10 +39,12 @@ public class ResizableMemoryBlock implements MemoryBlock {
protected ByteBuf buffer;
protected ResizableLimitSpec limitSpec;
+ private long memoryAddress;
public ResizableMemoryBlock(ByteBuf buffer, ResizableLimitSpec limitSpec) {
this.buffer = buffer.order(ByteOrder.LITTLE_ENDIAN);
this.limitSpec = limitSpec;
+ this.memoryAddress = this.buffer.hasMemoryAddress() ? this.buffer.memoryAddress() : 0;
}
public ResizableMemoryBlock(ByteBuf buffer) {
@@ -50,13 +52,13 @@ public class ResizableMemoryBlock implements MemoryBlock {
}
public ResizableMemoryBlock(ByteBuffer buffer) {
- this.buffer = Unpooled.wrappedBuffer(buffer).order(ByteOrder.LITTLE_ENDIAN);
- this.limitSpec = new ResizableLimitSpec(buffer.capacity());
+ this(Unpooled.wrappedBuffer(buffer), new ResizableLimitSpec(buffer.capacity(), buffer.capacity()));
}
public ResizableMemoryBlock(ResizableLimitSpec limitSpec, boolean isDirect) {
if (isDirect) {
this.buffer = BufferPool.directBuffer((int) limitSpec.initialSize(), (int) limitSpec.limit());
+ this.memoryAddress = buffer.memoryAddress();
} else {
this.buffer = BufferPool.heapBuffer((int) limitSpec.initialSize(), (int) limitSpec.limit());
}
@@ -65,7 +67,7 @@ public class ResizableMemoryBlock implements MemoryBlock {
@Override
public long address() {
- return buffer.memoryAddress();
+ return memoryAddress;
}
@Override
@@ -123,15 +125,14 @@ public class ResizableMemoryBlock implements MemoryBlock {
return buffer.writerIndex();
}
-
@Override
public void ensureSize(int size) {
if (!buffer.isWritable(size)) {
- if (!limitSpec.canIncrease(buffer.capacity())) {
+ if (!limitSpec.canIncrease(size)) {
throw new RuntimeException("Cannot increase RowBlock anymore.");
}
- int newBlockSize = limitSpec.increasedSize(buffer.capacity());
+ int newBlockSize = limitSpec.increasedSize(size);
resize(newBlockSize);
LOG.info("Increase DirectRowBlock to " + FileUtil.humanReadableByteCount(newBlockSize, false));
}
@@ -144,17 +145,18 @@ public class ResizableMemoryBlock implements MemoryBlock {
throw new RuntimeException("Resize cannot exceed the capacity limit");
}
- if (newSize < buffer.capacity()) {
+ if (newSize < buffer.writableBytes()) {
LOG.warn("The capacity reduction is ignored.");
}
int newBlockSize = UnsafeUtil.alignedSize(newSize);
buffer = BufferPool.ensureWritable(buffer, newBlockSize);
+ memoryAddress = buffer.memoryAddress();
}
@Override
public void release() {
- buffer.release();
+ if(buffer.refCnt() > 0) buffer.release();
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowBlock.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowBlock.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowBlock.java
index 68902fb..1ab1042 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowBlock.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowBlock.java
@@ -25,11 +25,31 @@ import java.io.IOException;
import java.nio.channels.ScatteringByteChannel;
public interface RowBlock {
+ /**
+ * a data format for de/serialization
+ */
+ String getDataFormat();
+ /**
+ * reset the memory and writer
+ */
void clear();
+ /**
+ * @return the number of bytes this memory block can contain.
+ */
int capacity();
+ /**
+ * @return the number of written bytes in this memory block
+ */
+ int usedMem();
+
+ /**
+ * @return the percentage of written bytes in this memory block
+ */
+ float usage();
+
void setRows(int rowNum);
int rows();
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTuple.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTuple.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTuple.java
index ec167f8..26f7df3 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTuple.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTuple.java
@@ -42,22 +42,22 @@ import static org.apache.tajo.common.TajoDataTypes.DataType;
public class UnSafeTuple extends ZeroCopyTuple {
private static final Unsafe UNSAFE = UnsafeUtil.unsafe;
- private long address;
+ private MemoryBlock memoryBlock;
private DataType[] types;
@Override
- public void set(MemoryBlock memoryBlock, int relativePos, int length, DataType[] types) {
+ public void set(MemoryBlock memoryBlock, int relativePos, DataType[] types) {
Preconditions.checkArgument(memoryBlock.hasAddress());
- this.address = memoryBlock.address();
+ this.memoryBlock = memoryBlock;
this.types = types;
- super.set(relativePos, length);
+ super.set(relativePos);
}
public void set(UnSafeTuple tuple) {
- this.address = tuple.address;
+ this.memoryBlock = tuple.memoryBlock;
this.types = tuple.types;
- super.set(tuple.getRelativePos(), tuple.getLength());
+ super.set(tuple.getRelativePos());
}
@Override
@@ -66,6 +66,11 @@ public class UnSafeTuple extends ZeroCopyTuple {
}
@Override
+ public int getLength() {
+ return PlatformDependent.getInt(address());
+ }
+
+ @Override
public TajoDataTypes.Type type(int fieldId) {
return types[fieldId].getType();
}
@@ -93,7 +98,7 @@ public class UnSafeTuple extends ZeroCopyTuple {
}
public long address() {
- return address + getRelativePos();
+ return memoryBlock.address() + getRelativePos();
}
public HeapTuple toHeapTuple() {
@@ -110,8 +115,9 @@ public class UnSafeTuple extends ZeroCopyTuple {
public long getFieldAddr(int fieldId) {
int fieldOffset = getFieldOffset(fieldId);
- if (fieldOffset == -1) {
- throw new RuntimeException("Invalid Field Access: " + fieldId);
+ if (fieldOffset < 0 || fieldOffset > getLength()) {
+ throw new RuntimeException("Invalid Access. Field : " + fieldId
+ + ", Offset:" + fieldOffset + ", Record length:" + getLength());
}
return address() + fieldOffset;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ZeroCopyTuple.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ZeroCopyTuple.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ZeroCopyTuple.java
index 1f4f57e..e9108f2 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ZeroCopyTuple.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/ZeroCopyTuple.java
@@ -24,22 +24,18 @@ import org.apache.tajo.storage.Tuple;
public abstract class ZeroCopyTuple implements Tuple {
protected int relativePos;
- protected int length;
- public abstract void set(MemoryBlock memoryBlock, int relativePos, int length, DataType[] types);
+ public abstract void set(MemoryBlock memoryBlock, int relativePos, DataType[] types);
- void set(int relativePos, int length) {
+ void set(int relativePos) {
this.relativePos = relativePos;
- this.length = length;
}
public int getRelativePos() {
return relativePos;
}
- public int getLength() {
- return length;
- }
+ public abstract int getLength();
@Override
public Tuple clone() throws CloneNotSupportedException {
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-common/src/main/java/org/apache/tajo/util/BitArray.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/BitArray.java b/tajo-common/src/main/java/org/apache/tajo/util/BitArray.java
index e62496a..973266b 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/BitArray.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/BitArray.java
@@ -19,6 +19,7 @@
package org.apache.tajo.util;
import java.nio.ByteBuffer;
+import java.util.Arrays;
public class BitArray {
private byte [] data;
@@ -60,9 +61,7 @@ public class BitArray {
}
public void clear() {
- for (int i = 0; i < data.length; i++) {
- data[i] = 0;
- }
+ Arrays.fill(data, (byte) 0);
}
public int bytesLength() {
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core-tests/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-core-tests/pom.xml b/tajo-core-tests/pom.xml
index 2037f0b..6de5546 100644
--- a/tajo-core-tests/pom.xml
+++ b/tajo-core-tests/pom.xml
@@ -375,7 +375,7 @@
<forkCount>${maven.fork.count}</forkCount>
<reuseForks>true</reuseForks>
<trimStackTrace>false</trimStackTrace>
- <argLine>-Xms128m -Xmx800m -XX:+CMSClassUnloadingEnabled -Dfile.encoding=UTF-8 -Dderby.storage.pageSize=1024 -Dderby.stream.error.file=/dev/null</argLine>
+ <argLine>-Xms128m -Xmx800m -XX:MinMetaspaceFreeRatio=10 -XX:MaxMetaspaceFreeRatio=10 -XX:+CMSClassUnloadingEnabled -Dfile.encoding=UTF-8 -Dderby.storage.pageSize=1024 -Dderby.stream.error.file=/dev/null</argLine>
<useSystemClassLoader>true</useSystemClassLoader>
<useManifestOnlyJar>true</useManifestOnlyJar>
<systemProperties>
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
index 4ee2c9c..30c46e5 100644
--- a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
@@ -122,7 +122,7 @@ public class TestExternalSortExec {
public final void testNext() throws IOException, TajoException {
conf.setIntVar(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT, 2);
QueryContext queryContext = LocalTajoTestingUtility.createDummyContext(conf);
- queryContext.setLong(SessionVars.EXTSORT_BUFFER_SIZE, 1024*1024);
+ queryContext.setInt(SessionVars.EXTSORT_BUFFER_SIZE, 1);
FileFragment[] frags = FileTablespace.splitNG(conf, "default.employee", employee.getMeta(),
new Path(employee.getUri()), Integer.MAX_VALUE);
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java
index a4afa7f..2f4d66f 100644
--- a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java
@@ -216,6 +216,7 @@ public class TestHashJoinExec {
HashJoinExec joinExec = proj.getChild();
assertCheckInnerJoinRelatedFunctions(ctx, phyPlanner, joinNode, joinExec);
+ exec.close();
}
/**
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
index bcecff7..adeb250 100644
--- a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
@@ -573,6 +573,7 @@ public class TestPhysicalPlanner {
LogicalNode rootNode = optimizer.optimize(plan);
PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf);
PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
+ exec.close();
assertTrue(exec instanceof SortBasedColPartitionStoreExec);
}
@@ -597,6 +598,7 @@ public class TestPhysicalPlanner {
PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf);
PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
+ exec.close();
assertTrue(exec instanceof HashBasedColPartitionStoreExec);
}
@@ -621,6 +623,7 @@ public class TestPhysicalPlanner {
PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf);
PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
+ exec.close();
assertTrue(exec instanceof SortBasedColPartitionStoreExec);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java
index 349aec0..b1f53da 100644
--- a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java
@@ -20,10 +20,7 @@ package org.apache.tajo.engine.planner.physical;
import org.apache.hadoop.fs.Path;
-import org.apache.tajo.LocalTajoTestingUtility;
-import org.apache.tajo.SessionVars;
-import org.apache.tajo.TajoConstants;
-import org.apache.tajo.TajoTestingCluster;
+import org.apache.tajo.*;
import org.apache.tajo.algebra.Expr;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.statistics.TableStats;
@@ -42,6 +39,7 @@ import org.apache.tajo.plan.LogicalPlanner;
import org.apache.tajo.plan.logical.LogicalNode;
import org.apache.tajo.storage.*;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.CommonTestingUtil;
import org.apache.tajo.worker.TaskAttemptContext;
import org.junit.After;
@@ -65,7 +63,7 @@ public class TestProgressExternalSortExec {
private LogicalPlanner planner;
private Path testDir;
- private final int numTuple = 5000;
+ private final int numTuple = 50000;
private Random rnd = new Random(System.currentTimeMillis());
private TableDesc employee;
@@ -87,8 +85,8 @@ public class TestProgressExternalSortExec {
schema.addColumn("empid", TajoDataTypes.Type.INT4);
schema.addColumn("deptname", TajoDataTypes.Type.TEXT);
- TableMeta employeeMeta = CatalogUtil.newTableMeta("RAW");
- Path employeePath = new Path(testDir, "employee.csv");
+ TableMeta employeeMeta = CatalogUtil.newTableMeta(BuiltinStorages.RAW);
+ Path employeePath = new Path(testDir, "employee.raw");
Appender appender = ((FileTablespace) TablespaceManager.getLocalFs())
.getAppender(employeeMeta, schema, employeePath);
appender.enableStats();
@@ -126,18 +124,24 @@ public class TestProgressExternalSortExec {
@Test
public void testExternalSortExecProgressWithMemTableScanner() throws Exception {
- testProgress(testDataStats.getNumBytes() * 20); //multiply 20 for memory fit
+ QueryContext queryContext = LocalTajoTestingUtility.createDummyContext(conf);
+ int bufferSize = (int) (testDataStats.getNumBytes() * 20) / StorageUnit.MB; //multiply 2 for memory fit
+ queryContext.setInt(SessionVars.EXTSORT_BUFFER_SIZE, bufferSize);
+
+ testProgress(queryContext);
}
@Test
public void testExternalSortExecProgressWithPairWiseMerger() throws Exception {
- testProgress(testDataStats.getNumBytes());
+ QueryContext queryContext = LocalTajoTestingUtility.createDummyContext(conf);
+ int bufferSize = (int) Math.max((testDataStats.getNumBytes() / StorageUnit.MB), 1);
+ queryContext.setInt(SessionVars.EXTSORT_BUFFER_SIZE, bufferSize);
+
+ testProgress(queryContext);
}
- private void testProgress(long sortBufferBytesNum) throws Exception {
+ private void testProgress(QueryContext queryContext) throws Exception {
conf.setIntVar(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT, 2);
- QueryContext queryContext = LocalTajoTestingUtility.createDummyContext(conf);
- queryContext.setLong(SessionVars.EXTSORT_BUFFER_SIZE, sortBufferBytesNum);
FileFragment[] frags = FileTablespace.splitNG(conf, "default.employee", employee.getMeta(),
new Path(employee.getUri()), Integer.MAX_VALUE);
@@ -191,9 +195,9 @@ public class TestProgressExternalSortExec {
TableStats tableStats = exec.getInputStats();
assertNotNull(tableStats);
assertEquals(testDataStats.getNumBytes().longValue(), tableStats.getNumBytes().longValue());
- assertEquals(cnt, testDataStats.getNumRows().longValue());
- assertEquals(cnt, tableStats.getNumRows().longValue());
- assertEquals(testDataStats.getNumBytes().longValue(), tableStats.getReadBytes().longValue());
+ assertEquals(testDataStats.getNumRows().longValue(), cnt);
+ assertEquals(testDataStats.getNumRows().longValue(), tableStats.getNumRows().longValue());
+ assertTrue(testDataStats.getNumBytes().longValue() <= tableStats.getReadBytes().longValue());
// for rescan test
preVal = null;
@@ -216,9 +220,10 @@ public class TestProgressExternalSortExec {
tableStats = exec.getInputStats();
assertNotNull(tableStats);
assertEquals(testDataStats.getNumBytes().longValue(), tableStats.getNumBytes().longValue());
- assertEquals(cnt, testDataStats.getNumRows().longValue());
- assertEquals(cnt, tableStats.getNumRows().longValue());
- assertEquals(testDataStats.getNumBytes().longValue(), tableStats.getReadBytes().longValue());
+ assertEquals(testDataStats.getNumRows().longValue(), cnt);
+ assertEquals(testDataStats.getNumRows().longValue(), tableStats.getNumRows().longValue());
+ //'ReadBytes' is actual read bytes
+ assertTrue(testDataStats.getNumBytes().longValue() <= tableStats.getReadBytes().longValue());
conf.setIntVar(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT, ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT.defaultIntVal);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core-tests/src/test/java/org/apache/tajo/querymaster/TestTaskStatusUpdate.java
----------------------------------------------------------------------
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/querymaster/TestTaskStatusUpdate.java b/tajo-core-tests/src/test/java/org/apache/tajo/querymaster/TestTaskStatusUpdate.java
index f845bb3..9f5ecb5 100644
--- a/tajo-core-tests/src/test/java/org/apache/tajo/querymaster/TestTaskStatusUpdate.java
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/querymaster/TestTaskStatusUpdate.java
@@ -106,7 +106,7 @@ public class TestTaskStatusUpdate extends QueryTestCaseBase {
res = executeQuery();
// in/out * stage(4)
- long[] expectedNumRows = new long[]{5, 5, 2, 2, 7, 2, 2, 2};
+ long[] expectedNumRows = new long[]{5, 5, 2, 2, 2, 2, 2, 2};
long[] expectedNumBytes = new long[]{20, 75, 8, 34, 109, 34, 34, 64};
long[] expectedReadBytes = new long[]{20, 20, 8, 8, 109, 0, 34, 0};
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result
----------------------------------------------------------------------
diff --git a/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result b/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result
index d65c346..ec8344f 100644
--- a/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result
+++ b/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result
@@ -28,7 +28,8 @@ Available Session Variables:
\set TABLE_PARTITION_PER_SHUFFLE_SIZE [int value] - shuffle output size for partition table write (mb)
\set GROUPBY_MULTI_LEVEL_ENABLED [true or false] - Multiple level groupby enabled
\set QUERY_EXECUTE_PARALLEL [int value] - Maximum parallel running of execution blocks for a query
-\set EXTSORT_BUFFER_SIZE [long value] - sort buffer size for external sort (mb)
+\set EXTSORT_BUFFER_SIZE [int value] - sort buffer size for external sort (mb)
+\set HASH_SHUFFLE_BUFFER_SIZE [int value] - hash-shuffle buffer size for local disk I/O (mb)
\set HASH_JOIN_SIZE_LIMIT [long value] - limited size for hash join (mb)
\set INNER_HASH_JOIN_SIZE_LIMIT [long value] - limited size for hash inner join (mb)
\set OUTER_HASH_JOIN_SIZE_LIMIT [long value] - limited size for hash outer join (mb)
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index c6db96d..e0da9e6 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -904,14 +904,12 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
public PhysicalExec createScanPlan(TaskAttemptContext ctx, ScanNode scanNode, Stack<LogicalNode> node)
throws IOException {
// check if an input is sorted in the same order to the subsequence sort operator.
- // TODO - it works only if input files are raw files. We should check the file format.
- // Since the default intermediate file format is raw file, it is not problem right now.
if (checkIfSortEquivalance(ctx, scanNode, node)) {
if (ctx.getTable(scanNode.getCanonicalName()) == null) {
return new SeqScanExec(ctx, scanNode, null);
}
FragmentProto [] fragments = ctx.getTables(scanNode.getCanonicalName());
- return new ExternalSortExec(ctx, (SortNode) node.peek(), fragments);
+ return new ExternalSortExec(ctx, (SortNode) node.peek(), scanNode, fragments);
} else {
Enforcer enforcer = ctx.getEnforcer();
http://git-wip-us.apache.org/repos/asf/tajo/blob/411a26d5/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
index b9ab344..8abafec 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
@@ -24,6 +24,7 @@ import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.io.IOUtils;
+import org.apache.tajo.BuiltinStorages;
import org.apache.tajo.SessionVars;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.Schema;
@@ -32,6 +33,7 @@ import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.engine.planner.PhysicalPlanningException;
+import org.apache.tajo.plan.logical.ScanNode;
import org.apache.tajo.plan.logical.SortNode;
import org.apache.tajo.storage.*;
import org.apache.tajo.storage.fragment.FileFragment;
@@ -49,7 +51,6 @@ import java.util.List;
import java.util.concurrent.*;
import static org.apache.tajo.storage.RawFile.RawFileAppender;
-import static org.apache.tajo.storage.RawFile.RawFileScanner;
/**
* This external sort algorithm can be characterized by the followings:
@@ -69,7 +70,8 @@ public class ExternalSortExec extends SortExec {
private static final String INTERMEDIATE_FILE_PREFIX = "@interFile_";
private SortNode plan;
- private final TableMeta meta;
+ /** the data format of intermediate file*/
+ private TableMeta intermediateMeta;
/** the defaultFanout of external sort */
private final int defaultFanout;
/** It's the size of in-memory table. If memory consumption exceeds it, store the memory table into a disk. */
@@ -87,9 +89,9 @@ public class ExternalSortExec extends SortExec {
/** local file system */
private final RawLocalFileSystem localFS;
/** final output files which are used for cleaning */
- private List<FileFragment> finalOutputFiles = null;
+ private List<Chunk> finalOutputFiles = null;
/** for directly merging sorted inputs */
- private List<FileFragment> mergedInputFragments = null;
+ private List<Chunk> mergedInputFragments = null;
///////////////////////////////////////////////////
// transient variables
@@ -108,8 +110,6 @@ public class ExternalSortExec extends SortExec {
super(context, plan.getInSchema(), plan.getOutSchema(), null, plan.getSortKeys());
this.plan = plan;
- this.meta = CatalogUtil.newTableMeta("ROWFILE");
-
this.defaultFanout = context.getConf().getIntVar(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT);
if (defaultFanout < 2) {
throw new PhysicalPlanningException(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT.varname + " cannot be lower than 2");
@@ -121,18 +121,19 @@ public class ExternalSortExec extends SortExec {
this.inMemoryTable = new TupleList(context.getQueryContext().getInt(SessionVars.SORT_LIST_SIZE));
this.sortTmpDir = getExecutorTmpDir();
- localDirAllocator = new LocalDirAllocator(ConfVars.WORKER_TEMPORAL_DIR.varname);
- localFS = new RawLocalFileSystem();
+ this.localDirAllocator = new LocalDirAllocator(ConfVars.WORKER_TEMPORAL_DIR.varname);
+ this.localFS = new RawLocalFileSystem();
+ this.intermediateMeta = CatalogUtil.newTableMeta(BuiltinStorages.RAW); //TODO change to SHUFFLE_FILE_FORMAT
}
- public ExternalSortExec(final TaskAttemptContext context,final SortNode plan,
+ public ExternalSortExec(final TaskAttemptContext context,final SortNode plan, final ScanNode scanNode,
final CatalogProtos.FragmentProto[] fragments) throws PhysicalPlanningException {
this(context, plan);
mergedInputFragments = TUtil.newList();
for (CatalogProtos.FragmentProto proto : fragments) {
FileFragment fragment = FragmentConvertor.convert(FileFragment.class, proto);
- mergedInputFragments.add(fragment);
+ mergedInputFragments.add(new Chunk(fragment, scanNode.getTableDesc().getMeta()));
}
}
@@ -154,9 +155,9 @@ public class ExternalSortExec extends SortExec {
/**
* Sort a tuple block and store them into a chunk file
*/
- private Path sortAndStoreChunk(int chunkId, TupleList tupleBlock)
+ private Chunk sortAndStoreChunk(int chunkId, TupleList tupleBlock)
throws IOException {
- TableMeta meta = CatalogUtil.newTableMeta("RAW");
+
int rowNum = tupleBlock.size();
long sortStart = System.currentTimeMillis();
@@ -165,7 +166,9 @@ public class ExternalSortExec extends SortExec {
long chunkWriteStart = System.currentTimeMillis();
Path outputPath = getChunkPathForWrite(0, chunkId);
- final RawFileAppender appender = new RawFileAppender(context.getConf(), null, inSchema, meta, outputPath);
+ final RawFileAppender appender =
+ new RawFileAppender(context.getConf(), null, inSchema, intermediateMeta, outputPath);
+
appender.init();
for (Tuple t : sorted) {
appender.addTuple(t);
@@ -179,7 +182,10 @@ public class ExternalSortExec extends SortExec {
FileUtil.humanReadableByteCount(appender.getOffset(), false) + " bytes, " + rowNum + " rows, " +
"sort time: " + (sortEnd - sortStart) + " msec, " +
"write time: " + (chunkWriteEnd - chunkWriteStart) + " msec)");
- return outputPath;
+
+ FileFragment frag = new FileFragment("", outputPath, 0,
+ new File(localFS.makeQualified(outputPath).toUri()).length());
+ return new Chunk(frag, intermediateMeta);
}
/**
@@ -188,10 +194,10 @@ public class ExternalSortExec extends SortExec {
* @return All paths of chunks
* @throws java.io.IOException
*/
- private List<Path> sortAndStoreAllChunks() throws IOException {
+ private List<Chunk> sortAndStoreAllChunks() throws IOException {
Tuple tuple;
long memoryConsumption = 0;
- List<Path> chunkPaths = TUtil.newList();
+ List<Chunk> chunkPaths = TUtil.newList();
int chunkId = 0;
long runStartTime = System.currentTimeMillis();
@@ -248,7 +254,8 @@ public class ExternalSortExec extends SortExec {
* Get a local path from all temporal paths in round-robin manner.
*/
private synchronized Path getChunkPathForWrite(int level, int chunkId) throws IOException {
- return localDirAllocator.getLocalPathForWrite(sortTmpDir + "/" + level +"_" + chunkId, context.getConf());
+ return localFS.makeQualified(localDirAllocator.getLocalPathForWrite(
+ sortTmpDir + "/" + level + "_" + chunkId, context.getConf()));
}
@Override
@@ -266,7 +273,7 @@ public class ExternalSortExec extends SortExec {
} else {
// Try to sort all data, and store them as multiple chunks if memory exceeds
long startTimeOfChunkSplit = System.currentTimeMillis();
- List<Path> chunks = sortAndStoreAllChunks();
+ List<Chunk> chunks = sortAndStoreAllChunks();
long endTimeOfChunkSplit = System.currentTimeMillis();
info(LOG, "Chunks creation time: " + (endTimeOfChunkSplit - startTimeOfChunkSplit) + " msec");
@@ -276,14 +283,7 @@ public class ExternalSortExec extends SortExec {
} else { // if input data exceeds main-memory at least once
try {
- List<FileFragment> fragments = TUtil.newList();
- for (Path chunk : chunks) {
- FileFragment frag = new FileFragment("", chunk, 0,
- new File(localFS.makeQualified(chunk).toUri()).length());
- fragments.add(frag);
- }
-
- this.result = externalMergeAndSort(fragments);
+ this.result = externalMergeAndSort(chunks);
} catch (Exception e) {
throw new PhysicalPlanningException(e);
}
@@ -324,11 +324,11 @@ public class ExternalSortExec extends SortExec {
return computedFanout;
}
- private Scanner externalMergeAndSort(List<FileFragment> chunks)
+ private Scanner externalMergeAndSort(List<Chunk> chunks)
throws IOException, ExecutionException, InterruptedException {
int level = 0;
- final List<FileFragment> inputFiles = TUtil.newList(chunks);
- final List<FileFragment> outputFiles = TUtil.newList();
+ final List<Chunk> inputFiles = TUtil.newList(chunks);
+ final List<Chunk> outputFiles = TUtil.newList();
int remainRun = inputFiles.size();
int chunksSize = chunks.size();
@@ -341,7 +341,7 @@ public class ExternalSortExec extends SortExec {
int remainInputRuns = inputFiles.size();
int outChunkId = 0;
int outputFileNum = 0;
- List<Future<FileFragment>> futures = TUtil.newList();
+ List<Future<Chunk>> futures = TUtil.newList();
// the number of files being merged in threads.
List<Integer> numberOfMergingFiles = TUtil.newList();
@@ -364,7 +364,7 @@ public class ExternalSortExec extends SortExec {
info(LOG, "Unbalanced merge possibility detected: number of remain input (" + remainInputRuns
+ ") and output files (" + outputFileNum + ") <= " + defaultFanout);
- List<FileFragment> switched = TUtil.newList();
+ List<Chunk> switched = TUtil.newList();
// switch the remain inputs to the next outputs
for (int j = startIdx; j < inputFiles.size(); j++) {
switched.add(inputFiles.get(j));
@@ -379,7 +379,7 @@ public class ExternalSortExec extends SortExec {
// wait for all sort runners
int finishedMerger = 0;
int index = 0;
- for (Future<FileFragment> future : futures) {
+ for (Future<Chunk> future : futures) {
outputFiles.add(future.get());
// Getting the number of merged files
finishedMerger += numberOfMergingFiles.get(index++);
@@ -405,12 +405,12 @@ public class ExternalSortExec extends SortExec {
* deleted at this point. However, for the ease of future code maintenance, we delete only type-C fragments here
*/
int numDeletedFiles = 0;
- for (FileFragment frag : inputFiles) {
- if (frag.getTableName().contains(INTERMEDIATE_FILE_PREFIX)) {
- localFS.delete(frag.getPath(), true);
+ for (Chunk chunk : inputFiles) {
+ if (chunk.getFragment().getTableName().contains(INTERMEDIATE_FILE_PREFIX)) {
+ localFS.delete(chunk.getFragment().getPath(), true);
numDeletedFiles++;
- if(LOG.isDebugEnabled()) LOG.debug("Delete merged intermediate file: " + frag);
+ if(LOG.isDebugEnabled()) LOG.debug("Delete merged intermediate file: " + chunk.getFragment());
}
}
info(LOG, numDeletedFiles + " merged intermediate files deleted");
@@ -436,15 +436,15 @@ public class ExternalSortExec extends SortExec {
/**
* Merge Thread
*/
- private class KWayMergerCaller implements Callable<FileFragment> {
+ private class KWayMergerCaller implements Callable<Chunk> {
final int level;
final int nextRunId;
- final List<FileFragment> inputFiles;
+ final List<Chunk> inputFiles;
final int startIdx;
final int mergeFanout;
final boolean updateInputStats;
- public KWayMergerCaller(final int level, final int nextRunId, final List<FileFragment> inputFiles,
+ public KWayMergerCaller(final int level, final int nextRunId, final List<Chunk> inputFiles,
final int startIdx, final int mergeFanout, final boolean updateInputStats) {
this.level = level;
this.nextRunId = nextRunId;
@@ -455,11 +455,12 @@ public class ExternalSortExec extends SortExec {
}
@Override
- public FileFragment call() throws Exception {
+ public Chunk call() throws Exception {
final Path outputPath = getChunkPathForWrite(level + 1, nextRunId);
info(LOG, mergeFanout + " files are being merged to an output file " + outputPath.getName());
long mergeStartTime = System.currentTimeMillis();
- final RawFileAppender output = new RawFileAppender(context.getConf(), null, inSchema, meta, outputPath);
+ final RawFileAppender output =
+ new RawFileAppender(context.getConf(), null, inSchema, intermediateMeta, outputPath);
output.init();
final Scanner merger = createKWayMerger(inputFiles, startIdx, mergeFanout);
merger.init();
@@ -475,7 +476,7 @@ public class ExternalSortExec extends SortExec {
+ " bytes, " + (mergeEndTime - mergeStartTime) + " msec)");
File f = new File(localFS.makeQualified(outputPath).toUri());
FileFragment frag = new FileFragment(INTERMEDIATE_FILE_PREFIX + outputPath.getName(), outputPath, 0, f.length());
- return frag;
+ return new Chunk(frag, intermediateMeta);
}
}
@@ -489,7 +490,7 @@ public class ExternalSortExec extends SortExec {
/**
* Create a merged file scanner or k-way merge scanner.
*/
- private Scanner createFinalMerger(List<FileFragment> inputs) throws IOException {
+ private Scanner createFinalMerger(List<Chunk> inputs) throws IOException {
if (inputs.size() == 1) {
this.result = getFileScanner(inputs.get(0));
} else {
@@ -498,11 +499,11 @@ public class ExternalSortExec extends SortExec {
return result;
}
- private Scanner getFileScanner(FileFragment frag) throws IOException {
- return new RawFileScanner(context.getConf(), plan.getInSchema(), meta, frag);
+ private Scanner getFileScanner(Chunk chunk) throws IOException {
+ return TablespaceManager.getLocalFs().getScanner(chunk.getMeta(), inSchema, chunk.getFragment(), outSchema);
}
- private Scanner createKWayMerger(List<FileFragment> inputs, final int startChunkId, final int num) throws IOException {
+ private Scanner createKWayMerger(List<Chunk> inputs, final int startChunkId, final int num) throws IOException {
final Scanner [] sources = new Scanner[num];
for (int i = 0; i < num; i++) {
sources[i] = getFileScanner(inputs.get(startChunkId + i));
@@ -773,6 +774,8 @@ public class ExternalSortExec extends SortExec {
@Override
public void close() throws IOException {
+ super.close();
+
if (result != null) {
result.close();
try {
@@ -784,7 +787,8 @@ public class ExternalSortExec extends SortExec {
}
if (finalOutputFiles != null) {
- for (FileFragment frag : finalOutputFiles) {
+ for (Chunk chunk : finalOutputFiles) {
+ FileFragment frag = chunk.getFragment();
File tmpFile = new File(localFS.makeQualified(frag.getPath()).toUri());
if (frag.getStartKey() == 0 && frag.getLength() == tmpFile.length()) {
localFS.delete(frag.getPath(), true);
@@ -804,7 +808,6 @@ public class ExternalSortExec extends SortExec {
}
plan = null;
- super.close();
}
@Override
@@ -833,4 +836,22 @@ public class ExternalSortExec extends SortExec {
return inputStats;
}
}
+
+ private static class Chunk {
+ private FileFragment fragment;
+ private TableMeta meta;
+
+ public Chunk(FileFragment fragment, TableMeta meta) {
+ this.fragment = fragment;
+ this.meta = meta;
+ }
+
+ public FileFragment getFragment() {
+ return fragment;
+ }
+
+ public TableMeta getMeta() {
+ return meta;
+ }
+ }
}