From 5e7fecc15b366510a69f5479e8af13d1613ccc02 Mon Sep 17 00:00:00 2001
From: stack <stack@apache.org>
Date: Wed, 17 Feb 2016 22:08:24 -0800
Subject: [PATCH] HBASE-15279 OrderedBytes.isEncodedValue does not check for
 int8 and int16 types (Robert Yokota)

---
 .../hadoop/hbase/util/OrderedBytes.java       | 23 +++++++-
 .../hadoop/hbase/util/TestOrderedBytes.java   | 58 +++++++++++++++++++
 2 files changed, 79 insertions(+), 2 deletions(-)

diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/OrderedBytes.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/OrderedBytes.java
index 499e34c3dea..a0c739095dc 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/OrderedBytes.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/OrderedBytes.java
@@ -1503,7 +1503,8 @@ public class OrderedBytes {
    * false otherwise.
    */
   public static boolean isEncodedValue(PositionedByteRange src) {
-    return isNull(src) || isNumeric(src) || isFixedInt32(src) || isFixedInt64(src)
+    return isNull(src) || isNumeric(src) || isFixedInt8(src) || isFixedInt16(src)
+        || isFixedInt32(src) || isFixedInt64(src)
         || isFixedFloat32(src) || isFixedFloat64(src) || isText(src) || isBlobCopy(src)
         || isBlobVar(src);
   }
@@ -1553,6 +1554,24 @@ public class OrderedBytes {
         (-1 == Integer.signum(src.peek()) ? DESCENDING : ASCENDING).apply(src.peek());
   }
 
+  /**
+   * Return true when the next encoded value in {@code src} uses fixed-width
+   * Int8 encoding, false otherwise.
+   */
+  public static boolean isFixedInt8(PositionedByteRange src) {
+    return FIXED_INT8 ==
+        (-1 == Integer.signum(src.peek()) ? DESCENDING : ASCENDING).apply(src.peek());
+  }
+
+  /**
+   * Return true when the next encoded value in {@code src} uses fixed-width
+   * Int16 encoding, false otherwise.
+   */
+  public static boolean isFixedInt16(PositionedByteRange src) {
+    return FIXED_INT16 ==
+        (-1 == Integer.signum(src.peek()) ? DESCENDING : ASCENDING).apply(src.peek());
+  }
+
   /**
    * Return true when the next encoded value in {@code src} uses fixed-width
    * Int32 encoding, false otherwise.
@@ -1734,7 +1753,7 @@ public class OrderedBytes {
         new SimplePositionedMutableByteRange(buff.getBytes(), buff.getOffset(), buff.getLength());
     b.setPosition(buff.getPosition());
     int cnt = 0;
-    for (; isEncodedValue(b); skip(buff), cnt++)
+    for (; isEncodedValue(b); skip(b), cnt++)
       ;
     return cnt;
   }
diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestOrderedBytes.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestOrderedBytes.java
index 37a30e10775..8f213b77cdf 100644
--- a/hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestOrderedBytes.java
+++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestOrderedBytes.java
@@ -1185,4 +1185,62 @@ public class TestOrderedBytes {
       assertEquals(o, OrderedBytes.skip(buff));
     }
   }
+
+  /**
+   * Test encoded value check
+   */
+  @Test
+  public void testEncodedValueCheck() {
+    BigDecimal longMax = BigDecimal.valueOf(Long.MAX_VALUE);
+    double negInf = Double.NEGATIVE_INFINITY;
+    BigDecimal negLarge = longMax.multiply(longMax).negate();
+    BigDecimal negMed = new BigDecimal("-10.0");
+    BigDecimal negSmall = new BigDecimal("-0.0010");
+    long zero = 0l;
+    BigDecimal posSmall = negSmall.negate();
+    BigDecimal posMed = negMed.negate();
+    BigDecimal posLarge = negLarge.negate();
+    double posInf = Double.POSITIVE_INFINITY;
+    double nan = Double.NaN;
+    byte int8 = 100;
+    short int16 = 100;
+    int int32 = 100;
+    long int64 = 100l;
+    float float32 = 100.0f;
+    double float64 = 100.0d;
+    String text = "hello world.";
+    byte[] blobVar = Bytes.toBytes("foo");
+
+    int cnt = 0;
+    PositionedByteRange buff = new SimplePositionedMutableByteRange(1024);
+    for (Order ord : new Order[] { Order.ASCENDING, Order.DESCENDING }) {
+      int o;
+      o = OrderedBytes.encodeNull(buff, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, negInf, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, negLarge, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, negMed, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, negSmall, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, zero, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, posSmall, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, posMed, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, posLarge, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, posInf, ord); cnt++;
+      o = OrderedBytes.encodeNumeric(buff, nan, ord); cnt++;
+      o = OrderedBytes.encodeInt8(buff, int8, ord); cnt++;
+      o = OrderedBytes.encodeInt16(buff, int16, ord); cnt++;
+      o = OrderedBytes.encodeInt32(buff, int32, ord); cnt++;
+      o = OrderedBytes.encodeInt64(buff, int64, ord); cnt++;
+      o = OrderedBytes.encodeFloat32(buff, float32, ord); cnt++;
+      o = OrderedBytes.encodeFloat64(buff, float64, ord); cnt++;
+      o = OrderedBytes.encodeString(buff, text, ord); cnt++;
+      o = OrderedBytes.encodeBlobVar(buff, blobVar, ord); cnt++;
+    }
+
+    buff.setPosition(0);
+    assertEquals(OrderedBytes.length(buff), cnt);
+    for (int i = 0; i < cnt; i++) {
+      assertEquals(OrderedBytes.isEncodedValue(buff), true);
+      OrderedBytes.skip(buff);
+    }
+  }
 }