From 3749158b551e71650fd95d87f1a41c246e3273e4 Mon Sep 17 00:00:00 2001
From: Lukasz Osipiuk <lukasz@osipiuk.net>
Date: Tue, 25 Oct 2016 10:39:36 +0200
Subject: [PATCH] Make RCFile reader independent of internal decimal
 representation

---
 .../presto/rcfile/binary/DecimalEncoding.java | 51 +++++--------------
 .../type/UnscaledDecimal128Arithmetic.java    | 16 ++++--
 2 files changed, 24 insertions(+), 43 deletions(-)

diff --git a/presto-rcfile/src/main/java/com/facebook/presto/rcfile/binary/DecimalEncoding.java b/presto-rcfile/src/main/java/com/facebook/presto/rcfile/binary/DecimalEncoding.java
index 890ef190b7ac..30377265a957 100644
--- a/presto-rcfile/src/main/java/com/facebook/presto/rcfile/binary/DecimalEncoding.java
+++ b/presto-rcfile/src/main/java/com/facebook/presto/rcfile/binary/DecimalEncoding.java
@@ -20,6 +20,7 @@
 import com.facebook.presto.spi.block.BlockBuilder;
 import com.facebook.presto.spi.block.BlockBuilderStatus;
 import com.facebook.presto.spi.type.DecimalType;
+import com.facebook.presto.spi.type.Decimals;
 import com.facebook.presto.spi.type.Type;
 import io.airlift.slice.Slice;
 import io.airlift.slice.SliceOutput;
@@ -30,7 +31,6 @@
 import static com.facebook.presto.rcfile.RcFileDecoderUtils.decodeVIntSize;
 import static com.facebook.presto.rcfile.RcFileDecoderUtils.readVInt;
 import static com.facebook.presto.rcfile.RcFileDecoderUtils.writeVInt;
-import static com.facebook.presto.spi.type.Decimals.encodeUnscaledValue;
 import static com.facebook.presto.spi.type.Decimals.isShortDecimal;
 import static com.facebook.presto.spi.type.Decimals.rescale;
 import static com.google.common.base.Preconditions.checkState;
@@ -159,7 +159,7 @@ private long parseLong(Slice slice, int offset)
 
     private Slice parseSlice(Slice slice, int offset)
     {
-        // first vint is scale, which is ignored
+        // first vint is scale
         int scale = toIntExact(readVInt(slice, offset));
         offset += decodeVIntSize(slice, offset);
 
@@ -181,10 +181,13 @@ private Slice parseSlice(Slice slice, int offset)
 
         resultSlice.setBytes(BYTES_IN_LONG_DECIMAL - length, slice, offset, length);
 
+        // todo get rid of BigInteger
+        BigInteger decimal = new BigInteger(resultBytes);
         if (scale != type.getScale()) {
-            return encodeUnscaledValue(rescale(new BigInteger(resultBytes), scale, type.getScale()));
+            decimal = Decimals.rescale(decimal, scale, type.getScale());
         }
-        return resultSlice;
+
+        return Decimals.encodeUnscaledValue(decimal);
     }
 
     private void writeLong(SliceOutput output, long value)
@@ -219,43 +222,15 @@ private void writeSlice(SliceOutput output, Block block, int position)
         writeVInt(output, type.getScale());
 
         // second vint is length
-        int length = getWriteByteCount(block, position);
-        writeVInt(output, length);
+        // todo get rid of BigInteger
+        BigInteger decimal = Decimals.decodeUnscaledValue(block.getSlice(position, 0, BYTES_IN_LONG_DECIMAL));
+        byte[] decimalBytes = decimal.toByteArray();
+        writeVInt(output, decimalBytes.length);
 
         // write value (big endian)
         // NOTE: long decimals are stored in a slice in big endian encoding
-        for (int i = BYTES_IN_LONG_DECIMAL - length; i < BYTES_IN_LONG_DECIMAL; i++) {
-            output.write(block.getByte(position, i));
-        }
-    }
-
-    private static int getWriteByteCount(Block block, int position)
-    {
-        int length = BYTES_IN_LONG_DECIMAL;
-        if (block.getByte(position, 0) < 0) {
-            for (int i = 0; i < BYTES_IN_LONG_DECIMAL - 1; i++) {
-                int aByte = block.getByte(position, i);
-                if (aByte != 0xFFFF_FFFF) {
-                    if (aByte >= 0) {
-                        length++;
-                    }
-                    break;
-                }
-                length--;
-            }
-        }
-        else {
-            for (int i = 0; i < BYTES_IN_LONG_DECIMAL - 1; i++) {
-                int aByte = block.getByte(position, i);
-                if (aByte != 0) {
-                    if (aByte < 0) {
-                        length++;
-                    }
-                    break;
-                }
-                length--;
-            }
+        for (byte decimalByte : decimalBytes) {
+            output.write(decimalByte);
         }
-        return length;
     }
 }
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/type/UnscaledDecimal128Arithmetic.java b/presto-spi/src/main/java/com/facebook/presto/spi/type/UnscaledDecimal128Arithmetic.java
index bd99517f5e34..7a7f847995bb 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/type/UnscaledDecimal128Arithmetic.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/type/UnscaledDecimal128Arithmetic.java
@@ -132,6 +132,13 @@ public static Slice unscaledDecimal(String unscaledValue)
 
     public static Slice unscaledDecimal(BigInteger unscaledValue)
     {
+        Slice decimal = Slices.allocate(UNSCALED_DECIMAL_128_SLICE_LENGTH);
+        return pack(unscaledValue, decimal);
+    }
+
+    public static Slice pack(BigInteger unscaledValue, Slice result)
+    {
+        pack(0, 0, false, result);
         byte[] bytes = unscaledValue.abs().toByteArray();
 
         if (bytes.length > UNSCALED_DECIMAL_128_SLICE_LENGTH
@@ -142,15 +149,14 @@ public static Slice unscaledDecimal(BigInteger unscaledValue)
         // convert to little-endian order
         reverse(bytes);
 
-        Slice decimal = Slices.allocate(UNSCALED_DECIMAL_128_SLICE_LENGTH);
-        decimal.setBytes(0, bytes);
+        result.setBytes(0, bytes);
         if (unscaledValue.signum() < 0) {
-            setNegative(decimal, true);
+            setNegative(result, true);
         }
 
-        throwIfOverflows(decimal);
+        throwIfOverflows(result);
 
-        return decimal;
+        return result;
     }
 
     public static Slice unscaledDecimal(long unscaledValue)