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 <d...@cloudera.com>


Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/4f34b69d
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/4f34b69d
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/4f34b69d

Branch: refs/heads/master
Commit: 4f34b69d4969c5a63885067a131c4d85016a173e
Parents: 118e8f3
Author: Grant Henke <granthe...@gmail.com>
Authored: Mon Dec 18 23:24:40 2017 -0600
Committer: Grant Henke <granthe...@gmail.com>
Committed: Tue Feb 13 22:20:03 2018 +0000

----------------------------------------------------------------------
 .../main/java/org/apache/kudu/ColumnSchema.java |  48 ++++-
 .../org/apache/kudu/ColumnTypeAttributes.java   | 162 +++++++++++++++
 .../src/main/java/org/apache/kudu/Schema.java   |   4 +-
 .../src/main/java/org/apache/kudu/Type.java     |  66 ++++++-
 .../apache/kudu/client/AsyncKuduScanner.java    |   1 +
 .../main/java/org/apache/kudu/client/Bytes.java | 196 ++++++++++++++++++-
 .../kudu/client/ColumnRangePredicate.java       |  35 +++-
 .../java/org/apache/kudu/client/KeyEncoder.java |  34 +++-
 .../org/apache/kudu/client/KuduPredicate.java   | 102 +++++++++-
 .../org/apache/kudu/client/KuduScanToken.java   |   2 +-
 .../java/org/apache/kudu/client/Operation.java  |   4 +-
 .../java/org/apache/kudu/client/PartialRow.java | 127 +++++++++++-
 .../org/apache/kudu/client/ProtobufHelper.java  |  48 ++++-
 .../java/org/apache/kudu/client/RowResult.java  |  47 ++++-
 .../java/org/apache/kudu/util/DecimalUtil.java  | 152 ++++++++++++++
 .../org/apache/kudu/client/BaseKuduTest.java    |   6 +-
 .../java/org/apache/kudu/client/TestBytes.java  |  45 ++++-
 .../kudu/client/TestColumnRangePredicate.java   |  18 +-
 .../org/apache/kudu/client/TestKeyEncoding.java |  76 +++++--
 .../org/apache/kudu/client/TestKuduClient.java  |  60 ++++++
 .../apache/kudu/client/TestKuduPredicate.java   | 111 ++++++++++-
 .../org/apache/kudu/client/TestPartialRow.java  | 119 ++++++++++-
 .../org/apache/kudu/client/TestRowResult.java   |   6 +
 .../apache/kudu/client/TestScanPredicate.java   |  95 +++++++++
 24 files changed, 1502 insertions(+), 62 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/ColumnSchema.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/ColumnSchema.java 
b/java/kudu-client/src/main/java/org/apache/kudu/ColumnSchema.java
index cf0273a..17dec97 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/ColumnSchema.java
+++ b/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 class ColumnSchema {
 
   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 @@ public class ColumnSchema {
     this.desiredBlockSize = desiredBlockSize;
     this.encoding = encoding;
     this.compressionAlgorithm = compressionAlgorithm;
+    this.typeAttributes = typeAttributes;
+    this.typeSize = type.getSize(typeAttributes);
   }
 
   /**
@@ -168,6 +174,21 @@ public class ColumnSchema {
     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 class ColumnSchema {
     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 class ColumnSchema {
     private int blockSize = 0;
     private Encoding encoding = null;
     private CompressionAlgorithm compressionAlgorithm = null;
+    private ColumnTypeAttributes typeAttributes = null;
 
     /**
      * Constructor for the required parameters.
@@ -310,13 +332,25 @@ public class ColumnSchema {
     }
 
     /**
+     * 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);
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/ColumnTypeAttributes.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/ColumnTypeAttributes.java 
b/java/kudu-client/src/main/java/org/apache/kudu/ColumnTypeAttributes.java
new file mode 100644
index 0000000..07da4a4
--- /dev/null
+++ b/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);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/Schema.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/Schema.java 
b/java/kudu-client/src/main/java/org/apache/kudu/Schema.java
index f38bcfb..8e19f38 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/Schema.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/Schema.java
@@ -111,7 +111,7 @@ public class Schema {
 
       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 @@ public class Schema {
     int totalSize = 0;
     boolean hasNullables = false;
     for (ColumnSchema column : columns) {
-      totalSize += column.getType().getSize();
+      totalSize += column.getTypeSize();
       hasNullables |= column.isNullable();
     }
     if (hasNullables) {

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/Type.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/Type.java 
b/java/kudu-client/src/main/java/org/apache/kudu/Type.java
index 123c577..4173c9b 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/Type.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/Type.java
@@ -19,12 +19,19 @@ package org.apache.kudu;
 
 import static org.apache.kudu.Common.DataType;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
 import com.google.common.primitives.Ints;
 import com.google.common.primitives.Longs;
 import com.google.common.primitives.Shorts;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.yetus.audience.InterfaceStability;
 
+import org.apache.kudu.ColumnTypeAttributes;
+import org.apache.kudu.util.DecimalUtil;
+
 /**
  * Describes all the types available to build table schemas.
  */
@@ -41,9 +48,10 @@ public enum Type {
   BOOL(DataType.BOOL, "bool"),
   FLOAT(DataType.FLOAT, "float"),
   DOUBLE(DataType.DOUBLE, "double"),
-  UNIXTIME_MICROS(DataType.UNIXTIME_MICROS, "unixtime_micros");
+  UNIXTIME_MICROS(DataType.UNIXTIME_MICROS, "unixtime_micros"),
+  DECIMAL(Arrays.asList(DataType.DECIMAL32, DataType.DECIMAL64, 
DataType.DECIMAL128), "decimal");
 
-  private final DataType dataType;
+  private final List<DataType> dataTypes;
   private final String name;
   private final int size;
 
@@ -53,17 +61,40 @@ public enum Type {
    * @param name string representation of the type
    */
   private Type(DataType dataType, String name) {
-    this.dataType = dataType;
+    this.dataTypes = Collections.singletonList(dataType);
+    this.name = name;
+    this.size = getTypeSize(dataType);
+  }
+
+  private Type(List<DataType> dataTypes, String name) {
+    this.dataTypes = dataTypes;
     this.name = name;
-    this.size = getTypeSize(this.dataType);
+    this.size = -1;
   }
 
   /**
    * Get the data type from the common's pb
    * @return A DataType
+   * @deprecated use {@link #getDataType(ColumnTypeAttributes)}
    */
   public DataType getDataType() {
-    return this.dataType;
+    if (this == DECIMAL) {
+      throw new IllegalStateException("Please use the newer 
getDataType(ColumnTypeAttributes) " +
+          "to support the Decimal data type");
+    }
+    return this.dataTypes.get(0);
+  }
+
+  /**
+   * Get the data type from the common's pb
+   * @param typeAttributes the additional attributes of the type.
+   * @return A DataType
+   */
+  public DataType getDataType(ColumnTypeAttributes typeAttributes) {
+    if (this == DECIMAL) {
+      return DecimalUtil.precisionToDataType(typeAttributes.getPrecision());
+    }
+    return this.dataTypes.get(0);
   }
 
   /**
@@ -77,14 +108,31 @@ public enum Type {
   /**
    * The size of this type on the wire
    * @return A size
+   * @deprecated use {@link #getSize(ColumnTypeAttributes)}
    */
   public int getSize() {
+    if (this == DECIMAL) {
+      throw new IllegalStateException("Please use the newer 
getSize(ColumnTypeAttributes) " +
+          "to support the Decimal data type");
+    }
+    return this.size;
+  }
+
+  /**
+   * The size of this type on the wire
+   * @param typeAttributes the additional attributes of the type.
+   * @return A size
+   */
+  public int getSize(ColumnTypeAttributes typeAttributes) {
+    if (this == DECIMAL) {
+      return DecimalUtil.precisionToSize(typeAttributes.getPrecision());
+    }
     return this.size;
   }
 
   @Override
   public String toString() {
-    return "Type: " + this.name + ", size: " + this.size;
+    return "Type: " + this.name;
   }
 
   /**
@@ -92,7 +140,7 @@ public enum Type {
    * @param type pb type
    * @return size in bytes
    */
-  static int getTypeSize(DataType type) {
+  private static int getTypeSize(DataType type) {
     switch (type) {
       case STRING:
       case BINARY:
@@ -131,6 +179,10 @@ public enum Type {
       case UNIXTIME_MICROS: return UNIXTIME_MICROS;
       case FLOAT: return FLOAT;
       case DOUBLE: return DOUBLE;
+      case DECIMAL32:
+      case DECIMAL64:
+      case DECIMAL128:
+        return DECIMAL;
       default:
         throw new IllegalArgumentException("The provided data type doesn't 
map" +
             " to know any known one: " + 
type.getDescriptorForType().getFullName());

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
index 5e7736c..9f5c137 100644
--- 
a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
+++ 
b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
@@ -303,6 +303,7 @@ public final class AsyncKuduScanner {
   private static ColumnSchema getStrippedColumnSchema(ColumnSchema 
columnToClone) {
     return new ColumnSchema.ColumnSchemaBuilder(columnToClone.getName(), 
columnToClone.getType())
         .nullable(columnToClone.isNullable())
+        .typeAttributes(columnToClone.getTypeAttributes())
         .build();
   }
 

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/Bytes.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/Bytes.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/Bytes.java
index d5a5889..11b8849 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/Bytes.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/Bytes.java
@@ -32,6 +32,7 @@ import java.io.IOException;
 import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
@@ -43,6 +44,7 @@ import org.apache.yetus.audience.InterfaceAudience;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.util.CharsetUtil;
 
+import org.apache.kudu.util.DecimalUtil;
 import org.apache.kudu.util.Slice;
 
 /**
@@ -57,6 +59,11 @@ public final class Bytes {
   // biginteger-java
   private static final BigInteger TWO_COMPL_REF = BigInteger.ONE.shiftLeft(64);
 
+  private static final BigInteger BIGINT32_MAX = 
BigInteger.valueOf(Integer.MAX_VALUE);
+  private static final BigInteger BIGINT32_MIN = 
BigInteger.valueOf(Integer.MIN_VALUE);
+  private static final BigInteger BIGINT64_MAX = 
BigInteger.valueOf(Long.MAX_VALUE);
+  private static final BigInteger BIGINT64_MIN = 
BigInteger.valueOf(Long.MIN_VALUE);
+
   private Bytes() {  // Can't instantiate.
   }
 
@@ -472,7 +479,7 @@ public final class Bytes {
    */
   public static BigInteger getUnsignedLong(final byte[] b, final int offset) {
     long l = getLong(b, offset);
-    BigInteger bi = new BigInteger(l + "");
+    BigInteger bi = BigInteger.valueOf(l);
     if (bi.compareTo(BigInteger.ZERO) < 0) {
       bi = bi.add(TWO_COMPL_REF);
     }
@@ -579,6 +586,89 @@ public final class Bytes {
   }
 
   /**
+   * Reads a little-endian 16-byte integer from the beginning of the given 
array.
+   * @param b The array to read from.
+   * @return A BigInteger.
+   * @throws IndexOutOfBoundsException if the byte array is too small.
+   */
+  public static BigInteger getBigInteger(final byte[] b) {
+    return getBigInteger(b, 0);
+  }
+
+  /**
+   * Reads a little-endian 16-byte integer from an offset in the given array.
+   * @param b The array to read from.
+   * @param offset The offset in the array to start reading from.
+   * @return A BigInteger.
+   * @throws IndexOutOfBoundsException if the byte array is too small.
+   */
+  public static BigInteger getBigInteger(final byte[] b, final int offset) {
+    // TODO: Support larger/smaller than 16 bytes (int128)
+    byte[] bytes = Arrays.copyOfRange(b, offset, offset + 16);
+    // BigInteger expects big-endian order.
+    reverseBytes(bytes);
+    return new BigInteger(bytes);
+  }
+
+  /**
+   * Writes a little-endian 16-byte BigInteger at the beginning of the given 
array.
+   * @param b The array to write to.
+   * @param n A BigInteger.
+   * @throws IndexOutOfBoundsException if the byte array is too small.
+   */
+  public static void setBigInteger(final byte[] b, final BigInteger n) {
+    setBigInteger(b, n, 0);
+  }
+
+  /**
+   * Writes a little-endian 16-byte BigInteger at an offset in the given array.
+   * @param b The zeroed byte array to write to.
+   * @param n A BigInteger.
+   * @param offset The offset in the array to start writing at.
+   * @throws IndexOutOfBoundsException if the byte array is too small.
+   */
+  public static void setBigInteger(final byte[] b, final BigInteger n, final 
int offset) {
+    byte[] bytes = n.toByteArray();
+    // TODO: Support larger/smaller than 16 bytes (int128)
+    // Guard against values that are too large.
+    if (bytes.length > 16) {
+      throw new IllegalArgumentException("Value is larger than the maximum 16 
bytes: " + n);
+    }
+    // BigInteger is big-endian order.
+    reverseBytes(bytes);
+    System.arraycopy(bytes, 0, b, offset, bytes.length);
+    // If the value is negative trail with set bits.
+    if (n.compareTo(BigInteger.ZERO) < 0) {
+      Arrays.fill(b, offset + bytes.length, offset + 16, (byte) 0xff);
+    }
+  }
+
+  /**
+   * Creates a new byte array containing a little-endian 16-byte BigInteger.
+   * @param n A BigInteger.
+   * @return A new byte array containing the given value.
+   */
+  public static byte[] fromBigInteger(final BigInteger n) {
+    // TODO: Support larger/smaller than 16 bytes (int128)
+    final byte[] b = new byte[16];
+    setBigInteger(b, n);
+    return b;
+  }
+
+  /**
+   * Reverses the passed byte array in place.
+   * @param b The array to reverse.
+   */
+  private static void reverseBytes(final byte[] b) {
+    // Swaps the items until the mid-point is reached.
+    for(int i = 0; i < b.length / 2; i++) {
+      byte temp = b[i];
+      b[i] = b[b.length - i - 1];
+      b[b.length - i - 1] = temp;
+    }
+  }
+
+  /**
    * Reads a little-endian 4-byte float from the beginning of the given array.
    * @param b The array to read from.
    * @return a float
@@ -684,6 +774,110 @@ public final class Bytes {
     return b;
   }
 
+  /**
+   * Reads a decimal from the beginning of the given array.
+   * @param b The array to read from.
+   * @param precision The precision of the decimal value.
+   * @return A BigDecimal.
+   * @throws IndexOutOfBoundsException if the byte array is too small.
+   */
+  public static BigDecimal getDecimal(final byte[] b, int precision, int 
scale) {
+    return getDecimal(b, 0, precision, scale);
+  }
+
+  /**
+   * Reads a decimal from the beginning of the given array.
+   * @param b The array to read from.
+   * @param offset The offset in the array to start reading from.
+   * @param precision The precision of the decimal value.
+   * @return A BigDecimal.
+   * @throws IndexOutOfBoundsException if the byte array is too small.
+   */
+  public static BigDecimal getDecimal(final byte[] b, final int offset, int 
precision, int scale) {
+    int size = DecimalUtil.precisionToSize(precision);
+    switch (size) {
+      case  DecimalUtil.DECIMAL32_SIZE:
+        int intVal = getInt(b, offset);
+        return BigDecimal.valueOf(intVal, scale);
+      case DecimalUtil.DECIMAL64_SIZE:
+        long longVal = getLong(b, offset);
+        return BigDecimal.valueOf(longVal, scale);
+      case DecimalUtil.DECIMAL128_SIZE:
+         BigInteger int128Val = getBigInteger(b, offset);
+        return new BigDecimal(int128Val, scale);
+      default:
+        throw new IllegalArgumentException("Unsupported decimal type size: " + 
size);
+    }
+  }
+
+  /**
+   * Writes a BigDecimal at the beginning of the given array.
+   *
+   * @param b The array to write to.
+   * @param n A BigDecimal.
+   * @param precision The target precision of the decimal value.
+   * @throws IndexOutOfBoundsException if the byte array is too small.
+   */
+  public static void setBigDecimal(final byte[] b,  final BigDecimal n, int 
precision) {
+    setBigDecimal(b, n, precision, 0);
+  }
+
+  /**
+   * Writes a BigDecimal at an offset in the given array.
+   * @param b The array to write to.
+   * @param n A BigDecimal.
+   * @param precision The target precision of the decimal value.
+   * @param offset The offset in the array to start writing at.
+   * @throws IndexOutOfBoundsException if the byte array is too small.
+   */
+  public static void setBigDecimal(final byte[] b, final BigDecimal n, int 
precision, final int offset) {
+    int size = DecimalUtil.precisionToSize(precision);
+    BigInteger bigInt = n.unscaledValue();
+    switch (size) {
+      case  DecimalUtil.DECIMAL32_SIZE:
+        // TODO: use n.unscaledValue().intValueExact() when we drop Java7 
support.
+        if (bigInt.compareTo(BIGINT32_MIN) >= 0 && 
bigInt.compareTo(BIGINT32_MAX) <= 0) {
+          setInt(b, bigInt.intValue(), offset);
+        } else {
+          throw new ArithmeticException("BigInteger out of int range");
+        }
+        break;
+      case DecimalUtil.DECIMAL64_SIZE:
+        // TODO: use n.unscaledValue().intValueExact() when we drop Java7 
support.
+        if (bigInt.compareTo(BIGINT64_MIN) >= 0 && 
bigInt.compareTo(BIGINT64_MAX) <= 0) {
+          setLong(b, bigInt.longValue(), offset);
+        } else {
+          throw new ArithmeticException("BigInteger out of int range");
+        }
+        break;
+      case DecimalUtil.DECIMAL128_SIZE:
+        setBigInteger(b, bigInt, offset);
+        break;
+      default:
+        throw new IllegalArgumentException("Unsupported decimal type size: " + 
size);
+    }
+  }
+
+  /**
+   * Creates a new byte array containing a little-endian BigDecimal.
+   * @param n A BigDecimal.
+   * @param precision The target precision of the decimal value.
+   * @return A new byte array containing the given value.
+   */
+  public static byte[] fromBigDecimal(final BigDecimal n, int precision) {
+    int size = DecimalUtil.precisionToSize(precision);
+    switch (size) {
+      case  DecimalUtil.DECIMAL32_SIZE:
+        return fromInt(n.unscaledValue().intValue());
+      case DecimalUtil.DECIMAL64_SIZE:
+        return fromLong(n.unscaledValue().longValue());
+      case DecimalUtil.DECIMAL128_SIZE:
+        return fromBigInteger(n.unscaledValue());
+      default:
+        throw new IllegalArgumentException("Unsupported decimal type size: " + 
size);
+    }
+  }
+
   // CHECKSTYLE:OFF
   /** Transforms a string into an UTF-8 encoded byte array.  */
   public static byte[] UTF8(final String s) {

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/ColumnRangePredicate.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/ColumnRangePredicate.java
 
b/java/kudu-client/src/main/java/org/apache/kudu/client/ColumnRangePredicate.java
index b2fa774..6a9fbef 100644
--- 
a/java/kudu-client/src/main/java/org/apache/kudu/client/ColumnRangePredicate.java
+++ 
b/java/kudu-client/src/main/java/org/apache/kudu/client/ColumnRangePredicate.java
@@ -17,6 +17,7 @@
 
 package org.apache.kudu.client;
 
+import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -26,8 +27,10 @@ import com.google.protobuf.UnsafeByteOperations;
 import org.apache.yetus.audience.InterfaceAudience;
 
 import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.ColumnTypeAttributes;
 import org.apache.kudu.Type;
 import org.apache.kudu.tserver.Tserver;
+import org.apache.kudu.util.DecimalUtil;
 
 /**
  * A range predicate on one of the columns in the underlying data.
@@ -76,7 +79,7 @@ public class ColumnRangePredicate {
     if (bound == null) {
       return null;
     }
-    switch (column.getType().getDataType()) {
+    switch (column.getType().getDataType(column.getTypeAttributes())) {
       case BOOL:
         return KuduPredicate.newComparisonPredicate(column, op, 
Bytes.getBoolean(bound));
       case INT8:
@@ -96,6 +99,12 @@ public class ColumnRangePredicate {
         return KuduPredicate.newComparisonPredicate(column, op, 
Bytes.getString(bound));
       case BINARY:
         return KuduPredicate.newComparisonPredicate(column, op, bound);
+      case DECIMAL32:
+      case DECIMAL64:
+      case DECIMAL128:
+        ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+        return KuduPredicate.newComparisonPredicate(column, op,
+            Bytes.getDecimal(bound, typeAttributes.getPrecision(), 
typeAttributes.getScale()));
       default:
         throw new IllegalStateException(String.format("unknown column type 
%s", column.getType()));
     }
@@ -206,6 +215,18 @@ public class ColumnRangePredicate {
   }
 
   /**
+   * Set a BigDecimal for the lower bound
+   * @param lowerBound value for the lower bound
+   */
+  public void setLowerBound(BigDecimal lowerBound) {
+    checkColumn(Type.DECIMAL);
+    int precision = column.getTypeAttributes().getPrecision();
+    int scale = column.getTypeAttributes().getScale();
+    BigDecimal coercedVal = DecimalUtil.coerce(lowerBound, precision, scale);
+    setLowerBoundInternal(Bytes.fromBigDecimal(coercedVal, precision));
+  }
+
+  /**
    * Set a boolean for the upper bound
    * @param upperBound value for the upper bound
    */
@@ -291,6 +312,18 @@ public class ColumnRangePredicate {
   }
 
   /**
+   * Set a BigDecimal for the upper bound
+   * @param upperBound value for the upper bound
+   */
+  public void setUpperBound(BigDecimal upperBound) {
+    checkColumn(Type.DECIMAL);
+    int precision = column.getTypeAttributes().getPrecision();
+    int scale = column.getTypeAttributes().getScale();
+    BigDecimal coercedVal = DecimalUtil.coerce(upperBound, precision, scale);
+    setUpperBoundInternal(Bytes.fromBigDecimal(coercedVal, precision));
+  }
+
+  /**
    * Get the column used by this predicate
    * @return the column
    */

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/KeyEncoder.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/KeyEncoder.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/KeyEncoder.java
index e58876e..724a270 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KeyEncoder.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KeyEncoder.java
@@ -17,6 +17,8 @@
 
 package org.apache.kudu.client;
 
+import java.math.BigDecimal;
+import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.ArrayList;
@@ -34,6 +36,7 @@ import org.apache.kudu.Schema;
 import org.apache.kudu.Type;
 import org.apache.kudu.client.PartitionSchema.HashBucketSchema;
 import org.apache.kudu.util.ByteVec;
+import org.apache.kudu.util.DecimalUtil;
 import org.apache.kudu.util.Pair;
 
 /**
@@ -42,6 +45,8 @@ import org.apache.kudu.util.Pair;
 @InterfaceAudience.Private
 class KeyEncoder {
 
+  private static final BigInteger MIN_VALUE_128 = 
BigInteger.valueOf(-2).pow(127);
+
   /** Non-constructable utility class. */
   private KeyEncoder() {
   }
@@ -139,13 +144,12 @@ class KeyEncoder {
                                                     column.getName()));
     }
     final Type type = column.getType();
-
     if (type == Type.STRING || type == Type.BINARY) {
       encodeBinary(row.getVarLengthData().get(columnIdx), isLast, buf);
     } else {
       encodeSignedInt(row.getRowAlloc(),
                       schema.getColumnOffset(columnIdx),
-                      type.getSize(),
+                      column.getTypeSize(),
                       buf);
     }
   }
@@ -313,7 +317,8 @@ class KeyEncoder {
    */
   private static void decodeColumn(ByteBuffer buf, PartialRow row, int idx, 
boolean isLast) {
     Schema schema = row.getSchema();
-    switch (schema.getColumnByIndex(idx).getType()) {
+    ColumnSchema column = schema.getColumnByIndex(idx);
+    switch (column.getType()) {
       case INT8:
         row.addByte(idx, (byte) (buf.get() ^ Byte.MIN_VALUE));
         break;
@@ -337,6 +342,29 @@ class KeyEncoder {
         row.addStringUtf8(idx, binary);
         break;
       }
+      case DECIMAL: {
+        int scale = column.getTypeAttributes().getScale();
+        int size = column.getTypeSize();
+        switch (size) {
+          case  DecimalUtil.DECIMAL32_SIZE:
+            int intVal = buf.getInt() ^ Integer.MIN_VALUE;
+            row.addDecimal(idx, BigDecimal.valueOf(intVal, scale));
+            break;
+          case DecimalUtil.DECIMAL64_SIZE:
+            long longVal = buf.getLong() ^ Long.MIN_VALUE;
+            row.addDecimal(idx, BigDecimal.valueOf(longVal, scale));
+            break;
+          case DecimalUtil.DECIMAL128_SIZE:
+            byte[] bytes = new byte[size];
+            buf.get(bytes);
+            BigInteger bigIntVal = new BigInteger(bytes).xor(MIN_VALUE_128);
+            row.addDecimal(idx, new BigDecimal(bigIntVal, scale));
+            break;
+          default:
+            throw new IllegalArgumentException("Unsupported decimal type size: 
" + size);
+        }
+        break;
+      }
       default:
         throw new IllegalArgumentException(String.format(
             "The column type %s is not a valid key component type",

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPredicate.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPredicate.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPredicate.java
index b346664..19a73bd 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPredicate.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPredicate.java
@@ -17,6 +17,8 @@
 
 package org.apache.kudu.client;
 
+import java.math.BigInteger;
+import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -35,9 +37,11 @@ import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.yetus.audience.InterfaceStability;
 
 import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.ColumnTypeAttributes;
 import org.apache.kudu.Common;
 import org.apache.kudu.Schema;
 import org.apache.kudu.Type;
+import org.apache.kudu.util.DecimalUtil;
 
 /**
  * A predicate which can be used to filter rows based on the value of a column.
@@ -227,6 +231,68 @@ public class KuduPredicate {
   }
 
   /**
+   * Creates a new comparison predicate on a Decimal column.
+   * @param column the column schema
+   * @param op the comparison operation
+   * @param value the value to compare against
+   */
+  public static KuduPredicate newComparisonPredicate(ColumnSchema column,
+                                                     ComparisonOp op,
+                                                     BigDecimal value) {
+    checkColumn(column, Type.DECIMAL);
+    ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+    int precision = typeAttributes.getPrecision();
+    int scale = typeAttributes.getScale();
+
+    BigDecimal minValue = DecimalUtil.minValue(precision, scale);
+    BigDecimal maxValue = DecimalUtil.maxValue(precision, scale);
+    Preconditions.checkArgument(value.compareTo(maxValue) <= 0 && 
value.compareTo(minValue) >= 0,
+        "Decimal value out of range for %s column: %s",
+        column.getType(), value);
+    BigDecimal smallestValue = DecimalUtil.smallestValue(scale);
+
+    if (op == ComparisonOp.LESS_EQUAL) {
+      if (value.equals(maxValue)) {
+        // If the value can't be incremented because it is at the top end of 
the
+        // range, then substitute the predicate with an IS NOT NULL predicate.
+        // This has the same effect as an inclusive upper bound on the maximum
+        // value. If the column is not nullable then the IS NOT NULL predicate
+        // is ignored.
+        return newIsNotNullPredicate(column);
+      }
+      value = value.add(smallestValue);
+      op = ComparisonOp.LESS;
+    } else if (op == ComparisonOp.GREATER) {
+      if (value == maxValue) {
+        return none(column);
+      }
+      value = value.add(smallestValue);
+      op = ComparisonOp.GREATER_EQUAL;
+    }
+
+    byte[] bytes = Bytes.fromBigDecimal(value, precision);
+
+    switch (op) {
+      case GREATER_EQUAL:
+        if (value.equals(minValue)) {
+          return newIsNotNullPredicate(column);
+        } else if (value.equals(maxValue)) {
+          return new KuduPredicate(PredicateType.EQUALITY, column, bytes, 
null);
+        }
+        return new KuduPredicate(PredicateType.RANGE, column, bytes, null);
+      case EQUAL:
+        return new KuduPredicate(PredicateType.EQUALITY, column, bytes, null);
+      case LESS:
+        if (value.equals(minValue)) {
+          return none(column);
+        }
+        return new KuduPredicate(PredicateType.RANGE, column, null, bytes);
+      default:
+        throw new RuntimeException("unknown comparison op");
+    }
+  }
+
+  /**
    * Creates a new comparison predicate on a float column.
    * @param column the column schema
    * @param op the comparison operation
@@ -449,6 +515,12 @@ public class KuduPredicate {
       for (T value : values) {
         vals.add(Bytes.fromDouble((Double) value));
       }
+    } else if (t instanceof BigDecimal) {
+        checkColumn(column, Type.DECIMAL);
+        for (T value : values) {
+          vals.add(Bytes.fromBigDecimal((BigDecimal) value,
+              column.getTypeAttributes().getPrecision()));
+        }
     } else if (t instanceof String) {
       checkColumn(column, Type.STRING);
       for (T value : values) {
@@ -647,7 +719,8 @@ public class KuduPredicate {
    */
   private static KuduPredicate buildInList(ColumnSchema column, 
Collection<byte[]> values) {
     // IN (true, false) predicates can be simplified to IS NOT NULL.
-    if (column.getType().getDataType() == Common.DataType.BOOL && 
values.size() > 1) {
+    if (column.getType().getDataType(column.getTypeAttributes()) ==
+        Common.DataType.BOOL && values.size() > 1) {
       return newIsNotNullPredicate(column);
     }
 
@@ -785,7 +858,7 @@ public class KuduPredicate {
    * @return the comparison of the serialized values based on the column type
    */
   private static int compare(ColumnSchema column, byte[] a, byte[] b) {
-    switch (column.getType().getDataType()) {
+    switch (column.getType().getDataType(column.getTypeAttributes())) {
       case BOOL:
         return Boolean.compare(Bytes.getBoolean(a), Bytes.getBoolean(b));
       case INT8:
@@ -793,9 +866,11 @@ public class KuduPredicate {
       case INT16:
         return Short.compare(Bytes.getShort(a), Bytes.getShort(b));
       case INT32:
+      case DECIMAL32:
         return Integer.compare(Bytes.getInt(a), Bytes.getInt(b));
       case INT64:
       case UNIXTIME_MICROS:
+      case DECIMAL64:
         return Long.compare(Bytes.getLong(a), Bytes.getLong(b));
       case FLOAT:
         return Float.compare(Bytes.getFloat(a), Bytes.getFloat(b));
@@ -804,6 +879,8 @@ public class KuduPredicate {
       case STRING:
       case BINARY:
         return UnsignedBytes.lexicographicalComparator().compare(a, b);
+      case DECIMAL128:
+        return Bytes.getBigInteger(a).compareTo(Bytes.getBigInteger(b));
       default:
         throw new IllegalStateException(String.format("unknown column type 
%s", column.getType()));
     }
@@ -816,7 +893,7 @@ public class KuduPredicate {
    * @return true if increment(a) == b
    */
   private boolean areConsecutive(byte[] a, byte[] b) {
-    switch (column.getType().getDataType()) {
+    switch (column.getType().getDataType(column.getTypeAttributes())) {
       case BOOL: return false;
       case INT8: {
         byte m = Bytes.getByte(a);
@@ -828,13 +905,15 @@ public class KuduPredicate {
         short n = Bytes.getShort(b);
         return m < n && m + 1 == n;
       }
-      case INT32: {
+      case INT32:
+      case DECIMAL32:{
         int m = Bytes.getInt(a);
         int n = Bytes.getInt(b);
         return m < n && m + 1 == n;
       }
       case INT64:
-      case UNIXTIME_MICROS: {
+      case UNIXTIME_MICROS:
+      case DECIMAL64:  {
         long m = Bytes.getLong(a);
         long n = Bytes.getLong(b);
         return m < n && m + 1 == n;
@@ -861,6 +940,11 @@ public class KuduPredicate {
         }
         return true;
       }
+      case DECIMAL128: {
+        BigInteger m = Bytes.getBigInteger(a);
+        BigInteger n = Bytes.getBigInteger(b);
+        return m.compareTo(n) < 0 && m.add(BigInteger.ONE).equals(n);
+      }
       default:
         throw new IllegalStateException(String.format("unknown column type 
%s", column.getType()));
     }
@@ -953,7 +1037,7 @@ public class KuduPredicate {
    * @return the text representation of the value
    */
   private String valueToString(byte[] value) {
-    switch (column.getType().getDataType()) {
+    switch (column.getType().getDataType(column.getTypeAttributes())) {
       case BOOL: return Boolean.toString(Bytes.getBoolean(value));
       case INT8: return Byte.toString(Bytes.getByte(value));
       case INT16: return Short.toString(Bytes.getShort(value));
@@ -971,6 +1055,12 @@ public class KuduPredicate {
         return sb.toString();
       }
       case BINARY: return Bytes.hex(value);
+      case DECIMAL32:
+      case DECIMAL64:
+      case DECIMAL128:
+       ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+       return Bytes.getDecimal(value, typeAttributes.getPrecision(),
+           typeAttributes.getScale()).toString();
       default:
         throw new IllegalStateException(String.format("unknown column type 
%s", column.getType()));
     }

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
index b85c37f..0f520f9 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
@@ -168,7 +168,7 @@ public class KuduScanToken implements 
Comparable<KuduScanToken> {
     for (Common.ColumnSchemaPB column : message.getProjectedColumnsList()) {
       int columnIdx = table.getSchema().getColumnIndex(column.getName());
       ColumnSchema schema = table.getSchema().getColumnByIndex(columnIdx);
-      if (column.getType() != schema.getType().getDataType()) {
+      if (column.getType() != 
schema.getType().getDataType(schema.getTypeAttributes())) {
         throw new IllegalStateException(String.format(
             "invalid type %s for column '%s' in scan token, expected: %s",
             column.getType().name(), column.getName(), 
schema.getType().name()));

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
index 4cd38f3..b6ca888 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
@@ -303,10 +303,10 @@ public abstract class Operation extends 
KuduRpc<OperationResponse> {
             indirectWrittenBytes += bbSize;
           } else {
             // This is for cols other than strings
-            rows.put(rowData, currentRowOffset, col.getType().getSize());
+            rows.put(rowData, currentRowOffset, col.getTypeSize());
           }
         }
-        currentRowOffset += col.getType().getSize();
+        currentRowOffset += col.getTypeSize();
         colIdx++;
       }
     }

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java
index 4b1f965..10b8642 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java
@@ -17,6 +17,7 @@
 
 package org.apache.kudu.client;
 
+import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -31,8 +32,10 @@ import org.apache.yetus.audience.InterfaceStability;
 import org.jboss.netty.util.CharsetUtil;
 
 import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.ColumnTypeAttributes;
 import org.apache.kudu.Schema;
 import org.apache.kudu.Type;
+import org.apache.kudu.util.DecimalUtil;
 import org.apache.kudu.util.StringUtil;
 
 /**
@@ -479,6 +482,69 @@ public class PartialRow {
   }
 
   /**
+   * Add a Decimal for the specified column.
+   * @param columnIndex the column's index in the schema
+   * @param val value to add
+   * @throws IllegalArgumentException if the value doesn't match the column's 
type
+   * @throws IllegalStateException if the row was already applied
+   * @throws IndexOutOfBoundsException if the column doesn't exist
+   */
+  public void addDecimal(int columnIndex, BigDecimal val) {
+    checkNotFrozen();
+    ColumnSchema column = schema.getColumnByIndex(columnIndex);
+    ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+    checkColumn(column, Type.DECIMAL);
+    BigDecimal coercedVal = 
DecimalUtil.coerce(val,typeAttributes.getPrecision(),
+        typeAttributes.getScale());
+    Bytes.setBigDecimal(rowAlloc, coercedVal, typeAttributes.getPrecision(),
+        getPositionInRowAllocAndSetBitSet(columnIndex));
+  }
+
+  /**
+   * Add a Decimal for the specified column.
+   *
+   * @param columnName Name of the column
+   * @param val value to add
+   * @throws IllegalArgumentException if the column doesn't exist
+   * or if the value doesn't match the column's type
+   * @throws IllegalStateException if the row was already applied
+   */
+  public void addDecimal(String columnName, BigDecimal val) {
+    addDecimal(schema.getColumnIndex(columnName), val);
+  }
+
+  /**
+   * Get the specified column's BigDecimal
+   *
+   * @param columnName name of the column to get data for
+   * @return a BigDecimal
+   * @throws IllegalArgumentException if the column doesn't exist,
+   * is null, is unset, or the type doesn't match the column's type
+   */
+  public BigDecimal getDecimal(String columnName) {
+    return getDecimal(this.schema.getColumnIndex(columnName));
+  }
+
+  /**
+   * Get the specified column's Decimal.
+   *
+   * @param columnIndex Column index in the schema
+   * @return a BigDecimal
+   * @throws IllegalArgumentException if the column is null, is unset,
+   * or if the type doesn't match the column's type
+   * @throws IndexOutOfBoundsException if the column doesn't exist
+   */
+  public BigDecimal getDecimal(int columnIndex) {
+    checkColumn(schema.getColumnByIndex(columnIndex), Type.DECIMAL);
+    checkColumnExists(schema.getColumnByIndex(columnIndex));
+    checkValue(columnIndex);
+    ColumnSchema column = schema.getColumnByIndex(columnIndex);
+    ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+    return Bytes.getDecimal(rowAlloc, schema.getColumnOffset(columnIndex),
+        typeAttributes.getPrecision(), typeAttributes.getScale());
+  }
+
+  /**
    * Add a String for the specified column.
    * @param columnIndex the column's index in the schema
    * @param val value to add
@@ -878,6 +944,9 @@ public class PartialRow {
 
       ColumnSchema col = schema.getColumnByIndex(idx);
       sb.append(col.getType().getName());
+      if (col.getTypeAttributes() != null) {
+        sb.append(col.getTypeAttributes().toStringForType(col.getType()));
+      }
       sb.append(' ');
       sb.append(col.getName());
       sb.append('=');
@@ -992,6 +1061,11 @@ public class PartialRow {
       case DOUBLE:
         sb.append(Bytes.getDouble(rowAlloc, schema.getColumnOffset(idx)));
         return;
+      case DECIMAL:
+        ColumnTypeAttributes typeAttributes = col.getTypeAttributes();
+        sb.append(Bytes.getDecimal(rowAlloc, schema.getColumnOffset(idx),
+            typeAttributes.getPrecision(), typeAttributes.getScale()));
+        return;
       case BINARY:
       case STRING:
         ByteBuffer value = getVarLengthData().get(idx).duplicate();
@@ -1016,7 +1090,8 @@ public class PartialRow {
    * @param index the index of the column to set to the minimum
    */
   void setMin(int index) {
-    Type type = schema.getColumnByIndex(index).getType();
+    ColumnSchema column = schema.getColumnByIndex(index);
+    Type type = column.getType();
     switch (type) {
       case BOOL:
         addBoolean(index, false);
@@ -1040,6 +1115,10 @@ public class PartialRow {
       case DOUBLE:
         addDouble(index, -Double.MAX_VALUE);
         break;
+      case DECIMAL:
+        ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+        addDecimal(index,
+            DecimalUtil.minValue(typeAttributes.getPrecision(), 
typeAttributes.getScale()));
       case STRING:
         addStringUtf8(index, AsyncKuduClient.EMPTY_ARRAY);
         break;
@@ -1057,7 +1136,8 @@ public class PartialRow {
    * @param value the raw value
    */
   void setRaw(int index, byte[] value) {
-    Type type = schema.getColumnByIndex(index).getType();
+    ColumnSchema column = schema.getColumnByIndex(index);
+    Type type = column.getType();
     switch (type) {
       case BOOL:
       case INT8:
@@ -1066,8 +1146,9 @@ public class PartialRow {
       case INT64:
       case UNIXTIME_MICROS:
       case FLOAT:
-      case DOUBLE: {
-        Preconditions.checkArgument(value.length == type.getSize());
+      case DOUBLE:
+      case DECIMAL: {
+        Preconditions.checkArgument(value.length == column.getTypeSize());
         System.arraycopy(value, 0, rowAlloc,
             getPositionInRowAllocAndSetBitSet(index), value.length);
         break;
@@ -1091,7 +1172,8 @@ public class PartialRow {
    *         it is already the maximum value
    */
   boolean incrementColumn(int index) {
-    Type type = schema.getColumnByIndex(index).getType();
+    ColumnSchema column = schema.getColumnByIndex(index);
+    Type type = column.getType();
     Preconditions.checkState(isSet(index));
     int offset = schema.getColumnOffset(index);
     switch (type) {
@@ -1143,7 +1225,7 @@ public class PartialRow {
         return true;
       }
       case DOUBLE: {
-        double existing = Bytes.getFloat(rowAlloc, offset);
+        double existing = Bytes.getDouble(rowAlloc, offset);
         double incremented = Math.nextAfter(existing, 
Double.POSITIVE_INFINITY);
         if (existing == incremented) {
           return false;
@@ -1151,6 +1233,18 @@ public class PartialRow {
         Bytes.setDouble(rowAlloc, incremented, offset);
         return true;
       }
+      case DECIMAL: {
+        int precision = column.getTypeAttributes().getPrecision();
+        int scale = column.getTypeAttributes().getScale();
+        BigDecimal existing = Bytes.getDecimal(rowAlloc, offset, precision, 
scale);
+        BigDecimal max = DecimalUtil.maxValue(precision, scale);
+        if (existing.equals(max)) {
+          return false;
+        }
+        BigDecimal smallest = DecimalUtil.smallestValue(scale);
+        Bytes.setBigDecimal(rowAlloc, existing.add(smallest), precision, 
offset);
+        return true;
+      }
       case STRING:
       case BINARY: {
         ByteBuffer data = varLengthData.get(index);
@@ -1212,7 +1306,8 @@ public class PartialRow {
     Preconditions.checkArgument(a.isSet(index));
     Preconditions.checkArgument(b.isSet(index));
 
-    Type type = a.getSchema().getColumnByIndex(index).getType();
+    ColumnSchema column = a.getSchema().getColumnByIndex(index);
+    Type type = column.getType();
     int offset = a.getSchema().getColumnOffset(index);
 
     switch (type) {
@@ -1231,6 +1326,12 @@ public class PartialRow {
         return Bytes.getFloat(a.rowAlloc, offset) == 
Bytes.getFloat(b.rowAlloc, offset);
       case DOUBLE:
         return Bytes.getDouble(a.rowAlloc, offset) == 
Bytes.getDouble(b.rowAlloc, offset);
+      case DECIMAL:
+        ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+        int precision = typeAttributes.getPrecision();
+        int scale = typeAttributes.getScale();
+        return Bytes.getDecimal(a.rowAlloc, offset, precision, scale)
+            .equals(Bytes.getDecimal(b.rowAlloc, offset, precision, scale));
       case STRING:
       case BINARY: {
         ByteBuffer aData = a.varLengthData.get(index).duplicate();
@@ -1272,7 +1373,8 @@ public class PartialRow {
     Preconditions.checkArgument(lower.isSet(index));
     Preconditions.checkArgument(upper.isSet(index));
 
-    Type type = lower.getSchema().getColumnByIndex(index).getType();
+    ColumnSchema column = lower.getSchema().getColumnByIndex(index);
+    Type type = column.getType();
     int offset = lower.getSchema().getColumnOffset(index);
 
     switch (type) {
@@ -1306,6 +1408,15 @@ public class PartialRow {
                Math.nextAfter(val, Double.POSITIVE_INFINITY) ==
                    Bytes.getDouble(upper.rowAlloc, offset);
       }
+      case DECIMAL: {
+        ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+        int precision = typeAttributes.getPrecision();
+        int scale = typeAttributes.getScale();
+        BigDecimal val = Bytes.getDecimal(lower.rowAlloc, offset, precision, 
scale);
+        BigDecimal smallestVal = DecimalUtil.smallestValue(scale);
+        return val.add(smallestVal).equals(
+                Bytes.getDecimal(upper.rowAlloc, offset, precision, scale));
+      }
       case STRING:
       case BINARY: {
         // Check that b is 1 byte bigger than a, the extra byte is 0, and the 
other bytes are equal.

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/ProtobufHelper.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/ProtobufHelper.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/ProtobufHelper.java
index c816e3d..b76194f 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ProtobufHelper.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ProtobufHelper.java
@@ -17,6 +17,7 @@
 
 package org.apache.kudu.client;
 
+import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.ArrayList;
@@ -31,9 +32,11 @@ import com.google.protobuf.UnsafeByteOperations;
 import org.apache.yetus.audience.InterfaceAudience;
 
 import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.ColumnTypeAttributes;
 import org.apache.kudu.Common;
 import org.apache.kudu.Schema;
 import org.apache.kudu.Type;
+import org.apache.kudu.util.DecimalUtil;
 
 @InterfaceAudience.Private
 public class ProtobufHelper {
@@ -68,7 +71,7 @@ public class ProtobufHelper {
                                                  ColumnSchema column) {
     schemaBuilder
         .setName(column.getName())
-        .setType(column.getType().getDataType())
+        .setType(column.getType().getDataType(column.getTypeAttributes()))
         .setIsKey(column.isKey())
         .setIsNullable(column.isNullable())
         .setCfileBlockSize(column.getDesiredBlockSize());
@@ -82,13 +85,31 @@ public class ProtobufHelper {
       schemaBuilder.setReadDefaultValue(UnsafeByteOperations.unsafeWrap(
           objectToWireFormat(column, column.getDefaultValue())));
     }
+    if(column.getTypeAttributes() != null) {
+      schemaBuilder.setTypeAttributes(
+          columnTypeAttributesToPb(Common.ColumnTypeAttributesPB.newBuilder(), 
column));
+    }
     return schemaBuilder.build();
   }
 
+  public static Common.ColumnTypeAttributesPB columnTypeAttributesToPb(
+      Common.ColumnTypeAttributesPB.Builder builder, ColumnSchema column) {
+    ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+    if (typeAttributes.hasPrecision()) {
+      builder.setPrecision(typeAttributes.getPrecision());
+    }
+    if (typeAttributes.hasScale()) {
+      builder.setScale(typeAttributes.getScale());
+    }
+    return builder.build();
+  }
+
   public static ColumnSchema pbToColumnSchema(Common.ColumnSchemaPB pb) {
     Type type = Type.getTypeForDataType(pb.getType());
+    ColumnTypeAttributes typeAttributes = pb.hasTypeAttributes() ?
+        pbToColumnTypeAttributes(pb.getTypeAttributes()) : null;
     Object defaultValue = pb.hasWriteDefaultValue() ?
-        byteStringToObject(type, pb.getWriteDefaultValue()) : null;
+        byteStringToObject(type, typeAttributes, pb.getWriteDefaultValue()) : 
null;
     ColumnSchema.Encoding encoding = 
ColumnSchema.Encoding.valueOf(pb.getEncoding().name());
     ColumnSchema.CompressionAlgorithm compressionAlgorithm =
         ColumnSchema.CompressionAlgorithm.valueOf(pb.getCompression().name());
@@ -100,9 +121,22 @@ public class ProtobufHelper {
                            .encoding(encoding)
                            .compressionAlgorithm(compressionAlgorithm)
                            .desiredBlockSize(desiredBlockSize)
+                           .typeAttributes(typeAttributes)
                            .build();
   }
 
+  public static ColumnTypeAttributes 
pbToColumnTypeAttributes(Common.ColumnTypeAttributesPB pb) {
+    ColumnTypeAttributes.ColumnTypeAttributesBuilder builder =
+        new ColumnTypeAttributes.ColumnTypeAttributesBuilder();
+    if(pb.hasPrecision()) {
+      builder.precision(pb.getPrecision());
+    }
+    if(pb.hasScale()) {
+      builder.scale(pb.getScale());
+    }
+    return builder.build();
+  }
+
   public static Schema pbToSchema(Common.SchemaPB schema) {
     List<ColumnSchema> columns = new ArrayList<>(schema.getColumnsCount());
     List<Integer> columnIds = new ArrayList<>(schema.getColumnsCount());
@@ -205,13 +239,16 @@ public class ProtobufHelper {
         return Bytes.fromFloat((Float) value);
       case DOUBLE:
         return Bytes.fromDouble((Double) value);
+      case DECIMAL:
+        return Bytes.fromBigDecimal((BigDecimal) value, 
col.getTypeAttributes().getPrecision());
       default:
         throw new IllegalArgumentException("The column " + col.getName() + " 
is of type " + col
             .getType() + " which is unknown");
     }
   }
 
-  private static Object byteStringToObject(Type type, ByteString value) {
+  private static Object byteStringToObject(Type type, ColumnTypeAttributes 
typeAttributes,
+                                           ByteString value) {
     ByteBuffer buf = value.asReadOnlyByteBuffer();
     buf.order(ByteOrder.LITTLE_ENDIAN);
     switch (type) {
@@ -234,6 +271,9 @@ public class ProtobufHelper {
         return value.toStringUtf8();
       case BINARY:
         return value.toByteArray();
+      case DECIMAL:
+        return Bytes.getDecimal(value.toByteArray(),
+            typeAttributes.getPrecision(), typeAttributes.getScale());
       default:
         throw new IllegalArgumentException("This type is unknown: " + type);
     }
@@ -270,6 +310,8 @@ public class ProtobufHelper {
       bytes = Bytes.fromFloat((Float) value);
     } else if (value instanceof Double) {
       bytes = Bytes.fromDouble((Double) value);
+    } else if (value instanceof BigDecimal) {
+      bytes = Bytes.fromBigDecimal((BigDecimal) value, 
DecimalUtil.MAX_DECIMAL_PRECISION);
     } else {
       throw new IllegalArgumentException("The default value provided for " +
           "column " + colName + " is of class " + value.getClass().getName() +

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/client/RowResult.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/client/RowResult.java 
b/java/kudu-client/src/main/java/org/apache/kudu/client/RowResult.java
index a0ca2b5..12edbf4 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/RowResult.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/RowResult.java
@@ -17,6 +17,7 @@
 
 package org.apache.kudu.client;
 
+import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.text.DateFormat;
 import java.text.FieldPosition;
@@ -29,6 +30,7 @@ import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.yetus.audience.InterfaceStability;
 
 import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.ColumnTypeAttributes;
 import org.apache.kudu.Schema;
 import org.apache.kudu.Type;
 import org.apache.kudu.util.Slice;
@@ -89,7 +91,8 @@ public class RowResult {
     // Pre-compute the columns offsets in rowData for easier lookups later.
     // If the schema has nullables, we also add the offset for the null bitmap 
at the end.
     for (int i = 1; i < columnOffsetsSize; i++) {
-      int previousSize = schema.getColumnByIndex(i - 1).getType().getSize();
+      org.apache.kudu.ColumnSchema column = schema.getColumnByIndex(i - 1);
+      int previousSize = column.getTypeSize();
       columnOffsets[i] = previousSize + currentOffset;
       currentOffset += previousSize;
     }
@@ -325,6 +328,36 @@ public class RowResult {
   }
 
   /**
+   * Get the specified column's Decimal.
+   *
+   * @param columnName name of the column to get data for
+   * @return a BigDecimal
+   * @throws IllegalArgumentException if the column doesn't exist or is null
+   */
+  public BigDecimal getDecimal(String columnName) {
+    return getDecimal(this.schema.getColumnIndex(columnName));
+  }
+
+  /**
+   * Get the specified column's Decimal.
+   *
+   * @param columnIndex Column index in the schema
+   * @return a BigDecimal.
+   * @throws IllegalArgumentException if the column is null
+   * @throws IndexOutOfBoundsException if the column doesn't exist
+   */
+  public BigDecimal getDecimal(int columnIndex) {
+    checkValidColumn(columnIndex);
+    checkNull(columnIndex);
+    checkType(columnIndex, Type.DECIMAL);
+    ColumnSchema column = schema.getColumnByIndex(columnIndex);
+    ColumnTypeAttributes typeAttributes = column.getTypeAttributes();
+    return Bytes.getDecimal(this.rowData.getRawArray(),
+        this.rowData.getRawOffset() + 
getCurrentRowDataOffsetForColumn(columnIndex),
+            typeAttributes.getPrecision(), typeAttributes.getScale());
+  }
+
+  /**
    * Get the schema used for this scanner's column projection.
    * @return a column projection as a schema.
    */
@@ -561,8 +594,13 @@ public class RowResult {
       if (i != 0) {
         buf.append(", ");
       }
-      buf.append(col.getType().name());
-      buf.append(" ").append(col.getName()).append("=");
+      Type type = col.getType();
+      buf.append(type.name());
+      buf.append(" ").append(col.getName());
+      if (col.getTypeAttributes() != null) {
+        buf.append(col.getTypeAttributes().toStringForType(type));
+      }
+      buf.append("=");
       if (isNull(i)) {
         buf.append("NULL");
       } else {
@@ -594,6 +632,9 @@ public class RowResult {
           case DOUBLE:
             buf.append(getDouble(i));
             break;
+          case DECIMAL:
+            buf.append(getDecimal(i));
+            break;
           case BOOL:
             buf.append(getBoolean(i));
             break;

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/main/java/org/apache/kudu/util/DecimalUtil.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/main/java/org/apache/kudu/util/DecimalUtil.java 
b/java/kudu-client/src/main/java/org/apache/kudu/util/DecimalUtil.java
new file mode 100644
index 0000000..ae76481
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/util/DecimalUtil.java
@@ -0,0 +1,152 @@
+// 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.util;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.MathContext;
+import java.math.RoundingMode;
+
+import com.google.common.base.Strings;
+
+import org.apache.kudu.ColumnTypeAttributes;
+
+import static org.apache.kudu.Common.DataType;
+
+public class DecimalUtil {
+  public static final int MAX_DECIMAL32_PRECISION = 9;
+  public static final int MAX_UNSCALED_DECIMAL32 = 999999999;
+  public static final int MIN_UNSCALED_DECIMAL32 = -MAX_UNSCALED_DECIMAL32;
+  public static final int DECIMAL32_SIZE = 32 / Byte.SIZE;
+
+  public static final int MAX_DECIMAL64_PRECISION = 18;
+  public static final long MAX_UNSCALED_DECIMAL64 = 999999999999999999L;
+  public static final long MIN_UNSCALED_DECIMAL64 = -MAX_UNSCALED_DECIMAL64;
+  public static final int DECIMAL64_SIZE = 64 / Byte.SIZE;
+
+  public static final int MAX_DECIMAL128_PRECISION = 38;
+  public static final BigInteger MAX_UNSCALED_DECIMAL128 =
+      new BigInteger(Strings.repeat("9", MAX_DECIMAL128_PRECISION));
+  public static final BigInteger MIN_UNSCALED_DECIMAL128 = 
MAX_UNSCALED_DECIMAL128.negate();
+  public static final int DECIMAL128_SIZE = 128 / Byte.SIZE;
+
+  public static final int MAX_DECIMAL_PRECISION = MAX_DECIMAL128_PRECISION;
+
+  /**
+   * Given a precision, returns the size of the Decimal in Bytes.
+   * @return the size in Bytes.
+   */
+  public static int precisionToSize(int precision) {
+    if (precision <= MAX_DECIMAL32_PRECISION) {
+      return DECIMAL32_SIZE;
+    } else if (precision <= MAX_DECIMAL64_PRECISION) {
+      return DECIMAL64_SIZE;
+    } else if (precision <= MAX_DECIMAL128_PRECISION) {
+      return DECIMAL128_SIZE;
+    } else {
+      throw new IllegalArgumentException("Unsupported decimal type precision: 
" + precision);
+    }
+  }
+
+  /**
+   * Given a precision, returns the smallest unscaled data type.
+   * @return the smallest valid DataType.
+   */
+  public static DataType precisionToDataType(int precision) {
+    if (precision <= MAX_DECIMAL32_PRECISION) {
+      return DataType.DECIMAL32;
+    } else if (precision <= MAX_DECIMAL64_PRECISION) {
+      return DataType.DECIMAL64;
+    } else if (precision <= MAX_DECIMAL128_PRECISION) {
+      return DataType.DECIMAL128;
+    } else {
+      throw new IllegalArgumentException("Unsupported decimal type precision: 
" + precision);
+    }
+  }
+
+  /**
+   * Returns the maximum value of a Decimal give a precision and scale.
+   * @param precision the precision of the decimal.
+   * @param scale the scale of the decimal.
+   * @return the maximum decimal value.
+   */
+  public static BigDecimal maxValue(int precision, int scale) {
+    String maxPrecision = Strings.repeat("9", precision);
+    return new BigDecimal(new BigInteger(maxPrecision), scale);
+  }
+
+  /**
+   * Returns the minimum value of a Decimal give a precision and scale.
+   * @param precision the precision of the decimal.
+   * @param scale the scale of the decimal.
+   * @return the minimum decimal value.
+   */
+  public static BigDecimal minValue(int precision, int scale) {
+    return maxValue(precision, scale).negate();
+  }
+
+  /**
+   * Returns the smallest value of a Decimal give a precision and scale.
+   * This value can be useful for incrementing a Decimal.
+   * @param scale the scale of the decimal.
+   * @return the smallest decimal value.
+   */
+  public static BigDecimal smallestValue(int scale) {
+    return new BigDecimal(BigInteger.ONE, scale);
+  }
+
+  /**
+   * Attempts to coerce a big decimal to a target precision and scale and
+   * returns the result. Throws an {@link IllegalArgumentException} if the 
value
+   * can't be coerced without rounding or exceeding the targetPrecision.
+   *
+   * @param val the BigDecimal value to coerce.
+   * @param targetPrecision the target precision of the coerced value.
+   * @param targetScale the target scale of the coerced value.
+   * @return the coerced BigDecimal value.
+   */
+  public static BigDecimal coerce(BigDecimal val, int targetPrecision, int 
targetScale) {
+    if (val.scale() != targetScale) {
+      try {
+        val = val.setScale(targetScale, BigDecimal.ROUND_UNNECESSARY);
+      } catch (ArithmeticException ex) {
+        throw new IllegalArgumentException("Value scale " + val.scale() +
+            " can't be coerced to target scale " +  targetScale + ". ");
+      }
+    }
+    if (val.precision() > targetPrecision) {
+      throw new IllegalArgumentException("Value precision " + val.precision() +
+          " (after scale coercion) can't be coerced to target precision " +  
targetPrecision + ". ");
+    }
+    return val;
+  }
+
+  /**
+   * Convenience method to create column type attributes for decimal columns.
+   * @param precision the precision.
+   * @param scale the scale.
+   * @return the column type attributes.
+   */
+  public static ColumnTypeAttributes typeAttributes(int precision, int scale) {
+    return new ColumnTypeAttributes.ColumnTypeAttributesBuilder()
+        .precision(precision)
+        .scale(scale)
+        .build();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java 
b/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java
index c1f8564..56b6127 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/BaseKuduTest.java
@@ -38,11 +38,13 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.ColumnTypeAttributes;
 import org.apache.kudu.Common.HostPortPB;
 import org.apache.kudu.Schema;
 import org.apache.kudu.Type;
 import org.apache.kudu.client.LocatedTablet.Replica;
 import org.apache.kudu.master.Master;
+import org.apache.kudu.util.DecimalUtil;
 
 public class BaseKuduTest {
 
@@ -237,7 +239,9 @@ public class BaseKuduTest {
             new ColumnSchema.ColumnSchemaBuilder("binary-array", 
Type.BINARY).build(),
             new ColumnSchema.ColumnSchemaBuilder("binary-bytebuffer", 
Type.BINARY).build(),
             new ColumnSchema.ColumnSchemaBuilder("null", 
Type.STRING).nullable(true).build(),
-            new ColumnSchema.ColumnSchemaBuilder("timestamp", 
Type.UNIXTIME_MICROS).build());
+            new ColumnSchema.ColumnSchemaBuilder("timestamp", 
Type.UNIXTIME_MICROS).build(),
+            new ColumnSchema.ColumnSchemaBuilder("decimal", Type.DECIMAL)
+                .typeAttributes(DecimalUtil.typeAttributes(5, 3)).build());
 
     return new Schema(columns);
   }

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/test/java/org/apache/kudu/client/TestBytes.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/test/java/org/apache/kudu/client/TestBytes.java 
b/java/kudu-client/src/test/java/org/apache/kudu/client/TestBytes.java
index ced6144..6291cf0 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestBytes.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestBytes.java
@@ -18,7 +18,12 @@ package org.apache.kudu.client;
 
 import static org.junit.Assert.assertEquals;
 
+import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.math.MathContext;
+import java.math.RoundingMode;
+
+import org.apache.kudu.util.DecimalUtil;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -27,7 +32,7 @@ public class TestBytes {
 
   @Test
   public void test() {
-    byte[] bytes = new byte[8];
+    byte[] bytes = new byte[16];
 
     // Boolean
     Bytes.setUnsignedByte(bytes, (short) 1);
@@ -94,6 +99,44 @@ public class TestBytes {
     double aDouble = 123.456;
     Bytes.setDouble(bytes, aDouble);
     assertEquals(aDouble, Bytes.getDouble(bytes), 0.001);
+
+    // DECIMAL (32 bits)
+    BigDecimal smallDecimal = new BigDecimal(BigInteger.valueOf(123456789), 0,
+        new MathContext(DecimalUtil.MAX_DECIMAL32_PRECISION, 
RoundingMode.UNNECESSARY));
+    Bytes.setBigDecimal(bytes, smallDecimal, 
DecimalUtil.MAX_DECIMAL32_PRECISION);
+    assertEquals(smallDecimal,
+        Bytes.getDecimal(bytes, 0, DecimalUtil.MAX_DECIMAL32_PRECISION, 0));
+    BigDecimal nSmallDecimal = new BigDecimal(BigInteger.valueOf(-123456789), 
0,
+        new MathContext(DecimalUtil.MAX_DECIMAL32_PRECISION, 
RoundingMode.UNNECESSARY));
+    Bytes.setBigDecimal(bytes, nSmallDecimal, 
DecimalUtil.MAX_DECIMAL32_PRECISION);
+    assertEquals(nSmallDecimal,
+        Bytes.getDecimal(bytes, 0, DecimalUtil.MAX_DECIMAL32_PRECISION, 0));
+
+    // DECIMAL (64 bits)
+    BigDecimal mediumDecimal = new BigDecimal(BigInteger.valueOf(123456789L), 
0,
+        new MathContext(DecimalUtil.MAX_DECIMAL64_PRECISION, 
RoundingMode.UNNECESSARY));
+    Bytes.setBigDecimal(bytes, mediumDecimal, 
DecimalUtil.MAX_DECIMAL64_PRECISION);
+    assertEquals(mediumDecimal,
+        Bytes.getDecimal(bytes, DecimalUtil.MAX_DECIMAL64_PRECISION, 0));
+    BigDecimal nMediumDecimal = new 
BigDecimal(BigInteger.valueOf(-123456789L), 0,
+        new MathContext(DecimalUtil.MAX_DECIMAL64_PRECISION, 
RoundingMode.UNNECESSARY));
+    Bytes.setBigDecimal(bytes, nMediumDecimal, 
DecimalUtil.MAX_DECIMAL64_PRECISION);
+    assertEquals(nMediumDecimal,
+        Bytes.getDecimal(bytes, DecimalUtil.MAX_DECIMAL64_PRECISION, 0));
+
+    // DECIMAL (128 bits)
+    BigDecimal largeDecimal =
+        new BigDecimal(new 
java.math.BigInteger("1234567891011121314151617181920212223"), 0,
+        new MathContext(DecimalUtil.MAX_DECIMAL128_PRECISION, 
RoundingMode.UNNECESSARY));
+    Bytes.setBigDecimal(bytes, largeDecimal, 
DecimalUtil.MAX_DECIMAL128_PRECISION);
+    assertEquals(largeDecimal,
+        Bytes.getDecimal(bytes, DecimalUtil.MAX_DECIMAL128_PRECISION, 0));
+    BigDecimal nLargeDecimal =
+        new BigDecimal(new 
java.math.BigInteger("-1234567891011121314151617181920212223"), 0,
+            new MathContext(DecimalUtil.MAX_DECIMAL128_PRECISION, 
RoundingMode.UNNECESSARY));
+    Bytes.setBigDecimal(bytes, nLargeDecimal, 
DecimalUtil.MAX_DECIMAL128_PRECISION);
+    assertEquals(nLargeDecimal,
+        Bytes.getDecimal(bytes, DecimalUtil.MAX_DECIMAL128_PRECISION, 0));
   }
 
   @Test

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java
 
b/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java
index e6022c1..979ce5d 100644
--- 
a/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java
+++ 
b/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java
@@ -20,12 +20,14 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
 
+import java.math.BigDecimal;
 import java.util.List;
 
 import com.google.common.collect.Lists;
 import org.junit.Test;
 
 import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.ColumnTypeAttributes;
 import org.apache.kudu.Type;
 import org.apache.kudu.tserver.Tserver;
 
@@ -36,6 +38,11 @@ public class TestColumnRangePredicate {
     ColumnSchema col1 = new ColumnSchema.ColumnSchemaBuilder("col1", 
Type.INT32).build();
     ColumnSchema col2 = new ColumnSchema.ColumnSchemaBuilder("col2", 
Type.STRING).build();
 
+    ColumnSchema col3 = new ColumnSchema.ColumnSchemaBuilder("col3", 
Type.DECIMAL)
+        .typeAttributes(new ColumnTypeAttributes.ColumnTypeAttributesBuilder()
+            .precision(6).scale(2).build()
+        ).build();
+
     ColumnRangePredicate pred1 = new ColumnRangePredicate(col1);
     pred1.setLowerBound(1);
 
@@ -46,7 +53,10 @@ public class TestColumnRangePredicate {
     pred3.setLowerBound("aaa");
     pred3.setUpperBound("bbb");
 
-    List<ColumnRangePredicate> preds = Lists.newArrayList(pred1, pred2, pred3);
+    ColumnRangePredicate pred4 = new ColumnRangePredicate(col3);
+    pred4.setLowerBound(BigDecimal.valueOf(12345, 2));
+
+    List<ColumnRangePredicate> preds = Lists.newArrayList(pred1, pred2, pred3, 
pred4);
 
     byte[] rawPreds = ColumnRangePredicate.toByteArray(preds);
 
@@ -57,7 +67,7 @@ public class TestColumnRangePredicate {
       fail("Couldn't decode: " + e.getMessage());
     }
 
-    assertEquals(3, decodedPreds.size());
+    assertEquals(4, decodedPreds.size());
 
     assertEquals(col1.getName(), decodedPreds.get(0).getColumn().getName());
     assertEquals(1, 
Bytes.getInt(decodedPreds.get(0).getLowerBound().toByteArray()));
@@ -70,5 +80,9 @@ public class TestColumnRangePredicate {
     assertEquals(col2.getName(), decodedPreds.get(2).getColumn().getName());
     assertEquals("aaa", 
Bytes.getString(decodedPreds.get(2).getLowerBound().toByteArray()));
     assertEquals("bbb", 
Bytes.getString(decodedPreds.get(2).getInclusiveUpperBound().toByteArray()));
+
+    assertEquals(col3.getName(), decodedPreds.get(3).getColumn().getName());
+    assertEquals(12345, 
Bytes.getInt(decodedPreds.get(3).getLowerBound().toByteArray()));
+    assertFalse(decodedPreds.get(0).hasInclusiveUpperBound());
   }
 }

http://git-wip-us.apache.org/repos/asf/kudu/blob/4f34b69d/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java
----------------------------------------------------------------------
diff --git 
a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java 
b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java
index 0ae3479..6122190 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java
@@ -19,6 +19,7 @@ package org.apache.kudu.client;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -33,6 +34,7 @@ import org.apache.kudu.Schema;
 import org.apache.kudu.Type;
 import org.apache.kudu.client.PartitionSchema.HashBucketSchema;
 import org.apache.kudu.client.PartitionSchema.RangeSchema;
+import org.apache.kudu.util.DecimalUtil;
 
 public class TestKeyEncoding extends BaseKuduTest {
 
@@ -164,6 +166,12 @@ public class TestKeyEncoding extends BaseKuduTest {
         new ColumnSchemaBuilder("int16", Type.INT16).key(true),
         new ColumnSchemaBuilder("int32", Type.INT32).key(true),
         new ColumnSchemaBuilder("int64", Type.INT64).key(true),
+        new ColumnSchemaBuilder("decimal32", Type.DECIMAL).key(true)
+            
.typeAttributes(DecimalUtil.typeAttributes(DecimalUtil.MAX_DECIMAL32_PRECISION, 
0)),
+        new ColumnSchemaBuilder("decimal64", Type.DECIMAL).key(true)
+            
.typeAttributes(DecimalUtil.typeAttributes(DecimalUtil.MAX_DECIMAL64_PRECISION, 
0)),
+        new ColumnSchemaBuilder("decimal128", Type.DECIMAL).key(true)
+            
.typeAttributes(DecimalUtil.typeAttributes(DecimalUtil.MAX_DECIMAL128_PRECISION,
 0)),
         new ColumnSchemaBuilder("string", Type.STRING).key(true),
         new ColumnSchemaBuilder("binary", Type.BINARY).key(true));
 
@@ -172,17 +180,26 @@ public class TestKeyEncoding extends BaseKuduTest {
     rowA.addShort("int16", Short.MIN_VALUE);
     rowA.addInt("int32", Integer.MIN_VALUE);
     rowA.addLong("int64", Long.MIN_VALUE);
+    // Note: The decimal value is not the minimum of the underlying int32, 
int64, int128 type so
+    // we don't use "minimum" values in the test.
+    rowA.addDecimal("decimal32", BigDecimal.valueOf(5));
+    rowA.addDecimal("decimal64", BigDecimal.valueOf(6));
+    rowA.addDecimal("decimal128", BigDecimal.valueOf(7));
     rowA.addString("string", "");
     rowA.addBinary("binary", "".getBytes(Charsets.UTF_8));
 
     byte[] rowAEncoded = rowA.encodePrimaryKey();
     assertBytesEquals(rowAEncoded,
-                      "\0"
-                    + "\0\0"
-                    + "\0\0\0\0"
-                    + "\0\0\0\0\0\0\0\0"
-                    + "\0\0"
-                    + "");
+                      new byte[] {
+                          0,
+                          0, 0,
+                          0, 0, 0, 0,
+                          0, 0, 0, 0, 0, 0, 0, 0,
+                          (byte) 0x80, 0, 0, 5,
+                          (byte) 0x80, 0, 0, 0, 0, 0, 0, 6,
+                          (byte) 0x80, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 
0, 7,
+                          0, 0
+                      });
     assertEquals(rowA.stringifyRowKey(),
                  KeyEncoder.decodePrimaryKey(schema, 
rowAEncoded).stringifyRowKey());
 
@@ -191,6 +208,11 @@ public class TestKeyEncoding extends BaseKuduTest {
     rowB.addShort("int16", Short.MAX_VALUE);
     rowB.addInt("int32", Integer.MAX_VALUE);
     rowB.addLong("int64", Long.MAX_VALUE);
+    // Note: The decimal value is not the maximum of the underlying int32, 
int64, int128 type so
+    // we don't use "minimum" values in the test.
+    rowB.addDecimal("decimal32", BigDecimal.valueOf(5));
+    rowB.addDecimal("decimal64", BigDecimal.valueOf(6));
+    rowB.addDecimal("decimal128", BigDecimal.valueOf(7));
     rowB.addString("string", "abc\1\0def");
     rowB.addBinary("binary", "\0\1binary".getBytes(Charsets.UTF_8));
 
@@ -201,6 +223,9 @@ public class TestKeyEncoding extends BaseKuduTest {
                           -1, -1,
                           -1, -1, -1, -1,
                           -1, -1, -1, -1, -1, -1, -1, -1,
+                          (byte) 0x80, 0, 0, 5,
+                          (byte) 0x80, 0, 0, 0, 0, 0, 0, 6,
+                          (byte) 0x80, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 
0, 7,
                           'a', 'b', 'c', 1, 0, 1, 'd', 'e', 'f', 0, 0,
                           0, 1, 'b', 'i', 'n', 'a', 'r', 'y',
                       });
@@ -212,6 +237,9 @@ public class TestKeyEncoding extends BaseKuduTest {
     rowC.addShort("int16", (short) 2);
     rowC.addInt("int32", 3);
     rowC.addLong("int64", 4);
+    rowC.addDecimal("decimal32", BigDecimal.valueOf(5));
+    rowC.addDecimal("decimal64", BigDecimal.valueOf(6));
+    rowC.addDecimal("decimal128", BigDecimal.valueOf(7));
     rowC.addString("string", "abc\n123");
     rowC.addBinary("binary", "\0\1\2\3\4\5".getBytes(Charsets.UTF_8));
 
@@ -222,6 +250,9 @@ public class TestKeyEncoding extends BaseKuduTest {
                           (byte) 0x80, 2,
                           (byte) 0x80, 0, 0, 3,
                           (byte) 0x80, 0, 0, 0, 0, 0, 0, 4,
+                          (byte) 0x80, 0, 0, 5,
+                          (byte) 0x80, 0, 0, 0, 0, 0, 0, 6,
+                          (byte) 0x80, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 
0, 7,
                           'a', 'b', 'c', '\n', '1', '2', '3', 0, 0,
                           0, 1, 2, 3, 4, 5,
                       });
@@ -233,6 +264,9 @@ public class TestKeyEncoding extends BaseKuduTest {
     rowD.addShort("int16", (short) -2);
     rowD.addInt("int32", -3);
     rowD.addLong("int64", -4);
+    rowD.addDecimal("decimal32", BigDecimal.valueOf(-5));
+    rowD.addDecimal("decimal64", BigDecimal.valueOf(-6));
+    rowD.addDecimal("decimal128", BigDecimal.valueOf(-7));
     rowD.addString("string", "\0abc\n\1\1\0 123\1\0");
     rowD.addBinary("binary", "\0\1\2\3\4\5\0".getBytes(Charsets.UTF_8));
 
@@ -243,6 +277,9 @@ public class TestKeyEncoding extends BaseKuduTest {
                           (byte) 127, -2,
                           (byte) 127, -1, -1, -3,
                           (byte) 127, -1, -1, -1, -1, -1, -1, -4,
+                          (byte) 127, -1, -1, -5,
+                          (byte) 127, -1, -1, -1, -1, -1, -1, -6,
+                          (byte) 127, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 
-1, -1, -1, -1, -7,
                           0, 1, 'a', 'b', 'c', '\n', 1, 1, 0, 1, ' ', '1', 
'2', '3', 1, 0, 1, 0, 0,
                           0, 1, 2, 3, 4, 5, 0,
                       });
@@ -325,6 +362,12 @@ public class TestKeyEncoding extends BaseKuduTest {
         new ColumnSchemaBuilder("string", Type.STRING).key(true),
         new ColumnSchemaBuilder("binary", Type.BINARY).key(true),
         new ColumnSchemaBuilder("timestamp", Type.UNIXTIME_MICROS).key(true),
+        new ColumnSchemaBuilder("decimal32", Type.DECIMAL).key(true)
+            
.typeAttributes(DecimalUtil.typeAttributes(DecimalUtil.MAX_DECIMAL32_PRECISION, 
0)),
+        new ColumnSchemaBuilder("decimal64", Type.DECIMAL).key(true)
+          
.typeAttributes(DecimalUtil.typeAttributes(DecimalUtil.MAX_DECIMAL64_PRECISION, 
0)),
+        new ColumnSchemaBuilder("decimal128", Type.DECIMAL).key(true)
+          
.typeAttributes(DecimalUtil.typeAttributes(DecimalUtil.MAX_DECIMAL128_PRECISION,
 0)),
         new ColumnSchemaBuilder("bool", Type.BOOL),       // not primary key 
type
         new ColumnSchemaBuilder("float", Type.FLOAT),     // not primary key 
type
         new ColumnSchemaBuilder("double", Type.DOUBLE));  // not primary key 
type
@@ -342,9 +385,12 @@ public class TestKeyEncoding extends BaseKuduTest {
     row.addString(4, "foo");
     row.addBinary(5, "bar".getBytes(Charsets.UTF_8));
     row.addLong(6, 6l);
-    row.addBoolean(7, true);
-    row.addFloat(8, 8.8f);
-    row.addDouble(9, 9.9);
+    row.addDecimal(7, BigDecimal.valueOf(DecimalUtil.MAX_UNSCALED_DECIMAL32));
+    row.addDecimal(8, BigDecimal.valueOf(DecimalUtil.MAX_UNSCALED_DECIMAL64));
+    row.addDecimal(9, new BigDecimal(DecimalUtil.MAX_UNSCALED_DECIMAL128));
+    row.addBoolean(10, true);
+    row.addFloat(11, 8.8f);
+    row.addDouble(12, 9.9);
     session.apply(insert);
     session.close();
 
@@ -360,9 +406,15 @@ public class TestKeyEncoding extends BaseKuduTest {
       assertBytesEquals(rr.getBinaryCopy(4), "foo");
       assertBytesEquals(rr.getBinaryCopy(5), "bar");
       assertEquals(6l, rr.getLong(6));
-      assertTrue(rr.getBoolean(7));
-      assertEquals(8.8f, rr.getFloat(8), .001f);
-      assertEquals(9.9, rr.getDouble(9), .001);
+      assertTrue(BigDecimal.valueOf(DecimalUtil.MAX_UNSCALED_DECIMAL32)
+          .compareTo(rr.getDecimal(7)) == 0);
+      assertTrue(BigDecimal.valueOf(DecimalUtil.MAX_UNSCALED_DECIMAL64)
+          .compareTo(rr.getDecimal(8)) == 0);
+      assertTrue(new BigDecimal(DecimalUtil.MAX_UNSCALED_DECIMAL128)
+          .compareTo(rr.getDecimal(9)) == 0);
+      assertTrue(rr.getBoolean(10));
+      assertEquals(8.8f, rr.getFloat(11), .001f);
+      assertEquals(9.9, rr.getDouble(12), .001);
     }
   }
 }

Reply via email to