JAPI-499 BinaryRecordReader support U8 to Decimal (hpcc-systems#599)

- Added flag to support treating U8s as BigDecimals
- Created JUnit for U8 to Decimal

Signed-off-by: James McMullan [email protected]

Signed-off-by: James McMullan [email protected]

Author: jpmcmu

commons-hpcc/src/main/java/org/hpccsystems/commons/utils/Utils.java

@@ -657,6 +657,8 @@ public static boolean isBooleanKeyWord(String str)
      */
     public static BigInteger extractUnsigned8Val(long unsigned8)
     {
+        // Shift upper 32 bits of the incoming long value down and use bitwise & to get unsigned value of upper 32 bits
+        // Shift upper 32 bits back once inside of BigInteger and add unsigned value of lower 32 bits
         return (BigInteger.valueOf((unsigned8 >> 32) & 0xffffffffL).shiftLeft(32)).add((BigInteger.valueOf(unsigned8 & 0xffffffffL)));
     }
 }

dfsclient/src/main/java/org/hpccsystems/dfs/client/BinaryRecordReader.java

@@ -28,6 +28,7 @@
 import org.hpccsystems.commons.ecl.HpccSrcType;
 import org.hpccsystems.commons.errors.HpccFileException;
 import org.hpccsystems.commons.errors.UnparsableContentException;
+import org.hpccsystems.commons.utils.Utils;
 
 class CountingInputStream extends InputStream
 {
@@ -129,6 +130,7 @@ public class BinaryRecordReader implements IRecordReader
     protected boolean            defaultLE;
     private long                 streamPosAfterLastRecord = 0;
     private boolean              isIndex = false;
+    private boolean              useDecimalForUnsigned8 = false;
 
     private byte[]               scratchBuffer = new byte[BUFFER_GROW_SIZE];
 
@@ -219,6 +221,16 @@ public void initialize(IRecordBuilder rb) throws Exception
         }
     }
 
+    /**
+     * Determines if unsigned 8 values should be parsed into BigDecimals to avoid long overflow.
+     * 
+     * @param useDecimal use decimal
+     */
+    public void setUseDecimalForUnsigned8(boolean useDecimal)
+    {
+        useDecimalForUnsigned8 = useDecimal;
+    }
+
     /**
      * Should be set if this record reader is reading an index file.
      * 
@@ -350,16 +362,27 @@ private Object parseFlatField(FieldDef fd, boolean isLittleEndian) throws Unpars
                 if (fd.isUnsigned())
                 {
                     intValue = getUnsigned((int) fd.getDataLen(), fd.getSourceType() == HpccSrcType.LITTLE_ENDIAN);
-                    if (intValue < 0)
+                    if (useDecimalForUnsigned8 && fd.getDataLen() == 8)
+                    {
+                        BigInteger bi = Utils.extractUnsigned8Val(intValue);
+                        fieldValue = new BigDecimal(bi);
+                    }
+                    else 
                     {
-                        messages.addMessage("Warning: Possible unsigned overflow in column: '" + fd.getFieldName() + "'. Convert values to BigInteger via org.hpccsystems.commons.utils.extractUnsigned8 if necessary." );
+                        fieldValue = Long.valueOf(intValue);
+                        if (intValue < 0)
+                        {
+                            messages.addMessage("Warning: Possible unsigned overflow in column: '" + fd.getFieldName() 
+                                            + "'. Convert values to BigInteger via org.hpccsystems.commons.utils.extractUnsigned8 if necessary, "
+                                            + " or call BinaryRecordReader.setUseDecimalForUnsigned8() before reading to convert unsigned8 values to BigDecimal values.");
+                        }
                     }
                 }
                 else
                 {
                     intValue = getInt((int) fd.getDataLen(), fd.getSourceType() == HpccSrcType.LITTLE_ENDIAN, fd.isBiased());
+                    fieldValue = Long.valueOf(intValue);
                 }
-                fieldValue = Long.valueOf(intValue);
                 break;
             case REAL:
                 // fixed number of bytes (4 or 8) in type info

dfsclient/src/main/java/org/hpccsystems/dfs/client/HpccRemoteFileReader.java

@@ -240,6 +240,11 @@ public FileReadResumeInfo getFileReadResumeInfo(Long streamPosition)
         return resumeInfo;
     }
 
+    /**
+     * Returns the number of messages created during the reading process
+     * 
+     * @return number of messages created
+     */
     public int getRemoteReadMessageCount()
     {
         int count = 0;
@@ -249,6 +254,11 @@ public int getRemoteReadMessageCount()
         return count;
     }
 
+    /**
+     * Returns messages created during the file reading process
+     * 
+     * @return Messages concatenated into a String
+     */
     public String getRemoteReadMessages()
     {
         String report = "";
@@ -329,7 +339,6 @@ public void close() throws Exception
     }
 
     /**
-     * getAvailable
      * Returns the number of bytes available to read immediately.
      *
      * @return the available
@@ -341,11 +350,21 @@ public int getAvailable() throws IOException
         return this.binaryRecordReader.getAvailable();
     }
 
+    /**
+     * Returns the RowServiceInputStream used to read the file from dafilesrv
+     * 
+     * @return the input stream
+     */
     public RowServiceInputStream getInputStream()
     {
         return this.inputStream;
     }
 
+    /**
+     * Returns the BinaryRecordReader used to construct records
+     * 
+     * @return the record reader
+     */
     public BinaryRecordReader getRecordReader()
     {
         return this.binaryRecordReader;

dfsclient/src/test/java/org/hpccsystems/dfs/client/DFSReadWriteTest.java

@@ -38,6 +38,7 @@
 import org.hpccsystems.commons.ecl.HpccSrcType;
 import org.hpccsystems.commons.ecl.RecordDefinitionTranslator;
 import org.hpccsystems.commons.errors.HpccFileException;
+import org.hpccsystems.commons.utils.Utils;
 import org.hpccsystems.ws.client.HPCCWsDFUClient;
 import org.hpccsystems.ws.client.HPCCWsWorkUnitsClient;
 import org.hpccsystems.ws.client.wrappers.wsworkunits.WorkunitWrapper;
@@ -334,6 +335,46 @@ public void integrationLargeRecordTest() throws Exception
             }
         }
     }
+    
+    @Test
+    public void unsigned8ToDecimalTest() throws Exception
+    {
+        // Create a large record dataset
+        FieldDef[] fieldDefs = new FieldDef[3];
+        fieldDefs[0] = new FieldDef("field1", FieldType.INTEGER, "UNSIGNED8", 8, true, true, HpccSrcType.LITTLE_ENDIAN, new FieldDef[0]);
+        fieldDefs[1] = new FieldDef("field2", FieldType.INTEGER, "UNSIGNED8", 8, true, true, HpccSrcType.LITTLE_ENDIAN, new FieldDef[0]);
+        fieldDefs[2] = new FieldDef("field3", FieldType.INTEGER, "UNSIGNED8", 8, true, true, HpccSrcType.LITTLE_ENDIAN, new FieldDef[0]);
+        FieldDef recordDef = new FieldDef("RootRecord", FieldType.RECORD, "rec", 4, false, false, HpccSrcType.LITTLE_ENDIAN, fieldDefs);
+
+        List<HPCCRecord> originalRecords = new ArrayList<HPCCRecord>();
+        for (int i = 0; i < 10; i++)
+        {
+            Object[] fields = new Object[3];
+            fields[0] = new BigDecimal(Utils.extractUnsigned8Val((long) Long.MIN_VALUE));       // Max U8 value
+            fields[1] = new BigDecimal(0);                                                  // Min U8 value
+            fields[2] = new BigDecimal(BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.ONE)); // First U8 value that would cause an overflow
+            HPCCRecord record = new HPCCRecord(fields, recordDef);
+            originalRecords.add(record);
+        }
+
+        String datasetName = "benchmark::unsigned8::10rows"; 
+        writeFile(originalRecords, datasetName, recordDef, connTO);
+
+        HPCCFile file = new HPCCFile(datasetName, connString , hpccUser, hpccPass);
+        List<HPCCRecord> readRecords = readFile(file, connTO, false, true);
+        if (readRecords.size() < 10)
+        {
+            Assert.fail("Failed to read " + datasetName);
+        }
+
+        for (int i = 0; i < 10; i++)
+        {
+            HPCCRecord readRecord = readRecords.get(i);
+            HPCCRecord originalRecord = originalRecords.get(i);
+            
+            assertEquals(readRecord, originalRecord);
+        }
+    }
 
     @Test
     public void numericOverflowTest() throws Exception
@@ -703,15 +744,22 @@ public void resumeFileReadTest() throws Exception
     }
 
     public List<HPCCRecord> readFile(HPCCFile file, Integer connectTimeoutMillis, boolean shouldForceTimeout) throws Exception
+    {
+        return readFile(file, connectTimeoutMillis, shouldForceTimeout, false);
+    }
+
+    public List<HPCCRecord> readFile(HPCCFile file, Integer connectTimeoutMillis, boolean shouldForceTimeout, boolean useDecimalForUnsigned8) throws Exception
     {
         if (file == null)
         {
             Assert.fail("HPCCFile construction failed.");
         }
+
         if (connectTimeoutMillis != null)
         {
             file.setFileAccessExpirySecs(connectTimeoutMillis/1000);
         }
+
         DataPartition[] fileParts = file.getFileParts();
         if (fileParts == null || fileParts.length == 0)
         {
@@ -731,6 +779,8 @@ public List<HPCCRecord> readFile(HPCCFile file, Integer connectTimeoutMillis, bo
             {
                 HPCCRecordBuilder recordBuilder = new HPCCRecordBuilder(file.getProjectedRecordDefinition());
                 HpccRemoteFileReader<HPCCRecord> fileReader = new HpccRemoteFileReader<HPCCRecord>(fileParts[i], originalRD, recordBuilder);
+                fileReader.getRecordReader().setUseDecimalForUnsigned8(useDecimalForUnsigned8);
+
                 fileReaders.add(fileReader);
             }
             catch (Exception e)