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/26 03:58:20 UTC
tajo git commit: TAJO-1983: Improve memory usage of ExternalSortExec.
Repository: tajo
Updated Branches:
refs/heads/master 4e0a7a33c -> 550c0189b
TAJO-1983: Improve memory usage of ExternalSortExec.
Closes #869
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/550c0189
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/550c0189
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/550c0189
Branch: refs/heads/master
Commit: 550c0189b9ac06f20d4ef70feb7ef743c0b06d0f
Parents: 4e0a7a3
Author: Jinho Kim <jh...@apache.org>
Authored: Thu Nov 26 11:57:02 2015 +0900
Committer: Jinho Kim <jh...@apache.org>
Committed: Thu Nov 26 11:57:02 2015 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../org/apache/tajo/tuple/BaseTupleBuilder.java | 23 +-
.../tuple/memory/CompactRowBlockWriter.java | 30 +-
.../tajo/tuple/memory/OffHeapRowBlockUtils.java | 74 ++-
.../tuple/memory/OffHeapRowBlockWriter.java | 25 +-
.../tajo/tuple/memory/OffHeapRowWriter.java | 16 +
.../tajo/tuple/memory/ResizableMemoryBlock.java | 10 +-
.../org/apache/tajo/tuple/memory/RowWriter.java | 9 +-
.../tajo/tuple/memory/UnSafeTupleList.java | 105 ++++
.../tajo/tuple/memory/TestMemoryRowBlock.java | 40 +-
.../physical/TestProgressExternalSortExec.java | 8 +-
.../planner/physical/TestUnSafeTuple.java | 104 ++++
.../tajo/querymaster/TestTaskStatusUpdate.java | 4 +-
.../planner/physical/ExternalSortExec.java | 482 +++++++++++++------
.../engine/planner/physical/PhysicalExec.java | 4 +
.../tajo/worker/ExecutionBlockContext.java | 5 +
.../java/org/apache/tajo/worker/Fetcher.java | 17 +-
.../java/org/apache/tajo/worker/TaskImpl.java | 29 +-
.../tajo/pullserver/TajoPullServerService.java | 24 +-
.../tajo/storage/BaseTupleComparator.java | 45 +-
.../org/apache/tajo/storage/NullScanner.java | 1 -
21 files changed, 831 insertions(+), 226 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index b9d73c8..00b9129 100644
--- a/CHANGES
+++ b/CHANGES
@@ -8,6 +8,8 @@ Release 0.12.0 - unreleased
IMPROVEMENT
+ TAJO-1983: Improve memory usage of ExternalSortExec. (jinho)
+
TAJO-1986: Rename the name 'option' to 'property' in TableMeta. (hyunsik)
TAJO-1988: Remove some duplicated codes in toString() of Projectable.
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/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 ebdcc26..d962218 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
@@ -19,6 +19,7 @@
package org.apache.tajo.tuple;
import org.apache.tajo.common.TajoDataTypes.DataType;
+import org.apache.tajo.exception.ValueOutOfRangeException;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.tuple.memory.*;
import org.apache.tajo.unit.StorageUnit;
@@ -54,6 +55,11 @@ public class BaseTupleBuilder extends OffHeapRowWriter implements TupleBuilder,
}
@Override
+ public void backward(int length) {
+ memoryBlock.writerPosition(memoryBlock.writerPosition() - length);
+ }
+
+ @Override
public boolean startRow() {
memoryBlock.writerPosition(0);
return super.startRow();
@@ -65,13 +71,18 @@ public class BaseTupleBuilder extends OffHeapRowWriter implements TupleBuilder,
}
@Override
- public void addTuple(Tuple tuple) {
- if (tuple instanceof UnSafeTuple) {
- UnSafeTuple unSafeTuple = TUtil.checkTypeAndGet(tuple, UnSafeTuple.class);
- addTuple(unSafeTuple);
- } else {
- OffHeapRowBlockUtils.convert(tuple, this);
+ public boolean addTuple(Tuple tuple) {
+ try {
+ if (tuple instanceof UnSafeTuple) {
+ UnSafeTuple unSafeTuple = TUtil.checkTypeAndGet(tuple, UnSafeTuple.class);
+ addTuple(unSafeTuple);
+ } else {
+ OffHeapRowBlockUtils.convert(tuple, this);
+ }
+ } catch (ValueOutOfRangeException e) {
+ return false;
}
+ return true;
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/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
index a88d2f1..67d5f8c 100644
--- 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
@@ -25,6 +25,7 @@ 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.exception.ValueOutOfRangeException;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.util.BitArray;
import org.apache.tajo.util.SizeOf;
@@ -182,6 +183,15 @@ public class CompactRowBlockWriter implements RowWriter {
rowBlock.getMemory().writerPosition(rowBlock.getMemory().writerPosition() + length);
}
+ /**
+ * Backward the address;
+ *
+ * @param length Length to be backwarded
+ */
+ public void backward(int length) {
+ rowBlock.getMemory().writerPosition(rowBlock.getMemory().writerPosition() - length);
+ }
+
public void ensureSize(int size) {
rowBlock.getMemory().ensureSize(size);
}
@@ -232,7 +242,7 @@ public class CompactRowBlockWriter implements RowWriter {
return true;
}
-
+ @Override
public void endRow() {
long rowHeaderPos = recordStartAddr();
// curOffset is equivalent to a byte length of this row.
@@ -249,6 +259,15 @@ public class CompactRowBlockWriter implements RowWriter {
}
@Override
+ public void cancelRow() {
+ // curOffset is equivalent to a byte length of current row.
+ backward(curOffset);
+ curOffset = 0;
+ nullFlags.clear();
+ curFieldIdx = 0;
+ }
+
+ @Override
public void skipField() {
// set null flag
nullFlags.set(curFieldIdx);
@@ -402,7 +421,12 @@ public class CompactRowBlockWriter implements RowWriter {
}
@Override
- public void addTuple(Tuple tuple) {
- OffHeapRowBlockUtils.convert(tuple, this);
+ public boolean addTuple(Tuple tuple) {
+ try {
+ OffHeapRowBlockUtils.convert(tuple, this);
+ } catch (ValueOutOfRangeException e) {
+ return false;
+ }
+ return true;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/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 1aca22f..3f27763 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
@@ -19,10 +19,13 @@
package org.apache.tajo.tuple.memory;
import com.google.common.collect.Lists;
+import com.google.common.primitives.*;
+import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.datum.IntervalDatum;
import org.apache.tajo.datum.ProtobufDatum;
import org.apache.tajo.exception.TajoRuntimeException;
import org.apache.tajo.exception.UnsupportedException;
+import org.apache.tajo.exception.ValueOutOfRangeException;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.tuple.RowBlockReader;
@@ -62,6 +65,11 @@ public class OffHeapRowBlockUtils {
return tupleList;
}
+ public static List<UnSafeTuple> sort(UnSafeTupleList list, Comparator<UnSafeTuple> comparator) {
+ Collections.sort(list, comparator);
+ return list;
+ }
+
public static Tuple[] sortToArray(MemoryRowBlock rowBlock, Comparator<Tuple> comparator) {
Tuple[] tuples = new Tuple[rowBlock.rows()];
@@ -85,18 +93,74 @@ public class OffHeapRowBlockUtils {
return tuples;
}
+ public static final int compareColumn(UnSafeTuple tuple1, UnSafeTuple tuple2, int index, TajoDataTypes.Type type,
+ boolean ascending, boolean nullFirst) {
+ final boolean n1 = tuple1.isBlankOrNull(index);
+ final boolean n2 = tuple2.isBlankOrNull(index);
+ if (n1 && n2) {
+ return 0;
+ }
+
+ if (n1 ^ n2) {
+ return nullFirst ? (n1 ? -1 : 1) : (n1 ? 1 : -1);
+ }
+
+ int compare;
+ switch (type) {
+ case BOOLEAN:
+ compare = Booleans.compare(tuple1.getBool(index), tuple2.getBool(index));
+ break;
+ case BIT:
+ compare = tuple1.getByte(index) - tuple2.getByte(index);
+ break;
+ case INT1:
+ case INT2:
+ compare = Shorts.compare(tuple1.getInt2(index), tuple2.getInt2(index));
+ break;
+ case DATE:
+ case INT4:
+ compare = Ints.compare(tuple1.getInt4(index), tuple2.getInt4(index));
+ break;
+ case INET4:
+ compare = UnsignedInts.compare(tuple1.getInt4(index), tuple2.getInt4(index));
+ break;
+ case TIME:
+ case TIMESTAMP:
+ case INT8:
+ compare = Longs.compare(tuple1.getInt8(index), tuple2.getInt8(index));
+ break;
+ case FLOAT4:
+ compare = Floats.compare(tuple1.getFloat4(index), tuple2.getFloat4(index));
+ break;
+ case FLOAT8:
+ compare = Doubles.compare(tuple1.getFloat8(index), tuple2.getFloat8(index));
+ break;
+ case CHAR:
+ case TEXT:
+ case BLOB:
+ compare = UnSafeTupleBytesComparator.compare(tuple1.getFieldAddr(index), tuple2.getFieldAddr(index));
+ break;
+ default:
+ throw new TajoRuntimeException(
+ new UnsupportedException("unknown data type '" + type.name() + "'"));
+ }
+ return ascending ? compare : -compare;
+ }
/**
* This class is tuple converter to the RowBlock
*/
public static class TupleConverter {
public void convert(Tuple tuple, RowWriter writer) {
- writer.startRow();
-
- for (int i = 0; i < writer.dataTypes().length; i++) {
- writeField(i, tuple, writer);
+ try {
+ writer.startRow();
+ for (int i = 0; i < writer.dataTypes().length; i++) {
+ writeField(i, tuple, writer);
+ }
+ } catch (ValueOutOfRangeException e) {
+ writer.cancelRow();
+ throw e;
}
-
writer.endRow();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockWriter.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockWriter.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockWriter.java
index 9f3d8a2..594f927 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockWriter.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowBlockWriter.java
@@ -20,6 +20,7 @@ package org.apache.tajo.tuple.memory;
import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.exception.TajoInternalError;
+import org.apache.tajo.exception.ValueOutOfRangeException;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.util.TUtil;
@@ -48,6 +49,11 @@ public class OffHeapRowBlockWriter extends OffHeapRowWriter {
rowBlock.getMemory().writerPosition(rowBlock.getMemory().writerPosition() + length);
}
+ @Override
+ public void backward(int length) {
+ rowBlock.getMemory().writerPosition(rowBlock.getMemory().writerPosition() - length);
+ }
+
public void ensureSize(int size) {
rowBlock.getMemory().ensureSize(size);
}
@@ -65,13 +71,18 @@ public class OffHeapRowBlockWriter extends OffHeapRowWriter {
@Override
- public void addTuple(Tuple tuple) {
- if (tuple instanceof UnSafeTuple) {
- UnSafeTuple unSafeTuple = TUtil.checkTypeAndGet(tuple, UnSafeTuple.class);
- addTuple(unSafeTuple);
- rowBlock.setRows(rowBlock.rows() + 1);
- } else {
- OffHeapRowBlockUtils.convert(tuple, this);
+ public boolean addTuple(Tuple tuple) {
+ try {
+ if (tuple instanceof UnSafeTuple) {
+ UnSafeTuple unSafeTuple = TUtil.checkTypeAndGet(tuple, UnSafeTuple.class);
+ addTuple(unSafeTuple);
+ rowBlock.setRows(rowBlock.rows() + 1);
+ } else {
+ OffHeapRowBlockUtils.convert(tuple, this);
+ }
+ } catch (ValueOutOfRangeException e) {
+ return false;
}
+ return true;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowWriter.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowWriter.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowWriter.java
index f082762..e992aed 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowWriter.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/OffHeapRowWriter.java
@@ -94,6 +94,12 @@ public abstract class OffHeapRowWriter implements RowWriter {
*/
public abstract void forward(int length);
+ /**
+ * Backward the address;
+ *
+ * @param length Length to be backwarded
+ */
+ public abstract void backward(int length);
@Override
public void clear() {
@@ -130,6 +136,16 @@ public abstract class OffHeapRowWriter implements RowWriter {
PlatformDependent.putInt(rowHeaderPos, MemoryRowBlock.NULL_FIELD_OFFSET);
rowHeaderPos += SizeOf.SIZE_OF_INT;
}
+ curOffset = 0;
+ }
+
+ @Override
+ public void cancelRow() {
+ // curOffset is equivalent to a byte length of current row.
+ backward(curOffset);
+ curOffset = 0;
+ curFieldOffset = SizeOf.SIZE_OF_INT;
+ curFieldIdx = 0;
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/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 09faff9..6b06486 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
@@ -23,6 +23,7 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.exception.ValueOutOfRangeException;
import org.apache.tajo.storage.BufferPool;
import org.apache.tajo.util.FileUtil;
import org.apache.tajo.util.UnsafeUtil;
@@ -128,11 +129,12 @@ public class ResizableMemoryBlock implements MemoryBlock {
@Override
public void ensureSize(int size) {
if (!buffer.isWritable(size)) {
- if (!limitSpec.canIncrease(size)) {
- throw new RuntimeException("Cannot increase RowBlock anymore.");
+ int newBlockSize = limitSpec.increasedSize(size);
+
+ if (!limitSpec.canIncrease(buffer.writableBytes() + newBlockSize)) {
+ throw new ValueOutOfRangeException("Cannot increase RowBlock anymore.");
}
- int newBlockSize = limitSpec.increasedSize(size);
resize(newBlockSize);
LOG.info("Increase DirectRowBlock to " + FileUtil.humanReadableByteCount(newBlockSize, false));
}
@@ -142,7 +144,7 @@ public class ResizableMemoryBlock implements MemoryBlock {
Preconditions.checkArgument(newSize > 0, "Size must be greater than 0 bytes");
if (newSize > limitSpec.limit()) {
- throw new RuntimeException("Resize cannot exceed the capacity limit");
+ throw new ValueOutOfRangeException("Resize cannot exceed the capacity limit");
}
if (newSize < buffer.writableBytes()) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowWriter.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowWriter.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowWriter.java
index 0393714..e055ab7 100644
--- a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowWriter.java
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/RowWriter.java
@@ -30,6 +30,11 @@ import org.apache.tajo.storage.Tuple;
* startRow() --> skipField() or putXXX --> endRow()
* </pre>
*
+ * If you want to cancel the current row, should be as follows:
+ * <pre>
+ * startRow() --> skipField() or putXXX --> cancelRow()
+ * </pre>
+ *
* The total number of skipField and putXXX invocations must be equivalent to the number of fields.
*/
public interface RowWriter {
@@ -40,6 +45,8 @@ public interface RowWriter {
void endRow();
+ void cancelRow();
+
void skipField();
void clear();
@@ -76,5 +83,5 @@ public interface RowWriter {
void putProtoDatum(ProtobufDatum datum);
- void addTuple(Tuple tuple);
+ boolean addTuple(Tuple tuple);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTupleList.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTupleList.java b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTupleList.java
new file mode 100644
index 0000000..4c4a6cb
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/tuple/memory/UnSafeTupleList.java
@@ -0,0 +1,105 @@
+/**
+ * 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 com.google.common.collect.Lists;
+import org.apache.tajo.common.TajoDataTypes.DataType;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.unit.StorageUnit;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * In UnSafeTupleList, input tuples are copied to off-heap memory page whenever the add() method is called.
+ * The memory pages are automatically added, if memory of a page are exceeded.
+ * This instance must be released
+ */
+public class UnSafeTupleList extends ArrayList<UnSafeTuple> {
+
+ private final DataType[] dataTypes;
+ private List<MemoryRowBlock> rowBlocks;
+ private MemoryRowBlock currentRowBlock;
+ private int totalUsedMem;
+ private int pageSize;
+
+ public UnSafeTupleList(DataType[] dataTypes, int initialArraySize) {
+ this(dataTypes, initialArraySize, StorageUnit.MB);
+
+ }
+
+ public UnSafeTupleList(DataType[] dataTypes, int initialArraySize, int pageSize) {
+ super(initialArraySize);
+ this.dataTypes = dataTypes;
+ this.pageSize = pageSize;
+ this.rowBlocks = Lists.newArrayList();
+ this.currentRowBlock = new MemoryRowBlock(dataTypes, new FixedSizeLimitSpec(pageSize), true);
+ this.rowBlocks.add(currentRowBlock);
+
+ }
+
+ @Override
+ public boolean add(UnSafeTuple tuple) {
+ return addTuple(tuple);
+ }
+
+ public boolean addTuple(Tuple tuple) {
+
+ int prevPos = currentRowBlock.getMemory().writerPosition();
+ if (currentRowBlock.getWriter().addTuple(tuple)) {
+ UnSafeTuple unSafeTuple = new UnSafeTuple();
+ unSafeTuple.set(currentRowBlock.getMemory(), prevPos, dataTypes);
+ return super.add(unSafeTuple);
+ } else {
+ this.totalUsedMem += currentRowBlock.usedMem();
+ this.currentRowBlock = new MemoryRowBlock(dataTypes, new FixedSizeLimitSpec(pageSize), true);
+ this.rowBlocks.add(currentRowBlock);
+ return this.addTuple(tuple);
+ }
+ }
+
+ /**
+ * Release the cached pages
+ */
+ public void release() {
+ for (MemoryRowBlock rowBlock : rowBlocks) {
+ rowBlock.release();
+ }
+ super.clear();
+ rowBlocks.clear();
+ totalUsedMem = 0;
+ }
+
+ /**
+ * Total used memory
+ */
+ public int usedMem() {
+ return totalUsedMem + currentRowBlock.usedMem();
+ }
+
+ /**
+ * Release and reset
+ */
+ @Override
+ public void clear() {
+ release();
+ this.currentRowBlock = new MemoryRowBlock(dataTypes, new FixedSizeLimitSpec(pageSize), true);
+ this.rowBlocks.add(currentRowBlock);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-common/src/test/java/org/apache/tajo/tuple/memory/TestMemoryRowBlock.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/test/java/org/apache/tajo/tuple/memory/TestMemoryRowBlock.java b/tajo-common/src/test/java/org/apache/tajo/tuple/memory/TestMemoryRowBlock.java
index a6003c7..4e77d68 100644
--- a/tajo-common/src/test/java/org/apache/tajo/tuple/memory/TestMemoryRowBlock.java
+++ b/tajo-common/src/test/java/org/apache/tajo/tuple/memory/TestMemoryRowBlock.java
@@ -24,6 +24,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.datum.ProtobufDatum;
+import org.apache.tajo.exception.ValueOutOfRangeException;
import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.storage.VTuple;
@@ -39,9 +40,7 @@ import java.util.Comparator;
import java.util.List;
import static org.apache.tajo.common.TajoDataTypes.Type;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
public class TestMemoryRowBlock {
private static final Log LOG = LogFactory.getLog(TestMemoryRowBlock.class);
@@ -115,6 +114,41 @@ public class TestMemoryRowBlock {
}
@Test
+ public void testPutAndCancelValidation() {
+ VTuple vTuple = new VTuple(schema.length);
+ fillVTuple(0, vTuple);
+
+ //get memory size of 1 row
+ MemoryRowBlock rowBlock = new MemoryRowBlock(schema);
+ fillRow(0, rowBlock.getWriter());
+ int rowSize = rowBlock.usedMem();
+ rowBlock.release();
+
+ rowBlock = new MemoryRowBlock(schema, new FixedSizeLimitSpec(rowSize / 2, 0.0f), true);
+ assertFalse(rowBlock.getWriter().addTuple(vTuple));
+ try {
+ OffHeapRowBlockUtils.convert(vTuple, rowBlock.getWriter());
+ fail();
+ } catch (Exception e) {
+ assertEquals(ValueOutOfRangeException.class, e.getClass());
+ }
+ rowBlock.release();
+
+ //allow 1 row
+ rowBlock = new MemoryRowBlock(schema, new FixedSizeLimitSpec(rowSize, 0.0f), true);
+ assertTrue(rowBlock.getWriter().addTuple(vTuple));
+ assertFalse(rowBlock.getWriter().addTuple(vTuple));
+ assertEquals(1, rowBlock.rows());
+
+ ZeroCopyTuple tuple = new UnSafeTuple();
+ RowBlockReader reader = rowBlock.getReader();
+ assertTrue(reader.next(tuple));
+ validateTupleResult(0, tuple);
+ assertFalse(reader.next(tuple));
+ rowBlock.release();
+ }
+
+ @Test
public void testNullityValidation() {
int rowNum = 1000;
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/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 b1f53da..15250a3 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
@@ -110,6 +110,7 @@ public class TestProgressExternalSortExec {
catalog.createTable(employee);
analyzer = new SQLAnalyzer();
planner = new LogicalPlanner(catalog, TablespaceManager.getInstance());
+ employeePath.getFileSystem(conf).deleteOnExit(employeePath);
}
@After
@@ -125,9 +126,8 @@ public class TestProgressExternalSortExec {
@Test
public void testExternalSortExecProgressWithMemTableScanner() throws Exception {
QueryContext queryContext = LocalTajoTestingUtility.createDummyContext(conf);
- int bufferSize = (int) (testDataStats.getNumBytes() * 20) / StorageUnit.MB; //multiply 2 for memory fit
+ int bufferSize = (int) (testDataStats.getNumBytes() * 2) / StorageUnit.MB; //multiply 2 for memory fit
queryContext.setInt(SessionVars.EXTSORT_BUFFER_SIZE, bufferSize);
-
testProgress(queryContext);
}
@@ -197,7 +197,7 @@ public class TestProgressExternalSortExec {
assertEquals(testDataStats.getNumBytes().longValue(), tableStats.getNumBytes().longValue());
assertEquals(testDataStats.getNumRows().longValue(), cnt);
assertEquals(testDataStats.getNumRows().longValue(), tableStats.getNumRows().longValue());
- assertTrue(testDataStats.getNumBytes().longValue() <= tableStats.getReadBytes().longValue());
+ assertTrue(testDataStats.getNumBytes() <= tableStats.getReadBytes());
// for rescan test
preVal = null;
@@ -223,7 +223,7 @@ public class TestProgressExternalSortExec {
assertEquals(testDataStats.getNumRows().longValue(), cnt);
assertEquals(testDataStats.getNumRows().longValue(), tableStats.getNumRows().longValue());
//'ReadBytes' is actual read bytes
- assertTrue(testDataStats.getNumBytes().longValue() <= tableStats.getReadBytes().longValue());
+ assertTrue(testDataStats.getNumBytes() <= tableStats.getReadBytes());
conf.setIntVar(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT, ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT.defaultIntVal);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestUnSafeTuple.java
----------------------------------------------------------------------
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestUnSafeTuple.java b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestUnSafeTuple.java
new file mode 100644
index 0000000..48170f6
--- /dev/null
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/engine/planner/physical/TestUnSafeTuple.java
@@ -0,0 +1,104 @@
+/**
+ * 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.engine.planner.physical;
+
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SchemaUtil;
+import org.apache.tajo.common.TajoDataTypes.Type;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+import org.apache.tajo.tuple.memory.UnSafeTupleList;
+import org.apache.tajo.unit.StorageUnit;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.Random;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestUnSafeTuple {
+
+ private static final Random rnd = new Random(-1);
+ private static Schema schema;
+
+ @BeforeClass
+ public static void setupClass() {
+ Column col0 = new Column("col0", Type.BOOLEAN);
+ Column col1 = new Column("col1", Type.INT4);
+ Column col2 = new Column("col2", Type.INT8);
+ Column col3 = new Column("col3", Type.FLOAT4);
+ Column col4 = new Column("col4", Type.FLOAT8);
+
+ schema = new Schema(new Column[]{col0, col1, col2, col3, col4});
+ }
+
+ @Test
+ public final void testMemoryPageAndValidation() {
+
+ Datum[] datums = new Datum[]{
+ DatumFactory.createBool(rnd.nextBoolean()),
+ DatumFactory.createInt4(rnd.nextInt()),
+ DatumFactory.createInt8(rnd.nextLong()),
+ DatumFactory.createFloat4(rnd.nextFloat()),
+ DatumFactory.createFloat8(rnd.nextDouble())};
+ Tuple tuple = new VTuple(datums);
+
+ int pageSize = StorageUnit.KB;
+ UnSafeTupleList unSafeTupleList = new UnSafeTupleList(SchemaUtil.toDataTypes(schema), 100, StorageUnit.KB);
+ assertEquals(0, unSafeTupleList.usedMem());
+ assertEquals(0, unSafeTupleList.size());
+
+ unSafeTupleList.addTuple(tuple);
+ //get the memory bytes of tuple
+ int tupleSize = unSafeTupleList.usedMem();
+ assertEquals(1, unSafeTupleList.size());
+ assertEquals(tuple, unSafeTupleList.get(0));
+
+ unSafeTupleList.clear();
+ assertEquals(0, unSafeTupleList.usedMem());
+ assertEquals(0, unSafeTupleList.size());
+
+ //test only 1 page
+ int testCount = pageSize / tupleSize;
+ Tuple[] tuples = new Tuple[testCount];
+
+ for (int i = 0; i < testCount; i++) {
+ datums = new Datum[]{
+ DatumFactory.createBool(rnd.nextBoolean()),
+ DatumFactory.createInt4(rnd.nextInt()),
+ DatumFactory.createInt8(rnd.nextLong()),
+ DatumFactory.createFloat4(rnd.nextFloat()),
+ DatumFactory.createFloat8(rnd.nextDouble())};
+ tuples[i] = new VTuple(datums);
+ unSafeTupleList.addTuple(tuples[i]);
+ }
+
+ assertEquals(testCount, unSafeTupleList.size());
+ assertEquals(tupleSize * testCount, unSafeTupleList.usedMem());
+
+ for (int i = 0; i < testCount; i++) {
+ assertEquals(tuples[i], unSafeTupleList.get(i));
+ }
+
+ unSafeTupleList.release();
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/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 9f5ecb5..425a7d6 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
@@ -79,7 +79,7 @@ public class TestTaskStatusUpdate extends QueryTestCaseBase {
// tpch/lineitem.tbl
long[] expectedNumRows = new long[]{5, 2, 2, 2, 2, 2};
long[] expectedNumBytes = new long[]{604, 162, 162, 138, 138, 236};
- long[] expectedReadBytes = new long[]{604, 604, 162, 0, 138, 0};
+ long[] expectedReadBytes = new long[]{604, 604, 236, 0, 138, 0};
QueryId queryId = getQueryId(res);
assertStatus(queryId, 3, expectedNumRows, expectedNumBytes, expectedReadBytes);
@@ -108,7 +108,7 @@ public class TestTaskStatusUpdate extends QueryTestCaseBase {
// in/out * stage(4)
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};
+ long[] expectedReadBytes = new long[]{20, 20, 8, 8, 64, 0, 34, 0};
QueryId queryId = getQueryId(res);
assertStatus(queryId, 4, expectedNumRows, expectedNumBytes, expectedReadBytes);
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/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 8abafec..69631b5 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
@@ -18,6 +18,9 @@
package org.apache.tajo.engine.planner.physical;
+import com.google.common.base.Preconditions;
+import com.google.common.primitives.*;
+import com.google.common.util.concurrent.SettableFuture;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.LocalDirAllocator;
@@ -26,18 +29,24 @@ 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;
-import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.datum.TextDatum;
import org.apache.tajo.engine.planner.PhysicalPlanningException;
+import org.apache.tajo.exception.TajoRuntimeException;
+import org.apache.tajo.exception.UnsupportedException;
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;
import org.apache.tajo.storage.fragment.FragmentConvertor;
+import org.apache.tajo.storage.rawfile.DirectRawFileWriter;
+import org.apache.tajo.tuple.memory.OffHeapRowBlockUtils;
+import org.apache.tajo.tuple.memory.UnSafeTuple;
+import org.apache.tajo.tuple.memory.UnSafeTupleList;
import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.FileUtil;
import org.apache.tajo.util.TUtil;
@@ -48,9 +57,10 @@ import java.io.IOException;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
-import java.util.concurrent.*;
-
-import static org.apache.tajo.storage.RawFile.RawFileAppender;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
/**
* This external sort algorithm can be characterized by the followings:
@@ -75,15 +85,19 @@ public class ExternalSortExec extends SortExec {
/** 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. */
- private long sortBufferBytesNum;
+ private int sortBufferBytesNum;
/** the number of available cores */
private final int allocatedCoreNum;
/** If there are available multiple cores, it tries parallel merge. */
private ExecutorService executorService;
/** used for in-memory sort of each chunk. */
- private TupleList inMemoryTable;
+ private UnSafeTupleList inMemoryTable;
+ /** for zero copy tuple comparison */
+ private Comparator<UnSafeTuple> unSafeComparator;
+ /** for other type tuple comparison */
+ private Comparator<Tuple> primitiveComparator;
/** temporal dir */
- private final Path sortTmpDir;
+ private Path sortTmpDir;
/** It enables round-robin disks allocation */
private final LocalDirAllocator localDirAllocator;
/** local file system */
@@ -98,12 +112,10 @@ public class ExternalSortExec extends SortExec {
///////////////////////////////////////////////////
/** already sorted or not */
private boolean sorted = false;
- /** a flag to point whether sorted data resides in memory or not */
- private boolean memoryResident = true;
/** the final result */
private Scanner result;
/** total bytes of input data */
- private long sortAndStoredBytes;
+ private long inputBytes;
private ExternalSortExec(final TaskAttemptContext context, final SortNode plan)
throws PhysicalPlanningException {
@@ -115,15 +127,12 @@ public class ExternalSortExec extends SortExec {
throw new PhysicalPlanningException(ConfVars.EXECUTOR_EXTERNAL_SORT_FANOUT.varname + " cannot be lower than 2");
}
// TODO - sort buffer and core num should be changed to use the allocated container resource.
- this.sortBufferBytesNum = context.getQueryContext().getLong(SessionVars.EXTSORT_BUFFER_SIZE) * StorageUnit.MB;
+ this.sortBufferBytesNum = context.getQueryContext().getInt(SessionVars.EXTSORT_BUFFER_SIZE) * StorageUnit.MB;
this.allocatedCoreNum = context.getConf().getIntVar(ConfVars.EXECUTOR_EXTERNAL_SORT_THREAD_NUM);
- this.executorService = Executors.newFixedThreadPool(this.allocatedCoreNum);
- this.inMemoryTable = new TupleList(context.getQueryContext().getInt(SessionVars.SORT_LIST_SIZE));
-
- this.sortTmpDir = getExecutorTmpDir();
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
+ this.intermediateMeta = CatalogUtil.newTableMeta(BuiltinStorages.DRAW);
+ this.inputStats = new TableStats();
}
public ExternalSortExec(final TaskAttemptContext context,final SortNode plan, final ScanNode scanNode,
@@ -133,7 +142,7 @@ public class ExternalSortExec extends SortExec {
mergedInputFragments = TUtil.newList();
for (CatalogProtos.FragmentProto proto : fragments) {
FileFragment fragment = FragmentConvertor.convert(FileFragment.class, proto);
- mergedInputFragments.add(new Chunk(fragment, scanNode.getTableDesc().getMeta()));
+ mergedInputFragments.add(new Chunk(inSchema, fragment, scanNode.getTableDesc().getMeta()));
}
}
@@ -143,8 +152,19 @@ public class ExternalSortExec extends SortExec {
setChild(child);
}
+ @Override
public void init() throws IOException {
- inputStats = new TableStats();
+ if(allocatedCoreNum > 1) {
+ this.executorService = Executors.newFixedThreadPool(this.allocatedCoreNum);
+ }
+
+ this.sortTmpDir = getExecutorTmpDir();
+
+ int initialArraySize = context.getQueryContext().getInt(SessionVars.SORT_LIST_SIZE);
+ this.inMemoryTable = new UnSafeTupleList(SchemaUtil.toDataTypes(inSchema), initialArraySize);
+ this.unSafeComparator = new UnSafeComparator(inSchema, sortSpecs);
+ this.primitiveComparator = new PrimitiveComparator(inSchema, sortSpecs);
+
super.init();
}
@@ -155,29 +175,25 @@ public class ExternalSortExec extends SortExec {
/**
* Sort a tuple block and store them into a chunk file
*/
- private Chunk sortAndStoreChunk(int chunkId, TupleList tupleBlock)
+ private Chunk sortAndStoreChunk(int chunkId, UnSafeTupleList tupleBlock)
throws IOException {
-
int rowNum = tupleBlock.size();
long sortStart = System.currentTimeMillis();
- Iterable<Tuple> sorted = getSorter(tupleBlock).sort();
+ OffHeapRowBlockUtils.sort(tupleBlock, unSafeComparator);
long sortEnd = System.currentTimeMillis();
long chunkWriteStart = System.currentTimeMillis();
Path outputPath = getChunkPathForWrite(0, chunkId);
- final RawFileAppender appender =
- new RawFileAppender(context.getConf(), null, inSchema, intermediateMeta, outputPath);
-
+ final DirectRawFileWriter appender =
+ new DirectRawFileWriter(context.getConf(), null, inSchema, intermediateMeta, outputPath);
appender.init();
- for (Tuple t : sorted) {
+ for (Tuple t : tupleBlock) {
appender.addTuple(t);
}
appender.close();
- tupleBlock.clear();
long chunkWriteEnd = System.currentTimeMillis();
-
info(LOG, "Chunk #" + chunkId + " sort and written (" +
FileUtil.humanReadableByteCount(appender.getOffset(), false) + " bytes, " + rowNum + " rows, " +
"sort time: " + (sortEnd - sortStart) + " msec, " +
@@ -185,7 +201,7 @@ public class ExternalSortExec extends SortExec {
FileFragment frag = new FileFragment("", outputPath, 0,
new File(localFS.makeQualified(outputPath).toUri()).length());
- return new Chunk(frag, intermediateMeta);
+ return new Chunk(inSchema, frag, intermediateMeta);
}
/**
@@ -196,26 +212,23 @@ public class ExternalSortExec extends SortExec {
*/
private List<Chunk> sortAndStoreAllChunks() throws IOException {
Tuple tuple;
- long memoryConsumption = 0;
List<Chunk> chunkPaths = TUtil.newList();
int chunkId = 0;
long runStartTime = System.currentTimeMillis();
+
while (!context.isStopped() && (tuple = child.next()) != null) { // partition sort start
- inMemoryTable.add(tuple);
- memoryConsumption += MemoryUtil.calculateMemorySize(tuple);
+ inMemoryTable.addTuple(tuple);
- if (memoryConsumption > sortBufferBytesNum) {
+ if (inMemoryTable.usedMem() > sortBufferBytesNum) { // if input data exceeds main-memory at least once
long runEndTime = System.currentTimeMillis();
- info(LOG, chunkId + " run loading time: " + (runEndTime - runStartTime) + " msec");
+ info(LOG, "Chunk #" + chunkId + " run loading time: " + (runEndTime - runStartTime) + " msec");
runStartTime = runEndTime;
- info(LOG, "Memory consumption exceeds " + sortBufferBytesNum + " bytes");
- memoryResident = false;
+ info(LOG, "Memory consumption exceeds " + FileUtil.humanReadableByteCount(inMemoryTable.usedMem(), false));
chunkPaths.add(sortAndStoreChunk(chunkId, inMemoryTable));
-
- memoryConsumption = 0;
+ inMemoryTable.clear();
chunkId++;
// When the volume of sorting data once exceed the size of sort buffer,
@@ -233,19 +246,15 @@ public class ExternalSortExec extends SortExec {
}
}
- if (!memoryResident && !inMemoryTable.isEmpty()) { // if there are at least one or more input tuples
- // check if data exceeds a sort buffer. If so, it store the remain data into a chunk.
- long start = System.currentTimeMillis();
- int rowNum = inMemoryTable.size();
- chunkPaths.add(sortAndStoreChunk(chunkId, inMemoryTable));
- long end = System.currentTimeMillis();
- info(LOG, "Last Chunk #" + chunkId + " " + rowNum + " rows written (" + (end - start) + " msec)");
+ if(inMemoryTable.size() > 0) { //if there are at least one or more input tuples
+ //store the remain data into a memory chunk.
+ chunkPaths.add(new Chunk(inSchema, inMemoryTable, intermediateMeta));
}
// get total loaded (or stored) bytes and total row numbers
TableStats childTableStats = child.getInputStats();
if (childTableStats != null) {
- sortAndStoredBytes = childTableStats.getNumBytes();
+ inputBytes = childTableStats.getNumBytes();
}
return chunkPaths;
}
@@ -267,6 +276,7 @@ public class ExternalSortExec extends SortExec {
if (mergedInputFragments != null) {
try {
this.result = externalMergeAndSort(mergedInputFragments);
+ this.inputBytes = result.getInputStats().getNumBytes();
} catch (Exception e) {
throw new PhysicalPlanningException(e);
}
@@ -275,19 +285,16 @@ public class ExternalSortExec extends SortExec {
long startTimeOfChunkSplit = System.currentTimeMillis();
List<Chunk> chunks = sortAndStoreAllChunks();
long endTimeOfChunkSplit = System.currentTimeMillis();
- info(LOG, "Chunks creation time: " + (endTimeOfChunkSplit - startTimeOfChunkSplit) + " msec");
-
- if (memoryResident) { // if all sorted data reside in a main-memory table.
- TupleSorter sorter = getSorter(inMemoryTable);
- result = new MemTableScanner(sorter.sort(), inMemoryTable.size(), sortAndStoredBytes);
- } else { // if input data exceeds main-memory at least once
+ info(LOG, chunks.size() + " Chunks creation time: " + (endTimeOfChunkSplit - startTimeOfChunkSplit) + " msec");
+ if(chunks.size() == 0) {
+ this.result = new NullScanner(context.getConf(), inSchema, intermediateMeta, null);
+ } else {
try {
this.result = externalMergeAndSort(chunks);
} catch (Exception e) {
throw new PhysicalPlanningException(e);
}
-
}
}
@@ -324,8 +331,7 @@ public class ExternalSortExec extends SortExec {
return computedFanout;
}
- private Scanner externalMergeAndSort(List<Chunk> chunks)
- throws IOException, ExecutionException, InterruptedException {
+ private Scanner externalMergeAndSort(List<Chunk> chunks) throws Exception {
int level = 0;
final List<Chunk> inputFiles = TUtil.newList(chunks);
final List<Chunk> outputFiles = TUtil.newList();
@@ -352,8 +358,14 @@ public class ExternalSortExec extends SortExec {
// how many files are merged in ith thread?
numberOfMergingFiles.add(fanout);
// launch a merger runner
- futures.add(executorService.submit(
- new KWayMergerCaller(level, outChunkId++, inputFiles, startIdx, fanout, false)));
+ if(allocatedCoreNum > 1) {
+ futures.add(executorService.submit(
+ new KWayMergerCaller(level, outChunkId++, inputFiles, startIdx, fanout, false)));
+ } else {
+ final SettableFuture<Chunk> future = SettableFuture.create();
+ future.set(new KWayMergerCaller(level, outChunkId++, inputFiles, startIdx, fanout, false).call());
+ futures.add(future);
+ }
outputFileNum++;
startIdx += fanout;
@@ -406,14 +418,23 @@ public class ExternalSortExec extends SortExec {
*/
int numDeletedFiles = 0;
for (Chunk chunk : inputFiles) {
- if (chunk.getFragment().getTableName().contains(INTERMEDIATE_FILE_PREFIX)) {
+ if (chunk.isMemory()) {
+ if (LOG.isDebugEnabled()) {
+ debug(LOG, "Remove intermediate memory tuples: " + chunk.getMemoryTuples().usedMem());
+ }
+ chunk.getMemoryTuples().release();
+ } else if (chunk.getFragment().getTableName().contains(INTERMEDIATE_FILE_PREFIX)) {
localFS.delete(chunk.getFragment().getPath(), true);
numDeletedFiles++;
- if(LOG.isDebugEnabled()) LOG.debug("Delete merged intermediate file: " + chunk.getFragment());
+ if (LOG.isDebugEnabled()) {
+ debug(LOG, "Delete merged intermediate file: " + chunk.getFragment());
+ }
}
}
- info(LOG, numDeletedFiles + " merged intermediate files deleted");
+ if(LOG.isDebugEnabled()) {
+ debug(LOG, numDeletedFiles + " merged intermediate files deleted");
+ }
// switch input files to output files, and then clear outputFiles
inputFiles.clear();
@@ -459,11 +480,14 @@ public class ExternalSortExec extends SortExec {
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, intermediateMeta, outputPath);
- output.init();
+
final Scanner merger = createKWayMerger(inputFiles, startIdx, mergeFanout);
merger.init();
+
+ final DirectRawFileWriter output =
+ new DirectRawFileWriter(context.getConf(), null, inSchema, intermediateMeta, outputPath);
+ output.init();
+
Tuple mergeTuple;
while((mergeTuple = merger.next()) != null) {
output.addTuple(mergeTuple);
@@ -476,7 +500,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 new Chunk(frag, intermediateMeta);
+ return new Chunk(inSchema, frag, intermediateMeta);
}
}
@@ -492,23 +516,35 @@ public class ExternalSortExec extends SortExec {
*/
private Scanner createFinalMerger(List<Chunk> inputs) throws IOException {
if (inputs.size() == 1) {
- this.result = getFileScanner(inputs.get(0));
+ this.result = getScanner(inputs.get(0));
} else {
this.result = createKWayMerger(inputs, 0, inputs.size());
}
return result;
}
- private Scanner getFileScanner(Chunk chunk) throws IOException {
- return TablespaceManager.getLocalFs().getScanner(chunk.getMeta(), inSchema, chunk.getFragment(), outSchema);
+ private Scanner getScanner(Chunk chunk) throws IOException {
+ if (chunk.isMemory()) {
+ long sortStart = System.currentTimeMillis();
+
+ OffHeapRowBlockUtils.sort(inMemoryTable, unSafeComparator);
+ Scanner scanner = new MemTableScanner<>(inMemoryTable, inMemoryTable.size(), inMemoryTable.usedMem());
+ if(LOG.isDebugEnabled()) {
+ debug(LOG, "Memory Chunk sort (" + FileUtil.humanReadableByteCount(inMemoryTable.usedMem(), false)
+ + " bytes, " + inMemoryTable.size() + " rows, sort time: "
+ + (System.currentTimeMillis() - sortStart) + " msec)");
+ }
+ return scanner;
+ } else {
+ return TablespaceManager.getLocalFs().getScanner(chunk.meta, chunk.schema, chunk.fragment, chunk.schema);
+ }
}
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));
+ sources[i] = getScanner(inputs.get(startChunkId + i));
}
-
return createKWayMergerInternal(sources, 0, num);
}
@@ -518,27 +554,24 @@ public class ExternalSortExec extends SortExec {
final int mid = (int) Math.ceil((float)num / 2);
Scanner left = createKWayMergerInternal(sources, startIdx, mid);
Scanner right = createKWayMergerInternal(sources, startIdx + mid, num - mid);
- if (ComparableVector.isVectorizable(sortSpecs)) {
- return new VectorComparePairWiseMerger(inSchema, left, right, comparator);
- }
- return new PairWiseMerger(inSchema, left, right, comparator);
+ return new PairWiseMerger(inSchema, left, right, primitiveComparator);
} else {
return sources[startIdx];
}
}
- private static class MemTableScanner extends AbstractScanner {
- final Iterable<Tuple> iterable;
+ private static class MemTableScanner<T extends Tuple> extends AbstractScanner {
+ final Iterable<T> iterable;
final long sortAndStoredBytes;
final int totalRecords;
- Iterator<Tuple> iterator;
+ Iterator<T> iterator;
// for input stats
float scannerProgress;
int numRecords;
TableStats scannerTableStats;
- public MemTableScanner(Iterable<Tuple> iterable, int length, long inBytes) {
+ public MemTableScanner(Iterable<T> iterable, int length, long inBytes) {
this.iterable = iterable;
this.totalRecords = length;
this.sortAndStoredBytes = inBytes;
@@ -601,30 +634,6 @@ public class ExternalSortExec extends SortExec {
CLOSED
}
- private static class VectorComparePairWiseMerger extends PairWiseMerger {
-
- private ComparableVector comparable;
-
- public VectorComparePairWiseMerger(Schema schema, Scanner leftScanner, Scanner rightScanner,
- BaseTupleComparator comparator) throws IOException {
- super(schema, leftScanner, rightScanner, null);
- comparable = new ComparableVector(2, comparator.getSortSpecs(), comparator.getSortKeyIds());
- }
-
- @Override
- protected Tuple prepare(int index, Tuple tuple) {
- if (tuple != null) {
- comparable.set(index, tuple);
- }
- return tuple;
- }
-
- @Override
- protected int compare() {
- return comparable.compare(0, 1);
- }
- }
-
/**
* Two-way merger scanner that reads two input sources and outputs one output tuples sorted in some order.
*/
@@ -638,8 +647,10 @@ public class ExternalSortExec extends SortExec {
private Tuple leftTuple;
private Tuple rightTuple;
+ private boolean leftEOF;
+ private boolean rightEOF;
- private final Tuple outTuple;
+ private Tuple outTuple;
private float mergerProgress;
private TableStats mergerInputStats;
@@ -652,7 +663,6 @@ public class ExternalSortExec extends SortExec {
this.leftScan = leftScanner;
this.rightScan = rightScanner;
this.comparator = comparator;
- this.outTuple = new VTuple(schema.size());
}
private void setState(State state) {
@@ -665,8 +675,6 @@ public class ExternalSortExec extends SortExec {
leftScan.init();
rightScan.init();
- prepareTuplesForFirstComparison();
-
mergerInputStats = new TableStats();
mergerProgress = 0.0f;
@@ -676,31 +684,46 @@ public class ExternalSortExec extends SortExec {
}
}
- private void prepareTuplesForFirstComparison() throws IOException {
- leftTuple = prepare(0, leftScan.next());
- rightTuple = prepare(1, rightScan.next());
- }
-
- protected Tuple prepare(int index, Tuple tuple) {
- return tuple;
- }
-
protected int compare() {
return comparator.compare(leftTuple, rightTuple);
}
@Override
public Tuple next() throws IOException {
- if (leftTuple == null && rightTuple == null) {
- return null;
+ if(!leftEOF && leftTuple == null) {
+ leftTuple = leftScan.next();
}
- if (rightTuple == null || (leftTuple != null && compare() < 0)) {
- outTuple.put(leftTuple.getValues());
- leftTuple = prepare(0, leftScan.next());
+
+ if(!rightEOF && rightTuple == null) {
+ rightTuple = rightScan.next();
+ }
+
+ if (leftTuple != null && rightTuple != null) {
+ if (compare() < 0) {
+ outTuple = leftTuple;
+ leftTuple = null;
+ } else {
+ outTuple = rightTuple;
+ rightTuple = null;
+ }
return outTuple;
}
- outTuple.put(rightTuple.getValues());
- rightTuple = prepare(1, rightScan.next());
+
+ if (leftTuple == null) {
+ leftEOF = true;
+
+ if (rightTuple != null) {
+ outTuple = rightTuple;
+ rightTuple = null;
+ } else {
+ rightEOF = true;
+ outTuple = null;
+ }
+ } else {
+ rightEOF = true;
+ outTuple = leftTuple;
+ leftTuple = null;
+ }
return outTuple;
}
@@ -712,8 +735,11 @@ public class ExternalSortExec extends SortExec {
leftTuple = null;
rightTuple = null;
+ outTuple = null;
+
+ leftEOF = false;
+ rightEOF = false;
- prepareTuplesForFirstComparison();
} else {
throw new IllegalStateException("Illegal State: init() is not allowed in " + state.name());
}
@@ -778,27 +804,25 @@ public class ExternalSortExec extends SortExec {
if (result != null) {
result.close();
- try {
- inputStats = (TableStats)result.getInputStats().clone();
- } catch (CloneNotSupportedException e) {
- LOG.warn(e.getMessage());
- }
- result = null;
}
if (finalOutputFiles != null) {
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);
- LOG.info("Delete file: " + frag);
+ if (!chunk.isMemory()) {
+ 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);
+ if(LOG.isDebugEnabled()) {
+ debug(LOG, "Delete file: " + frag);
+ }
+ }
}
}
}
- if(inMemoryTable != null){
- inMemoryTable.clear();
+ if(inMemoryTable != null) {
+ inMemoryTable.release();
inMemoryTable = null;
}
@@ -831,21 +855,35 @@ public class ExternalSortExec extends SortExec {
@Override
public TableStats getInputStats() {
if (result != null) {
- return result.getInputStats();
- } else {
- return inputStats;
+
+ TableStats tableStats = result.getInputStats();
+ inputStats.setNumRows(tableStats.getNumRows());
+ inputStats.setNumBytes(inputBytes);
+ inputStats.setReadBytes(tableStats.getReadBytes());
}
+ return inputStats;
}
private static class Chunk {
private FileFragment fragment;
private TableMeta meta;
+ private Schema schema;
+ private UnSafeTupleList memoryTuples;
+ private boolean isMemory;
- public Chunk(FileFragment fragment, TableMeta meta) {
+ public Chunk(Schema schema, FileFragment fragment, TableMeta meta) {
+ this.schema = schema;
this.fragment = fragment;
this.meta = meta;
}
+ public Chunk(Schema schema, UnSafeTupleList tuples, TableMeta meta) {
+ this.memoryTuples = tuples;
+ this.isMemory = true;
+ this.schema = schema;
+ this.meta = meta;
+ }
+
public FileFragment getFragment() {
return fragment;
}
@@ -853,5 +891,169 @@ public class ExternalSortExec extends SortExec {
public TableMeta getMeta() {
return meta;
}
+
+ public UnSafeTupleList getMemoryTuples() {
+ return memoryTuples;
+ }
+
+ public boolean isMemory() {
+ return isMemory;
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+ }
+
+ /**
+ * The Comparator class for UnSafeTuples
+ *
+ * @see UnSafeTuple
+ */
+ static class UnSafeComparator implements Comparator<UnSafeTuple> {
+ private final int[] sortKeyIds;
+ private final TajoDataTypes.Type[] sortKeyTypes;
+ private final boolean[] asc;
+ private final boolean[] nullFirsts;
+
+ /**
+ * @param schema The schema of input tuples
+ * @param sortKeys The description of sort keys
+ */
+ public UnSafeComparator(Schema schema, SortSpec[] sortKeys) {
+ Preconditions.checkArgument(sortKeys.length > 0,
+ "At least one sort key must be specified.");
+
+ this.sortKeyIds = new int[sortKeys.length];
+ this.sortKeyTypes = new TajoDataTypes.Type[sortKeys.length];
+ this.asc = new boolean[sortKeys.length];
+ this.nullFirsts = new boolean[sortKeys.length];
+ for (int i = 0; i < sortKeys.length; i++) {
+ if (sortKeys[i].getSortKey().hasQualifier()) {
+ this.sortKeyIds[i] = schema.getColumnId(sortKeys[i].getSortKey().getQualifiedName());
+ } else {
+ this.sortKeyIds[i] = schema.getColumnIdByName(sortKeys[i].getSortKey().getSimpleName());
+ }
+
+ this.asc[i] = sortKeys[i].isAscending();
+ this.nullFirsts[i] = sortKeys[i].isNullsFirst();
+ this.sortKeyTypes[i] = sortKeys[i].getSortKey().getDataType().getType();
+ }
+ }
+
+ @Override
+ public int compare(UnSafeTuple tuple1, UnSafeTuple tuple2) {
+ for (int i = 0; i < sortKeyIds.length; i++) {
+ int compare = OffHeapRowBlockUtils.compareColumn(tuple1, tuple2,
+ sortKeyIds[i], sortKeyTypes[i], asc[i], nullFirsts[i]);
+
+ if (compare != 0) {
+ return compare;
+ }
+ }
+ return 0;
+ }
+ }
+
+ /**
+ * The Comparator class for raw file
+ */
+ static class PrimitiveComparator implements Comparator<Tuple> {
+ private final int[] sortKeyIds;
+ private final TajoDataTypes.Type[] sortKeyTypes;
+ private final boolean[] asc;
+ private final boolean[] nullFirsts;
+
+ /**
+ * @param schema The schema of input tuples
+ * @param sortKeys The description of sort keys
+ */
+ public PrimitiveComparator(Schema schema, SortSpec[] sortKeys) {
+ Preconditions.checkArgument(sortKeys.length > 0,
+ "At least one sort key must be specified.");
+
+ this.sortKeyIds = new int[sortKeys.length];
+ this.sortKeyTypes = new TajoDataTypes.Type[sortKeys.length];
+ this.asc = new boolean[sortKeys.length];
+ this.nullFirsts = new boolean[sortKeys.length];
+ for (int i = 0; i < sortKeys.length; i++) {
+ if (sortKeys[i].getSortKey().hasQualifier()) {
+ this.sortKeyIds[i] = schema.getColumnId(sortKeys[i].getSortKey().getQualifiedName());
+ } else {
+ this.sortKeyIds[i] = schema.getColumnIdByName(sortKeys[i].getSortKey().getSimpleName());
+ }
+
+ this.asc[i] = sortKeys[i].isAscending();
+ this.nullFirsts[i] = sortKeys[i].isNullsFirst();
+ this.sortKeyTypes[i] = sortKeys[i].getSortKey().getDataType().getType();
+ }
+ }
+
+ @Override
+ public int compare(Tuple tuple1, Tuple tuple2) {
+ for (int i = 0; i < sortKeyIds.length; i++) {
+ int compare = compareColumn(tuple1, tuple2,
+ sortKeyIds[i], sortKeyTypes[i], asc[i], nullFirsts[i]);
+
+ if (compare != 0) {
+ return compare;
+ }
+ }
+ return 0;
+ }
+
+ public int compareColumn(Tuple tuple1, Tuple tuple2, int index, TajoDataTypes.Type type,
+ boolean ascending, boolean nullFirst) {
+ final boolean n1 = tuple1.isBlankOrNull(index);
+ final boolean n2 = tuple2.isBlankOrNull(index);
+ if (n1 && n2) {
+ return 0;
+ }
+
+ if (n1 ^ n2) {
+ return nullFirst ? (n1 ? -1 : 1) : (n1 ? 1 : -1);
+ }
+
+ int compare;
+ switch (type) {
+ case BOOLEAN:
+ compare = Booleans.compare(tuple1.getBool(index), tuple2.getBool(index));
+ break;
+ case BIT:
+ compare = tuple1.getByte(index) - tuple2.getByte(index);
+ break;
+ case INT1:
+ case INT2:
+ compare = Shorts.compare(tuple1.getInt2(index), tuple2.getInt2(index));
+ break;
+ case DATE:
+ case INT4:
+ compare = Ints.compare(tuple1.getInt4(index), tuple2.getInt4(index));
+ break;
+ case INET4:
+ compare = UnsignedInts.compare(tuple1.getInt4(index), tuple2.getInt4(index));
+ break;
+ case TIME:
+ case TIMESTAMP:
+ case INT8:
+ compare = Longs.compare(tuple1.getInt8(index), tuple2.getInt8(index));
+ break;
+ case FLOAT4:
+ compare = Floats.compare(tuple1.getFloat4(index), tuple2.getFloat4(index));
+ break;
+ case FLOAT8:
+ compare = Doubles.compare(tuple1.getFloat8(index), tuple2.getFloat8(index));
+ break;
+ case CHAR:
+ case TEXT:
+ case BLOB:
+ compare = TextDatum.COMPARATOR.compare(tuple1.getBytes(index), tuple2.getBytes(index));
+ break;
+ default:
+ throw new TajoRuntimeException(
+ new UnsupportedException("unknown data type '" + type.name() + "'"));
+ }
+ return ascending ? compare : -compare;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java
index 87a19a9..c70e1ff 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java
@@ -70,6 +70,10 @@ public abstract class PhysicalExec implements SchemaObject {
log.info("["+ context.getTaskId() + "] " + message);
}
+ protected void debug(Log log, String message) {
+ log.debug("["+ context.getTaskId() + "] " + message);
+ }
+
protected void warn(Log log, String message) {
log.warn("[" + context.getTaskId() + "] " + message);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java b/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java
index 9feae7e..65cb6ac 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java
@@ -166,6 +166,11 @@ public class ExecutionBlockContext {
return;
}
+ LOG.info("Worker's task counter - total:" + completedTasksNum.intValue() +
+ ", succeeded: " + succeededTasksNum.intValue()
+ + ", killed: " + killedTasksNum.intValue()
+ + ", failed: " + failedTasksNum.intValue());
+
try {
reporter.stop();
} catch (InterruptedException e) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java b/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java
index 2e1639f..b5abffe 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Fetcher.java
@@ -26,6 +26,7 @@ import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.ReadTimeoutException;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.util.ReferenceCountUtil;
+import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.IOUtils;
@@ -152,7 +153,9 @@ public class Fetcher {
request.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);
request.headers().set(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP);
- LOG.info("Status: " + getState() + ", URI:" + uri);
+ if(LOG.isDebugEnabled()) {
+ LOG.info("Status: " + getState() + ", URI:" + uri);
+ }
// Send the HTTP request.
channel.writeAndFlush(request);
@@ -168,7 +171,17 @@ public class Fetcher {
}
this.finishTime = System.currentTimeMillis();
- LOG.info("Fetcher finished:" + (finishTime - startTime) + " ms, " + getState() + ", URI:" + uri);
+ long elapsedMills = finishTime - startTime;
+ String transferSpeed;
+ if(elapsedMills > 1000) {
+ long bytePerSec = (fileChunk.length() * 1000) / elapsedMills;
+ transferSpeed = FileUtils.byteCountToDisplaySize(bytePerSec);
+ } else {
+ transferSpeed = FileUtils.byteCountToDisplaySize(Math.max(fileChunk.length(), 0));
+ }
+
+ LOG.info(String.format("Fetcher :%d ms elapsed. %s/sec, len:%d, state:%s, URL:%s",
+ elapsedMills, transferSpeed, fileChunk.length(), getState(), uri));
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/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 fbd1948..873d9e0 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
@@ -110,7 +110,9 @@ public class TaskImpl implements Task {
this.descs = Maps.newHashMap();
Path baseDirPath = executionBlockContext.createBaseDir();
- LOG.info("Task basedir is created (" + baseDirPath +")");
+
+ if(LOG.isDebugEnabled()) LOG.debug("Task basedir is created (" + baseDirPath +")");
+
TaskAttemptId taskAttemptId = request.getId();
this.taskDir = StorageUtil.concatPath(baseDirPath,
@@ -146,24 +148,21 @@ public class TaskImpl implements Task {
}
this.localChunks = Collections.synchronizedList(new ArrayList<>());
- LOG.info("==================================");
- LOG.info("* Stage " + request.getId() + " is initialized");
- LOG.info("* InterQuery: " + interQuery
- + (interQuery ? ", Use " + this.shuffleType + " shuffle" : "") +
- ", Fragments (num: " + request.getFragments().size() + ")" +
- ", Fetches (total:" + request.getFetches().size() + ") :");
+
+ LOG.info(String.format("* Task %s is initialized. InterQuery: %b, Shuffle: %s, Fragments: %d, Fetches:%d, " +
+ "Local dir: %s", request.getId(), interQuery, shuffleType, request.getFragments().size(),
+ request.getFetches().size(), taskDir));
if(LOG.isDebugEnabled()) {
for (FetchImpl f : request.getFetches()) {
LOG.debug("Table Id: " + f.getName() + ", Simple URIs: " + f.getSimpleURIs());
}
}
- LOG.info("* Local task dir: " + taskDir);
+
if(LOG.isDebugEnabled()) {
LOG.debug("* plan:\n");
LOG.debug(plan.toString());
}
- LOG.info("==================================");
}
private void updateDescsForScanNodes(NodeType nodeType) {
@@ -191,7 +190,6 @@ public class TaskImpl implements Task {
@Override
public void init() throws IOException {
- LOG.info("Initializing: " + getId());
initPlan();
startScriptExecutors();
@@ -210,8 +208,10 @@ public class TaskImpl implements Task {
for (String inputTable : context.getInputTables()) {
tableDir = new Path(inputTableBaseDir, inputTable);
if (!localFS.exists(tableDir)) {
- LOG.info("the directory is created " + tableDir.toUri());
localFS.mkdirs(tableDir);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("the directory is created " + tableDir.toUri());
+ }
}
}
}
@@ -456,11 +456,8 @@ public class TaskImpl implements Task {
queryMasterStub.done(null, report, NullCallback.get());
}
endTime = System.currentTimeMillis();
- LOG.info(context.getTaskId() + " completed. " +
- "Worker's task counter - total:" + executionBlockContext.completedTasksNum.intValue() +
- ", succeeded: " + executionBlockContext.succeededTasksNum.intValue()
- + ", killed: " + executionBlockContext.killedTasksNum.intValue()
- + ", failed: " + executionBlockContext.failedTasksNum.intValue());
+ LOG.info(String.format("%s is complete. %d ms elapsed, final state:%s",
+ context.getTaskId(), endTime - startTime, context.getState()));
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
----------------------------------------------------------------------
diff --git a/tajo-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java b/tajo-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
index b3ab85d..6bcb7b4 100644
--- a/tajo-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
+++ b/tajo-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
@@ -392,10 +392,7 @@ public class TajoPullServerService extends AbstractService {
public void decrementRemainFiles(FileRegion filePart, long fileStartTime) {
long fileSendTime = System.currentTimeMillis() - fileStartTime;
- if (fileSendTime > 20 * 1000) {
- LOG.info("PullServer send too long time: filePos=" + filePart.position() + ", fileLen=" + filePart.count());
- SLOW_FILE_UPDATER.compareAndSet(this, numSlowFile, numSlowFile+ 1);
- }
+
if (fileSendTime > maxTime) {
maxTime = fileSendTime;
}
@@ -403,12 +400,20 @@ public class TajoPullServerService extends AbstractService {
minTime = fileSendTime;
}
+ if (fileSendTime > 20 * 1000) {
+ LOG.warn("Sending data takes too long. " + fileSendTime + "ms elapsed, " +
+ "length:" + (filePart.count() - filePart.position()) + ", URI:" + requestUri);
+ SLOW_FILE_UPDATER.compareAndSet(this, numSlowFile, numSlowFile + 1);
+ }
+
REMAIN_FILE_UPDATER.compareAndSet(this, remainFiles, remainFiles - 1);
if (REMAIN_FILE_UPDATER.get(this) <= 0) {
processingStatusMap.remove(requestUri);
- LOG.info("PullServer processing status: totalTime=" + (System.currentTimeMillis() - startTime) + " ms, "
- + "makeFileListTime=" + makeFileListTime + " ms, minTime=" + minTime + " ms, maxTime=" + maxTime + " ms, "
- + "numFiles=" + numFiles + ", numSlowFile=" + numSlowFile);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("PullServer processing status: totalTime=" + (System.currentTimeMillis() - startTime) + " ms, "
+ + "makeFileListTime=" + makeFileListTime + " ms, minTime=" + minTime + " ms, maxTime=" + maxTime + " ms, "
+ + "numFiles=" + numFiles + ", numSlowFile=" + numSlowFile);
+ }
}
}
}
@@ -431,7 +436,10 @@ public class TajoPullServerService extends AbstractService {
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
accepted.add(ctx.channel());
- LOG.info(String.format("Current number of shuffle connections (%d)", accepted.size()));
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Current number of shuffle connections (%d)", accepted.size()));
+ }
super.channelRegistered(ctx);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java
index c8f46a6..6d4b137 100644
--- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java
+++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java
@@ -40,11 +40,7 @@ public class BaseTupleComparator extends TupleComparator implements ProtoObject<
private final int[] sortKeyIds;
private final boolean[] asc;
@SuppressWarnings("unused")
- private final boolean[] nullFirsts;
-
- private Datum left;
- private Datum right;
- private int compVal;
+ private final boolean[] nullFirsts;
/**
* @param schema The schema of input tuples
@@ -110,29 +106,30 @@ public class BaseTupleComparator extends TupleComparator implements ProtoObject<
@Override
public int compare(Tuple tuple1, Tuple tuple2) {
+
for (int i = 0; i < sortKeyIds.length; i++) {
- left = tuple1.asDatum(sortKeyIds[i]);
- right = tuple2.asDatum(sortKeyIds[i]);
-
- if (left.isNull() || right.isNull()) {
- if (!left.equals(right)) {
- if (left.isNull()) {
- compVal = nullFirsts[i] ? -1 : 1;
- } else {
- compVal = nullFirsts[i] ? 1 : -1;
- }
- } else {
- compVal = 0;
- }
+ final boolean n1 = tuple1.isBlankOrNull(sortKeyIds[i]);
+ final boolean n2 = tuple2.isBlankOrNull(sortKeyIds[i]);
+
+ if (n1 && n2) {
+ continue;
+ }
+
+ if (n1 ^ n2) {
+ return nullFirsts[i] ? (n1 ? -1 : 1) : (n1 ? 1 : -1);
+ }
+
+ Datum left = tuple1.asDatum(sortKeyIds[i]);
+ Datum right = tuple2.asDatum(sortKeyIds[i]);
+
+ int compVal;
+ if (asc[i]) {
+ compVal = left.compareTo(right);
} else {
- if (asc[i]) {
- compVal = left.compareTo(right);
- } else {
- compVal = right.compareTo(left);
- }
+ compVal = right.compareTo(left);
}
- if (compVal < 0 || compVal > 0) {
+ if (compVal != 0) {
return compVal;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/550c0189/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java
----------------------------------------------------------------------
diff --git a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java
index a20adf7..7c91197 100644
--- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java
+++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java
@@ -59,7 +59,6 @@ public class NullScanner implements Scanner {
@Override
public Tuple next() throws IOException {
- progress = 1.0f;
return null;
}