TAJO-1554 Implement hash-theta join

tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java

@@ -311,6 +311,15 @@ public boolean containsAll(Collection<Column> columns) {
     return fields.containsAll(columns);
   }
 
+  public boolean containsAny(Collection<Column> columns) {
+    for (Column column : columns) {
+      if (contains(column)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   public synchronized Schema addColumn(String name, TypeDesc typeDesc) {
     String normalized = name;
     if(fieldsByQualifiedName.containsKey(normalized)) {
@@ -383,6 +392,17 @@ public SchemaProto getProto() {
     return builder.build();
 	}
 
+  public Set<String> getAliases() {
+    Set<String> aliases = new HashSet<String>();
+    for (Column column : fields) {
+      if (column.hasQualifier()) {
+        String qualifier = column.getQualifier();
+        aliases.add(qualifier.startsWith("default.") ? qualifier.substring(8) : qualifier);
+      }
+    }
+    return aliases;
+  }
+
   private static class SchemaProtoBuilder implements ColumnVisitor {
     private SchemaProto.Builder builder;
     public SchemaProtoBuilder(SchemaProto.Builder builder) {

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

@@ -18,17 +18,12 @@
 
 package org.apache.tajo.storage;
 
-import org.apache.tajo.datum.Datum;
-import org.apache.tajo.datum.NullDatum;
-import org.apache.tajo.datum.ProtobufDatum;
-
 /* This class doesn’t have content datum. if selected column is zero, this is useful
 *  e.g. select count(*) from table
 * */
-public class EmptyTuple implements Tuple, Cloneable {
+public class EmptyTuple extends NullTuple {
 
   private static EmptyTuple tuple;
-  private static Datum[] EMPTY_VALUES = new Datum[0];
 
   static {
     tuple = new EmptyTuple();
@@ -39,138 +34,11 @@ public static EmptyTuple get() {
   }
 
   private EmptyTuple() {
+    super(0);
   }
 
   @Override
-  public int size() {
-    return 0;
-  }
-
-  public boolean contains(int fieldId) {
-    return false;
-  }
-
-  @Override
-  public boolean isNull(int fieldid) {
-    return true;
-  }
-
-  @Override
-  public boolean isNotNull(int fieldid) {
-    return false;
-  }
-
-  @Override
-  public void clear() {
-  }
-
-  @Override
-  public void put(int fieldId, Datum value) {
-    throw new UnsupportedOperationException();
-  }
-
-  @Override
-  public void put(int fieldId, Datum[] values) {
-    throw new UnsupportedOperationException();
-  }
-
-  @Override
-  public void put(int fieldId, Tuple tuple) {
-    throw new UnsupportedOperationException();
-  }
-
-  @Override
-  public void put(Datum[] values) {
-    throw new UnsupportedOperationException();
-  }
-
-  @Override
-  public Datum get(int fieldId) {
-    return NullDatum.get();
-  }
-
-  @Override
-  public void setOffset(long offset) {
-
-  }
-
-  @Override
-  public long getOffset() {
-    return -1;
-  }
-
-  @Override
-  public boolean getBool(int fieldId) {
-    return NullDatum.get().asBool();
-  }
-
-  @Override
-  public byte getByte(int fieldId) {
-    return NullDatum.get().asByte();
-  }
-
-  @Override
-  public char getChar(int fieldId) {
-    return NullDatum.get().asChar();
-  }
-
-  @Override
-  public byte[] getBytes(int fieldId) {
-    return NullDatum.get().asByteArray();
-  }
-
-  @Override
-  public short getInt2(int fieldId) {
-    return NullDatum.get().asInt2();
-  }
-
-  @Override
-  public int getInt4(int fieldId) {
-    return NullDatum.get().asInt4();
-  }
-
-  @Override
-  public long getInt8(int fieldId) {
-    return NullDatum.get().asInt8();
-  }
-
-  @Override
-  public float getFloat4(int fieldId) {
-    return NullDatum.get().asFloat4();
-  }
-
-  @Override
-  public double getFloat8(int fieldId) {
-    return NullDatum.get().asFloat8();
-  }
-
-  @Override
-  public String getText(int fieldId) {
-    return NullDatum.get().asChars();
-  }
-
-  @Override
-  public ProtobufDatum getProtobufDatum(int fieldId) {
-    throw new UnsupportedOperationException();
-  }
-
-  @Override
-  public Datum getInterval(int fieldId) {
-    return NullDatum.get();
-  }
-
-  @Override
-  public char[] getUnicodeChars(int fieldId) {
-    return NullDatum.get().asUnicodeChars();
-  }
-
-  @Override
-  public Tuple clone() throws CloneNotSupportedException {
-    throw new CloneNotSupportedException();
-  }
-
-  @Override
-  public Datum[] getValues() {
-    return EMPTY_VALUES;
+  public Tuple clone() {
+    return this;
   }
 }

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

@@ -0,0 +1,172 @@
+/**
+ * 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 org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.datum.ProtobufDatum;
+
+import java.util.Arrays;
+
+public class NullTuple implements Tuple, Cloneable {
+
+  public static Tuple create(int size) {
+    return new NullTuple(size);
+  }
+
+  private final int size;
+
+  NullTuple(int size) {
+    this.size = size;
+  }
+
+  @Override
+  public int size() {
+    return size;
+  }
+
+  public boolean contains(int fieldId) {
+    return fieldId < size;
+  }
+
+  @Override
+  public boolean isNull(int fieldid) {
+    return true;
+  }
+
+  @Override
+  public boolean isNotNull(int fieldid) {
+    return false;
+  }
+
+  @Override
+  public void clear() {
+  }
+
+  @Override
+  public void put(int fieldId, Datum value) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void put(int fieldId, Datum[] values) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void put(int fieldId, Tuple tuple) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void put(Datum[] values) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Datum get(int fieldId) {
+    return NullDatum.get();
+  }
+
+  @Override
+  public void setOffset(long offset) {
+  }
+
+  @Override
+  public long getOffset() {
+    return 0;
+  }
+
+  @Override
+  public boolean getBool(int fieldId) {
+    return NullDatum.get().asBool();
+  }
+
+  @Override
+  public byte getByte(int fieldId) {
+    return NullDatum.get().asByte();
+  }
+
+  @Override
+  public char getChar(int fieldId) {
+    return NullDatum.get().asChar();
+  }
+
+  @Override
+  public byte[] getBytes(int fieldId) {
+    return NullDatum.get().asByteArray();
+  }
+
+  @Override
+  public short getInt2(int fieldId) {
+    return NullDatum.get().asInt2();
+  }
+
+  @Override
+  public int getInt4(int fieldId) {
+    return NullDatum.get().asInt4();
+  }
+
+  @Override
+  public long getInt8(int fieldId) {
+    return NullDatum.get().asInt8();
+  }
+
+  @Override
+  public float getFloat4(int fieldId) {
+    return NullDatum.get().asFloat4();
+  }
+
+  @Override
+  public double getFloat8(int fieldId) {
+    return NullDatum.get().asFloat8();
+  }
+
+  @Override
+  public String getText(int fieldId) {
+    return NullDatum.get().asChars();
+  }
+
+  @Override
+  public ProtobufDatum getProtobufDatum(int fieldId) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Datum getInterval(int fieldId) {
+    return NullDatum.get();
+  }
+
+  @Override
+  public char[] getUnicodeChars(int fieldId) {
+    return NullDatum.get().asUnicodeChars();
+  }
+
+  @Override
+  public Tuple clone() throws CloneNotSupportedException {
+    return new NullTuple(size);
+  }
+
+  @Override
+  public Datum[] getValues() {
+    Datum[] datum = new Datum[size];
+    Arrays.fill(datum, NullDatum.get());
+    return datum;
+  }
+}