TAJO-2129: Apply new type implementation to Schema and Catalog.

tajo-algebra/src/main/java/org/apache/tajo/algebra/ColumnDefinition.java

@@ -24,11 +24,6 @@ public  class ColumnDefinition extends DataTypeExpr {
   @Expose @SerializedName("ColumnDefName")
   String columnName;
 
-  public ColumnDefinition(String columnName, String dataType) {
-    super(dataType);
-    this.columnName = columnName;
-  }
-
   public ColumnDefinition(String columnName, DataTypeExpr dataType) {
     super(dataType.getTypeName());
 
@@ -42,15 +37,17 @@ public ColumnDefinition(String columnName, DataTypeExpr dataType) {
       }
     }
 
-    // nested records
     if (dataType.isRecordType()) {
       this.recordType = dataType.recordType;
     }
 
-    // map type
     if (dataType.isMapType()) {
       this.mapType = dataType.mapType;
     }
+
+    if (dataType.isArrayType()) {
+      this.arrayType = dataType.arrayType;
+    }
   }
 
   public String getColumnName() {

tajo-algebra/src/main/java/org/apache/tajo/algebra/DataTypeExpr.java

@@ -19,37 +19,52 @@
 package org.apache.tajo.algebra;
 
 import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
 import com.google.gson.annotations.Expose;
 import com.google.gson.annotations.SerializedName;
+import org.apache.tajo.Assert;
 import org.apache.tajo.common.TajoDataTypes.Type;
 import org.apache.tajo.util.TUtil;
 
+import static org.apache.tajo.Assert.assertNotNull;
+
 public class DataTypeExpr extends Expr {
   @Expose @SerializedName("DataTypeName")
   String typeName;
   @Expose @SerializedName("LengthOrPrecision")
   Integer lengthOrPrecision;
   @Expose @SerializedName("Scale")
   Integer scale;
+  @Expose @SerializedName("Array")
+  ArrayType arrayType; // not null if the type is ARRAY
   @Expose @SerializedName("Record")
   RecordType recordType; // not null if the type is RECORD
   @Expose @SerializedName("Map")
   MapType mapType;
 
   public DataTypeExpr(String typeName) {
     super(OpType.DataType);
+    assertNotNull(typeName);
     this.typeName = typeName;
   }
 
+  public DataTypeExpr(ArrayType array) {
+    super(OpType.DataType);
+    assertNotNull(array);
+    this.typeName = Type.ARRAY.name();
+    this.arrayType = array;
+  }
+
   public DataTypeExpr(RecordType record) {
     super(OpType.DataType);
+    assertNotNull(record);
     this.typeName = Type.RECORD.name();
     this.recordType = record;
   }
 
   public DataTypeExpr(MapType map) {
     super(OpType.DataType);
-    // RECORD = 51 in DataTypes.proto
+    assertNotNull(map);
     this.typeName = Type.MAP.name();
     this.mapType = map;
   }
@@ -59,15 +74,34 @@ public String getTypeName() {
   }
 
   public boolean isPrimitiveType() {
-    return !this.isRecordType() && !isMapType();
+    return !isArrayType()&& !isRecordType() && !isMapType();
+  }
+
+  public boolean isArrayType() {
+    return arrayType != null;
   }
 
   public boolean isRecordType() {
-    return this.typeName.equals(Type.RECORD.name());
+    return recordType != null;
   }
 
   public boolean isMapType() {
-    return this.typeName.equals(Type.MAP.name());
+    return mapType != null;
+  }
+
+  public DataTypeExpr getElementType() {
+    Preconditions.checkState(isArrayType());
+    return arrayType.type;
+  }
+
+  public DataTypeExpr getKeyType() {
+    Preconditions.checkState(isMapType());
+    return mapType.keyType;
+  }
+
+  public DataTypeExpr getValueType() {
+    Preconditions.checkState(isMapType());
+    return mapType.valueType;
   }
 
   public ColumnDefinition [] getNestedRecordTypes() {
@@ -125,6 +159,27 @@ public Object clone() throws CloneNotSupportedException {
     return dataType;
   }
 
+  public static class ArrayType implements JsonSerializable, Cloneable {
+    @Expose
+    @SerializedName("type")
+    DataTypeExpr type;
+
+    public ArrayType(DataTypeExpr elementType) {
+      this.type = elementType;
+    }
+
+    @Override
+    public String toJson() {
+      return JsonHelper.toJson(this);
+    }
+
+    public Object clone() throws CloneNotSupportedException {
+      ArrayType newMap = (ArrayType) super.clone();
+      newMap.type = type;
+      return newMap;
+    }
+  }
+
   public static class RecordType implements JsonSerializable, Cloneable {
     @Expose @SerializedName("Schema")
     ColumnDefinition [] schema; // not null if the type is RECORD

tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/AbstractCatalogClient.java

@@ -32,6 +32,7 @@
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.NullProto;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.ReturnState;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.StringListResponse;
+import org.apache.tajo.schema.IdentifierUtil;
 import org.apache.tajo.util.ProtoUtil;
 
 import java.io.Closeable;
@@ -281,7 +282,7 @@ public final TableDesc getTableDesc(final String databaseName, final String tabl
 
   @Override
   public TableDesc getTableDesc(String qualifiedName) throws UndefinedTableException {
-    String[] splitted = CatalogUtil.splitFQTableName(qualifiedName);
+    String[] splitted = IdentifierUtil.splitFQTableName(qualifiedName);
     return getTableDesc(splitted[0], splitted[1]);
   }
 
@@ -596,7 +597,7 @@ public final void createTable(final TableDesc desc)
   public void dropTable(String tableName)
       throws UndefinedDatabaseException, UndefinedTableException, InsufficientPrivilegeException {
 
-    String[] splitted = CatalogUtil.splitFQTableName(tableName);
+    String[] splitted = IdentifierUtil.splitFQTableName(tableName);
     final String databaseName = splitted[0];
     final String simpleName = splitted[1];
 
@@ -617,7 +618,7 @@ public void dropTable(String tableName)
 
   @Override
   public final boolean existsTable(final String databaseName, final String tableName) {
-    if (CatalogUtil.isFQTableName(tableName)) {
+    if (IdentifierUtil.isFQTableName(tableName)) {
       throw new IllegalArgumentException(
           "tableName cannot be composed of multiple parts, but it is \"" + tableName + "\"");
     }
@@ -640,7 +641,7 @@ public final boolean existsTable(final String databaseName, final String tableNa
 
   @Override
   public final boolean existsTable(final String tableName) {
-    String[] splitted = CatalogUtil.splitFQTableName(tableName);
+    String[] splitted = IdentifierUtil.splitFQTableName(tableName);
     return existsTable(splitted[0], splitted[1]);
   }
 

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

@@ -87,8 +87,8 @@ public static boolean isUpperCastable(TajoDataTypes.Type define, TajoDataTypes.T
    */
   public static org.apache.tajo.type.Type determineType(org.apache.tajo.type.Type left,
                                                         org.apache.tajo.type.Type right) {
-    TajoDataTypes.Type rhsBaseType = right.baseType();
-    switch (left.baseType()) {
+    TajoDataTypes.Type rhsBaseType = right.kind();
+    switch (left.kind()) {
 
     case INT1:
     case INT2:

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

@@ -19,10 +19,6 @@
 package org.apache.tajo.catalog;
 
 public class CatalogConstants {
-  public final static String IDENTIFIER_DELIMITER_REGEXP = "\\.";
-  public final static String IDENTIFIER_DELIMITER = ".";
-  public final static String IDENTIFIER_QUOTE_STRING = "\"";
-  public final static int MAX_IDENTIFIER_LENGTH = 128;
   // Linux and BSD's max username length is 32. For compatibility with other systems, we should follow it.
   public final static int MAX_USERNAME_LENGTH = 32;
   public final static int MAX_STATEMENT_LENGTH = 128 * 1024;