You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hy...@apache.org on 2014/11/04 09:51:00 UTC
[1/3] TAJO-1153: Merge off-heap package in block_iteration branch to
master branch.
Repository: tajo
Updated Branches:
refs/heads/master 080f4e145 -> 514ed8474
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestBSTIndex.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestBSTIndex.java b/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestBSTIndex.java
index bf13f3c..7900195 100644
--- a/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestBSTIndex.java
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestBSTIndex.java
@@ -115,7 +115,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("long", Type.INT8));
keySchema.addColumn(new Column("double", Type.FLOAT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "testFindValue_" + storeType + ".idx"),
@@ -189,7 +189,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("long", Type.INT8));
keySchema.addColumn(new Column("double", Type.FLOAT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "testBuildIndexWithAppender_" + storeType + ".idx"),
@@ -281,7 +281,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("long", Type.INT8));
keySchema.addColumn(new Column("double", Type.FLOAT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "testFindOmittedValue_" + storeType + ".idx"),
@@ -352,7 +352,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("int", Type.INT4));
keySchema.addColumn(new Column("long", Type.INT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "testFindNextKeyValue_" + storeType + ".idx"),
@@ -442,7 +442,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("int", Type.INT4));
keySchema.addColumn(new Column("long", Type.INT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "testFindNextKeyOmittedValue_" + storeType + ".idx"),
@@ -522,7 +522,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("long", Type.INT8));
keySchema.addColumn(new Column("double", Type.FLOAT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "testFindMinValue_" + storeType + ".idx"),
BSTIndex.TWO_LEVEL_INDEX, keySchema, comp);
@@ -604,7 +604,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("int", Type.INT4));
keySchema.addColumn(new Column("long", Type.INT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "testMinMax_" + storeType + ".idx"),
@@ -709,7 +709,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("int", Type.INT4));
keySchema.addColumn(new Column("long", Type.INT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "testConcurrentAccess_" + storeType + ".idx"),
@@ -789,7 +789,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("long", Type.INT8));
keySchema.addColumn(new Column("double", Type.FLOAT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
@@ -880,7 +880,7 @@ public class TestBSTIndex {
keySchema.addColumn(new Column("int", Type.INT4));
keySchema.addColumn(new Column("long", Type.INT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir,
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestSingleCSVFileBSTIndex.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestSingleCSVFileBSTIndex.java b/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestSingleCSVFileBSTIndex.java
index 78b16c3..c0dda1f 100644
--- a/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestSingleCSVFileBSTIndex.java
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/index/TestSingleCSVFileBSTIndex.java
@@ -104,7 +104,7 @@ public class TestSingleCSVFileBSTIndex {
keySchema.addColumn(new Column("long", Type.INT8));
keySchema.addColumn(new Column("double", Type.FLOAT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir,
@@ -193,7 +193,7 @@ public class TestSingleCSVFileBSTIndex {
keySchema.addColumn(new Column("int", Type.INT4));
keySchema.addColumn(new Column("long", Type.INT8));
- TupleComparator comp = new TupleComparator(keySchema, sortKeys);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortKeys);
BSTIndex bst = new BSTIndex(conf);
BSTIndexWriter creater = bst.getIndexWriter(new Path(testDir, "FindNextKeyValueInCSV.idx"),
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/test/java/org/apache/tajo/tuple/TestBaseTupleBuilder.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/tuple/TestBaseTupleBuilder.java b/tajo-storage/src/test/java/org/apache/tajo/tuple/TestBaseTupleBuilder.java
new file mode 100644
index 0000000..b332364
--- /dev/null
+++ b/tajo-storage/src/test/java/org/apache/tajo/tuple/TestBaseTupleBuilder.java
@@ -0,0 +1,76 @@
+/**
+ * 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;
+
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.tuple.offheap.*;
+import org.junit.Test;
+
+public class TestBaseTupleBuilder {
+
+ @Test
+ public void testBuild() {
+ BaseTupleBuilder builder = new BaseTupleBuilder(TestOffHeapRowBlock.schema);
+
+ OffHeapRowBlock rowBlock = TestOffHeapRowBlock.createRowBlock(10248);
+ OffHeapRowBlockReader reader = rowBlock.getReader();
+
+ ZeroCopyTuple inputTuple = new ZeroCopyTuple();
+
+ HeapTuple heapTuple = null;
+ ZeroCopyTuple zcTuple = null;
+ int i = 0;
+ while(reader.next(inputTuple)) {
+ RowStoreUtil.convert(inputTuple, builder);
+
+ heapTuple = builder.buildToHeapTuple();
+ TestOffHeapRowBlock.validateTupleResult(i, heapTuple);
+
+ zcTuple = builder.buildToZeroCopyTuple();
+ TestOffHeapRowBlock.validateTupleResult(i, zcTuple);
+
+ i++;
+ }
+ }
+
+ @Test
+ public void testBuildWithNull() {
+ BaseTupleBuilder builder = new BaseTupleBuilder(TestOffHeapRowBlock.schema);
+
+ OffHeapRowBlock rowBlock = TestOffHeapRowBlock.createRowBlockWithNull(10248);
+ OffHeapRowBlockReader reader = rowBlock.getReader();
+
+ ZeroCopyTuple inputTuple = new ZeroCopyTuple();
+
+ HeapTuple heapTuple = null;
+ ZeroCopyTuple zcTuple = null;
+ int i = 0;
+ while(reader.next(inputTuple)) {
+ RowStoreUtil.convert(inputTuple, builder);
+
+ heapTuple = builder.buildToHeapTuple();
+ TestOffHeapRowBlock.validateNullity(i, heapTuple);
+
+ zcTuple = builder.buildToZeroCopyTuple();
+ TestOffHeapRowBlock.validateNullity(i, zcTuple);
+
+ i++;
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestHeapTuple.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestHeapTuple.java b/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestHeapTuple.java
new file mode 100644
index 0000000..96f465a
--- /dev/null
+++ b/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestHeapTuple.java
@@ -0,0 +1,45 @@
+/***
+ * 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.offheap;
+
+import org.apache.tajo.catalog.SchemaUtil;
+import org.junit.Test;
+
+public class TestHeapTuple {
+
+ @Test
+ public void testHeapTuple() {
+ OffHeapRowBlock rowBlock = TestOffHeapRowBlock.createRowBlock(1024);
+
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(rowBlock);
+
+ ZeroCopyTuple zcTuple = new ZeroCopyTuple();
+ int i = 0;
+ while (reader.next(zcTuple)) {
+ byte [] bytes = new byte[zcTuple.nioBuffer().limit()];
+ zcTuple.nioBuffer().get(bytes);
+
+ HeapTuple heapTuple = new HeapTuple(bytes, SchemaUtil.toDataTypes(TestOffHeapRowBlock.schema));
+ TestOffHeapRowBlock.validateTupleResult(i, heapTuple);
+ i++;
+ }
+
+ rowBlock.release();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestOffHeapRowBlock.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestOffHeapRowBlock.java b/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestOffHeapRowBlock.java
new file mode 100644
index 0000000..c43ba38
--- /dev/null
+++ b/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestOffHeapRowBlock.java
@@ -0,0 +1,577 @@
+/***
+ * 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.offheap;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.catalog.*;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.datum.ProtobufDatum;
+import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
+import org.apache.tajo.storage.BaseTupleComparator;
+import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+import org.apache.tajo.unit.StorageUnit;
+import org.apache.tajo.util.FileUtil;
+import org.apache.tajo.util.ProtoUtil;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.tajo.common.TajoDataTypes.Type;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestOffHeapRowBlock {
+ private static final Log LOG = LogFactory.getLog(TestOffHeapRowBlock.class);
+ public static String UNICODE_FIELD_PREFIX = "abc_가나다_";
+ public static Schema schema;
+
+ static {
+ schema = new Schema();
+ schema.addColumn("col0", Type.BOOLEAN);
+ schema.addColumn("col1", Type.INT2);
+ schema.addColumn("col2", Type.INT4);
+ schema.addColumn("col3", Type.INT8);
+ schema.addColumn("col4", Type.FLOAT4);
+ schema.addColumn("col5", Type.FLOAT8);
+ schema.addColumn("col6", Type.TEXT);
+ schema.addColumn("col7", Type.TIMESTAMP);
+ schema.addColumn("col8", Type.DATE);
+ schema.addColumn("col9", Type.TIME);
+ schema.addColumn("col10", Type.INTERVAL);
+ schema.addColumn("col11", Type.INET4);
+ schema.addColumn("col12",
+ CatalogUtil.newDataType(TajoDataTypes.Type.PROTOBUF, PrimitiveProtos.StringProto.class.getName()));
+ }
+
+ private void explainRowBlockAllocation(OffHeapRowBlock rowBlock, long startTime, long endTime) {
+ LOG.info(FileUtil.humanReadableByteCount(rowBlock.size(), true) + " bytes allocated "
+ + (endTime - startTime) + " msec");
+ }
+
+ @Test
+ public void testPutAndReadValidation() {
+ int rowNum = 1000;
+
+ long allocStart = System.currentTimeMillis();
+ OffHeapRowBlock rowBlock = new OffHeapRowBlock(schema, 1024);
+ long allocEnd = System.currentTimeMillis();
+ explainRowBlockAllocation(rowBlock, allocStart, allocEnd);
+
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(rowBlock);
+
+ ZeroCopyTuple tuple = new ZeroCopyTuple();
+ long writeStart = System.currentTimeMillis();
+ for (int i = 0; i < rowNum; i++) {
+ fillRow(i, rowBlock.getWriter());
+
+ reader.reset();
+ int j = 0;
+ while(reader.next(tuple)) {
+ validateTupleResult(j, tuple);
+
+ j++;
+ }
+ }
+ long writeEnd = System.currentTimeMillis();
+ LOG.info("writing and validating take " + (writeEnd - writeStart) + " msec");
+
+ long readStart = System.currentTimeMillis();
+ tuple = new ZeroCopyTuple();
+ int j = 0;
+ reader.reset();
+ while(reader.next(tuple)) {
+ validateTupleResult(j, tuple);
+ j++;
+ }
+ long readEnd = System.currentTimeMillis();
+ LOG.info("reading takes " + (readEnd - readStart) + " msec");
+
+ rowBlock.release();
+ }
+
+ @Test
+ public void testNullityValidation() {
+ int rowNum = 1000;
+
+ long allocStart = System.currentTimeMillis();
+ OffHeapRowBlock rowBlock = new OffHeapRowBlock(schema, 1024);
+ long allocEnd = System.currentTimeMillis();
+ explainRowBlockAllocation(rowBlock, allocStart, allocEnd);
+
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(rowBlock);
+ ZeroCopyTuple tuple = new ZeroCopyTuple();
+ long writeStart = System.currentTimeMillis();
+ for (int i = 0; i < rowNum; i++) {
+ fillRowBlockWithNull(i, rowBlock.getWriter());
+
+ reader.reset();
+ int j = 0;
+ while(reader.next(tuple)) {
+ validateNullity(j, tuple);
+
+ j++;
+ }
+ }
+ long writeEnd = System.currentTimeMillis();
+ LOG.info("writing and nullity validating take " + (writeEnd - writeStart) +" msec");
+
+ long readStart = System.currentTimeMillis();
+ tuple = new ZeroCopyTuple();
+ int j = 0;
+ reader.reset();
+ while(reader.next(tuple)) {
+ validateNullity(j, tuple);
+
+ j++;
+ }
+ long readEnd = System.currentTimeMillis();
+ LOG.info("reading takes " + (readEnd - readStart) + " msec");
+
+ rowBlock.release();
+ }
+
+ @Test
+ public void testEmptyRow() {
+ int rowNum = 1000;
+
+ long allocStart = System.currentTimeMillis();
+ OffHeapRowBlock rowBlock = new OffHeapRowBlock(schema, StorageUnit.MB * 10);
+ long allocEnd = System.currentTimeMillis();
+ explainRowBlockAllocation(rowBlock, allocStart, allocEnd);
+
+ long writeStart = System.currentTimeMillis();
+ for (int i = 0; i < rowNum; i++) {
+ rowBlock.getWriter().startRow();
+ // empty columns
+ rowBlock.getWriter().endRow();
+ }
+ long writeEnd = System.currentTimeMillis();
+ LOG.info("writing tooks " + (writeEnd - writeStart) + " msec");
+
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(rowBlock);
+
+ long readStart = System.currentTimeMillis();
+ ZeroCopyTuple tuple = new ZeroCopyTuple();
+ int j = 0;
+ reader.reset();
+ while(reader.next(tuple)) {
+ j++;
+ }
+ long readEnd = System.currentTimeMillis();
+ LOG.info("reading takes " + (readEnd - readStart) + " msec");
+ rowBlock.release();
+
+ assertEquals(rowNum, j);
+ assertEquals(rowNum, rowBlock.rows());
+ }
+
+ @Test
+ public void testSortBenchmark() {
+ int rowNum = 1000;
+
+ OffHeapRowBlock rowBlock = createRowBlock(rowNum);
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(rowBlock);
+
+ List<ZeroCopyTuple> unSafeTuples = Lists.newArrayList();
+
+ long readStart = System.currentTimeMillis();
+ ZeroCopyTuple tuple = new ZeroCopyTuple();
+ reader.reset();
+ while(reader.next(tuple)) {
+ unSafeTuples.add(tuple);
+ tuple = new ZeroCopyTuple();
+ }
+ long readEnd = System.currentTimeMillis();
+ LOG.info("reading takes " + (readEnd - readStart) + " msec");
+
+ SortSpec sortSpec = new SortSpec(new Column("col2", Type.INT4));
+ BaseTupleComparator comparator = new BaseTupleComparator(schema, new SortSpec[] {sortSpec});
+
+ long sortStart = System.currentTimeMillis();
+ Collections.sort(unSafeTuples, comparator);
+ long sortEnd = System.currentTimeMillis();
+ LOG.info("sorting took " + (sortEnd - sortStart) + " msec");
+ rowBlock.release();
+ }
+
+ @Test
+ public void testVTuplePutAndGetBenchmark() {
+ int rowNum = 1000;
+
+ List<VTuple> rowBlock = Lists.newArrayList();
+ long writeStart = System.currentTimeMillis();
+ VTuple tuple;
+ for (int i = 0; i < rowNum; i++) {
+ tuple = new VTuple(schema.size());
+ fillVTuple(i, tuple);
+ rowBlock.add(tuple);
+ }
+ long writeEnd = System.currentTimeMillis();
+ LOG.info("Writing takes " + (writeEnd - writeStart) + " msec");
+
+ long readStart = System.currentTimeMillis();
+ int j = 0;
+ for (VTuple t : rowBlock) {
+ validateTupleResult(j, t);
+ j++;
+ }
+ long readEnd = System.currentTimeMillis();
+ LOG.info("reading takes " + (readEnd - readStart) + " msec");
+
+ int count = 0;
+ for (int l = 0; l < rowBlock.size(); l++) {
+ for(int m = 0; m < schema.size(); m++ ) {
+ if (rowBlock.get(l).contains(m) && rowBlock.get(l).get(m).type() == Type.INT4) {
+ count ++;
+ }
+ }
+ }
+ // For preventing unnecessary code elimination optimization.
+ LOG.info("The number of INT4 values is " + count + ".");
+ }
+
+ @Test
+ public void testVTuplePutAndGetBenchmarkViaDirectRowEncoder() {
+ int rowNum = 1000;
+
+ OffHeapRowBlock rowBlock = new OffHeapRowBlock(schema, StorageUnit.MB * 100);
+
+ long writeStart = System.currentTimeMillis();
+ VTuple tuple = new VTuple(schema.size());
+ for (int i = 0; i < rowNum; i++) {
+ fillVTuple(i, tuple);
+
+ RowStoreUtil.convert(tuple, rowBlock.getWriter());
+ }
+ long writeEnd = System.currentTimeMillis();
+ LOG.info("Writing takes " + (writeEnd - writeStart) + " msec");
+
+ validateResults(rowBlock);
+ rowBlock.release();
+ }
+
+ @Test
+ public void testSerDerOfRowBlock() {
+ int rowNum = 1000;
+
+ OffHeapRowBlock rowBlock = createRowBlock(rowNum);
+
+ ByteBuffer bb = rowBlock.nioBuffer();
+ OffHeapRowBlock restoredRowBlock = new OffHeapRowBlock(schema, bb);
+ validateResults(restoredRowBlock);
+ rowBlock.release();
+ }
+
+ @Test
+ public void testSerDerOfZeroCopyTuple() {
+ int rowNum = 1000;
+
+ OffHeapRowBlock rowBlock = createRowBlock(rowNum);
+
+ ByteBuffer bb = rowBlock.nioBuffer();
+ OffHeapRowBlock restoredRowBlock = new OffHeapRowBlock(schema, bb);
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(restoredRowBlock);
+
+ long readStart = System.currentTimeMillis();
+ ZeroCopyTuple tuple = new ZeroCopyTuple();
+ ZeroCopyTuple copyTuple = new ZeroCopyTuple();
+ int j = 0;
+ reader.reset();
+ while(reader.next(tuple)) {
+ ByteBuffer copy = tuple.nioBuffer();
+ copyTuple.set(copy, SchemaUtil.toDataTypes(schema));
+
+ validateTupleResult(j, copyTuple);
+
+ j++;
+ }
+ long readEnd = System.currentTimeMillis();
+ LOG.info("reading takes " + (readEnd - readStart) + " msec");
+
+ rowBlock.release();
+ }
+
+ public static OffHeapRowBlock createRowBlock(int rowNum) {
+ long allocateStart = System.currentTimeMillis();
+ OffHeapRowBlock rowBlock = new OffHeapRowBlock(schema, StorageUnit.MB * 8);
+ long allocatedEnd = System.currentTimeMillis();
+ LOG.info(FileUtil.humanReadableByteCount(rowBlock.size(), true) + " bytes allocated "
+ + (allocatedEnd - allocateStart) + " msec");
+
+ long writeStart = System.currentTimeMillis();
+ for (int i = 0; i < rowNum; i++) {
+ fillRow(i, rowBlock.getWriter());
+ }
+ long writeEnd = System.currentTimeMillis();
+ LOG.info("writing takes " + (writeEnd - writeStart) + " msec");
+
+ return rowBlock;
+ }
+
+ public static OffHeapRowBlock createRowBlockWithNull(int rowNum) {
+ long allocateStart = System.currentTimeMillis();
+ OffHeapRowBlock rowBlock = new OffHeapRowBlock(schema, StorageUnit.MB * 8);
+ long allocatedEnd = System.currentTimeMillis();
+ LOG.info(FileUtil.humanReadableByteCount(rowBlock.size(), true) + " bytes allocated "
+ + (allocatedEnd - allocateStart) + " msec");
+
+ long writeStart = System.currentTimeMillis();
+ for (int i = 0; i < rowNum; i++) {
+ fillRowBlockWithNull(i, rowBlock.getWriter());
+ }
+ long writeEnd = System.currentTimeMillis();
+ LOG.info("writing and validating take " + (writeEnd - writeStart) + " msec");
+
+ return rowBlock;
+ }
+
+ public static void fillRow(int i, RowWriter builder) {
+ builder.startRow();
+ builder.putBool(i % 1 == 0 ? true : false); // 0
+ builder.putInt2((short) 1); // 1
+ builder.putInt4(i); // 2
+ builder.putInt8(i); // 3
+ builder.putFloat4(i); // 4
+ builder.putFloat8(i); // 5
+ builder.putText((UNICODE_FIELD_PREFIX + i).getBytes()); // 6
+ builder.putTimestamp(DatumFactory.createTimestamp("2014-04-16 08:48:00").asInt8() + i); // 7
+ builder.putDate(DatumFactory.createDate("2014-04-16").asInt4() + i); // 8
+ builder.putTime(DatumFactory.createTime("08:48:00").asInt8() + i); // 9
+ builder.putInterval(DatumFactory.createInterval((i + 1) + " hours")); // 10
+ builder.putInet4(DatumFactory.createInet4("192.168.0.1").asInt4() + i); // 11
+ builder.putProtoDatum(new ProtobufDatum(ProtoUtil.convertString(i + ""))); // 12
+ builder.endRow();
+ }
+
+ public static void fillRowBlockWithNull(int i, RowWriter writer) {
+ writer.startRow();
+
+ if (i == 0) {
+ writer.skipField();
+ } else {
+ writer.putBool(i % 1 == 0 ? true : false); // 0
+ }
+ if (i % 1 == 0) {
+ writer.skipField();
+ } else {
+ writer.putInt2((short) 1); // 1
+ }
+
+ if (i % 2 == 0) {
+ writer.skipField();
+ } else {
+ writer.putInt4(i); // 2
+ }
+
+ if (i % 3 == 0) {
+ writer.skipField();
+ } else {
+ writer.putInt8(i); // 3
+ }
+
+ if (i % 4 == 0) {
+ writer.skipField();
+ } else {
+ writer.putFloat4(i); // 4
+ }
+
+ if (i % 5 == 0) {
+ writer.skipField();
+ } else {
+ writer.putFloat8(i); // 5
+ }
+
+ if (i % 6 == 0) {
+ writer.skipField();
+ } else {
+ writer.putText((UNICODE_FIELD_PREFIX + i).getBytes()); // 6
+ }
+
+ if (i % 7 == 0) {
+ writer.skipField();
+ } else {
+ writer.putTimestamp(DatumFactory.createTimestamp("2014-04-16 08:48:00").asInt8() + i); // 7
+ }
+
+ if (i % 8 == 0) {
+ writer.skipField();
+ } else {
+ writer.putDate(DatumFactory.createDate("2014-04-16").asInt4() + i); // 8
+ }
+
+ if (i % 9 == 0) {
+ writer.skipField();
+ } else {
+ writer.putTime(DatumFactory.createTime("08:48:00").asInt8() + i); // 9
+ }
+
+ if (i % 10 == 0) {
+ writer.skipField();
+ } else {
+ writer.putInterval(DatumFactory.createInterval((i + 1) + " hours")); // 10
+ }
+
+ if (i % 11 == 0) {
+ writer.skipField();
+ } else {
+ writer.putInet4(DatumFactory.createInet4("192.168.0.1").asInt4() + i); // 11
+ }
+
+ if (i % 12 == 0) {
+ writer.skipField();
+ } else {
+ writer.putProtoDatum(new ProtobufDatum(ProtoUtil.convertString(i + ""))); // 12
+ }
+
+ writer.endRow();
+ }
+
+ public static void fillVTuple(int i, VTuple tuple) {
+ tuple.put(0, DatumFactory.createBool(i % 1 == 0));
+ tuple.put(1, DatumFactory.createInt2((short) 1));
+ tuple.put(2, DatumFactory.createInt4(i));
+ tuple.put(3, DatumFactory.createInt8(i));
+ tuple.put(4, DatumFactory.createFloat4(i));
+ tuple.put(5, DatumFactory.createFloat8(i));
+ tuple.put(6, DatumFactory.createText((UNICODE_FIELD_PREFIX + i).getBytes()));
+ tuple.put(7, DatumFactory.createTimestamp(DatumFactory.createTimestamp("2014-04-16 08:48:00").asInt8() + i)); // 7
+ tuple.put(8, DatumFactory.createDate(DatumFactory.createDate("2014-04-16").asInt4() + i)); // 8
+ tuple.put(9, DatumFactory.createTime(DatumFactory.createTime("08:48:00").asInt8() + i)); // 9
+ tuple.put(10, DatumFactory.createInterval((i + 1) + " hours")); // 10
+ tuple.put(11, DatumFactory.createInet4(DatumFactory.createInet4("192.168.0.1").asInt4() + i)); // 11
+ tuple.put(12, new ProtobufDatum(ProtoUtil.convertString(i + ""))); // 12;
+ }
+
+ public static void validateResults(OffHeapRowBlock rowBlock) {
+ long readStart = System.currentTimeMillis();
+ ZeroCopyTuple tuple = new ZeroCopyTuple();
+ int j = 0;
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(rowBlock);
+ reader.reset();
+ while(reader.next(tuple)) {
+ validateTupleResult(j, tuple);
+ j++;
+ }
+ long readEnd = System.currentTimeMillis();
+ LOG.info("Reading takes " + (readEnd - readStart) + " msec");
+ }
+
+ public static void validateTupleResult(int j, Tuple t) {
+ assertTrue((j % 1 == 0) == t.getBool(0));
+ assertTrue(1 == t.getInt2(1));
+ assertEquals(j, t.getInt4(2));
+ assertEquals(j, t.getInt8(3));
+ assertTrue(j == t.getFloat4(4));
+ assertTrue(j == t.getFloat8(5));
+ assertEquals(new String(UNICODE_FIELD_PREFIX + j), t.getText(6));
+ assertEquals(DatumFactory.createTimestamp("2014-04-16 08:48:00").asInt8() + (long) j, t.getInt8(7));
+ assertEquals(DatumFactory.createDate("2014-04-16").asInt4() + j, t.getInt4(8));
+ assertEquals(DatumFactory.createTime("08:48:00").asInt8() + j, t.getInt8(9));
+ assertEquals(DatumFactory.createInterval((j + 1) + " hours"), t.getInterval(10));
+ assertEquals(DatumFactory.createInet4("192.168.0.1").asInt4() + j, t.getInt4(11));
+ assertEquals(new ProtobufDatum(ProtoUtil.convertString(j + "")), t.getProtobufDatum(12));
+ }
+
+ public static void validateNullity(int j, Tuple tuple) {
+ if (j == 0) {
+ tuple.isNull(0);
+ } else {
+ assertTrue((j % 1 == 0) == tuple.getBool(0));
+ }
+
+ if (j % 1 == 0) {
+ tuple.isNull(1);
+ } else {
+ assertTrue(1 == tuple.getInt2(1));
+ }
+
+ if (j % 2 == 0) {
+ tuple.isNull(2);
+ } else {
+ assertEquals(j, tuple.getInt4(2));
+ }
+
+ if (j % 3 == 0) {
+ tuple.isNull(3);
+ } else {
+ assertEquals(j, tuple.getInt8(3));
+ }
+
+ if (j % 4 == 0) {
+ tuple.isNull(4);
+ } else {
+ assertTrue(j == tuple.getFloat4(4));
+ }
+
+ if (j % 5 == 0) {
+ tuple.isNull(5);
+ } else {
+ assertTrue(j == tuple.getFloat8(5));
+ }
+
+ if (j % 6 == 0) {
+ tuple.isNull(6);
+ } else {
+ assertEquals(new String(UNICODE_FIELD_PREFIX + j), tuple.getText(6));
+ }
+
+ if (j % 7 == 0) {
+ tuple.isNull(7);
+ } else {
+ assertEquals(DatumFactory.createTimestamp("2014-04-16 08:48:00").asInt8() + (long) j, tuple.getInt8(7));
+ }
+
+ if (j % 8 == 0) {
+ tuple.isNull(8);
+ } else {
+ assertEquals(DatumFactory.createDate("2014-04-16").asInt4() + j, tuple.getInt4(8));
+ }
+
+ if (j % 9 == 0) {
+ tuple.isNull(9);
+ } else {
+ assertEquals(DatumFactory.createTime("08:48:00").asInt8() + j, tuple.getInt8(9));
+ }
+
+ if (j % 10 == 0) {
+ tuple.isNull(10);
+ } else {
+ assertEquals(DatumFactory.createInterval((j + 1) + " hours"), tuple.getInterval(10));
+ }
+
+ if (j % 11 == 0) {
+ tuple.isNull(11);
+ } else {
+ assertEquals(DatumFactory.createInet4("192.168.0.1").asInt4() + j, tuple.getInt4(11));
+ }
+
+ if (j % 12 == 0) {
+ tuple.isNull(12);
+ } else {
+ assertEquals(new ProtobufDatum(ProtoUtil.convertString(j + "")), tuple.getProtobufDatum(12));
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestResizableSpec.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestResizableSpec.java b/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestResizableSpec.java
new file mode 100644
index 0000000..1eb9c17
--- /dev/null
+++ b/tajo-storage/src/test/java/org/apache/tajo/tuple/offheap/TestResizableSpec.java
@@ -0,0 +1,59 @@
+/*
+ * 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.offheap;
+
+import org.apache.tajo.unit.StorageUnit;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class TestResizableSpec {
+
+ @Test
+ public void testResizableLimit() {
+ ResizableLimitSpec limit = new ResizableLimitSpec(10 * StorageUnit.MB, 1000 * StorageUnit.MB, 0.1f, 1.0f);
+
+ long expectedMaxSize = (long) (1000 * StorageUnit.MB + (1000 * StorageUnit.MB * 0.1f));
+
+ assertTrue(limit.limit() == 1000 * StorageUnit.MB + (1000 * StorageUnit.MB * 0.1f));
+
+ assertEquals(20971520, limit.increasedSize(10 * StorageUnit.MB));
+
+ assertEquals(expectedMaxSize, limit.increasedSize(1600 * StorageUnit.MB));
+
+ assertEquals(0.98f, limit.remainRatio(980 * StorageUnit.MB), 0.1);
+
+ assertFalse(limit.canIncrease(limit.limit()));
+ }
+
+ @Test
+ public void testFixedLimit() {
+ FixedSizeLimitSpec limit = new FixedSizeLimitSpec(100 * StorageUnit.MB, 0.0f);
+
+ assertEquals(limit.limit(), 100 * StorageUnit.MB);
+
+ assertEquals(100 * StorageUnit.MB, limit.increasedSize(1000));
+
+ assertEquals(100 * StorageUnit.MB, limit.increasedSize(1600 * StorageUnit.MB));
+
+ assertTrue(0.98f == limit.remainRatio(98 * StorageUnit.MB));
+
+ assertFalse(limit.canIncrease(limit.limit()));
+ }
+}
\ No newline at end of file
[3/3] git commit: TAJO-1153: Merge off-heap package in
block_iteration branch to master branch.
Posted by hy...@apache.org.
TAJO-1153: Merge off-heap package in block_iteration branch to master branch.
Closes #225
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/514ed847
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/514ed847
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/514ed847
Branch: refs/heads/master
Commit: 514ed8474a843c92b2fb772971e41f53010f5a4e
Parents: 080f4e1
Author: Hyunsik Choi <hy...@apache.org>
Authored: Tue Nov 4 00:49:39 2014 -0800
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Tue Nov 4 00:49:39 2014 -0800
----------------------------------------------------------------------
CHANGES | 3 +
.../org/apache/tajo/catalog/SchemaUtil.java | 103 ++++
.../java/org/apache/tajo/datum/TextDatum.java | 6 +-
.../java/org/apache/tajo/storage/Tuple.java | 12 +-
.../java/org/apache/tajo/storage/VTuple.java | 55 +-
.../org/apache/tajo/util/Deallocatable.java | 23 +
.../main/java/org/apache/tajo/util/SizeOf.java | 159 +++++
.../org/apache/tajo/util/UnsafeComparer.java | 160 +++++
.../java/org/apache/tajo/util/UnsafeUtil.java | 115 ++++
.../engine/planner/PhysicalPlannerImpl.java | 5 +-
.../tajo/engine/planner/global/DataChannel.java | 2 +-
.../planner/physical/ExternalSortExec.java | 130 ++++-
.../planner/physical/HashFullOuterJoinExec.java | 2 +-
.../engine/planner/physical/HashJoinExec.java | 2 +-
.../planner/physical/HashLeftOuterJoinExec.java | 2 +-
.../engine/planner/physical/MergeJoinExec.java | 2 +-
.../planner/physical/PhysicalPlanUtil.java | 9 +-
.../physical/RangeShuffleFileWriteExec.java | 2 +-
.../physical/RightOuterMergeJoinExec.java | 2 +-
.../engine/planner/physical/SeqScanExec.java | 2 +-
.../tajo/engine/planner/physical/SortExec.java | 7 +-
.../engine/planner/physical/WindowAggExec.java | 5 +-
.../tajo/worker/RangeRetrieverHandler.java | 6 +-
.../main/java/org/apache/tajo/worker/Task.java | 3 +-
.../apache/tajo/engine/eval/ExprTestBase.java | 2 +-
.../tajo/engine/planner/TestPlannerUtil.java | 4 +-
.../planner/TestUniformRangePartition.java | 4 +-
.../planner/physical/TestBSTIndexExec.java | 4 +-
.../planner/physical/TestExternalSortExec.java | 2 +-
.../planner/physical/TestPhysicalPlanner.java | 2 +-
.../physical/TestProgressExternalSortExec.java | 2 +-
.../apache/tajo/engine/util/TestTupleUtil.java | 2 +-
.../tajo/worker/TestRangeRetrieverHandler.java | 4 +-
.../org/apache/tajo/jdbc/MetaDataTuple.java | 17 +-
.../tajo/plan/LogicalPlanPreprocessor.java | 2 +-
.../org/apache/tajo/plan/LogicalPlanner.java | 2 +-
.../GreedyHeuristicJoinOrderAlgorithm.java | 2 +-
.../org/apache/tajo/plan/logical/ScanNode.java | 2 +-
.../tajo/plan/logical/TableSubQueryNode.java | 2 +-
.../rewrite/rules/ProjectionPushDownRule.java | 2 +-
.../org/apache/tajo/plan/util/PlannerUtil.java | 1 +
.../org/apache/tajo/plan/util/SchemaUtil.java | 88 ---
.../tajo/storage/BaseTupleComparator.java | 206 +++++++
.../org/apache/tajo/storage/FrameTuple.java | 14 +-
.../java/org/apache/tajo/storage/LazyTuple.java | 13 +-
.../org/apache/tajo/storage/RowStoreUtil.java | 202 +++++--
.../apache/tajo/storage/TupleComparator.java | 159 +----
.../org/apache/tajo/storage/TupleRange.java | 2 +-
.../apache/tajo/storage/index/IndexMethod.java | 1 +
.../apache/tajo/storage/index/bst/BSTIndex.java | 10 +-
.../org/apache/tajo/tuple/BaseTupleBuilder.java | 112 ++++
.../org/apache/tajo/tuple/RowBlockReader.java | 33 ++
.../org/apache/tajo/tuple/TupleBuilder.java | 26 +
.../tajo/tuple/offheap/DirectBufTuple.java | 41 ++
.../tajo/tuple/offheap/FixedSizeLimitSpec.java | 32 +
.../apache/tajo/tuple/offheap/HeapTuple.java | 269 +++++++++
.../tajo/tuple/offheap/OffHeapMemory.java | 102 ++++
.../tajo/tuple/offheap/OffHeapRowBlock.java | 176 ++++++
.../tuple/offheap/OffHeapRowBlockReader.java | 63 ++
.../tuple/offheap/OffHeapRowBlockUtils.java | 54 ++
.../tuple/offheap/OffHeapRowBlockWriter.java | 58 ++
.../tajo/tuple/offheap/OffHeapRowWriter.java | 232 ++++++++
.../tajo/tuple/offheap/ResizableLimitSpec.java | 142 +++++
.../apache/tajo/tuple/offheap/RowWriter.java | 73 +++
.../apache/tajo/tuple/offheap/UnSafeTuple.java | 308 ++++++++++
.../offheap/UnSafeTupleBytesComparator.java | 99 ++++
.../tajo/tuple/offheap/ZeroCopyTuple.java | 35 ++
tajo-storage/src/main/proto/IndexProtos.proto | 4 +-
.../tajo/storage/TestTupleComparator.java | 2 +-
.../apache/tajo/storage/index/TestBSTIndex.java | 20 +-
.../index/TestSingleCSVFileBSTIndex.java | 4 +-
.../apache/tajo/tuple/TestBaseTupleBuilder.java | 76 +++
.../tajo/tuple/offheap/TestHeapTuple.java | 45 ++
.../tajo/tuple/offheap/TestOffHeapRowBlock.java | 577 +++++++++++++++++++
.../tajo/tuple/offheap/TestResizableSpec.java | 59 ++
75 files changed, 3789 insertions(+), 421 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 943869b..19dc1f6 100644
--- a/CHANGES
+++ b/CHANGES
@@ -44,6 +44,9 @@ Release 0.9.1 - unreleased
TASKS
+ TAJO-1153: Merge off-heap package in block_iteration branch to master
+ branch. (hyunsik)
+
TAJO-1032: Improve TravisCI scripts to adjust log4j log level. (jinho)
TAJO-1141: Refactor the packages hierarchy of tajo-client. (hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SchemaUtil.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SchemaUtil.java
new file mode 100644
index 0000000..ee670ef
--- /dev/null
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SchemaUtil.java
@@ -0,0 +1,103 @@
+/**
+ * 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.catalog;
+
+import static org.apache.tajo.common.TajoDataTypes.DataType;
+import static org.apache.tajo.common.TajoDataTypes.Type;
+
+public class SchemaUtil {
+ // See TAJO-914 bug.
+ //
+ // Its essential problem is that constant value is evaluated multiple times at each scan.
+ // As a result, join nodes can take the child nodes which have the same named fields.
+ // Because current schema does not allow the same name and ignore the duplicated schema,
+ // it finally causes the in-out schema mismatch between the parent and child nodes.
+ //
+ // tmpColumnSeq is a hack to avoid the above problem by keeping duplicated constant values as different name fields.
+ // The essential solution would be https://issues.apache.org/jira/browse/TAJO-895.
+ static int tmpColumnSeq = 0;
+ public static Schema merge(Schema left, Schema right) {
+ Schema merged = new Schema();
+ for(Column col : left.getColumns()) {
+ if (!merged.containsByQualifiedName(col.getQualifiedName())) {
+ merged.addColumn(col);
+ }
+ }
+ for(Column col : right.getColumns()) {
+ if (merged.containsByQualifiedName(col.getQualifiedName())) {
+ merged.addColumn("?fake" + (tmpColumnSeq++), col.getDataType());
+ } else {
+ merged.addColumn(col);
+ }
+ }
+
+ // if overflow
+ if (tmpColumnSeq < 0) {
+ tmpColumnSeq = 0;
+ }
+ return merged;
+ }
+
+ /**
+ * Get common columns to be used as join keys of natural joins.
+ */
+ public static Schema getNaturalJoinColumns(Schema left, Schema right) {
+ Schema common = new Schema();
+ for (Column outer : left.getColumns()) {
+ if (!common.containsByName(outer.getSimpleName()) && right.containsByName(outer.getSimpleName())) {
+ common.addColumn(new Column(outer.getSimpleName(), outer.getDataType()));
+ }
+ }
+
+ return common;
+ }
+
+ public static Schema getQualifiedLogicalSchema(TableDesc tableDesc, String tableName) {
+ Schema logicalSchema = new Schema(tableDesc.getLogicalSchema());
+ if (tableName != null) {
+ logicalSchema.setQualifier(tableName);
+ }
+ return logicalSchema;
+ }
+
+ public static <T extends Schema> T clone(Schema schema) {
+ try {
+ T copy = (T) schema.clone();
+ return copy;
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static DataType [] toDataTypes(Schema schema) {
+ DataType[] types = new DataType[schema.size()];
+ for (int i = 0; i < schema.size(); i++) {
+ types[i] = schema.getColumn(i).getDataType();
+ }
+ return types;
+ }
+
+ public static Type [] toTypes(Schema schema) {
+ Type [] types = new Type[schema.size()];
+ for (int i = 0; i < schema.size(); i++) {
+ types[i] = schema.getColumn(i).getDataType().getType();
+ }
+ return types;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-common/src/main/java/org/apache/tajo/datum/TextDatum.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/datum/TextDatum.java b/tajo-common/src/main/java/org/apache/tajo/datum/TextDatum.java
index ca76ed2..9a9b37e 100644
--- a/tajo-common/src/main/java/org/apache/tajo/datum/TextDatum.java
+++ b/tajo-common/src/main/java/org/apache/tajo/datum/TextDatum.java
@@ -30,7 +30,7 @@ import java.nio.charset.Charset;
import java.util.Comparator;
public class TextDatum extends Datum {
- static Charset defaultCharset = Charset.forName("UTF-8");
+ public static Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
@Expose private final int size;
/* encoded in UTF-8 */
@@ -47,7 +47,7 @@ public class TextDatum extends Datum {
}
public TextDatum(String string) {
- this(string.getBytes(defaultCharset));
+ this(string.getBytes(DEFAULT_CHARSET));
}
@Override
@@ -92,7 +92,7 @@ public class TextDatum extends Datum {
@Override
public String asChars() {
- return new String(this.bytes, defaultCharset);
+ return new String(this.bytes, DEFAULT_CHARSET);
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java b/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java
index c183171..1ba1926 100644
--- a/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java
+++ b/tajo-common/src/main/java/org/apache/tajo/storage/Tuple.java
@@ -19,7 +19,6 @@
package org.apache.tajo.storage;
import org.apache.tajo.datum.Datum;
-import org.apache.tajo.datum.ProtobufDatum;
public interface Tuple extends Cloneable {
@@ -28,16 +27,19 @@ public interface Tuple extends Cloneable {
public boolean contains(int fieldid);
public boolean isNull(int fieldid);
+
+ @SuppressWarnings("unused")
+ public boolean isNotNull(int fieldid);
public void clear();
public void put(int fieldId, Datum value);
- public void put(int fieldId, Datum [] values);
+ public void put(int fieldId, Datum[] values);
public void put(int fieldId, Tuple tuple);
- public void put(Datum [] values);
+ public void put(Datum[] values);
public Datum get(int fieldId);
@@ -65,7 +67,9 @@ public interface Tuple extends Cloneable {
public String getText(int fieldId);
- public ProtobufDatum getProtobufDatum(int fieldId);
+ public Datum getProtobufDatum(int fieldId);
+
+ public Datum getInterval(int fieldId);
public char [] getUnicodeChars(int fieldId);
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-common/src/main/java/org/apache/tajo/storage/VTuple.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/storage/VTuple.java b/tajo-common/src/main/java/org/apache/tajo/storage/VTuple.java
index 4fb35f9..6304734 100644
--- a/tajo-common/src/main/java/org/apache/tajo/storage/VTuple.java
+++ b/tajo-common/src/main/java/org/apache/tajo/storage/VTuple.java
@@ -21,7 +21,7 @@ package org.apache.tajo.storage;
import com.google.gson.annotations.Expose;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.Inet4Datum;
-import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.datum.IntervalDatum;
import org.apache.tajo.datum.ProtobufDatum;
import org.apache.tajo.exception.UnimplementedException;
@@ -38,10 +38,9 @@ public class VTuple implements Tuple, Cloneable {
public VTuple(Tuple tuple) {
this.values = tuple.getValues().clone();
- this.offset = ((VTuple)tuple).offset;
}
- public VTuple(Datum [] datum) {
+ public VTuple(Datum[] datum) {
this(datum.length);
put(datum);
}
@@ -57,7 +56,12 @@ public class VTuple implements Tuple, Cloneable {
@Override
public boolean isNull(int fieldid) {
- return values[fieldid] instanceof NullDatum;
+ return values[fieldid].isNull();
+ }
+
+ @Override
+ public boolean isNotNull(int fieldid) {
+ return !isNull(fieldid);
}
@Override
@@ -179,6 +183,11 @@ public class VTuple implements Tuple, Cloneable {
}
@Override
+ public IntervalDatum getInterval(int fieldId) {
+ return (IntervalDatum) values[fieldId];
+ }
+
+ @Override
public char[] getUnicodeChars(int fieldId) {
return values[fieldId].asUnicodeChars();
}
@@ -193,23 +202,7 @@ public class VTuple implements Tuple, Cloneable {
}
public String toString() {
- boolean first = true;
- StringBuilder str = new StringBuilder();
- str.append("(");
- for(int i=0; i < values.length; i++) {
- if(values[i] != null) {
- if(first) {
- first = false;
- } else {
- str.append(", ");
- }
- str.append(i)
- .append("=>")
- .append(values[i]);
- }
- }
- str.append(")");
- return str.toString();
+ return toDisplayString(getValues());
}
@Override
@@ -230,4 +223,24 @@ public class VTuple implements Tuple, Cloneable {
}
return false;
}
+
+ public static String toDisplayString(Datum [] values) {
+ boolean first = true;
+ StringBuilder str = new StringBuilder();
+ str.append("(");
+ for(int i=0; i < values.length; i++) {
+ if(values[i] != null) {
+ if(first) {
+ first = false;
+ } else {
+ str.append(", ");
+ }
+ str.append(i)
+ .append("=>")
+ .append(values[i]);
+ }
+ }
+ str.append(")");
+ return str.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-common/src/main/java/org/apache/tajo/util/Deallocatable.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/Deallocatable.java b/tajo-common/src/main/java/org/apache/tajo/util/Deallocatable.java
new file mode 100644
index 0000000..8202a4b
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/util/Deallocatable.java
@@ -0,0 +1,23 @@
+/*
+ * 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.util;
+
+public interface Deallocatable {
+ void release();
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-common/src/main/java/org/apache/tajo/util/SizeOf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/SizeOf.java b/tajo-common/src/main/java/org/apache/tajo/util/SizeOf.java
new file mode 100644
index 0000000..b7ca10c
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/util/SizeOf.java
@@ -0,0 +1,159 @@
+/***
+ * 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.util;
+
+import static org.apache.tajo.util.UnsafeUtil.*;
+
+
+public final class SizeOf {
+
+ public static final int BITS_PER_BYTE = 8;
+ public static final int BYTES_PER_WORD = SizeOf.SIZE_OF_LONG;
+ public static final int BITS_PER_WORD = SizeOf.SIZE_OF_LONG * BITS_PER_BYTE;
+
+ public static final byte SIZE_OF_BOOL = 1;
+ public static final byte SIZE_OF_BYTE = 1;
+ public static final byte SIZE_OF_SHORT = 2;
+ public static final byte SIZE_OF_INT = 4;
+ public static final byte SIZE_OF_LONG = 8;
+ public static final byte SIZE_OF_FLOAT = 4;
+ public static final byte SIZE_OF_DOUBLE = 8;
+
+ public static long sizeOf(boolean[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_BOOLEAN_BASE_OFFSET + (((long) ARRAY_BOOLEAN_INDEX_SCALE) * array.length);
+ }
+
+ public static long sizeOf(byte[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_BYTE_BASE_OFFSET + (((long) ARRAY_BYTE_INDEX_SCALE) * array.length);
+ }
+
+ public static long sizeOf(short[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_SHORT_BASE_OFFSET + (((long) ARRAY_SHORT_INDEX_SCALE) * array.length);
+ }
+
+ public static long sizeOf(char[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_CHAR_BASE_OFFSET + (((long) ARRAY_CHAR_INDEX_SCALE) * array.length);
+ }
+
+ public static long sizeOf(int[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_INT_BASE_OFFSET + (((long) ARRAY_INT_INDEX_SCALE) * array.length);
+ }
+
+ public static long sizeOf(long[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_LONG_BASE_OFFSET + (((long) ARRAY_LONG_INDEX_SCALE) * array.length);
+ }
+
+ public static long sizeOf(float[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_FLOAT_BASE_OFFSET + (((long) ARRAY_FLOAT_INDEX_SCALE) * array.length);
+ }
+
+ public static long sizeOf(double[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_DOUBLE_BASE_OFFSET + (((long) ARRAY_DOUBLE_INDEX_SCALE) * array.length);
+ }
+
+ public static long sizeOf(Object[] array)
+ {
+ if (array == null) {
+ return 0;
+ }
+ return ARRAY_OBJECT_BASE_OFFSET + (((long) ARRAY_OBJECT_INDEX_SCALE) * array.length);
+ }
+
+
+ public static long sizeOfBooleanArray(int length)
+ {
+ return ARRAY_BOOLEAN_BASE_OFFSET + (((long) ARRAY_BOOLEAN_INDEX_SCALE) * length);
+ }
+
+ public static long sizeOfByteArray(int length)
+ {
+ return ARRAY_BYTE_BASE_OFFSET + (((long) ARRAY_BYTE_INDEX_SCALE) * length);
+ }
+
+ public static long sizeOfShortArray(int length)
+ {
+ return ARRAY_SHORT_BASE_OFFSET + (((long) ARRAY_SHORT_INDEX_SCALE) * length);
+ }
+
+ public static long sizeOfCharArray(int length)
+ {
+ return ARRAY_CHAR_BASE_OFFSET + (((long) ARRAY_CHAR_INDEX_SCALE) * length);
+ }
+
+ public static long sizeOfIntArray(int length)
+ {
+ return ARRAY_INT_BASE_OFFSET + (((long) ARRAY_INT_INDEX_SCALE) * length);
+ }
+
+ public static long sizeOfLongArray(int length)
+ {
+ return ARRAY_LONG_BASE_OFFSET + (((long) ARRAY_LONG_INDEX_SCALE) * length);
+ }
+
+ public static long sizeOfFloatArray(int length)
+ {
+ return ARRAY_FLOAT_BASE_OFFSET + (((long) ARRAY_FLOAT_INDEX_SCALE) * length);
+ }
+
+ public static long sizeOfDoubleArray(int length)
+ {
+ return ARRAY_DOUBLE_BASE_OFFSET + (((long) ARRAY_DOUBLE_INDEX_SCALE) * length);
+ }
+
+ public static long sizeOfObjectArray(int length)
+ {
+ return ARRAY_OBJECT_BASE_OFFSET + (((long) ARRAY_OBJECT_INDEX_SCALE) * length);
+ }
+
+ private SizeOf()
+ {
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-common/src/main/java/org/apache/tajo/util/UnsafeComparer.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/UnsafeComparer.java b/tajo-common/src/main/java/org/apache/tajo/util/UnsafeComparer.java
new file mode 100644
index 0000000..fec4e5a
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/util/UnsafeComparer.java
@@ -0,0 +1,160 @@
+/***
+ * 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.util;
+
+import com.google.common.primitives.Longs;
+import com.google.common.primitives.UnsignedBytes;
+import com.google.common.primitives.UnsignedLongs;
+import sun.misc.Unsafe;
+
+import java.lang.reflect.Field;
+import java.nio.ByteOrder;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.Comparator;
+
+/**
+ * This is borrowed from Guava's UnsignedBytes.
+ */
+public class UnsafeComparer implements Comparator<byte[]> {
+
+ public static final UnsafeComparer INSTANCE;
+
+ static {
+ INSTANCE = new UnsafeComparer();
+ }
+
+ public Comparator<byte []> get() {
+ return INSTANCE;
+ }
+
+ private UnsafeComparer() {}
+
+ static final boolean littleEndian =
+ ByteOrder.nativeOrder().equals(ByteOrder.LITTLE_ENDIAN);
+
+ /*
+ * The following static final fields exist for performance reasons.
+ *
+ * In UnsignedBytesBenchmark, accessing the following objects via static
+ * final fields is the fastest (more than twice as fast as the Java
+ * implementation, vs ~1.5x with non-final static fields, on x86_32)
+ * under the Hotspot server compiler. The reason is obviously that the
+ * non-final fields need to be reloaded inside the loop.
+ *
+ * And, no, defining (final or not) local variables out of the loop still
+ * isn't as good because the null check on the theUnsafe object remains
+ * inside the loop and BYTE_ARRAY_BASE_OFFSET doesn't get
+ * constant-folded.
+ *
+ * The compiler can treat static final fields as compile-time constants
+ * and can constant-fold them while (final or not) local variables are
+ * run time values.
+ */
+
+ static final Unsafe theUnsafe;
+
+ /** The offset to the first element in a byte array. */
+ static final int BYTE_ARRAY_BASE_OFFSET;
+
+ static {
+ theUnsafe = (Unsafe) AccessController.doPrivileged(
+ new PrivilegedAction<Object>() {
+ @Override
+ public Object run() {
+ try {
+ Field f = Unsafe.class.getDeclaredField("theUnsafe");
+ f.setAccessible(true);
+ return f.get(null);
+ } catch (NoSuchFieldException e) {
+ // It doesn't matter what we throw;
+ // it's swallowed in getBestComparator().
+ throw new Error();
+ } catch (IllegalAccessException e) {
+ throw new Error();
+ }
+ }
+ });
+
+ BYTE_ARRAY_BASE_OFFSET = theUnsafe.arrayBaseOffset(byte[].class);
+
+ // sanity check - this should never fail
+ if (theUnsafe.arrayIndexScale(byte[].class) != 1) {
+ throw new AssertionError();
+ }
+ }
+
+ @SuppressWarnings("unused")
+ public static int compareStatic(byte[] left, byte[] right) {
+ return INSTANCE.compare(left, right);
+ }
+
+ @Override public int compare(byte[] left, byte[] right) {
+ int minLength = Math.min(left.length, right.length);
+ int minWords = minLength / Longs.BYTES;
+
+ /*
+ * Compare 8 bytes at a time. Benchmarking shows comparing 8 bytes at a
+ * time is no slower than comparing 4 bytes at a time even on 32-bit.
+ * On the other hand, it is substantially faster on 64-bit.
+ */
+ for (int i = 0; i < minWords * Longs.BYTES; i += Longs.BYTES) {
+ long lw = theUnsafe.getLong(left, BYTE_ARRAY_BASE_OFFSET + (long) i);
+ long rw = theUnsafe.getLong(right, BYTE_ARRAY_BASE_OFFSET + (long) i);
+ long diff = lw ^ rw;
+
+ if (diff != 0) {
+ if (!littleEndian) {
+ return UnsignedLongs.compare(lw, rw);
+ }
+
+ // Use binary search
+ int n = 0;
+ int y;
+ int x = (int) diff;
+ if (x == 0) {
+ x = (int) (diff >>> 32);
+ n = 32;
+ }
+
+ y = x << 16;
+ if (y == 0) {
+ n += 16;
+ } else {
+ x = y;
+ }
+
+ y = x << 8;
+ if (y == 0) {
+ n += 8;
+ }
+ return (int) (((lw >>> n) & 0xFFL) - ((rw >>> n) & 0xFFL));
+ }
+ }
+
+ // The epilogue to cover the last (minLength % 8) elements.
+ for (int i = minWords * Longs.BYTES; i < minLength; i++) {
+ int result = UnsignedBytes.compare(left[i], right[i]);
+ if (result != 0) {
+ return result;
+ }
+ }
+ return left.length - right.length;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-common/src/main/java/org/apache/tajo/util/UnsafeUtil.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/UnsafeUtil.java b/tajo-common/src/main/java/org/apache/tajo/util/UnsafeUtil.java
new file mode 100644
index 0000000..ae88f73
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/util/UnsafeUtil.java
@@ -0,0 +1,115 @@
+/***
+ * 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.util;
+
+import com.google.common.base.Preconditions;
+import sun.misc.Cleaner;
+import sun.misc.Unsafe;
+import sun.nio.ch.DirectBuffer;
+
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+
+public class UnsafeUtil {
+ public static final Unsafe unsafe;
+
+ // offsets
+ public static final int ARRAY_BOOLEAN_BASE_OFFSET;
+ public static final int ARRAY_BYTE_BASE_OFFSET;
+ public static final int ARRAY_SHORT_BASE_OFFSET;
+ public static final int ARRAY_CHAR_BASE_OFFSET;
+ public static final int ARRAY_INT_BASE_OFFSET;
+ public static final int ARRAY_LONG_BASE_OFFSET;
+ public static final int ARRAY_FLOAT_BASE_OFFSET;
+ public static final int ARRAY_DOUBLE_BASE_OFFSET;
+ public static final int ARRAY_OBJECT_BASE_OFFSET;
+
+ // scale
+ public static final int ARRAY_BOOLEAN_INDEX_SCALE;
+ public static final int ARRAY_BYTE_INDEX_SCALE;
+ public static final int ARRAY_SHORT_INDEX_SCALE;
+ public static final int ARRAY_CHAR_INDEX_SCALE;
+ public static final int ARRAY_INT_INDEX_SCALE;
+ public static final int ARRAY_LONG_INDEX_SCALE;
+ public static final int ARRAY_FLOAT_INDEX_SCALE;
+ public static final int ARRAY_DOUBLE_INDEX_SCALE;
+ public static final int ARRAY_OBJECT_INDEX_SCALE;
+
+ static {
+ Field field;
+ try {
+ field = Unsafe.class.getDeclaredField("theUnsafe");
+
+ field.setAccessible(true);
+ unsafe = (Unsafe) field.get(null);
+ if (unsafe == null) {
+ throw new RuntimeException("Unsafe access not available");
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ ARRAY_BOOLEAN_BASE_OFFSET = unsafe.arrayBaseOffset(boolean[].class);
+ ARRAY_BYTE_BASE_OFFSET = unsafe.arrayBaseOffset(byte[].class);
+ ARRAY_SHORT_BASE_OFFSET = unsafe.arrayBaseOffset(short[].class);
+ ARRAY_CHAR_BASE_OFFSET = unsafe.arrayBaseOffset(char[].class);
+ ARRAY_INT_BASE_OFFSET = unsafe.arrayBaseOffset(int[].class);
+ ARRAY_LONG_BASE_OFFSET = unsafe.arrayBaseOffset(long[].class);
+ ARRAY_FLOAT_BASE_OFFSET = unsafe.arrayBaseOffset(float[].class);
+ ARRAY_DOUBLE_BASE_OFFSET = unsafe.arrayBaseOffset(double[].class);
+ ARRAY_OBJECT_BASE_OFFSET = unsafe.arrayBaseOffset(Object[].class);
+
+ ARRAY_BOOLEAN_INDEX_SCALE = unsafe.arrayIndexScale(boolean[].class);
+ ARRAY_BYTE_INDEX_SCALE = unsafe.arrayIndexScale(byte[].class);
+ ARRAY_SHORT_INDEX_SCALE = unsafe.arrayIndexScale(short[].class);
+ ARRAY_CHAR_INDEX_SCALE = unsafe.arrayIndexScale(char[].class);
+ ARRAY_INT_INDEX_SCALE = unsafe.arrayIndexScale(int[].class);
+ ARRAY_LONG_INDEX_SCALE = unsafe.arrayIndexScale(long[].class);
+ ARRAY_FLOAT_INDEX_SCALE = unsafe.arrayIndexScale(float[].class);
+ ARRAY_DOUBLE_INDEX_SCALE = unsafe.arrayIndexScale(double[].class);
+ ARRAY_OBJECT_INDEX_SCALE = unsafe.arrayIndexScale(Object[].class);
+ }
+
+ public static int alignedSize(int size) {
+ int remain = size % SizeOf.BYTES_PER_WORD;
+ if (remain > 0) {
+ return size + (SizeOf.BYTES_PER_WORD - remain);
+ } else {
+ return size;
+ }
+ }
+
+ public static long getAddress(ByteBuffer buffer) {
+ Preconditions.checkArgument(buffer instanceof DirectBuffer, "ByteBuffer must be DirectBuffer");
+ return ((DirectBuffer)buffer).address();
+ }
+
+ public static void free(Deallocatable obj) {
+ obj.release();
+ }
+
+ public static void free(ByteBuffer bb) {
+ Preconditions.checkNotNull(bb);
+ Preconditions.checkState(bb instanceof DirectBuffer);
+
+ Cleaner cleaner = ((DirectBuffer) bb).cleaner();
+ if (cleaner != null) {
+ cleaner.clean();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/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 f7f071f..99b79ec 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
@@ -45,8 +45,9 @@ import org.apache.tajo.ipc.TajoWorkerProtocol.DistinctGroupbyEnforcer.DistinctAg
import org.apache.tajo.ipc.TajoWorkerProtocol.DistinctGroupbyEnforcer.MultipleAggregationStage;
import org.apache.tajo.ipc.TajoWorkerProtocol.DistinctGroupbyEnforcer.SortSpecArray;
import org.apache.tajo.plan.LogicalPlan;
-import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.plan.logical.*;
+import org.apache.tajo.plan.util.PlannerUtil;
+import org.apache.tajo.storage.BaseTupleComparator;
import org.apache.tajo.storage.StorageConstants;
import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.storage.TupleComparator;
@@ -1185,7 +1186,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
String indexName = IndexUtil.getIndexNameOfFrag(fragments.get(0), annotation.getSortKeys());
Path indexPath = new Path(sm.getTablePath(annotation.getTableName()), "index");
- TupleComparator comp = new TupleComparator(annotation.getKeySchema(),
+ TupleComparator comp = new BaseTupleComparator(annotation.getKeySchema(),
annotation.getSortKeys());
return new BSTIndexScanExec(ctx, sm, annotation, fragments.get(0), new Path(indexPath, indexName),
annotation.getKeySchema(), comp, annotation.getDatum());
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java
index 4a2cd19..11548d3 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/global/DataChannel.java
@@ -22,7 +22,7 @@ import com.google.common.base.Preconditions;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.Schema;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.util.TUtil;
import static org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/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 4c0caea..121e6bd 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
@@ -529,9 +529,9 @@ public class ExternalSortExec extends SortExec {
throws IOException {
if (num > 1) {
final int mid = (int) Math.ceil((float)num / 2);
- return new PairWiseMerger(
+ return new PairWiseMerger(inSchema,
createKWayMergerInternal(sources, startIdx, mid),
- createKWayMergerInternal(sources, startIdx + mid, num - mid));
+ createKWayMergerInternal(sources, startIdx + mid, num - mid), getComparator());
} else {
return sources[startIdx];
}
@@ -626,6 +626,12 @@ public class ExternalSortExec extends SortExec {
}
}
+ enum State {
+ NEW,
+ INITED,
+ CLOSED
+ }
+
/**
* Two-way merger scanner that reads two input sources and outputs one output tuples sorted in some order.
*/
@@ -633,59 +639,132 @@ public class ExternalSortExec extends SortExec {
private Scanner leftScan;
private Scanner rightScan;
- private Tuple leftTuple;
- private Tuple rightTuple;
+ private VTuple outTuple;
+ private VTuple leftTuple;
+ private VTuple rightTuple;
- private final Comparator<Tuple> comparator = getComparator();
+ private final Schema schema;
+ private final Comparator<Tuple> comparator;
private float mergerProgress;
private TableStats mergerInputStats;
- public PairWiseMerger(Scanner leftScanner, Scanner rightScanner) throws IOException {
+ private State state = State.NEW;
+
+ public PairWiseMerger(Schema schema, Scanner leftScanner, Scanner rightScanner, Comparator<Tuple> comparator)
+ throws IOException {
+ this.schema = schema;
this.leftScan = leftScanner;
this.rightScan = rightScanner;
+ this.comparator = comparator;
+ }
+
+ private void setState(State state) {
+ this.state = state;
}
@Override
public void init() throws IOException {
- leftScan.init();
- rightScan.init();
+ if (state == State.NEW) {
+ leftScan.init();
+ rightScan.init();
+
+ prepareTuplesForFirstComparison();
+
+ mergerInputStats = new TableStats();
+ mergerProgress = 0.0f;
+
+ setState(State.INITED);
+ } else {
+ throw new IllegalStateException("Illegal State: init() is not allowed in " + state.name());
+ }
+ }
- leftTuple = leftScan.next();
- rightTuple = rightScan.next();
+ private void prepareTuplesForFirstComparison() throws IOException {
+ Tuple lt = leftScan.next();
+ if (lt != null) {
+ leftTuple = new VTuple(lt);
+ } else {
+ leftTuple = null; // TODO - missed free
+ }
- mergerInputStats = new TableStats();
- mergerProgress = 0.0f;
+ Tuple rt = rightScan.next();
+ if (rt != null) {
+ rightTuple = new VTuple(rt);
+ } else {
+ rightTuple = null; // TODO - missed free
+ }
}
public Tuple next() throws IOException {
- Tuple outTuple;
+
if (leftTuple != null && rightTuple != null) {
if (comparator.compare(leftTuple, rightTuple) < 0) {
- outTuple = leftTuple;
- leftTuple = leftScan.next();
+ outTuple = new VTuple(leftTuple);
+
+ Tuple lt = leftScan.next();
+ if (lt != null) {
+ leftTuple = new VTuple(lt);
+ } else {
+ leftTuple = null; // TODO - missed free
+ }
} else {
- outTuple = rightTuple;
- rightTuple = rightScan.next();
+ outTuple = new VTuple(rightTuple);
+
+ Tuple rt = rightScan.next();
+ if (rt != null) {
+ rightTuple = new VTuple(rt);
+ } else {
+ rightTuple = null; // TODO - missed free
+ }
}
return outTuple;
}
if (leftTuple == null) {
- outTuple = rightTuple;
- rightTuple = rightScan.next();
+ if (rightTuple != null) {
+ outTuple = new VTuple(rightTuple);
+ } else {
+ outTuple = null;
+ }
+
+ Tuple rt = rightScan.next();
+ if (rt != null) {
+ rightTuple = new VTuple(rt);
+ } else {
+ rightTuple = null; // TODO - missed free
+ }
} else {
- outTuple = leftTuple;
- leftTuple = leftScan.next();
+ if (leftTuple != null) {
+ outTuple = new VTuple(leftTuple);
+ } else {
+ outTuple = null;
+ }
+
+ Tuple lt = leftScan.next();
+ if (lt != null) {
+ leftTuple = new VTuple(lt);
+ } else {
+ leftTuple = null; // TODO - missed free
+ }
}
return outTuple;
}
@Override
public void reset() throws IOException {
- leftScan.reset();
- rightScan.reset();
- init();
+ if (state == State.INITED) {
+ leftScan.reset();
+ rightScan.reset();
+
+ outTuple = null;
+ leftTuple = null;
+ rightTuple = null;
+
+ prepareTuplesForFirstComparison();
+ } else {
+ throw new IllegalStateException("Illegal State: init() is not allowed in " + state.name());
+ }
}
public void close() throws IOException {
@@ -694,6 +773,7 @@ public class ExternalSortExec extends SortExec {
leftScan = null;
rightScan = null;
mergerProgress = 1.0f;
+ setState(State.CLOSED);
}
@Override
@@ -721,7 +801,7 @@ public class ExternalSortExec extends SortExec {
@Override
public Schema getSchema() {
- return inSchema;
+ return schema;
}
@Override
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashFullOuterJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashFullOuterJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashFullOuterJoinExec.java
index cf333b0..28d9a3e 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashFullOuterJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashFullOuterJoinExec.java
@@ -23,7 +23,7 @@ import org.apache.tajo.engine.codegen.CompilationError;
import org.apache.tajo.engine.planner.Projector;
import org.apache.tajo.engine.utils.TupleUtil;
import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.logical.JoinNode;
import org.apache.tajo.storage.FrameTuple;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
index 18be6b9..701297f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashJoinExec.java
@@ -22,7 +22,7 @@ import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.engine.planner.Projector;
import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.logical.JoinNode;
import org.apache.tajo.storage.FrameTuple;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
index c38be88..c1b6522 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/HashLeftOuterJoinExec.java
@@ -25,7 +25,7 @@ import org.apache.tajo.catalog.Column;
import org.apache.tajo.engine.planner.Projector;
import org.apache.tajo.engine.utils.TupleUtil;
import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.expr.AlgebraicUtil;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.expr.EvalTreeUtil;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java
index 20df210..13104ee 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/MergeJoinExec.java
@@ -52,7 +52,7 @@ public class MergeJoinExec extends BinaryPhysicalExec {
private Iterator<Tuple> innerIterator;
private JoinTupleComparator joincomparator = null;
- private TupleComparator[] tupleComparator = null;
+ private TupleComparator [] tupleComparator = null;
private final static int INITIAL_TUPLE_SLOT = 10000;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
index 1b824b5..ec5915f 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
@@ -34,6 +34,7 @@ import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.logical.NodeType;
import org.apache.tajo.plan.logical.PersistentStoreNode;
import org.apache.tajo.engine.query.QueryContext;
+import org.apache.tajo.storage.BaseTupleComparator;
import org.apache.tajo.storage.StorageConstants;
import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.storage.TupleComparator;
@@ -53,11 +54,11 @@ public class PhysicalPlanUtil {
return (T) new FindVisitor().visit(plan, new Stack<PhysicalExec>(), clazz);
}
- public static TupleComparator[] getComparatorsFromJoinQual(EvalNode joinQual, Schema leftSchema, Schema rightSchema) {
+ public static TupleComparator [] getComparatorsFromJoinQual(EvalNode joinQual, Schema leftSchema, Schema rightSchema) {
SortSpec[][] sortSpecs = PlannerUtil.getSortKeysFromJoinQual(joinQual, leftSchema, rightSchema);
- TupleComparator[] comparators = new TupleComparator[2];
- comparators[0] = new TupleComparator(leftSchema, sortSpecs[0]);
- comparators[1] = new TupleComparator(rightSchema, sortSpecs[1]);
+ BaseTupleComparator[] comparators = new BaseTupleComparator[2];
+ comparators[0] = new BaseTupleComparator(leftSchema, sortSpecs[0]);
+ comparators[1] = new BaseTupleComparator(rightSchema, sortSpecs[1]);
return comparators;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java
index dd72910..79993da 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RangeShuffleFileWriteExec.java
@@ -71,7 +71,7 @@ public class RangeShuffleFileWriteExec extends UnaryPhysicalExec {
}
BSTIndex bst = new BSTIndex(new TajoConf());
- this.comp = new TupleComparator(keySchema, sortSpecs);
+ this.comp = new BaseTupleComparator(keySchema, sortSpecs);
Path storeTablePath = new Path(context.getWorkDir(), "output");
LOG.info("Output data directory: " + storeTablePath);
this.meta = CatalogUtil.newTableMeta(context.getDataChannel() != null ?
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java
index 9d8122f..a02d00b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/RightOuterMergeJoinExec.java
@@ -51,7 +51,7 @@ public class RightOuterMergeJoinExec extends BinaryPhysicalExec {
private List<Tuple> innerTupleSlots;
private JoinTupleComparator joinComparator = null;
- private TupleComparator[] tupleComparator = null;
+ private TupleComparator [] tupleComparator = null;
private final static int INITIAL_TUPLE_SLOT = 10000;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
index c6a4f55..3cbb7c9 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java
@@ -30,7 +30,7 @@ import org.apache.tajo.engine.codegen.CompilationError;
import org.apache.tajo.engine.planner.Projector;
import org.apache.tajo.engine.utils.TupleCache;
import org.apache.tajo.engine.utils.TupleCacheKey;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.Target;
import org.apache.tajo.plan.expr.ConstEval;
import org.apache.tajo.plan.expr.EvalNode;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/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 a4a8d37..c0703dd 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
@@ -18,24 +18,25 @@
package org.apache.tajo.engine.planner.physical;
-import org.apache.tajo.worker.TaskAttemptContext;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.storage.BaseTupleComparator;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.storage.TupleComparator;
+import org.apache.tajo.worker.TaskAttemptContext;
import java.io.IOException;
import java.util.Comparator;
public abstract class SortExec extends UnaryPhysicalExec {
- private final Comparator<Tuple> comparator;
+ private final TupleComparator comparator;
private final SortSpec [] sortSpecs;
public SortExec(TaskAttemptContext context, Schema inSchema,
Schema outSchema, PhysicalExec child, SortSpec [] sortSpecs) {
super(context, inSchema, outSchema, child);
this.sortSpecs = sortSpecs;
- this.comparator = new TupleComparator(inSchema, sortSpecs);
+ this.comparator = new BaseTupleComparator(inSchema, sortSpecs);
}
public SortSpec[] getSortSpecs() {
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
index 3f4b22b..e36dcd8 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
@@ -27,6 +27,7 @@ import org.apache.tajo.plan.expr.WindowFunctionEval;
import org.apache.tajo.plan.function.FunctionContext;
import org.apache.tajo.plan.logical.WindowAggNode;
import org.apache.tajo.plan.logical.WindowSpec;
+import org.apache.tajo.storage.BaseTupleComparator;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.storage.TupleComparator;
import org.apache.tajo.storage.VTuple;
@@ -285,9 +286,9 @@ public class WindowAggExec extends UnaryPhysicalExec {
for (int idx = 0; idx < functions.length; idx++) {
if (orderedFuncFlags[idx]) {
- comp = new TupleComparator(inSchema, functions[idx].getSortSpecs());
+ comp = new BaseTupleComparator(inSchema, functions[idx].getSortSpecs());
Collections.sort(accumulatedInTuples, comp);
- comp = new TupleComparator(schemaForOrderBy, functions[idx].getSortSpecs());
+ comp = new BaseTupleComparator(schemaForOrderBy, functions[idx].getSortSpecs());
Collections.sort(evaluatedTuples, comp);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/worker/RangeRetrieverHandler.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/RangeRetrieverHandler.java b/tajo-core/src/main/java/org/apache/tajo/worker/RangeRetrieverHandler.java
index be33a12..2b58196 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/RangeRetrieverHandler.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/RangeRetrieverHandler.java
@@ -26,10 +26,10 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.storage.BaseTupleComparator;
import org.apache.tajo.storage.RowStoreUtil;
import org.apache.tajo.storage.RowStoreUtil.RowStoreDecoder;
import org.apache.tajo.storage.Tuple;
-import org.apache.tajo.storage.TupleComparator;
import org.apache.tajo.storage.index.bst.BSTIndex;
import org.apache.tajo.worker.dataserver.retriever.FileChunk;
import org.apache.tajo.worker.dataserver.retriever.RetrieverHandler;
@@ -57,10 +57,10 @@ public class RangeRetrieverHandler implements RetrieverHandler {
private final File file;
private final BSTIndex.BSTIndexReader idxReader;
private final Schema schema;
- private final TupleComparator comp;
+ private final BaseTupleComparator comp;
private final RowStoreDecoder decoder;
- public RangeRetrieverHandler(File outDir, Schema schema, TupleComparator comp) throws IOException {
+ public RangeRetrieverHandler(File outDir, Schema schema, BaseTupleComparator comp) throws IOException {
this.file = outDir;
BSTIndex index = new BSTIndex(new TajoConf());
this.schema = schema;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
index 3858c96..a7b52c7 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
@@ -52,6 +52,7 @@ import org.apache.tajo.plan.logical.*;
import org.apache.tajo.pullserver.TajoPullServerService;
import org.apache.tajo.pullserver.retriever.FileChunk;
import org.apache.tajo.rpc.NullCallback;
+import org.apache.tajo.storage.BaseTupleComparator;
import org.apache.tajo.storage.HashShuffleAppenderManager;
import org.apache.tajo.storage.StorageUtil;
import org.apache.tajo.storage.TupleComparator;
@@ -185,7 +186,7 @@ public class Task {
if (shuffleType == ShuffleType.RANGE_SHUFFLE) {
SortNode sortNode = PlannerUtil.findTopNode(plan, NodeType.SORT);
this.finalSchema = PlannerUtil.sortSpecsToSchema(sortNode.getSortKeys());
- this.sortComp = new TupleComparator(finalSchema, sortNode.getSortKeys());
+ this.sortComp = new BaseTupleComparator(finalSchema, sortNode.getSortKeys());
}
} else {
// The final result of a task will be written in a file named part-ss-nnnnnnn,
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java b/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
index 4287191..73d89ec 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/eval/ExprTestBase.java
@@ -41,7 +41,7 @@ import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.serder.EvalTreeProtoDeserializer;
import org.apache.tajo.plan.serder.EvalTreeProtoSerializer;
import org.apache.tajo.engine.query.QueryContext;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.serder.PlanProto;
import org.apache.tajo.plan.verifier.LogicalPlanVerifier;
import org.apache.tajo.plan.verifier.PreLogicalPlanVerifier;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java b/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java
index 90ec9b1..5cc89b8 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java
@@ -36,10 +36,10 @@ import org.apache.tajo.engine.function.builtin.SumInt;
import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.physical.PhysicalPlanUtil;
import org.apache.tajo.plan.LogicalPlanner;
-import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.plan.PlanningException;
import org.apache.tajo.plan.expr.*;
import org.apache.tajo.plan.logical.*;
+import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.storage.TupleComparator;
import org.apache.tajo.storage.VTuple;
@@ -299,7 +299,7 @@ public class TestPlannerUtil {
FieldEval f4 = new FieldEval("people.fid2", CatalogUtil.newSimpleDataType(Type.INT4));
EvalNode joinQual = new BinaryEval(EvalType.EQUAL, f1, f2);
- TupleComparator [] comparators = PhysicalPlanUtil.getComparatorsFromJoinQual(joinQual, outerSchema, innerSchema);
+ TupleComparator[] comparators = PhysicalPlanUtil.getComparatorsFromJoinQual(joinQual, outerSchema, innerSchema);
Tuple t1 = new VTuple(2);
t1.put(0, DatumFactory.createInt4(1));
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestUniformRangePartition.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestUniformRangePartition.java b/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestUniformRangePartition.java
index b044384..c5ef4ef 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestUniformRangePartition.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/planner/TestUniformRangePartition.java
@@ -23,8 +23,8 @@ import org.apache.tajo.catalog.SortSpec;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.plan.util.PlannerUtil;
+import org.apache.tajo.storage.BaseTupleComparator;
import org.apache.tajo.storage.Tuple;
-import org.apache.tajo.storage.TupleComparator;
import org.apache.tajo.storage.TupleRange;
import org.apache.tajo.storage.VTuple;
import org.junit.Test;
@@ -484,7 +484,7 @@ public class TestUniformRangePartition {
TupleRange expected = new TupleRange(sortSpecs, s, e);
UniformRangePartition partitioner = new UniformRangePartition(expected, sortSpecs);
- TupleComparator comp = new TupleComparator(schema, sortSpecs);
+ BaseTupleComparator comp = new BaseTupleComparator(schema, sortSpecs);
Tuple tuple = s;
Tuple prevTuple = null;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
index 868a82e..c7fbf39 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
@@ -68,7 +68,7 @@ public class TestBSTIndexExec {
private LogicalOptimizer optimizer;
private StorageManager sm;
private Schema idxSchema;
- private TupleComparator comp;
+ private BaseTupleComparator comp;
private BSTIndex.BSTIndexWriter writer;
private HashMap<Integer , Integer> randomValues ;
private int rndKey = -1;
@@ -104,7 +104,7 @@ public class TestBSTIndexExec {
idxSchema.addColumn("managerid", Type.INT4);
SortSpec[] sortKeys = new SortSpec[1];
sortKeys[0] = new SortSpec(idxSchema.getColumn("managerid"), true, false);
- this.comp = new TupleComparator(idxSchema, sortKeys);
+ this.comp = new BaseTupleComparator(idxSchema, sortKeys);
this.writer = new BSTIndex(conf).getIndexWriter(idxPath,
BSTIndex.TWO_LEVEL_INDEX, this.idxSchema, this.comp);
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
index 46f55c6..f09d104 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
@@ -152,7 +152,7 @@ public class TestExternalSortExec {
int cnt = 0;
exec.init();
long start = System.currentTimeMillis();
- TupleComparator comparator = new TupleComparator(proj.getSchema(),
+ BaseTupleComparator comparator = new BaseTupleComparator(proj.getSchema(),
new SortSpec[]{
new SortSpec(new Column("managerid", Type.INT4)),
new SortSpec(new Column("empid", Type.INT4))
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
index 341ce9e..42338a8 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestPhysicalPlanner.java
@@ -1073,7 +1073,7 @@ public class TestPhysicalPlanner {
keySchema.addColumn("?empId", Type.INT4);
SortSpec[] sortSpec = new SortSpec[1];
sortSpec[0] = new SortSpec(keySchema.getColumn(0), true, false);
- TupleComparator comp = new TupleComparator(keySchema, sortSpec);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortSpec);
BSTIndex bst = new BSTIndex(conf);
BSTIndex.BSTIndexReader reader = bst.getIndexReader(new Path(workDir, "output/index"),
keySchema, comp);
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java
index a02f2ce..9da9dee 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestProgressExternalSortExec.java
@@ -170,7 +170,7 @@ public class TestProgressExternalSortExec {
Tuple curVal;
int cnt = 0;
exec.init();
- TupleComparator comparator = new TupleComparator(proj.getSchema(),
+ BaseTupleComparator comparator = new BaseTupleComparator(proj.getSchema(),
new SortSpec[]{
new SortSpec(new Column("managerid", TajoDataTypes.Type.INT4)),
new SortSpec(new Column("empid", TajoDataTypes.Type.INT4))
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/engine/util/TestTupleUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/util/TestTupleUtil.java b/tajo-core/src/test/java/org/apache/tajo/engine/util/TestTupleUtil.java
index 62c959c..b8114e0 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/util/TestTupleUtil.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/util/TestTupleUtil.java
@@ -112,7 +112,7 @@ public class TestTupleUtil {
sortSpecs);
TupleRange [] ranges = partitioner.partition(5);
assertTrue(5 <= ranges.length);
- TupleComparator comp = new TupleComparator(schema, PlannerUtil.schemaToSortSpecs(schema));
+ BaseTupleComparator comp = new BaseTupleComparator(schema, PlannerUtil.schemaToSortSpecs(schema));
TupleRange prev = ranges[0];
for (int i = 1; i < ranges.length; i++) {
assertTrue(comp.compare(prev.getStart(), ranges[i].getStart()) < 0);
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-core/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java b/tajo-core/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java
index 124976a..e35b80c 100644
--- a/tajo-core/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java
+++ b/tajo-core/src/test/java/org/apache/tajo/worker/TestRangeRetrieverHandler.java
@@ -178,7 +178,7 @@ public class TestRangeRetrieverHandler {
exec.close();
Schema keySchema = PlannerUtil.sortSpecsToSchema(sortSpecs);
- TupleComparator comp = new TupleComparator(keySchema, sortSpecs);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortSpecs);
BSTIndex bst = new BSTIndex(conf);
BSTIndex.BSTIndexReader reader = bst.getIndexReader(
new Path(testDir, "output/index"), keySchema, comp);
@@ -302,7 +302,7 @@ public class TestRangeRetrieverHandler {
exec.close();
Schema keySchema = PlannerUtil.sortSpecsToSchema(sortSpecs);
- TupleComparator comp = new TupleComparator(keySchema, sortSpecs);
+ BaseTupleComparator comp = new BaseTupleComparator(keySchema, sortSpecs);
BSTIndex bst = new BSTIndex(conf);
BSTIndex.BSTIndexReader reader = bst.getIndexReader(
new Path(testDir, "output/index"), keySchema, comp);
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java
----------------------------------------------------------------------
diff --git a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java
index 5dae67e..6c8ef5d 100644
--- a/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java
+++ b/tajo-jdbc/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java
@@ -17,6 +17,7 @@ package org.apache.tajo.jdbc; /**
*/
import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.IntervalDatum;
import org.apache.tajo.datum.NullDatum;
import org.apache.tajo.datum.ProtobufDatum;
import org.apache.tajo.exception.UnsupportedException;
@@ -47,7 +48,12 @@ public class MetaDataTuple implements Tuple {
@Override
public boolean isNull(int fieldid) {
- return values.get(fieldid) == null || values.get(fieldid) instanceof NullDatum;
+ return values.get(fieldid) == null || values.get(fieldid).isNull();
+ }
+
+ @Override
+ public boolean isNotNull(int fieldid) {
+ return !isNull(fieldid);
}
@Override
@@ -142,7 +148,12 @@ public class MetaDataTuple implements Tuple {
@Override
public ProtobufDatum getProtobufDatum(int fieldId) {
- throw new UnsupportedException();
+ throw new UnsupportedException("getProtobufDatum");
+ }
+
+ @Override
+ public IntervalDatum getInterval(int fieldId) {
+ throw new UnsupportedException("getInterval");
}
@Override
@@ -157,6 +168,6 @@ public class MetaDataTuple implements Tuple {
@Override
public Datum[] getValues(){
- throw new UnsupportedException();
+ throw new UnsupportedException("getValues");
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java
index 6bdfb9d..68f2186 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java
@@ -32,7 +32,7 @@ import org.apache.tajo.plan.logical.*;
import org.apache.tajo.plan.nameresolver.NameResolver;
import org.apache.tajo.plan.nameresolver.NameResolvingMode;
import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.visitor.SimpleAlgebraVisitor;
import org.apache.tajo.util.TUtil;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java
index 84942a1..e953ccf 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java
@@ -47,7 +47,7 @@ import org.apache.tajo.plan.nameresolver.NameResolvingMode;
import org.apache.tajo.plan.rewrite.rules.ProjectionPushDownRule;
import org.apache.tajo.plan.util.ExprFinder;
import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.verifier.VerifyException;
import org.apache.tajo.util.KeyValueSet;
import org.apache.tajo.util.Pair;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-plan/src/main/java/org/apache/tajo/plan/joinorder/GreedyHeuristicJoinOrderAlgorithm.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/joinorder/GreedyHeuristicJoinOrderAlgorithm.java b/tajo-plan/src/main/java/org/apache/tajo/plan/joinorder/GreedyHeuristicJoinOrderAlgorithm.java
index a916d5c..0fad8a8 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/joinorder/GreedyHeuristicJoinOrderAlgorithm.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/joinorder/GreedyHeuristicJoinOrderAlgorithm.java
@@ -23,7 +23,7 @@ import org.apache.tajo.catalog.Schema;
import org.apache.tajo.plan.LogicalPlan;
import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.plan.PlanningException;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.expr.AlgebraicUtil;
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.util.TUtil;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-plan/src/main/java/org/apache/tajo/plan/logical/ScanNode.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/logical/ScanNode.java b/tajo-plan/src/main/java/org/apache/tajo/plan/logical/ScanNode.java
index a7ef543..3e4abe3 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/logical/ScanNode.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/logical/ScanNode.java
@@ -25,7 +25,7 @@ import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.plan.PlanString;
import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.Target;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.util.TUtil;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-plan/src/main/java/org/apache/tajo/plan/logical/TableSubQueryNode.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/logical/TableSubQueryNode.java b/tajo-plan/src/main/java/org/apache/tajo/plan/logical/TableSubQueryNode.java
index c8fbecd..4e5f41c 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/logical/TableSubQueryNode.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/logical/TableSubQueryNode.java
@@ -23,7 +23,7 @@ import com.google.gson.annotations.Expose;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.plan.PlanString;
import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.Target;
public class TableSubQueryNode extends RelationNode implements Projectable {
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java
index 167da5b..f7fd90d 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/ProjectionPushDownRule.java
@@ -31,7 +31,7 @@ import org.apache.tajo.plan.expr.*;
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.plan.rewrite.RewriteRule;
import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.util.SchemaUtil;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.plan.visitor.BasicLogicalPlanVisitor;
import org.apache.tajo.util.TUtil;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java b/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java
index 5fe1515..6868b6c 100644
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java
@@ -25,6 +25,7 @@ import org.apache.tajo.algebra.*;
import org.apache.tajo.annotation.Nullable;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SchemaUtil;
import org.apache.tajo.catalog.SortSpec;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.common.TajoDataTypes.DataType;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-plan/src/main/java/org/apache/tajo/plan/util/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/util/SchemaUtil.java b/tajo-plan/src/main/java/org/apache/tajo/plan/util/SchemaUtil.java
deleted file mode 100644
index 78135ab..0000000
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/util/SchemaUtil.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.tajo.plan.util;
-
-import org.apache.tajo.catalog.Column;
-import org.apache.tajo.catalog.Schema;
-import org.apache.tajo.catalog.TableDesc;
-
-public class SchemaUtil {
- // See TAJO-914 bug.
- //
- // Its essential problem is that constant value is evaluated multiple times at each scan.
- // As a result, join nodes can take the child nodes which have the same named fields.
- // Because current schema does not allow the same name and ignore the duplicated schema,
- // it finally causes the in-out schema mismatch between the parent and child nodes.
- //
- // tmpColumnSeq is a hack to avoid the above problem by keeping duplicated constant values as different name fields.
- // The essential solution would be https://issues.apache.org/jira/browse/TAJO-895.
- static int tmpColumnSeq = 0;
- public static Schema merge(Schema left, Schema right) {
- Schema merged = new Schema();
- for(Column col : left.getColumns()) {
- if (!merged.containsByQualifiedName(col.getQualifiedName())) {
- merged.addColumn(col);
- }
- }
- for(Column col : right.getColumns()) {
- if (merged.containsByQualifiedName(col.getQualifiedName())) {
- merged.addColumn("?fake" + (tmpColumnSeq++), col.getDataType());
- } else {
- merged.addColumn(col);
- }
- }
-
- // if overflow
- if (tmpColumnSeq < 0) {
- tmpColumnSeq = 0;
- }
- return merged;
- }
-
- /**
- * Get common columns to be used as join keys of natural joins.
- */
- public static Schema getNaturalJoinColumns(Schema left, Schema right) {
- Schema common = new Schema();
- for (Column outer : left.getColumns()) {
- if (!common.containsByName(outer.getSimpleName()) && right.containsByName(outer.getSimpleName())) {
- common.addColumn(new Column(outer.getSimpleName(), outer.getDataType()));
- }
- }
-
- return common;
- }
-
- public static Schema getQualifiedLogicalSchema(TableDesc tableDesc, String tableName) {
- Schema logicalSchema = new Schema(tableDesc.getLogicalSchema());
- if (tableName != null) {
- logicalSchema.setQualifier(tableName);
- }
- return logicalSchema;
- }
-
- public static <T extends Schema> T clone(Schema schema) {
- try {
- T copy = (T) schema.clone();
- return copy;
- } catch (CloneNotSupportedException e) {
- throw new RuntimeException(e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java b/tajo-storage/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java
new file mode 100644
index 0000000..b829f60
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/BaseTupleComparator.java
@@ -0,0 +1,206 @@
+/**
+ * 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.storage;
+
+import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.common.ProtoObject;
+import org.apache.tajo.datum.Datum;
+
+import static org.apache.tajo.catalog.proto.CatalogProtos.TupleComparatorSpecProto;
+import static org.apache.tajo.index.IndexProtos.TupleComparatorProto;
+
+/**
+ * The Comparator class for Tuples
+ *
+ * @see Tuple
+ */
+public class BaseTupleComparator extends TupleComparator implements ProtoObject<TupleComparatorProto> {
+ private final Schema schema;
+ private final SortSpec [] sortSpecs;
+ private final int[] sortKeyIds;
+ private final boolean[] asc;
+ @SuppressWarnings("unused")
+ private final boolean[] nullFirsts;
+
+ private Datum left;
+ private Datum right;
+ private int compVal;
+
+ /**
+ * @param schema The schema of input tuples
+ * @param sortKeys The description of sort keys
+ */
+ public BaseTupleComparator(Schema schema, SortSpec[] sortKeys) {
+ Preconditions.checkArgument(sortKeys.length > 0,
+ "At least one sort key must be specified.");
+
+ this.schema = schema;
+ this.sortSpecs = sortKeys;
+ this.sortKeyIds = new int[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].isNullFirst();
+ }
+ }
+
+ public BaseTupleComparator(TupleComparatorProto proto) {
+ this.schema = new Schema(proto.getSchema());
+
+ this.sortSpecs = new SortSpec[proto.getSortSpecsCount()];
+ for (int i = 0; i < proto.getSortSpecsCount(); i++) {
+ sortSpecs[i] = new SortSpec(proto.getSortSpecs(i));
+ }
+
+ this.sortKeyIds = new int[proto.getCompSpecsCount()];
+ this.asc = new boolean[proto.getCompSpecsCount()];
+ this.nullFirsts = new boolean[proto.getCompSpecsCount()];
+
+ for (int i = 0; i < proto.getCompSpecsCount(); i++) {
+ TupleComparatorSpecProto sortSepcProto = proto.getCompSpecs(i);
+ sortKeyIds[i] = sortSepcProto.getColumnId();
+ asc[i] = sortSepcProto.getAscending();
+ nullFirsts[i] = sortSepcProto.getNullFirst();
+ }
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+ public SortSpec [] getSortSpecs() {
+ return sortSpecs;
+ }
+
+ public int [] getSortKeyIds() {
+ return sortKeyIds;
+ }
+
+ @Override
+ public boolean isAscendingFirstKey() {
+ return this.asc[0];
+ }
+
+ @Override
+ public int compare(Tuple tuple1, Tuple tuple2) {
+ for (int i = 0; i < sortKeyIds.length; i++) {
+ left = tuple1.get(sortKeyIds[i]);
+ right = tuple2.get(sortKeyIds[i]);
+
+ if (left.isNull() || right.isNull()) {
+ if (!left.equals(right)) {
+ if (left.isNull()) {
+ compVal = 1;
+ } else if (right.isNull()) {
+ compVal = -1;
+ }
+ if (nullFirsts[i]) {
+ if (compVal != 0) {
+ compVal *= -1;
+ }
+ }
+ } else {
+ compVal = 0;
+ }
+ } else {
+ if (asc[i]) {
+ compVal = left.compareTo(right);
+ } else {
+ compVal = right.compareTo(left);
+ }
+ }
+
+ if (compVal < 0 || compVal > 0) {
+ return compVal;
+ }
+ }
+ return 0;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(sortKeyIds);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof BaseTupleComparator) {
+ BaseTupleComparator other = (BaseTupleComparator) obj;
+ if (sortKeyIds.length != other.sortKeyIds.length) {
+ return false;
+ }
+
+ for (int i = 0; i < sortKeyIds.length; i++) {
+ if (sortKeyIds[i] != other.sortKeyIds[i] ||
+ asc[i] != other.asc[i] ||
+ nullFirsts[i] != other.nullFirsts[i]) {
+ return false;
+ }
+ }
+
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public TupleComparatorProto getProto() {
+ TupleComparatorProto.Builder builder = TupleComparatorProto.newBuilder();
+ builder.setSchema(schema.getProto());
+ for (int i = 0; i < sortSpecs.length; i++) {
+ builder.addSortSpecs(sortSpecs[i].getProto());
+ }
+
+ TupleComparatorSpecProto.Builder sortSpecBuilder;
+ for (int i = 0; i < sortKeyIds.length; i++) {
+ sortSpecBuilder = TupleComparatorSpecProto.newBuilder();
+ sortSpecBuilder.setColumnId(sortKeyIds[i]);
+ sortSpecBuilder.setAscending(asc[i]);
+ sortSpecBuilder.setNullFirst(nullFirsts[i]);
+ builder.addCompSpecs(sortSpecBuilder);
+ }
+
+ return builder.build();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+
+ String prefix = "";
+ for (int i = 0; i < sortKeyIds.length; i++) {
+ sb.append(prefix).append("SortKeyId=").append(sortKeyIds[i])
+ .append(",Asc=").append(asc[i])
+ .append(",NullFirst=").append(nullFirsts[i]);
+ prefix = " ,";
+ }
+ return sb.toString();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/storage/FrameTuple.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/FrameTuple.java b/tajo-storage/src/main/java/org/apache/tajo/storage/FrameTuple.java
index e0f8a2e..8b7e2e0 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/FrameTuple.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/FrameTuple.java
@@ -23,7 +23,7 @@ package org.apache.tajo.storage;
import com.google.common.base.Preconditions;
import org.apache.tajo.datum.Datum;
-import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.datum.IntervalDatum;
import org.apache.tajo.datum.ProtobufDatum;
import org.apache.tajo.exception.UnsupportedException;
@@ -71,7 +71,12 @@ public class FrameTuple implements Tuple, Cloneable {
@Override
public boolean isNull(int fieldid) {
- return get(fieldid) instanceof NullDatum;
+ return get(fieldid).isNull();
+ }
+
+ @Override
+ public boolean isNotNull(int fieldid) {
+ return !isNull(fieldid);
}
@Override
@@ -177,6 +182,11 @@ public class FrameTuple implements Tuple, Cloneable {
}
@Override
+ public IntervalDatum getInterval(int fieldId) {
+ return (IntervalDatum) get(fieldId);
+ }
+
+ @Override
public char [] getUnicodeChars(int fieldId) {
return get(fieldId).asUnicodeChars();
}
[2/3] TAJO-1153: Merge off-heap package in block_iteration branch to
master branch.
Posted by hy...@apache.org.
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/storage/LazyTuple.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/LazyTuple.java b/tajo-storage/src/main/java/org/apache/tajo/storage/LazyTuple.java
index 167e4a8..bfbe478 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/LazyTuple.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/LazyTuple.java
@@ -20,6 +20,7 @@ package org.apache.tajo.storage;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.IntervalDatum;
import org.apache.tajo.datum.NullDatum;
import org.apache.tajo.datum.ProtobufDatum;
import org.apache.tajo.exception.UnsupportedException;
@@ -68,7 +69,12 @@ public class LazyTuple implements Tuple, Cloneable {
@Override
public boolean isNull(int fieldid) {
- return get(fieldid) instanceof NullDatum;
+ return get(fieldid).isNull();
+ }
+
+ @Override
+ public boolean isNotNull(int fieldid) {
+ return !isNull(fieldid);
}
@Override
@@ -199,6 +205,11 @@ public class LazyTuple implements Tuple, Cloneable {
}
@Override
+ public IntervalDatum getInterval(int fieldId) {
+ return (IntervalDatum) get(fieldId);
+ }
+
+ @Override
public char[] getUnicodeChars(int fieldId) {
return get(fieldId).asUnicodeChars();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java b/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java
index 70044ca..24b6280 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/RowStoreUtil.java
@@ -23,8 +23,10 @@ import org.apache.tajo.catalog.Schema;
import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.datum.IntervalDatum;
+import org.apache.tajo.datum.ProtobufDatum;
import org.apache.tajo.exception.UnsupportedException;
import org.apache.tajo.storage.exception.UnknownDataTypeException;
+import org.apache.tajo.tuple.offheap.RowWriter;
import org.apache.tajo.util.BitArray;
import java.nio.ByteBuffer;
@@ -177,7 +179,8 @@ public class RowStoreUtil {
nullFlags = new BitArray(schema.size());
headerSize = nullFlags.bytesLength();
}
- public byte [] toBytes(Tuple tuple) {
+
+ public byte[] toBytes(Tuple tuple) {
nullFlags.clear();
int size = estimateTupleDataSize(tuple);
ByteBuffer bb = ByteBuffer.allocate(size + headerSize);
@@ -191,42 +194,64 @@ public class RowStoreUtil {
col = schema.getColumn(i);
switch (col.getDataType().getType()) {
- case NULL_TYPE: nullFlags.set(i); break;
- case BOOLEAN: bb.put(tuple.get(i).asByte()); break;
- case BIT: bb.put(tuple.get(i).asByte()); break;
- case CHAR: bb.put(tuple.get(i).asByte()); break;
- case INT2: bb.putShort(tuple.get(i).asInt2()); break;
- case INT4: bb.putInt(tuple.get(i).asInt4()); break;
- case INT8: bb.putLong(tuple.get(i).asInt8()); break;
- case FLOAT4: bb.putFloat(tuple.get(i).asFloat4()); break;
- case FLOAT8: bb.putDouble(tuple.get(i).asFloat8()); break;
- case TEXT:
- byte [] _string = tuple.get(i).asByteArray();
- bb.putInt(_string.length);
- bb.put(_string);
- break;
- case DATE: bb.putInt(tuple.get(i).asInt4()); break;
- case TIME:
- case TIMESTAMP:
- bb.putLong(tuple.get(i).asInt8());
- break;
- case INTERVAL:
- IntervalDatum interval = (IntervalDatum) tuple.get(i);
- bb.putInt(interval.getMonths());
- bb.putLong(interval.getMilliSeconds());
- break;
- case BLOB:
- byte [] bytes = tuple.get(i).asByteArray();
- bb.putInt(bytes.length);
- bb.put(bytes);
- break;
- case INET4:
- byte [] ipBytes = tuple.get(i).asByteArray();
- bb.put(ipBytes);
- break;
- case INET6: bb.put(tuple.get(i).asByteArray()); break;
- default:
- throw new RuntimeException(new UnknownDataTypeException(col.getDataType().getType().name()));
+ case NULL_TYPE:
+ nullFlags.set(i);
+ break;
+ case BOOLEAN:
+ bb.put(tuple.get(i).asByte());
+ break;
+ case BIT:
+ bb.put(tuple.get(i).asByte());
+ break;
+ case CHAR:
+ bb.put(tuple.get(i).asByte());
+ break;
+ case INT2:
+ bb.putShort(tuple.get(i).asInt2());
+ break;
+ case INT4:
+ bb.putInt(tuple.get(i).asInt4());
+ break;
+ case INT8:
+ bb.putLong(tuple.get(i).asInt8());
+ break;
+ case FLOAT4:
+ bb.putFloat(tuple.get(i).asFloat4());
+ break;
+ case FLOAT8:
+ bb.putDouble(tuple.get(i).asFloat8());
+ break;
+ case TEXT:
+ byte[] _string = tuple.get(i).asByteArray();
+ bb.putInt(_string.length);
+ bb.put(_string);
+ break;
+ case DATE:
+ bb.putInt(tuple.get(i).asInt4());
+ break;
+ case TIME:
+ case TIMESTAMP:
+ bb.putLong(tuple.get(i).asInt8());
+ break;
+ case INTERVAL:
+ IntervalDatum interval = (IntervalDatum) tuple.get(i);
+ bb.putInt(interval.getMonths());
+ bb.putLong(interval.getMilliSeconds());
+ break;
+ case BLOB:
+ byte[] bytes = tuple.get(i).asByteArray();
+ bb.putInt(bytes.length);
+ bb.put(bytes);
+ break;
+ case INET4:
+ byte[] ipBytes = tuple.get(i).asByteArray();
+ bb.put(ipBytes);
+ break;
+ case INET6:
+ bb.put(tuple.get(i).asByteArray());
+ break;
+ default:
+ throw new RuntimeException(new UnknownDataTypeException(col.getDataType().getType().name()));
}
}
@@ -237,7 +262,7 @@ public class RowStoreUtil {
bb.position(finalPosition);
bb.flip();
- byte [] buf = new byte [bb.limit()];
+ byte[] buf = new byte[bb.limit()];
bb.get(buf);
return buf;
}
@@ -254,24 +279,38 @@ public class RowStoreUtil {
col = schema.getColumn(i);
switch (col.getDataType().getType()) {
- case BOOLEAN:
- case BIT:
- case CHAR: size += 1; break;
- case INT2: size += 2; break;
- case DATE:
- case INT4:
- case FLOAT4: size += 4; break;
- case TIME:
- case TIMESTAMP:
- case INT8:
- case FLOAT8: size += 8; break;
- case INTERVAL: size += 12; break;
- case TEXT:
- case BLOB: size += (4 + tuple.get(i).asByteArray().length); break;
- case INET4:
- case INET6: size += tuple.get(i).asByteArray().length; break;
- default:
- throw new RuntimeException(new UnknownDataTypeException(col.getDataType().getType().name()));
+ case BOOLEAN:
+ case BIT:
+ case CHAR:
+ size += 1;
+ break;
+ case INT2:
+ size += 2;
+ break;
+ case DATE:
+ case INT4:
+ case FLOAT4:
+ size += 4;
+ break;
+ case TIME:
+ case TIMESTAMP:
+ case INT8:
+ case FLOAT8:
+ size += 8;
+ break;
+ case INTERVAL:
+ size += 12;
+ break;
+ case TEXT:
+ case BLOB:
+ size += (4 + tuple.get(i).asByteArray().length);
+ break;
+ case INET4:
+ case INET6:
+ size += tuple.get(i).asByteArray().length;
+ break;
+ default:
+ throw new RuntimeException(new UnknownDataTypeException(col.getDataType().getType().name()));
}
}
@@ -284,4 +323,55 @@ public class RowStoreUtil {
return schema;
}
}
+
+ public static void convert(Tuple tuple, RowWriter writer) {
+ writer.startRow();
+
+ for (int i = 0; i < writer.dataTypes().length; i++) {
+ if (tuple.isNull(i)) {
+ writer.skipField();
+ continue;
+ }
+ switch (writer.dataTypes()[i].getType()) {
+ case BOOLEAN:
+ writer.putBool(tuple.getBool(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 TEXT:
+ writer.putText(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.skipField();
+ break;
+ default:
+ throw new UnsupportedException("Unknown data type: " + writer.dataTypes()[i]);
+ }
+ }
+ writer.endRow();
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/storage/TupleComparator.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/TupleComparator.java b/tajo-storage/src/main/java/org/apache/tajo/storage/TupleComparator.java
index 084c105..8dffd8d 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/TupleComparator.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/TupleComparator.java
@@ -1,4 +1,4 @@
-/**
+/***
* 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
@@ -18,164 +18,15 @@
package org.apache.tajo.storage;
-import com.google.common.base.Objects;
-import com.google.common.base.Preconditions;
-import org.apache.tajo.catalog.Schema;
-import org.apache.tajo.catalog.SortSpec;
import org.apache.tajo.common.ProtoObject;
-import org.apache.tajo.datum.Datum;
-import org.apache.tajo.datum.NullDatum;
import java.util.Comparator;
-import static org.apache.tajo.catalog.proto.CatalogProtos.TupleComparatorSpecProto;
import static org.apache.tajo.index.IndexProtos.TupleComparatorProto;
-/**
- * The Comparator class for Tuples
- *
- * @see Tuple
- */
-public class TupleComparator implements Comparator<Tuple>, ProtoObject<TupleComparatorProto> {
- private final int[] sortKeyIds;
- private final boolean[] asc;
- @SuppressWarnings("unused")
- private final boolean[] nullFirsts;
-
- /**
- * @param schema The schema of input tuples
- * @param sortKeys The description of sort keys
- */
- public TupleComparator(Schema schema, SortSpec[] sortKeys) {
- Preconditions.checkArgument(sortKeys.length > 0,
- "At least one sort key must be specified.");
-
- this.sortKeyIds = new int[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].isNullFirst();
- }
- }
-
- public TupleComparator(TupleComparatorProto proto) {
- this.sortKeyIds = new int[proto.getCompSpecsCount()];
- this.asc = new boolean[proto.getCompSpecsCount()];
- this.nullFirsts = new boolean[proto.getCompSpecsCount()];
-
- for (int i = 0; i < proto.getCompSpecsCount(); i++) {
- TupleComparatorSpecProto sortSepcProto = proto.getCompSpecs(i);
- sortKeyIds[i] = sortSepcProto.getColumnId();
- asc[i] = sortSepcProto.getAscending();
- nullFirsts[i] = sortSepcProto.getNullFirst();
- }
- }
-
- public boolean isAscendingFirstKey() {
- return this.asc[0];
- }
-
- @Override
- public int compare(Tuple tuple1, Tuple tuple2) {
- Datum left = null;
- Datum right = null;
- int compVal = 0;
-
- for (int i = 0; i < sortKeyIds.length; i++) {
- left = tuple1.get(sortKeyIds[i]);
- right = tuple2.get(sortKeyIds[i]);
-
- if (left instanceof NullDatum || right instanceof NullDatum) {
- if (!left.equals(right)) {
- if (left instanceof NullDatum) {
- compVal = 1;
- } else if (right instanceof NullDatum) {
- compVal = -1;
- }
- if (nullFirsts[i]) {
- if (compVal != 0) {
- compVal *= -1;
- }
- }
- } else {
- compVal = 0;
- }
- } else {
- if (asc[i]) {
- compVal = left.compareTo(right);
- } else {
- compVal = right.compareTo(left);
- }
- }
-
- if (compVal < 0 || compVal > 0) {
- return compVal;
- }
- }
- return 0;
- }
-
- @Override
- public int hashCode() {
- return Objects.hashCode(sortKeyIds);
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj instanceof TupleComparator) {
- TupleComparator other = (TupleComparator) obj;
- if (sortKeyIds.length != other.sortKeyIds.length) {
- return false;
- }
-
- for (int i = 0; i < sortKeyIds.length; i++) {
- if (sortKeyIds[i] != other.sortKeyIds[i] ||
- asc[i] != other.asc[i] ||
- nullFirsts[i] != other.nullFirsts[i]) {
- return false;
- }
- }
-
- return true;
- } else {
- return false;
- }
- }
-
- @Override
- public TupleComparatorProto getProto() {
- TupleComparatorProto.Builder builder = TupleComparatorProto.newBuilder();
- TupleComparatorSpecProto.Builder sortSpecBuilder;
-
- for (int i = 0; i < sortKeyIds.length; i++) {
- sortSpecBuilder = TupleComparatorSpecProto.newBuilder();
- sortSpecBuilder.setColumnId(sortKeyIds[i]);
- sortSpecBuilder.setAscending(asc[i]);
- sortSpecBuilder.setNullFirst(nullFirsts[i]);
- builder.addCompSpecs(sortSpecBuilder);
- }
-
- return builder.build();
- }
+public abstract class TupleComparator implements Comparator<Tuple>, ProtoObject<TupleComparatorProto> {
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
+ public abstract int compare(Tuple o1, Tuple o2);
- String prefix = "";
- for (int i = 0; i < sortKeyIds.length; i++) {
- sb.append(prefix).append("SortKeyId=").append(sortKeyIds[i])
- .append(",Asc=").append(asc[i])
- .append(",NullFirst=").append(nullFirsts[i]);
- prefix = " ,";
- }
- return sb.toString();
- }
-}
\ No newline at end of file
+ public abstract boolean isAscendingFirstKey();
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/storage/TupleRange.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/TupleRange.java b/tajo-storage/src/main/java/org/apache/tajo/storage/TupleRange.java
index 6cc09d4..e824b99 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/TupleRange.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/TupleRange.java
@@ -33,7 +33,7 @@ public class TupleRange implements Comparable<TupleRange>, Cloneable {
private final TupleComparator comp;
public TupleRange(final SortSpec[] sortSpecs, final Tuple start, final Tuple end) {
- this.comp = new TupleComparator(sortSpecsToSchema(sortSpecs), sortSpecs);
+ this.comp = new BaseTupleComparator(sortSpecsToSchema(sortSpecs), sortSpecs);
// if there is only one value, start == end
this.start = start;
this.end = end;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/storage/index/IndexMethod.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/index/IndexMethod.java b/tajo-storage/src/main/java/org/apache/tajo/storage/index/IndexMethod.java
index 74be7ff..ccba3be 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/index/IndexMethod.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/index/IndexMethod.java
@@ -20,6 +20,7 @@ package org.apache.tajo.storage.index;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.storage.BaseTupleComparator;
import org.apache.tajo.storage.TupleComparator;
import java.io.IOException;
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/storage/index/bst/BSTIndex.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/index/bst/BSTIndex.java b/tajo-storage/src/main/java/org/apache/tajo/storage/index/bst/BSTIndex.java
index 5d43bd5..f093f9d 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/index/bst/BSTIndex.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/index/bst/BSTIndex.java
@@ -27,12 +27,9 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.proto.CatalogProtos.SchemaProto;
-import org.apache.tajo.storage.RowStoreUtil;
+import org.apache.tajo.storage.*;
import org.apache.tajo.storage.RowStoreUtil.RowStoreDecoder;
import org.apache.tajo.storage.RowStoreUtil.RowStoreEncoder;
-import org.apache.tajo.storage.StorageUtil;
-import org.apache.tajo.storage.Tuple;
-import org.apache.tajo.storage.TupleComparator;
import org.apache.tajo.storage.index.IndexMethod;
import org.apache.tajo.storage.index.IndexWriter;
import org.apache.tajo.storage.index.OrderIndexReader;
@@ -72,8 +69,7 @@ public class BSTIndex implements IndexMethod {
}
@Override
- public BSTIndexReader getIndexReader(Path fileName, Schema keySchema,
- TupleComparator comparator) throws IOException {
+ public BSTIndexReader getIndexReader(Path fileName, Schema keySchema, TupleComparator comparator) throws IOException {
return new BSTIndexReader(fileName, keySchema, comparator);
}
@@ -350,7 +346,7 @@ public class BSTIndex implements IndexMethod {
TupleComparatorProto.Builder compProto = TupleComparatorProto.newBuilder();
compProto.mergeFrom(compBytes);
- this.comparator = new TupleComparator(compProto.build());
+ this.comparator = new BaseTupleComparator(compProto.build());
// level
this.level = indexIn.readInt();
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java
new file mode 100644
index 0000000..c1835df
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/BaseTupleBuilder.java
@@ -0,0 +1,112 @@
+/***
+ * 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;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SchemaUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.tuple.offheap.HeapTuple;
+import org.apache.tajo.tuple.offheap.OffHeapRowWriter;
+import org.apache.tajo.tuple.offheap.ZeroCopyTuple;
+import org.apache.tajo.unit.StorageUnit;
+import org.apache.tajo.util.Deallocatable;
+import org.apache.tajo.util.FileUtil;
+import org.apache.tajo.util.UnsafeUtil;
+import sun.misc.Unsafe;
+import sun.nio.ch.DirectBuffer;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+public class BaseTupleBuilder extends OffHeapRowWriter implements TupleBuilder, Deallocatable {
+ private static final Log LOG = LogFactory.getLog(BaseTupleBuilder.class);
+
+ private static final Unsafe UNSAFE = UnsafeUtil.unsafe;
+
+ // buffer
+ private ByteBuffer buffer;
+ private long address;
+
+ public BaseTupleBuilder(Schema schema) {
+ super(SchemaUtil.toDataTypes(schema));
+ buffer = ByteBuffer.allocateDirect(64 * StorageUnit.KB).order(ByteOrder.nativeOrder());
+ address = UnsafeUtil.getAddress(buffer);
+ }
+
+ @Override
+ public long address() {
+ return address;
+ }
+
+ public void ensureSize(int size) {
+ if (buffer.remaining() - size < 0) { // check the remain size
+ // enlarge new buffer and copy writing data
+ int newBlockSize = UnsafeUtil.alignedSize(buffer.capacity() * 2);
+ ByteBuffer newByteBuf = ByteBuffer.allocateDirect(newBlockSize);
+ long newAddress = ((DirectBuffer)newByteBuf).address();
+ UNSAFE.copyMemory(this.address, newAddress, buffer.limit());
+ LOG.debug("Increase the buffer size to " + FileUtil.humanReadableByteCount(newBlockSize, false));
+
+ // release existing buffer and replace variables
+ UnsafeUtil.free(buffer);
+ buffer = newByteBuf;
+ address = newAddress;
+ }
+ }
+
+ @Override
+ public int position() {
+ return 0;
+ }
+
+ @Override
+ public void forward(int length) {
+ }
+
+ @Override
+ public void endRow() {
+ super.endRow();
+ buffer.position(0).limit(offset());
+ }
+
+ @Override
+ public Tuple build() {
+ return buildToHeapTuple();
+ }
+
+ public HeapTuple buildToHeapTuple() {
+ byte [] bytes = new byte[buffer.limit()];
+ UNSAFE.copyMemory(null, address, bytes, UnsafeUtil.ARRAY_BOOLEAN_BASE_OFFSET, buffer.limit());
+ return new HeapTuple(bytes, dataTypes());
+ }
+
+ public ZeroCopyTuple buildToZeroCopyTuple() {
+ ZeroCopyTuple zcTuple = new ZeroCopyTuple();
+ zcTuple.set(buffer, 0, buffer.limit(), dataTypes());
+ return zcTuple;
+ }
+
+ public void release() {
+ UnsafeUtil.free(buffer);
+ buffer = null;
+ address = 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/RowBlockReader.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/RowBlockReader.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/RowBlockReader.java
new file mode 100644
index 0000000..be734e1
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/RowBlockReader.java
@@ -0,0 +1,33 @@
+/**
+ * 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;
+
+import org.apache.tajo.storage.Tuple;
+
+public interface RowBlockReader<T extends Tuple> {
+
+ /**
+ * Return for each tuple
+ *
+ * @return True if tuple block is filled with tuples. Otherwise, It will return false.
+ */
+ public boolean next(T tuple);
+
+ public void reset();
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/TupleBuilder.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/TupleBuilder.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/TupleBuilder.java
new file mode 100644
index 0000000..c43c018
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/TupleBuilder.java
@@ -0,0 +1,26 @@
+/***
+ * 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;
+
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.tuple.offheap.RowWriter;
+
+public interface TupleBuilder extends RowWriter {
+ public Tuple build();
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/DirectBufTuple.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/DirectBufTuple.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/DirectBufTuple.java
new file mode 100644
index 0000000..9662d5a
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/DirectBufTuple.java
@@ -0,0 +1,41 @@
+/*
+ * 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.offheap;
+
+import org.apache.tajo.util.Deallocatable;
+import org.apache.tajo.util.UnsafeUtil;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.tajo.common.TajoDataTypes.DataType;
+
+public class DirectBufTuple extends UnSafeTuple implements Deallocatable {
+ private ByteBuffer bb;
+
+ public DirectBufTuple(int length, DataType[] types) {
+ bb = ByteBuffer.allocateDirect(length).order(ByteOrder.nativeOrder());
+ set(bb, 0, length, types);
+ }
+
+ @Override
+ public void release() {
+ UnsafeUtil.free(bb);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/FixedSizeLimitSpec.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/FixedSizeLimitSpec.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/FixedSizeLimitSpec.java
new file mode 100644
index 0000000..a327123
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/FixedSizeLimitSpec.java
@@ -0,0 +1,32 @@
+/*
+ * 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.offheap;
+
+/**
+ * Fixed size limit specification
+ */
+public class FixedSizeLimitSpec extends ResizableLimitSpec {
+ public FixedSizeLimitSpec(long size) {
+ super(size, size);
+ }
+
+ public FixedSizeLimitSpec(long size, float allowedOverflowRatio) {
+ super(size, size, allowedOverflowRatio);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/HeapTuple.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/HeapTuple.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/HeapTuple.java
new file mode 100644
index 0000000..e38555c
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/HeapTuple.java
@@ -0,0 +1,269 @@
+/***
+ * 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.offheap;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import com.sun.tools.javac.util.Convert;
+import org.apache.tajo.datum.*;
+import org.apache.tajo.exception.UnsupportedException;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+import org.apache.tajo.util.SizeOf;
+import org.apache.tajo.util.UnsafeUtil;
+import sun.misc.Unsafe;
+
+import java.nio.ByteBuffer;
+
+import static org.apache.tajo.common.TajoDataTypes.DataType;
+
+public class HeapTuple implements Tuple {
+ private static final Unsafe UNSAFE = UnsafeUtil.unsafe;
+ private static final long BASE_OFFSET = UnsafeUtil.ARRAY_BYTE_BASE_OFFSET;
+
+ private final byte [] data;
+ private final DataType [] types;
+
+ public HeapTuple(final byte [] bytes, final DataType [] types) {
+ this.data = bytes;
+ this.types = types;
+ }
+
+ @Override
+ public int size() {
+ return data.length;
+ }
+
+ public ByteBuffer nioBuffer() {
+ return ByteBuffer.wrap(data);
+ }
+
+ private int getFieldOffset(int fieldId) {
+ return UNSAFE.getInt(data, BASE_OFFSET + SizeOf.SIZE_OF_INT + (fieldId * SizeOf.SIZE_OF_INT));
+ }
+
+ private int checkNullAndGetOffset(int fieldId) {
+ int offset = getFieldOffset(fieldId);
+ if (offset == OffHeapRowBlock.NULL_FIELD_OFFSET) {
+ throw new RuntimeException("Invalid Field Access: " + fieldId);
+ }
+ return offset;
+ }
+
+ @Override
+ public boolean contains(int fieldid) {
+ return getFieldOffset(fieldid) > OffHeapRowBlock.NULL_FIELD_OFFSET;
+ }
+
+ @Override
+ public boolean isNull(int fieldid) {
+ return getFieldOffset(fieldid) == OffHeapRowBlock.NULL_FIELD_OFFSET;
+ }
+
+ @Override
+ public boolean isNotNull(int fieldid) {
+ return getFieldOffset(fieldid) > OffHeapRowBlock.NULL_FIELD_OFFSET;
+ }
+
+ @Override
+ public void clear() {
+ // nothing to do
+ }
+
+ @Override
+ public void put(int fieldId, Datum value) {
+ throw new UnsupportedException("UnSafeTuple does not support put(int, Datum).");
+ }
+
+ @Override
+ public void put(int fieldId, Datum[] values) {
+ throw new UnsupportedException("UnSafeTuple does not support put(int, Datum []).");
+ }
+
+ @Override
+ public void put(int fieldId, Tuple tuple) {
+ throw new UnsupportedException("UnSafeTuple does not support put(int, Tuple).");
+ }
+
+ @Override
+ public void put(Datum[] values) {
+ throw new UnsupportedException("UnSafeTuple does not support put(Datum []).");
+ }
+
+ @Override
+ public Datum get(int fieldId) {
+ if (isNull(fieldId)) {
+ return NullDatum.get();
+ }
+
+ switch (types[fieldId].getType()) {
+ case BOOLEAN:
+ return DatumFactory.createBool(getBool(fieldId));
+ case INT1:
+ case INT2:
+ return DatumFactory.createInt2(getInt2(fieldId));
+ case INT4:
+ return DatumFactory.createInt4(getInt4(fieldId));
+ case INT8:
+ return DatumFactory.createInt8(getInt4(fieldId));
+ case FLOAT4:
+ return DatumFactory.createFloat4(getFloat4(fieldId));
+ case FLOAT8:
+ return DatumFactory.createFloat8(getFloat8(fieldId));
+ case TEXT:
+ return DatumFactory.createText(getText(fieldId));
+ case TIMESTAMP:
+ return DatumFactory.createTimestamp(getInt8(fieldId));
+ case DATE:
+ return DatumFactory.createDate(getInt4(fieldId));
+ case TIME:
+ return DatumFactory.createTime(getInt8(fieldId));
+ case INTERVAL:
+ return getInterval(fieldId);
+ case INET4:
+ return DatumFactory.createInet4(getInt4(fieldId));
+ case PROTOBUF:
+ return getProtobufDatum(fieldId);
+ default:
+ throw new UnsupportedException("Unknown type: " + types[fieldId]);
+ }
+ }
+
+ @Override
+ public void setOffset(long offset) {
+ }
+
+ @Override
+ public long getOffset() {
+ return 0;
+ }
+
+ @Override
+ public boolean getBool(int fieldId) {
+ return UNSAFE.getByte(data, BASE_OFFSET + checkNullAndGetOffset(fieldId)) == 0x01;
+ }
+
+ @Override
+ public byte getByte(int fieldId) {
+ return UNSAFE.getByte(data, BASE_OFFSET + checkNullAndGetOffset(fieldId));
+ }
+
+ @Override
+ public char getChar(int fieldId) {
+ return UNSAFE.getChar(data, BASE_OFFSET + checkNullAndGetOffset(fieldId));
+ }
+
+ @Override
+ public byte[] getBytes(int fieldId) {
+ long pos = checkNullAndGetOffset(fieldId);
+ int len = UNSAFE.getInt(data, BASE_OFFSET + pos);
+ pos += SizeOf.SIZE_OF_INT;
+
+ byte [] bytes = new byte[len];
+ UNSAFE.copyMemory(data, BASE_OFFSET + pos, bytes, UnsafeUtil.ARRAY_BYTE_BASE_OFFSET, len);
+ return bytes;
+ }
+
+ @Override
+ public short getInt2(int fieldId) {
+ return UNSAFE.getShort(data, BASE_OFFSET + checkNullAndGetOffset(fieldId));
+ }
+
+ @Override
+ public int getInt4(int fieldId) {
+ return UNSAFE.getInt(data, BASE_OFFSET + checkNullAndGetOffset(fieldId));
+ }
+
+ @Override
+ public long getInt8(int fieldId) {
+ return UNSAFE.getLong(data, BASE_OFFSET + checkNullAndGetOffset(fieldId));
+ }
+
+ @Override
+ public float getFloat4(int fieldId) {
+ return UNSAFE.getFloat(data, BASE_OFFSET + checkNullAndGetOffset(fieldId));
+ }
+
+ @Override
+ public double getFloat8(int fieldId) {
+ return UNSAFE.getDouble(data, BASE_OFFSET + checkNullAndGetOffset(fieldId));
+ }
+
+ @Override
+ public String getText(int fieldId) {
+ return new String(getBytes(fieldId));
+ }
+
+ public IntervalDatum getInterval(int fieldId) {
+ long pos = checkNullAndGetOffset(fieldId);
+ int months = UNSAFE.getInt(data, BASE_OFFSET + pos);
+ pos += SizeOf.SIZE_OF_INT;
+ long millisecs = UNSAFE.getLong(data, BASE_OFFSET + pos);
+ return new IntervalDatum(months, millisecs);
+ }
+
+ @Override
+ public Datum getProtobufDatum(int fieldId) {
+ byte [] bytes = getBytes(fieldId);
+
+ ProtobufDatumFactory factory = ProtobufDatumFactory.get(types[fieldId].getCode());
+ Message.Builder builder = factory.newBuilder();
+ try {
+ builder.mergeFrom(bytes);
+ } catch (InvalidProtocolBufferException e) {
+ return NullDatum.get();
+ }
+
+ return new ProtobufDatum(builder.build());
+ }
+
+ @Override
+ public char[] getUnicodeChars(int fieldId) {
+ long pos = checkNullAndGetOffset(fieldId);
+ int len = UNSAFE.getInt(data, BASE_OFFSET + pos);
+ pos += SizeOf.SIZE_OF_INT;
+
+ byte [] bytes = new byte[len];
+ UNSAFE.copyMemory(data, BASE_OFFSET + pos, bytes, UnsafeUtil.ARRAY_BYTE_BASE_OFFSET, len);
+ return Convert.utf2chars(bytes);
+ }
+
+ @Override
+ public Tuple clone() throws CloneNotSupportedException {
+ return this;
+ }
+
+ @Override
+ public Datum[] getValues() {
+ Datum [] datums = new Datum[size()];
+ for (int i = 0; i < size(); i++) {
+ if (contains(i)) {
+ datums[i] = get(i);
+ } else {
+ datums[i] = NullDatum.get();
+ }
+ }
+ return datums;
+ }
+
+ @Override
+ public String toString() {
+ return VTuple.toDisplayString(getValues());
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapMemory.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapMemory.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapMemory.java
new file mode 100644
index 0000000..2f8e349
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapMemory.java
@@ -0,0 +1,102 @@
+/**
+ * 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.offheap;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.util.Deallocatable;
+import org.apache.tajo.util.FileUtil;
+import org.apache.tajo.util.UnsafeUtil;
+import sun.misc.Unsafe;
+import sun.nio.ch.DirectBuffer;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+public class OffHeapMemory implements Deallocatable {
+ private static final Log LOG = LogFactory.getLog(OffHeapMemory.class);
+
+ protected static final Unsafe UNSAFE = UnsafeUtil.unsafe;
+
+ protected ByteBuffer buffer;
+ protected int memorySize;
+ protected ResizableLimitSpec limitSpec;
+ protected long address;
+
+ @VisibleForTesting
+ protected OffHeapMemory(ByteBuffer buffer, ResizableLimitSpec limitSpec) {
+ this.buffer = buffer;
+ this.address = ((DirectBuffer) buffer).address();
+ this.memorySize = buffer.limit();
+ this.limitSpec = limitSpec;
+ }
+
+ public OffHeapMemory(ResizableLimitSpec limitSpec) {
+ this(ByteBuffer.allocateDirect((int) limitSpec.initialSize()).order(ByteOrder.nativeOrder()), limitSpec);
+ }
+
+ public long address() {
+ return address;
+ }
+
+ public long size() {
+ return memorySize;
+ }
+
+ public void resize(int newSize) {
+ Preconditions.checkArgument(newSize > 0, "Size must be greater than 0 bytes");
+
+ if (newSize > limitSpec.limit()) {
+ throw new RuntimeException("Resize cannot exceed the size limit");
+ }
+
+ if (newSize < memorySize) {
+ LOG.warn("The size reduction is ignored.");
+ }
+
+ int newBlockSize = UnsafeUtil.alignedSize(newSize);
+ ByteBuffer newByteBuf = ByteBuffer.allocateDirect(newBlockSize);
+ long newAddress = ((DirectBuffer)newByteBuf).address();
+
+ UNSAFE.copyMemory(this.address, newAddress, memorySize);
+
+ UnsafeUtil.free(buffer);
+ this.memorySize = newSize;
+ this.buffer = newByteBuf;
+ this.address = newAddress;
+ }
+
+ public java.nio.Buffer nioBuffer() {
+ return (ByteBuffer) buffer.position(0).limit(memorySize);
+ }
+
+ @Override
+ public void release() {
+ UnsafeUtil.free(this.buffer);
+ this.buffer = null;
+ this.address = 0;
+ this.memorySize = 0;
+ }
+
+ public String toString() {
+ return "memory=" + FileUtil.humanReadableByteCount(memorySize, false) + "," + limitSpec;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlock.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlock.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlock.java
new file mode 100644
index 0000000..689efb7
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlock.java
@@ -0,0 +1,176 @@
+/***
+ * 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.offheap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SchemaUtil;
+import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.util.Deallocatable;
+import org.apache.tajo.util.FileUtil;
+import org.apache.tajo.util.SizeOf;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+
+import static org.apache.tajo.common.TajoDataTypes.DataType;
+
+public class OffHeapRowBlock extends OffHeapMemory implements Deallocatable {
+ private static final Log LOG = LogFactory.getLog(OffHeapRowBlock.class);
+
+ public static final int NULL_FIELD_OFFSET = -1;
+
+ DataType [] dataTypes;
+
+ // Basic States
+ private int maxRowNum = Integer.MAX_VALUE; // optional
+ private int rowNum;
+ protected int position = 0;
+
+ private OffHeapRowBlockWriter builder;
+
+ private OffHeapRowBlock(ByteBuffer buffer, Schema schema, ResizableLimitSpec limitSpec) {
+ super(buffer, limitSpec);
+ initialize(schema);
+ }
+
+ public OffHeapRowBlock(Schema schema, ResizableLimitSpec limitSpec) {
+ super(limitSpec);
+ initialize(schema);
+ }
+
+ private void initialize(Schema schema) {
+ dataTypes = SchemaUtil.toDataTypes(schema);
+
+ this.builder = new OffHeapRowBlockWriter(this);
+ }
+
+ @VisibleForTesting
+ public OffHeapRowBlock(Schema schema, int bytes) {
+ this(schema, new ResizableLimitSpec(bytes));
+ }
+
+ @VisibleForTesting
+ public OffHeapRowBlock(Schema schema, ByteBuffer buffer) {
+ this(buffer, schema, ResizableLimitSpec.DEFAULT_LIMIT);
+ }
+
+ public void position(int pos) {
+ this.position = pos;
+ }
+
+ public void clear() {
+ this.position = 0;
+ this.rowNum = 0;
+
+ builder.clear();
+ }
+
+ @Override
+ public ByteBuffer nioBuffer() {
+ return (ByteBuffer) buffer.position(0).limit(position);
+ }
+
+ public int position() {
+ return position;
+ }
+
+ public long usedMem() {
+ return position;
+ }
+
+ /**
+ * Ensure that this buffer has enough remaining space to add the size.
+ * Creates and copies to a new buffer if necessary
+ *
+ * @param size Size to add
+ */
+ public void ensureSize(int size) {
+ if (remain() - size < 0) {
+ if (!limitSpec.canIncrease(memorySize)) {
+ throw new RuntimeException("Cannot increase RowBlock anymore.");
+ }
+
+ int newBlockSize = limitSpec.increasedSize(memorySize);
+ resize(newBlockSize);
+ LOG.info("Increase DirectRowBlock to " + FileUtil.humanReadableByteCount(newBlockSize, false));
+ }
+ }
+
+ public long remain() {
+ return memorySize - position - builder.offset();
+ }
+
+ public int maxRowNum() {
+ return maxRowNum;
+ }
+ public int rows() {
+ return rowNum;
+ }
+
+ public void setRows(int rowNum) {
+ this.rowNum = rowNum;
+ }
+
+ public boolean copyFromChannel(FileChannel channel, TableStats stats) throws IOException {
+ if (channel.position() < channel.size()) {
+ clear();
+
+ buffer.clear();
+ channel.read(buffer);
+ memorySize = buffer.position();
+
+ while (position < memorySize) {
+ long recordPtr = address + position;
+
+ if (remain() < SizeOf.SIZE_OF_INT) {
+ channel.position(channel.position() - remain());
+ memorySize = (int) (memorySize - remain());
+ return true;
+ }
+
+ int recordSize = UNSAFE.getInt(recordPtr);
+
+ if (remain() < recordSize) {
+ channel.position(channel.position() - remain());
+ memorySize = (int) (memorySize - remain());
+ return true;
+ }
+
+ position += recordSize;
+ rowNum++;
+ }
+
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ public RowWriter getWriter() {
+ return builder;
+ }
+
+ public OffHeapRowBlockReader getReader() {
+ return new OffHeapRowBlockReader(this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockReader.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockReader.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockReader.java
new file mode 100644
index 0000000..4a9313f
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockReader.java
@@ -0,0 +1,63 @@
+/***
+ * 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.offheap;
+
+import org.apache.tajo.tuple.RowBlockReader;
+import org.apache.tajo.util.UnsafeUtil;
+import sun.misc.Unsafe;
+
+public class OffHeapRowBlockReader implements RowBlockReader<ZeroCopyTuple> {
+ private static final Unsafe UNSAFE = UnsafeUtil.unsafe;
+ OffHeapRowBlock rowBlock;
+
+ // Read States
+ private int curRowIdxForRead;
+ private int curPosForRead;
+
+ public OffHeapRowBlockReader(OffHeapRowBlock rowBlock) {
+ this.rowBlock = rowBlock;
+ }
+
+ public long remainForRead() {
+ return rowBlock.memorySize - curPosForRead;
+ }
+
+ @Override
+ public boolean next(ZeroCopyTuple tuple) {
+ if (curRowIdxForRead < rowBlock.rows()) {
+
+ long recordStartPtr = rowBlock.address() + curPosForRead;
+ int recordLen = UNSAFE.getInt(recordStartPtr);
+ tuple.set(rowBlock.buffer, curPosForRead, recordLen, rowBlock.dataTypes);
+
+ curPosForRead += recordLen;
+ curRowIdxForRead++;
+
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public void reset() {
+ curPosForRead = 0;
+ curRowIdxForRead = 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockUtils.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockUtils.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockUtils.java
new file mode 100644
index 0000000..dbc3188
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockUtils.java
@@ -0,0 +1,54 @@
+/*
+ * 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.offheap;
+
+import com.google.common.collect.Lists;
+import org.apache.tajo.storage.Tuple;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+
+public class OffHeapRowBlockUtils {
+
+ public static List<Tuple> sort(OffHeapRowBlock rowBlock, Comparator<Tuple> comparator) {
+ List<Tuple> tupleList = Lists.newArrayList();
+ ZeroCopyTuple zcTuple = new ZeroCopyTuple();
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(rowBlock);
+ while(reader.next(zcTuple)) {
+ tupleList.add(zcTuple);
+ zcTuple = new ZeroCopyTuple();
+ }
+ Collections.sort(tupleList, comparator);
+ return tupleList;
+ }
+
+ public static Tuple[] sortToArray(OffHeapRowBlock rowBlock, Comparator<Tuple> comparator) {
+ Tuple[] tuples = new Tuple[rowBlock.rows()];
+ ZeroCopyTuple zcTuple = new ZeroCopyTuple();
+ OffHeapRowBlockReader reader = new OffHeapRowBlockReader(rowBlock);
+ for (int i = 0; i < rowBlock.rows() && reader.next(zcTuple); i++) {
+ tuples[i] = zcTuple;
+ zcTuple = new ZeroCopyTuple();
+ }
+ Arrays.sort(tuples, comparator);
+ return tuples;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockWriter.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockWriter.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockWriter.java
new file mode 100644
index 0000000..d177e0c
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowBlockWriter.java
@@ -0,0 +1,58 @@
+/**
+ * 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.offheap;
+
+import org.apache.tajo.common.TajoDataTypes;
+
+public class OffHeapRowBlockWriter extends OffHeapRowWriter {
+ OffHeapRowBlock rowBlock;
+
+ OffHeapRowBlockWriter(OffHeapRowBlock rowBlock) {
+ super(rowBlock.dataTypes);
+ this.rowBlock = rowBlock;
+ }
+
+ public long address() {
+ return rowBlock.address();
+ }
+
+ public int position() {
+ return rowBlock.position();
+ }
+
+ @Override
+ public void forward(int length) {
+ rowBlock.position(position() + length);
+ }
+
+ public void ensureSize(int size) {
+ rowBlock.ensureSize(size);
+ }
+
+ @Override
+ public void endRow() {
+ super.endRow();
+ rowBlock.setRows(rowBlock.rows() + 1);
+ }
+
+ @Override
+ public TajoDataTypes.DataType[] dataTypes() {
+ return rowBlock.dataTypes;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowWriter.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowWriter.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowWriter.java
new file mode 100644
index 0000000..85c7e0b
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/OffHeapRowWriter.java
@@ -0,0 +1,232 @@
+/**
+ * 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.offheap;
+
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.datum.IntervalDatum;
+import org.apache.tajo.datum.ProtobufDatum;
+import org.apache.tajo.datum.TextDatum;
+import org.apache.tajo.util.SizeOf;
+import org.apache.tajo.util.UnsafeUtil;
+
+/**
+ *
+ * Row Record Structure
+ *
+ * | row length (4 bytes) | field 1 offset | field 2 offset | ... | field N offset| field 1 | field 2| ... | field N |
+ * 4 bytes 4 bytes 4 bytes
+ *
+ */
+public abstract class OffHeapRowWriter implements RowWriter {
+ /** record size + offset list */
+ private final int headerSize;
+ /** field offsets */
+ private final int [] fieldOffsets;
+ private final TajoDataTypes.DataType [] dataTypes;
+
+ private int curFieldIdx;
+ private int curOffset;
+
+ public OffHeapRowWriter(final TajoDataTypes.DataType [] dataTypes) {
+ this.dataTypes = dataTypes;
+ fieldOffsets = new int[dataTypes.length];
+ headerSize = SizeOf.SIZE_OF_INT * (dataTypes.length + 1);
+ }
+
+ public void clear() {
+ curOffset = 0;
+ curFieldIdx = 0;
+ }
+
+ public long recordStartAddr() {
+ return address() + position();
+ }
+
+ public abstract long address();
+
+ public abstract void ensureSize(int size);
+
+ public int offset() {
+ return curOffset;
+ }
+
+ /**
+ * Current position
+ *
+ * @return The position
+ */
+ public abstract int position();
+
+ /**
+ * Forward the address;
+ *
+ * @param length Length to be forwarded
+ */
+ public abstract void forward(int length);
+
+ @Override
+ public TajoDataTypes.DataType[] dataTypes() {
+ return dataTypes;
+ }
+
+ public boolean startRow() {
+ curOffset = headerSize;
+ curFieldIdx = 0;
+ return true;
+ }
+
+ public void endRow() {
+ long rowHeaderPos = address() + position();
+ OffHeapMemory.UNSAFE.putInt(rowHeaderPos, curOffset);
+ rowHeaderPos += SizeOf.SIZE_OF_INT;
+
+ for (int i = 0; i < curFieldIdx; i++) {
+ OffHeapMemory.UNSAFE.putInt(rowHeaderPos, fieldOffsets[i]);
+ rowHeaderPos += SizeOf.SIZE_OF_INT;
+ }
+ for (int i = curFieldIdx; i < dataTypes.length; i++) {
+ OffHeapMemory.UNSAFE.putInt(rowHeaderPos, OffHeapRowBlock.NULL_FIELD_OFFSET);
+ rowHeaderPos += SizeOf.SIZE_OF_INT;
+ }
+
+ // rowOffset is equivalent to a byte length of this row.
+ forward(curOffset);
+ }
+
+ public void skipField() {
+ fieldOffsets[curFieldIdx++] = OffHeapRowBlock.NULL_FIELD_OFFSET;
+ }
+
+ private void forwardField() {
+ fieldOffsets[curFieldIdx++] = curOffset;
+ }
+
+ public void putBool(boolean val) {
+ ensureSize(SizeOf.SIZE_OF_BOOL);
+ forwardField();
+
+ OffHeapMemory.UNSAFE.putByte(recordStartAddr() + curOffset, (byte) (val ? 0x01 : 0x00));
+
+ curOffset += SizeOf.SIZE_OF_BOOL;
+ }
+
+ public void putInt2(short val) {
+ ensureSize(SizeOf.SIZE_OF_SHORT);
+ forwardField();
+
+ OffHeapMemory.UNSAFE.putShort(recordStartAddr() + curOffset, val);
+ curOffset += SizeOf.SIZE_OF_SHORT;
+ }
+
+ public void putInt4(int val) {
+ ensureSize(SizeOf.SIZE_OF_INT);
+ forwardField();
+
+ OffHeapMemory.UNSAFE.putInt(recordStartAddr() + curOffset, val);
+ curOffset += SizeOf.SIZE_OF_INT;
+ }
+
+ public void putInt8(long val) {
+ ensureSize(SizeOf.SIZE_OF_LONG);
+ forwardField();
+
+ OffHeapMemory.UNSAFE.putLong(recordStartAddr() + curOffset, val);
+ curOffset += SizeOf.SIZE_OF_LONG;
+ }
+
+ public void putFloat4(float val) {
+ ensureSize(SizeOf.SIZE_OF_FLOAT);
+ forwardField();
+
+ OffHeapMemory.UNSAFE.putFloat(recordStartAddr() + curOffset, val);
+ curOffset += SizeOf.SIZE_OF_FLOAT;
+ }
+
+ public void putFloat8(double val) {
+ ensureSize(SizeOf.SIZE_OF_DOUBLE);
+ forwardField();
+
+ OffHeapMemory.UNSAFE.putDouble(recordStartAddr() + curOffset, val);
+ curOffset += SizeOf.SIZE_OF_DOUBLE;
+ }
+
+ public void putText(String val) {
+ byte[] bytes = val.getBytes(TextDatum.DEFAULT_CHARSET);
+ putText(bytes);
+ }
+
+ public void putText(byte[] val) {
+ int bytesLen = val.length;
+
+ ensureSize(SizeOf.SIZE_OF_INT + bytesLen);
+ forwardField();
+
+ OffHeapMemory.UNSAFE.putInt(recordStartAddr() + curOffset, bytesLen);
+ curOffset += SizeOf.SIZE_OF_INT;
+
+ OffHeapMemory.UNSAFE.copyMemory(val, UnsafeUtil.ARRAY_BYTE_BASE_OFFSET, null,
+ recordStartAddr() + curOffset, bytesLen);
+ curOffset += bytesLen;
+ }
+
+ public void putBlob(byte[] val) {
+ int bytesLen = val.length;
+
+ ensureSize(SizeOf.SIZE_OF_INT + bytesLen);
+ forwardField();
+
+ OffHeapMemory.UNSAFE.putInt(recordStartAddr() + curOffset, bytesLen);
+ curOffset += SizeOf.SIZE_OF_INT;
+
+ OffHeapMemory.UNSAFE.copyMemory(val, UnsafeUtil.ARRAY_BYTE_BASE_OFFSET, null,
+ recordStartAddr() + curOffset, bytesLen);
+ curOffset += bytesLen;
+ }
+
+ public void putTimestamp(long val) {
+ putInt8(val);
+ }
+
+ public void putDate(int val) {
+ putInt4(val);
+ }
+
+ public void putTime(long val) {
+ putInt8(val);
+ }
+
+ public void putInterval(IntervalDatum val) {
+ ensureSize(SizeOf.SIZE_OF_INT + SizeOf.SIZE_OF_LONG);
+ forwardField();
+
+ long offset = recordStartAddr() + curOffset;
+ OffHeapMemory.UNSAFE.putInt(offset, val.getMonths());
+ offset += SizeOf.SIZE_OF_INT;
+ OffHeapMemory.UNSAFE.putLong(offset, val.getMilliSeconds());
+ curOffset += SizeOf.SIZE_OF_INT + SizeOf.SIZE_OF_LONG;
+ }
+
+ public void putInet4(int val) {
+ putInt4(val);
+ }
+
+ public void putProtoDatum(ProtobufDatum val) {
+ putBlob(val.asByteArray());
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/ResizableLimitSpec.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/ResizableLimitSpec.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/ResizableLimitSpec.java
new file mode 100644
index 0000000..14e67b2
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/ResizableLimitSpec.java
@@ -0,0 +1,142 @@
+/*
+ * 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.offheap;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.util.FileUtil;
+
+/**
+ * It specifies the maximum size or increasing ratio. In addition,
+ * it guarantees that all numbers are less than or equal to Integer.MAX_VALUE 2^31
+ * due to ByteBuffer.
+ */
+public class ResizableLimitSpec {
+ private 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);
+
+ private final long initSize;
+ private final long limitBytes;
+ private final float incRatio;
+ private final float allowedOVerflowRatio;
+ private final static float DEFAULT_ALLOWED_OVERFLOW_RATIO = 0.1f;
+ private final static float DEFAULT_INCREASE_RATIO = 1.0f;
+
+ public ResizableLimitSpec(long initSize) {
+ this(initSize, MAX_SIZE_BYTES, DEFAULT_ALLOWED_OVERFLOW_RATIO);
+ }
+
+ public ResizableLimitSpec(long initSize, long limitBytes) {
+ this(initSize, limitBytes, DEFAULT_ALLOWED_OVERFLOW_RATIO);
+ }
+
+ public ResizableLimitSpec(long initSize, long limitBytes, float allowedOverflow) {
+ this(initSize, limitBytes, allowedOverflow, DEFAULT_INCREASE_RATIO);
+ }
+
+ public ResizableLimitSpec(long initSize, long limitBytes, float allowedOverflowRatio, float incRatio) {
+ Preconditions.checkArgument(initSize > 0, "initial size must be greater than 0 bytes.");
+ Preconditions.checkArgument(initSize <= MAX_SIZE_BYTES, "The maximum initial size is 2GB.");
+ Preconditions.checkArgument(limitBytes > 0, "The limit size must be greater than 0 bytes.");
+ Preconditions.checkArgument(limitBytes <= MAX_SIZE_BYTES, "The maximum limit size is 2GB.");
+ Preconditions.checkArgument(incRatio > 0.0f, "Increase Ratio must be greater than 0.");
+
+ if (initSize == limitBytes) {
+ long overflowedSize = (long) (initSize + (initSize * allowedOverflowRatio));
+
+ if (overflowedSize > Integer.MAX_VALUE) {
+ overflowedSize = Integer.MAX_VALUE;
+ }
+
+ this.initSize = overflowedSize;
+ this.limitBytes = overflowedSize;
+ } else {
+ this.initSize = initSize;
+ limitBytes = (long) (limitBytes + (limitBytes * allowedOverflowRatio));
+
+ if (limitBytes > Integer.MAX_VALUE) {
+ this.limitBytes = Integer.MAX_VALUE;
+ } else {
+ this.limitBytes = limitBytes;
+ }
+ }
+
+ this.allowedOVerflowRatio = allowedOverflowRatio;
+ this.incRatio = incRatio;
+ }
+
+ public long initialSize() {
+ return initSize;
+ }
+
+ public long limit() {
+ return limitBytes;
+ }
+
+ public float remainRatio(long currentSize) {
+ Preconditions.checkArgument(currentSize > 0, "Size must be greater than 0 bytes.");
+ if (currentSize > Integer.MAX_VALUE) {
+ currentSize = Integer.MAX_VALUE;
+ }
+ return (float)currentSize / (float)limitBytes;
+ }
+
+ public boolean canIncrease(long currentSize) {
+ return remain(currentSize) > 0;
+ }
+
+ public long remain(long currentSize) {
+ Preconditions.checkArgument(currentSize > 0, "Size must be greater than 0 bytes.");
+ return limitBytes > Integer.MAX_VALUE ? Integer.MAX_VALUE - currentSize : limitBytes - currentSize;
+ }
+
+ public int increasedSize(int currentSize) {
+ if (currentSize < initSize) {
+ 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;
+ }
+ long nextSize = (long) (currentSize + ((float) currentSize * incRatio));
+
+ if (nextSize > limitBytes) {
+ LOG.info("Increasing reaches size limit (" + FileUtil.humanReadableByteCount(limitBytes, false) + ")");
+ nextSize = limitBytes;
+ }
+
+ if (nextSize > Integer.MAX_VALUE) {
+ LOG.info("Increasing reaches maximum size (" + FileUtil.humanReadableByteCount(Integer.MAX_VALUE, false) + ")");
+ nextSize = Integer.MAX_VALUE;
+ }
+
+ return (int) nextSize;
+ }
+
+ @Override
+ public String toString() {
+ return "init=" + FileUtil.humanReadableByteCount(initSize, false) + ",limit="
+ + FileUtil.humanReadableByteCount(limitBytes, false) + ",overflow_ratio=" + allowedOVerflowRatio
+ + ",inc_ratio=" + incRatio;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/RowWriter.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/RowWriter.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/RowWriter.java
new file mode 100644
index 0000000..a2b2561
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/RowWriter.java
@@ -0,0 +1,73 @@
+/***
+ * 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.offheap;
+
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.datum.IntervalDatum;
+import org.apache.tajo.datum.ProtobufDatum;
+
+/**
+ * The call sequence should be as follows:
+ *
+ * <pre>
+ * startRow() --> skipField() or putXXX --> endRow()
+ * </pre>
+ *
+ * The total number of skipField and putXXX invocations must be equivalent to the number of fields.
+ */
+public interface RowWriter {
+
+ public TajoDataTypes.DataType [] dataTypes();
+
+ public boolean startRow();
+
+ public void endRow();
+
+ public void skipField();
+
+ public void putBool(boolean val);
+
+ public void putInt2(short val);
+
+ public void putInt4(int val);
+
+ public void putInt8(long val);
+
+ public void putFloat4(float val);
+
+ public void putFloat8(double val);
+
+ public void putText(String val);
+
+ public void putText(byte[] val);
+
+ public void putBlob(byte[] val);
+
+ public void putTimestamp(long val);
+
+ public void putTime(long val);
+
+ public void putDate(int val);
+
+ public void putInterval(IntervalDatum val);
+
+ public void putInet4(int val);
+
+ public void putProtoDatum(ProtobufDatum datum);
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/UnSafeTuple.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/UnSafeTuple.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/UnSafeTuple.java
new file mode 100644
index 0000000..138386f
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/UnSafeTuple.java
@@ -0,0 +1,308 @@
+/***
+ * 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.offheap;
+
+import com.google.common.base.Preconditions;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import com.sun.tools.javac.util.Convert;
+import org.apache.tajo.datum.*;
+import org.apache.tajo.exception.UnsupportedException;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.VTuple;
+import org.apache.tajo.util.SizeOf;
+import org.apache.tajo.util.UnsafeUtil;
+import sun.misc.Unsafe;
+import sun.nio.ch.DirectBuffer;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.tajo.common.TajoDataTypes.DataType;
+
+public abstract class UnSafeTuple implements Tuple {
+ private static final Unsafe UNSAFE = UnsafeUtil.unsafe;
+
+ private DirectBuffer bb;
+ private int relativePos;
+ private int length;
+ private DataType [] types;
+
+ protected void set(ByteBuffer bb, int relativePos, int length, DataType [] types) {
+ this.bb = (DirectBuffer) bb;
+ this.relativePos = relativePos;
+ this.length = length;
+ this.types = types;
+ }
+
+ void set(ByteBuffer bb, DataType [] types) {
+ set(bb, 0, bb.limit(), types);
+ }
+
+ @Override
+ public int size() {
+ return types.length;
+ }
+
+ public ByteBuffer nioBuffer() {
+ return ((ByteBuffer)((ByteBuffer)bb).duplicate().position(relativePos).limit(relativePos + length)).slice();
+ }
+
+ public HeapTuple toHeapTuple() {
+ byte [] bytes = new byte[length];
+ UNSAFE.copyMemory(null, bb.address() + relativePos, bytes, UnsafeUtil.ARRAY_BYTE_BASE_OFFSET, length);
+ return new HeapTuple(bytes, types);
+ }
+
+ public void copyFrom(UnSafeTuple tuple) {
+ Preconditions.checkNotNull(tuple);
+
+ ((ByteBuffer) bb).clear();
+ if (length < tuple.length) {
+ UnsafeUtil.free((ByteBuffer) bb);
+ bb = (DirectBuffer) ByteBuffer.allocateDirect(tuple.length).order(ByteOrder.nativeOrder());
+ this.relativePos = 0;
+ this.length = tuple.length;
+ }
+
+ ((ByteBuffer) bb).put(tuple.nioBuffer());
+ }
+
+ private int getFieldOffset(int fieldId) {
+ return UNSAFE.getInt(bb.address() + relativePos + SizeOf.SIZE_OF_INT + (fieldId * SizeOf.SIZE_OF_INT));
+ }
+
+ public long getFieldAddr(int fieldId) {
+ int fieldOffset = getFieldOffset(fieldId);
+ if (fieldOffset == -1) {
+ throw new RuntimeException("Invalid Field Access: " + fieldId);
+ }
+ return bb.address() + relativePos + fieldOffset;
+ }
+
+ @Override
+ public boolean contains(int fieldid) {
+ return getFieldOffset(fieldid) > OffHeapRowBlock.NULL_FIELD_OFFSET;
+ }
+
+ @Override
+ public boolean isNull(int fieldid) {
+ return getFieldOffset(fieldid) == OffHeapRowBlock.NULL_FIELD_OFFSET;
+ }
+
+ @Override
+ public boolean isNotNull(int fieldid) {
+ return getFieldOffset(fieldid) > OffHeapRowBlock.NULL_FIELD_OFFSET;
+ }
+
+ @Override
+ public void clear() {
+ // nothing to do
+ }
+
+ @Override
+ public void put(int fieldId, Datum value) {
+ throw new UnsupportedException("UnSafeTuple does not support put(int, Datum).");
+ }
+
+ @Override
+ public void put(int fieldId, Datum[] values) {
+ throw new UnsupportedException("UnSafeTuple does not support put(int, Datum []).");
+ }
+
+ @Override
+ public void put(int fieldId, Tuple tuple) {
+ throw new UnsupportedException("UnSafeTuple does not support put(int, Tuple).");
+ }
+
+ @Override
+ public void put(Datum[] values) {
+ throw new UnsupportedException("UnSafeTuple does not support put(Datum []).");
+ }
+
+ @Override
+ public Datum get(int fieldId) {
+ if (isNull(fieldId)) {
+ return NullDatum.get();
+ }
+
+ switch (types[fieldId].getType()) {
+ case BOOLEAN:
+ return DatumFactory.createBool(getBool(fieldId));
+ case INT1:
+ case INT2:
+ return DatumFactory.createInt2(getInt2(fieldId));
+ case INT4:
+ return DatumFactory.createInt4(getInt4(fieldId));
+ case INT8:
+ return DatumFactory.createInt8(getInt4(fieldId));
+ case FLOAT4:
+ return DatumFactory.createFloat4(getFloat4(fieldId));
+ case FLOAT8:
+ return DatumFactory.createFloat8(getFloat8(fieldId));
+ case TEXT:
+ return DatumFactory.createText(getText(fieldId));
+ case TIMESTAMP:
+ return DatumFactory.createTimestamp(getInt8(fieldId));
+ case DATE:
+ return DatumFactory.createDate(getInt4(fieldId));
+ case TIME:
+ return DatumFactory.createTime(getInt8(fieldId));
+ case INTERVAL:
+ return getInterval(fieldId);
+ case INET4:
+ return DatumFactory.createInet4(getInt4(fieldId));
+ case PROTOBUF:
+ return getProtobufDatum(fieldId);
+ default:
+ throw new UnsupportedException("Unknown type: " + types[fieldId]);
+ }
+ }
+
+ @Override
+ public void setOffset(long offset) {
+ }
+
+ @Override
+ public long getOffset() {
+ return 0;
+ }
+
+ @Override
+ public boolean getBool(int fieldId) {
+ return UNSAFE.getByte(getFieldAddr(fieldId)) == 0x01;
+ }
+
+ @Override
+ public byte getByte(int fieldId) {
+ return UNSAFE.getByte(getFieldAddr(fieldId));
+ }
+
+ @Override
+ public char getChar(int fieldId) {
+ return UNSAFE.getChar(getFieldAddr(fieldId));
+ }
+
+ @Override
+ public byte[] getBytes(int fieldId) {
+ long pos = getFieldAddr(fieldId);
+ int len = UNSAFE.getInt(pos);
+ pos += SizeOf.SIZE_OF_INT;
+
+ byte [] bytes = new byte[len];
+ UNSAFE.copyMemory(null, pos, bytes, UnsafeUtil.ARRAY_BYTE_BASE_OFFSET, len);
+ return bytes;
+ }
+
+ @Override
+ public short getInt2(int fieldId) {
+ long addr = getFieldAddr(fieldId);
+ return UNSAFE.getShort(addr);
+ }
+
+ @Override
+ public int getInt4(int fieldId) {
+ return UNSAFE.getInt(getFieldAddr(fieldId));
+ }
+
+ @Override
+ public long getInt8(int fieldId) {
+ return UNSAFE.getLong(getFieldAddr(fieldId));
+ }
+
+ @Override
+ public float getFloat4(int fieldId) {
+ return UNSAFE.getFloat(getFieldAddr(fieldId));
+ }
+
+ @Override
+ public double getFloat8(int fieldId) {
+ return UNSAFE.getDouble(getFieldAddr(fieldId));
+ }
+
+ @Override
+ public String getText(int fieldId) {
+ long pos = getFieldAddr(fieldId);
+ int len = UNSAFE.getInt(pos);
+ pos += SizeOf.SIZE_OF_INT;
+
+ byte [] bytes = new byte[len];
+ UNSAFE.copyMemory(null, pos, bytes, UnsafeUtil.ARRAY_BYTE_BASE_OFFSET, len);
+ return new String(bytes);
+ }
+
+ public IntervalDatum getInterval(int fieldId) {
+ long pos = getFieldAddr(fieldId);
+ int months = UNSAFE.getInt(pos);
+ pos += SizeOf.SIZE_OF_INT;
+ long millisecs = UNSAFE.getLong(pos);
+ return new IntervalDatum(months, millisecs);
+ }
+
+ @Override
+ public Datum getProtobufDatum(int fieldId) {
+ byte [] bytes = getBytes(fieldId);
+
+ ProtobufDatumFactory factory = ProtobufDatumFactory.get(types[fieldId].getCode());
+ Message.Builder builder = factory.newBuilder();
+ try {
+ builder.mergeFrom(bytes);
+ } catch (InvalidProtocolBufferException e) {
+ return NullDatum.get();
+ }
+
+ return new ProtobufDatum(builder.build());
+ }
+
+ @Override
+ public char[] getUnicodeChars(int fieldId) {
+ long pos = getFieldAddr(fieldId);
+ int len = UNSAFE.getInt(pos);
+ pos += SizeOf.SIZE_OF_INT;
+
+ byte [] bytes = new byte[len];
+ UNSAFE.copyMemory(null, pos, bytes, UnsafeUtil.ARRAY_BYTE_BASE_OFFSET, len);
+ return Convert.utf2chars(bytes);
+ }
+
+ @Override
+ public Tuple clone() throws CloneNotSupportedException {
+ return toHeapTuple();
+ }
+
+ @Override
+ public Datum[] getValues() {
+ Datum [] datums = new Datum[size()];
+ for (int i = 0; i < size(); i++) {
+ if (contains(i)) {
+ datums[i] = get(i);
+ } else {
+ datums[i] = NullDatum.get();
+ }
+ }
+ return datums;
+ }
+
+ @Override
+ public String toString() {
+ return VTuple.toDisplayString(getValues());
+ }
+
+ public abstract void release();
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/UnSafeTupleBytesComparator.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/UnSafeTupleBytesComparator.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/UnSafeTupleBytesComparator.java
new file mode 100644
index 0000000..73e1e2f
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/UnSafeTupleBytesComparator.java
@@ -0,0 +1,99 @@
+/***
+ * 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.offheap;
+
+import com.google.common.primitives.Longs;
+import com.google.common.primitives.UnsignedLongs;
+import org.apache.tajo.util.SizeOf;
+import org.apache.tajo.util.UnsafeUtil;
+import sun.misc.Unsafe;
+
+import java.nio.ByteOrder;
+
+/**
+ * It directly access UTF bytes in UnSafeTuple without any copy. It is used by compiled TupleComparator.
+ */
+public class UnSafeTupleBytesComparator {
+ private static final Unsafe UNSAFE = UnsafeUtil.unsafe;
+
+ static final boolean littleEndian =
+ ByteOrder.nativeOrder().equals(ByteOrder.LITTLE_ENDIAN);
+
+ public static int compare(long ptr1, long ptr2) {
+ int lstrLen = UNSAFE.getInt(ptr1);
+ int rstrLen = UNSAFE.getInt(ptr2);
+
+ ptr1 += SizeOf.SIZE_OF_INT;
+ ptr2 += SizeOf.SIZE_OF_INT;
+
+ int minLength = Math.min(lstrLen, rstrLen);
+ int minWords = minLength / Longs.BYTES;
+
+ /*
+ * Compare 8 bytes at a time. Benchmarking shows comparing 8 bytes at a
+ * time is no slower than comparing 4 bytes at a time even on 32-bit.
+ * On the other hand, it is substantially faster on 64-bit.
+ */
+ for (int i = 0; i < minWords * Longs.BYTES; i += Longs.BYTES) {
+ long lw = UNSAFE.getLong(ptr1);
+ long rw = UNSAFE.getLong(ptr2);
+ long diff = lw ^ rw;
+
+ if (diff != 0) {
+ if (!littleEndian) {
+ return UnsignedLongs.compare(lw, rw);
+ }
+
+ // Use binary search
+ int n = 0;
+ int y;
+ int x = (int) diff;
+ if (x == 0) {
+ x = (int) (diff >>> 32);
+ n = 32;
+ }
+
+ y = x << 16;
+ if (y == 0) {
+ n += 16;
+ } else {
+ x = y;
+ }
+
+ y = x << 8;
+ if (y == 0) {
+ n += 8;
+ }
+ return (int) (((lw >>> n) & 0xFFL) - ((rw >>> n) & 0xFFL));
+ }
+
+ ptr1 += SizeOf.SIZE_OF_LONG;
+ ptr2 += SizeOf.SIZE_OF_LONG;
+ }
+
+ // The epilogue to cover the last (minLength % 8) elements.
+ for (int i = minWords * Longs.BYTES; i < minLength; i++) {
+ int result = UNSAFE.getByte(ptr1++) - UNSAFE.getByte(ptr2++);
+ if (result != 0) {
+ return result;
+ }
+ }
+ return lstrLen - rstrLen;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/ZeroCopyTuple.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/ZeroCopyTuple.java b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/ZeroCopyTuple.java
new file mode 100644
index 0000000..51dbb29
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/tuple/offheap/ZeroCopyTuple.java
@@ -0,0 +1,35 @@
+/*
+ * 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.offheap;
+
+import java.nio.ByteBuffer;
+
+import static org.apache.tajo.common.TajoDataTypes.DataType;
+
+public class ZeroCopyTuple extends UnSafeTuple {
+
+ public void set(ByteBuffer bb, int relativePos, int length, DataType [] types) {
+ super.set(bb, relativePos, length, types);
+ }
+
+ @Override
+ public void release() {
+ // nothing to do
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/main/proto/IndexProtos.proto
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/proto/IndexProtos.proto b/tajo-storage/src/main/proto/IndexProtos.proto
index bcb0cbe..f5c8a08 100644
--- a/tajo-storage/src/main/proto/IndexProtos.proto
+++ b/tajo-storage/src/main/proto/IndexProtos.proto
@@ -25,5 +25,7 @@ option java_generate_equals_and_hash = true;
import "CatalogProtos.proto";
message TupleComparatorProto {
- repeated TupleComparatorSpecProto compSpecs = 1;
+ required SchemaProto schema = 1;
+ repeated SortSpecProto sortSpecs = 2;
+ repeated TupleComparatorSpecProto compSpecs = 3;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/514ed847/tajo-storage/src/test/java/org/apache/tajo/storage/TestTupleComparator.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/TestTupleComparator.java b/tajo-storage/src/test/java/org/apache/tajo/storage/TestTupleComparator.java
index ab7c2b2..639ca04 100644
--- a/tajo-storage/src/test/java/org/apache/tajo/storage/TestTupleComparator.java
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/TestTupleComparator.java
@@ -69,7 +69,7 @@ public class TestTupleComparator {
SortSpec sortKey1 = new SortSpec(schema.getColumn("col4"), true, false);
SortSpec sortKey2 = new SortSpec(schema.getColumn("col5"), true, false);
- TupleComparator tc = new TupleComparator(schema,
+ BaseTupleComparator tc = new BaseTupleComparator(schema,
new SortSpec[] {sortKey1, sortKey2});
assertEquals(-1, tc.compare(tuple1, tuple2));
assertEquals(1, tc.compare(tuple2, tuple1));