TAJO-2096: Apply offheap to the join operator.

tajo-common/src/main/java/org/apache/tajo/SessionVars.java

@@ -142,6 +142,8 @@ public enum SessionVars implements ConfigKey {
   SORT_LIST_SIZE(ConfVars.$SORT_LIST_SIZE, "The initial size of list for in-memory sort", DEFAULT),
   JOIN_HASH_TABLE_SIZE(ConfVars.$JOIN_HASH_TABLE_SIZE, "The initial size of hash table for in-memory hash join",
       DEFAULT),
+  EXECUTOR_DIRECT_MEMORY_ENABLE(ConfVars.$EXECUTOR_MEMORY_DIRECT,
+      "If true, the executor data will be kept in direct memory", DEFAULT),
 
   // for index
   INDEX_ENABLED(ConfVars.$INDEX_ENABLED, "index scan enabled", DEFAULT),

tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java

@@ -367,6 +367,7 @@ public static enum ConfVars implements ConfigKey {
     $AGG_HASH_TABLE_SIZE("tajo.executor.aggregate.hash-table.size", 10000),
     $SORT_LIST_SIZE("tajo.executor.sort.list.size", 100000),
     $JOIN_HASH_TABLE_SIZE("tajo.executor.join.hash-table.size", 100000),
+    $EXECUTOR_MEMORY_DIRECT("tajo.executor.memory.direct", true),
 
     // for index
     $INDEX_ENABLED("tajo.query.index.enabled", false),

tajo-common/src/main/java/org/apache/tajo/datum/BlobDatum.java

@@ -124,17 +124,20 @@ public int hashCode() {
     return bb.hashCode();
   }
 
-  @Override
-  public boolean equals(Object obj) {
-    if (obj instanceof BlobDatum) {
-      BlobDatum other = (BlobDatum) obj;
-      initFromBytes();
-      other.initFromBytes();
+	@Override
+	public boolean equals(Object obj) {
+		if (this == obj)
+			return true;
+
+		if (obj instanceof BlobDatum) {
+			BlobDatum other = (BlobDatum) obj;
+			initFromBytes();
+			other.initFromBytes();
 			return Arrays.equals(this.val, other.val);
 		}
-    
-    return false;
-  }
+
+		return false;
+	}
 
   @Override
   public Datum equalsTo(Datum datum) {

tajo-common/src/main/java/org/apache/tajo/datum/BooleanDatum.java

@@ -147,6 +147,9 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof BooleanDatum) {
       BooleanDatum other = (BooleanDatum) obj;
       return val == other.val;

tajo-common/src/main/java/org/apache/tajo/datum/CharDatum.java

@@ -21,6 +21,7 @@
 import com.google.common.primitives.UnsignedBytes;
 import com.google.gson.annotations.Expose;
 import org.apache.tajo.exception.InvalidOperationException;
+import org.apache.tajo.util.MurmurHash3_32;
 
 import java.util.Arrays;
 
@@ -49,12 +50,12 @@ public CharDatum(byte [] bytes) {
   }
 
   public CharDatum(String val) {
-    this(val.getBytes());
+    this(val.getBytes(TextDatum.DEFAULT_CHARSET));
   }
 
   private String getString() {
     if (chars == null) {
-      chars = new String(bytes);
+      chars = new String(bytes, TextDatum.DEFAULT_CHARSET);
     }
     return chars;
   }
@@ -115,11 +116,14 @@ public int size() {
 
   @Override
   public int hashCode() {
-    return getString().hashCode();
+    return MurmurHash3_32.hash(bytes);
   }
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof CharDatum) {
       CharDatum other = (CharDatum) obj;
       return this.size == other.size &&

tajo-common/src/main/java/org/apache/tajo/datum/DateDatum.java

@@ -25,6 +25,7 @@
 import org.apache.tajo.exception.InvalidOperationException;
 import org.apache.tajo.exception.TajoRuntimeException;
 import org.apache.tajo.util.Bytes;
+import org.apache.tajo.util.MurmurHash3_32;
 import org.apache.tajo.util.datetime.DateTimeConstants.DateStyle;
 import org.apache.tajo.util.datetime.DateTimeFormat;
 import org.apache.tajo.util.datetime.DateTimeUtil;
@@ -34,7 +35,7 @@ public class DateDatum extends Datum {
   public static final int SIZE = 4;
 
   // Dates are stored in UTC.
-  private int jdate;
+  private final int jdate;
 
   public DateDatum(int value) {
     super(TajoDataTypes.Type.DATE);
@@ -231,23 +232,19 @@ public int compareTo(Datum datum) {
 
   @Override
   public boolean equals(Object obj) {
-    TimeMeta tm = asTimeMeta();
+    if (this == obj)
+      return true;
+
     if (obj instanceof DateDatum) {
-      TimeMeta another = ((DateDatum) obj).asTimeMeta();
-      return tm.years == another.years && tm.monthOfYear == another.monthOfYear && tm.dayOfMonth == another.dayOfMonth;
+      DateDatum other = (DateDatum) obj;
+      return jdate == other.jdate;
     } else {
       return false;
     }
   }
 
   @Override
   public int hashCode() {
-    TimeMeta tm = asTimeMeta();
-    int total = 157;
-    total = 23 * total + tm.years;
-    total = 23 * total + tm.monthOfYear;
-    total = 23 * total + tm.dayOfMonth;
-
-    return total;
+    return MurmurHash3_32.hash(jdate);
   }
 }

tajo-common/src/main/java/org/apache/tajo/datum/Float4Datum.java

@@ -23,7 +23,7 @@
 import org.apache.tajo.exception.InvalidValueForCastException;
 import org.apache.tajo.exception.InvalidOperationException;
 import org.apache.tajo.exception.TajoRuntimeException;
-import org.apache.tajo.util.MurmurHash;
+import org.apache.tajo.util.MurmurHash3_32;
 import org.apache.tajo.util.NumberUtil;
 import org.apache.tajo.util.datetime.TimeMeta;
 
@@ -107,11 +107,14 @@ public int size() {
 
   @Override
   public int hashCode() {
-    return MurmurHash.hash(val);
+    return MurmurHash3_32.hash(val);
   }
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof Float4Datum) {
       Float4Datum other = (Float4Datum) obj;
       return val == other.val;

tajo-common/src/main/java/org/apache/tajo/datum/Float8Datum.java

@@ -22,7 +22,7 @@
 import org.apache.tajo.common.TajoDataTypes;
 import org.apache.tajo.exception.InvalidOperationException;
 import org.apache.tajo.util.Bytes;
-import org.apache.tajo.util.MurmurHash;
+import org.apache.tajo.util.MurmurHash3_32;
 import org.apache.tajo.util.NumberUtil;
 import org.apache.tajo.util.datetime.TimeMeta;
 
@@ -96,10 +96,13 @@ public int size() {
 
   @Override
   public int hashCode() {
-    return MurmurHash.hash(val);
+    return MurmurHash3_32.hash(val);
   }
 
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof Float8Datum) {
       Float8Datum other = (Float8Datum) obj;
       return val == other.val;

tajo-common/src/main/java/org/apache/tajo/datum/Inet4Datum.java

@@ -99,6 +99,9 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof Inet4Datum) {
       Inet4Datum other = (Inet4Datum) obj;
       return this.address == other.address;

tajo-common/src/main/java/org/apache/tajo/datum/Int2Datum.java

@@ -21,7 +21,7 @@
 import com.google.gson.annotations.Expose;
 import org.apache.tajo.common.TajoDataTypes;
 import org.apache.tajo.exception.InvalidOperationException;
-import org.apache.tajo.util.MurmurHash;
+import org.apache.tajo.util.MurmurHash3_32;
 import org.apache.tajo.util.NumberUtil;
 import org.apache.tajo.util.datetime.TimeMeta;
 
@@ -97,11 +97,14 @@ public int size() {
 
   @Override
   public int hashCode() {
-    return MurmurHash.hash(val);
+    return MurmurHash3_32.hash(val);
   }
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof Int2Datum) {
       Int2Datum other = (Int2Datum) obj;
       return val == other.val;

tajo-common/src/main/java/org/apache/tajo/datum/Int4Datum.java

@@ -21,7 +21,7 @@
 import com.google.gson.annotations.Expose;
 import org.apache.tajo.common.TajoDataTypes.Type;
 import org.apache.tajo.exception.InvalidOperationException;
-import org.apache.tajo.util.MurmurHash;
+import org.apache.tajo.util.MurmurHash3_32;
 import org.apache.tajo.util.NumberUtil;
 import org.apache.tajo.util.datetime.TimeMeta;
 
@@ -102,10 +102,13 @@ public int size() {
 
   @Override
   public int hashCode() {
-    return MurmurHash.hash(val);
+    return MurmurHash3_32.hash(val);
   }
 
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof Int4Datum) {
       Int4Datum other = (Int4Datum) obj;
       return val == other.val;

tajo-common/src/main/java/org/apache/tajo/datum/Int8Datum.java

@@ -23,7 +23,7 @@
 import org.apache.tajo.exception.InvalidValueForCastException;
 import org.apache.tajo.exception.InvalidOperationException;
 import org.apache.tajo.exception.TajoRuntimeException;
-import org.apache.tajo.util.MurmurHash;
+import org.apache.tajo.util.MurmurHash3_32;
 import org.apache.tajo.util.NumberUtil;
 import org.apache.tajo.util.datetime.TimeMeta;
 
@@ -109,11 +109,14 @@ public int size() {
 
   @Override
   public int hashCode() {
-    return MurmurHash.hash(val);
+    return MurmurHash3_32.hash(val);
   }
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof Int8Datum) {
       Int8Datum other = (Int8Datum) obj;
       return val == other.val;

tajo-common/src/main/java/org/apache/tajo/datum/IntervalDatum.java

@@ -412,6 +412,9 @@ public Datum equalsTo(Datum datum) {
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof IntervalDatum) {
       return asInt8() == ((IntervalDatum)obj).asInt8();
     } else {

tajo-common/src/main/java/org/apache/tajo/datum/ProtobufDatum.java

@@ -58,6 +58,9 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object object) {
+    if (this == object)
+      return true;
+
     if (object instanceof ProtobufDatum) {
       ProtobufDatum another = (ProtobufDatum) object;
       return value.equals(another.value);

tajo-common/src/main/java/org/apache/tajo/datum/TextDatum.java

@@ -25,7 +25,7 @@
 import org.apache.tajo.exception.InvalidValueForCastException;
 import org.apache.tajo.exception.InvalidOperationException;
 import org.apache.tajo.exception.TajoRuntimeException;
-import org.apache.tajo.util.MurmurHash;
+import org.apache.tajo.util.MurmurHash3_32;
 import org.apache.tajo.util.StringUtils;
 
 import java.nio.charset.Charset;
@@ -128,9 +128,12 @@ public int compareTo(Datum datum) {
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof TextDatum) {
       TextDatum o = (TextDatum) obj;
-      return COMPARATOR.compare(this.bytes, o.bytes) == 0;
+      return size() == o.size() && COMPARATOR.compare(this.bytes, o.bytes) == 0;
     }
 
     return false;
@@ -152,7 +155,7 @@ public Datum equalsTo(Datum datum) {
 
   @Override
   public int hashCode() {
-    return MurmurHash.hash(bytes);
+    return MurmurHash3_32.hash(bytes);
   }
 
   @Override

tajo-common/src/main/java/org/apache/tajo/datum/TimeDatum.java

@@ -179,6 +179,9 @@ public int compareTo(Datum datum) {
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof TimeDatum) {
       TimeDatum another = (TimeDatum) obj;
       return time == another.time;

tajo-common/src/main/java/org/apache/tajo/datum/TimestampDatum.java

@@ -32,7 +32,7 @@
 public class TimestampDatum extends Datum {
   public static final int SIZE = 8;
 
-  private long timestamp;
+  private final long timestamp;
 
   /**
    *
@@ -204,6 +204,9 @@ public int compareTo(Datum datum) {
 
   @Override
   public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+
     if (obj instanceof TimestampDatum) {
       TimestampDatum another = (TimestampDatum) obj;
       return timestamp == another.timestamp;

tajo-common/src/main/java/org/apache/tajo/storage/BufferPool.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.tajo.TajoConstants;
 import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.util.FileUtil;
 
 import java.lang.management.BufferPoolMXBean;
 import java.lang.management.ManagementFactory;
@@ -188,4 +189,18 @@ public static BufferPoolMXBean getMappedBufferPool() {
   private static List<BufferPoolMXBean> getBufferPools() {
     return ManagementFactory.getPlatformMXBeans(BufferPoolMXBean.class);
   }
+
+  public static String printDirectMemoryUsage() {
+    BufferPoolMXBean direct = BufferPool.getDirectBufferPool();
+    return String.format("%s (USED)/%s (TOTAL)",
+        FileUtil.humanReadableByteCount(direct.getMemoryUsed(), false),
+        FileUtil.humanReadableByteCount(direct.getTotalCapacity(), false));
+  }
+
+  public static String printHeapMemoryUsage() {
+    return String.format("%s (FREE)/%s (TOTAL)/%s (MAX)",
+        FileUtil.humanReadableByteCount(Runtime.getRuntime().freeMemory(), false),
+        FileUtil.humanReadableByteCount(Runtime.getRuntime().totalMemory(), false),
+        FileUtil.humanReadableByteCount(Runtime.getRuntime().maxMemory(), false));
+  }
 }