KUDU-721: [Java] Add DECIMAL column type support

This patch adds basic support to the Java client to
create, read, and write tables with DECIMAL columns.

Change-Id: I6240e3cfe0d6328b68c50099d442ffeeab6c9fd9
Reviewed-on: http://gerrit.cloudera.org:8080/8882
Tested-by: Kudu Jenkins
Reviewed-by: Dan Burkert <dan@cloudera.com>

java/kudu-client/src/main/java/org/apache/kudu/ColumnSchema.java

@@ -17,6 +17,8 @@
 
 package org.apache.kudu;
 
+import java.util.Objects;
+
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.yetus.audience.InterfaceStability;
 
@@ -39,6 +41,8 @@ public class ColumnSchema {
   private final int desiredBlockSize;
   private final Encoding encoding;
   private final CompressionAlgorithm compressionAlgorithm;
+  private final ColumnTypeAttributes typeAttributes;
+  private final int typeSize;
 
   /**
    * Specifies the encoding of data for a column on disk.
@@ -92,7 +96,7 @@ public CompressionType getInternalPbType() {
 
   private ColumnSchema(String name, Type type, boolean key, boolean nullable,
                        Object defaultValue, int desiredBlockSize, Encoding encoding,
-                       CompressionAlgorithm compressionAlgorithm) {
+                       CompressionAlgorithm compressionAlgorithm, ColumnTypeAttributes typeAttributes) {
     this.name = name;
     this.type = type;
     this.key = key;
@@ -101,6 +105,8 @@ private ColumnSchema(String name, Type type, boolean key, boolean nullable,
     this.desiredBlockSize = desiredBlockSize;
     this.encoding = encoding;
     this.compressionAlgorithm = compressionAlgorithm;
+    this.typeAttributes = typeAttributes;
+    this.typeSize = type.getSize(typeAttributes);
   }
 
   /**
@@ -168,6 +174,21 @@ public CompressionAlgorithm getCompressionAlgorithm() {
     return compressionAlgorithm;
   }
 
+  /**
+   * Return the column type attributes for the column, or null if it is not known.
+   */
+  public ColumnTypeAttributes getTypeAttributes() {
+    return typeAttributes;
+  }
+
+  /**
+   * The size of this type in bytes on the wire.
+   * @return A size
+   */
+  public int getTypeSize() {
+    return typeSize;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -188,21 +209,21 @@ public boolean equals(Object o) {
     if (!type.equals(that.type)) {
       return false;
     }
+    if (!typeAttributes.equals(that.typeAttributes)) {
+      return false;
+    }
 
     return true;
   }
 
   @Override
   public int hashCode() {
-    int result = name.hashCode();
-    result = 31 * result + type.hashCode();
-    result = 31 * result + (key ? 1 : 0);
-    return result;
+    return Objects.hash(name, type, key, typeAttributes);
   }
 
   @Override
   public String toString() {
-    return "Column name: " + name + ", type: " + type.getName();
+    return "Column name: " + name + ", type: " + type.getName() + typeAttributes.toStringForType(type);
   }
 
   /**
@@ -219,6 +240,7 @@ public static class ColumnSchemaBuilder {
     private int blockSize = 0;
     private Encoding encoding = null;
     private CompressionAlgorithm compressionAlgorithm = null;
+    private ColumnTypeAttributes typeAttributes = null;
 
     /**
      * Constructor for the required parameters.
@@ -309,14 +331,26 @@ public ColumnSchemaBuilder compressionAlgorithm(CompressionAlgorithm compression
       return this;
     }
 
+    /**
+     * Set the column type attributes for this column.
+     */
+    public ColumnSchemaBuilder typeAttributes(ColumnTypeAttributes typeAttributes) {
+      if (type != Type.DECIMAL && typeAttributes != null) {
+        throw new IllegalArgumentException(
+            "ColumnTypeAttributes are not used on " + type + " columns");
+      }
+      this.typeAttributes = typeAttributes;
+      return this;
+    }
+
     /**
      * Builds a {@link ColumnSchema} using the passed parameters.
      * @return a new {@link ColumnSchema}
      */
     public ColumnSchema build() {
       return new ColumnSchema(name, type,
                               key, nullable, defaultValue,
-                              blockSize, encoding, compressionAlgorithm);
+                              blockSize, encoding, compressionAlgorithm, typeAttributes);
     }
   }
 }

java/kudu-client/src/main/java/org/apache/kudu/ColumnTypeAttributes.java

@@ -0,0 +1,162 @@
+// 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.kudu;
+
+import java.util.Objects;
+
+/**
+ * Represents a Kudu Table column's type attributes.
+ */
+@org.apache.yetus.audience.InterfaceAudience.Public
+@org.apache.yetus.audience.InterfaceStability.Evolving
+public class ColumnTypeAttributes {
+
+  private final boolean hasPrecision;
+  private final int precision;
+
+  private final boolean hasScale;
+  private final int scale;
+
+  private ColumnTypeAttributes(boolean hasPrecision, int precision,
+                               boolean hasScale, int scale) {
+    this.hasPrecision = hasPrecision;
+    this.precision = precision;
+    this.hasScale = hasScale;
+    this.scale = scale;
+  }
+
+  /**
+   * Returns true if the precision is set;
+   */
+  public boolean hasPrecision() {
+    return hasPrecision;
+  }
+
+  /**
+   * Return the precision;
+   */
+  public int getPrecision() {
+    return precision;
+  }
+
+  /**
+   * Returns true if the scale is set;
+   */
+  public boolean hasScale() {
+    return hasScale;
+  }
+
+  /**
+   * Return the scale;
+   */
+  public int getScale() {
+    return scale;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    ColumnTypeAttributes that = (ColumnTypeAttributes) o;
+
+    if (hasPrecision != that.hasPrecision) {
+      return false;
+    }
+    if (precision != that.precision) {
+      return false;
+    }
+    if (hasScale != that.hasScale) {
+      return false;
+    }
+    if (scale != that.scale) {
+      return false;
+    }
+
+    return true;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(hasPrecision, precision, hasScale, scale);
+  }
+
+  /**
+   * Return a string representation appropriate for `type`.
+   * This is meant to be postfixed to the name of a primitive type to describe
+   * the full type, e.g. decimal(10, 4).
+   * @param type the type.
+   * @return a postfix string.
+   */
+  public String toStringForType(Type type) {
+    if (type == Type.DECIMAL) {
+      return "(" + precision + ", " + scale + ")";
+    } else {
+      return "";
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "hasPrecision: " + hasPrecision + ", precision: " + precision +
+        ", hasScale: " + hasScale + ", scale: " + scale;
+  }
+
+  /**
+   * Builder for ColumnTypeAttributes.
+   */
+  @org.apache.yetus.audience.InterfaceAudience.Public
+  @org.apache.yetus.audience.InterfaceStability.Evolving
+  public static class ColumnTypeAttributesBuilder {
+
+    private boolean hasPrecision;
+    private int precision;
+    private boolean hasScale;
+    private int scale;
+
+    /**
+     * Set the precision. Only used for Decimal columns.
+     */
+    public ColumnTypeAttributesBuilder precision(int precision) {
+      this.hasPrecision = true;
+      this.precision = precision;
+      return this;
+    }
+
+    /**
+     * Set the scale. Only used for Decimal columns.
+     */
+    public ColumnTypeAttributesBuilder scale(int scale) {
+      this.hasScale = true;
+      this.scale = scale;
+      return this;
+    }
+
+    /**
+     * Builds a {@link ColumnTypeAttributes} using the passed parameters.
+     * @return a new {@link ColumnTypeAttributes}
+     */
+    public ColumnTypeAttributes build() {
+      return new ColumnTypeAttributes(hasPrecision, precision, hasScale, scale);
+    }
+  }
+}

java/kudu-client/src/main/java/org/apache/kudu/Schema.java

@@ -111,7 +111,7 @@ public Schema(List<ColumnSchema> columns, List<Integer> columnIds) {
 
       hasNulls |= column.isNullable();
       columnOffsets[index] = offset;
-      offset += column.getType().getSize();
+      offset += column.getTypeSize();
       if (this.columnsByName.put(column.getName(), index) != null) {
         throw new IllegalArgumentException(
             String.format("Column names must be unique: %s", columns));
@@ -167,7 +167,7 @@ private static int getRowSize(List<ColumnSchema> columns) {
     int totalSize = 0;
     boolean hasNullables = false;
     for (ColumnSchema column : columns) {
-      totalSize += column.getType().getSize();
+      totalSize += column.getTypeSize();
       hasNullables |= column.isNullable();
     }
     if (hasNullables) {