HPCC4J-581 WsFS Client should add path delim only if needed

commons-hpcc/pom.xml

@@ -9,7 +9,7 @@
   <parent>
     <groupId>org.hpccsystems</groupId>
     <artifactId>hpcc4j</artifactId>
-    <version>9.2.65-0-SNAPSHOT</version>
+    <version>9.4.41-0-SNAPSHOT</version>
   </parent>
 
   <build>

dfsclient/pom.xml

@@ -9,7 +9,7 @@
   <parent>
     <groupId>org.hpccsystems</groupId>
     <artifactId>hpcc4j</artifactId>
-    <version>9.2.65-0-SNAPSHOT</version>
+    <version>9.4.41-0-SNAPSHOT</version>
   </parent>
 
   <build>

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

@@ -830,7 +830,15 @@ private BigDecimal getUnsignedDecimal(int numDigits, int precision, int dataLen)
         BigDecimal ret = new BigDecimal(0);
 
         int idx = 0;
-        int curDigit = numDigits - 1;
+        int curDigit = numDigits;
+
+        // If the # of digits is odd the top most nibble is unused and we don't want to include it
+        // in the scale calculations below. Due to how the scale calculation works below this means
+        // we decrement the starting value of curDigit in the case of even length decimals
+        if ((numDigits % 2) == 0)
+        {
+            curDigit--;
+        }
 
         while (idx < dataLen)
         {

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

@@ -16,6 +16,10 @@
 package org.hpccsystems.dfs.client;
 
 import java.io.Serializable;
+import java.security.InvalidParameterException;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Arrays;
 
 import org.hpccsystems.commons.ecl.FileFilter;
 import org.hpccsystems.commons.errors.HpccFileException;
@@ -261,11 +265,61 @@ public String[] getCopyLocations()
     public String getCopyIP(int copyindex)
     {
         int copiescount = copyLocations.length;
-        if (copyindex < 0 || copyindex >= copiescount) return null;
+        if (copyindex < 0 || copyindex >= copiescount)
+        {
+            return null;
+        }
 
         return copyLocations[copyindex];
     }
 
+    /**
+     * Set the copy IP
+     *
+     * @param copyIndex
+     *            the copyindex
+     * @param copyIP The IP of the file part copy
+     */
+    public void setCopyIP(int copyIndex, String copyIP)
+    {
+        if (copyIndex < 0 || copyIndex >= copyLocations.length)
+        {
+            return;
+        }
+
+        copyLocations[copyIndex] = copyIP;
+    }
+
+    /**
+     * Add file part copy
+     *
+     * @param index The index at which to insert the file part copy
+     * @param copyIP The IP of the new file part copy
+     * @param copyPath The path of the new file part copy
+     * @throws Exception The exception
+     */
+    public void add(int index, String copyIP, String copyPath) throws Exception
+    {
+        if (index < 0 || index > copyLocations.length)
+        {
+            throw new InvalidParameterException("Insertion index: " + index + " is invalid."
+                        + "Expected index in range of: [0," + copyLocations.length + "]");
+        }
+
+        if (copyIP == null || copyPath == null)
+        {
+            throw new InvalidParameterException("Copy IP or Path are invalid, must be non-null.");
+        }
+
+        List<String> copyLocationsList = new ArrayList<>(Arrays.asList(copyLocations));
+        copyLocationsList.add(index, copyIP);
+        copyLocations = copyLocationsList.toArray(new String[0]);
+
+        List<String> copyPathList = new ArrayList<>(Arrays.asList(copyPaths));
+        copyPathList.add(index, copyPath);
+        copyPaths = copyPathList.toArray(new String[0]);
+    }
+
     /**
      * Count of copies available for this file part.
      * @return copy locations size

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

@@ -417,6 +417,24 @@ private static Options getReadOptions()
         return options;
     }
 
+    private static Options getReadTestOptions()
+    {
+        Options options = new Options();
+        options.addRequiredOption("read_test", "Read test", true, "Specifies the file that should be read.");
+        options.addRequiredOption("url", "Source Cluster URL", true, "Specifies the URL of the ESP to connect to.");
+        options.addOption("user", true, "Specifies the username used to connect. Defaults to null.");
+        options.addOption("pass", true, "Specifies the password used to connect. Defaults to null.");
+        options.addOption("num_threads", true, "Specifies the number of parallel to use to perform operations.");
+
+        options.addOption(Option.builder("file_parts")
+                                .argName("_file_parts")
+                                .hasArgs()
+                                .valueSeparator(',')
+                                .desc("Specifies the file parts that should be read. Defaults to all file parts.")
+                                .build());
+        return options;
+    }
+
     private static Options getCopyOptions()
     {
         Options options = new Options();
@@ -463,6 +481,7 @@ private static Options getTopLevelOptions()
     {
         Options options = new Options();
         options.addOption("read", "Reads the specified file(s) and writes a copy of the files to the local directory.");
+        options.addOption("read_test", "Reads the specified file and/or particular file parts without writing it locally.");
         options.addOption("copy", "Copies the specified remote source file to the specified remote destination cluster / file.");
         options.addOption("write", "Writes the specified local source file to the specified remote destination cluster / file.");
 
@@ -660,6 +679,44 @@ public void run()
         }
     }
 
+    private static Runnable[] createReadTestTasks(DataPartition[] fileParts, FieldDef recordDef, TaskContext context) throws Exception
+    {
+        Runnable[] tasks = new Runnable[fileParts.length];
+        for (int i = 0; i < tasks.length; i++)
+        {
+            final int taskIndex = i;
+            final DataPartition filePart = fileParts[taskIndex];
+            final HpccRemoteFileReader<HPCCRecord> filePartReader = new HpccRemoteFileReader<HPCCRecord>(filePart, recordDef, new HPCCRecordBuilder(recordDef));
+
+            tasks[taskIndex] = new Runnable()
+            {
+                HpccRemoteFileReader<HPCCRecord> fileReader = filePartReader;
+
+                public void run()
+                {
+                    try
+                    {
+                        while (fileReader.hasNext())
+                        {
+                            HPCCRecord record = fileReader.next();
+                            context.recordsRead.incrementAndGet();
+                        }
+
+                        fileReader.close();
+                        context.bytesRead.addAndGet(fileReader.getStreamPosition());
+                    }
+                    catch (Exception e)
+                    {
+                        context.addError("Error while reading file part index: '" + filePart.getThisPart() + " Error message: " + e.getMessage());
+                        return;
+                    }
+                }
+            };
+        }
+
+        return tasks;
+    }
+
     private static Runnable[] createReadToThorTasks(DataPartition[] fileParts, SplitTable[] splitTables, String[] outFilePaths, FieldDef recordDef, TaskContext context) throws Exception
     {
         Runnable[] tasks = new Runnable[fileParts.length];
@@ -1159,6 +1216,165 @@ private static void performRead(String[] args, TaskContext context)
         }
     }
 
+    private static void performReadTest(String[] args, TaskContext context)
+    {
+        Options options = getReadTestOptions();
+        CommandLineParser parser = new DefaultParser();
+
+        CommandLine cmd = null;
+        try
+        {
+            cmd = parser.parse(options, args);
+        }
+        catch (ParseException e)
+        {
+            System.out.println("Error parsing commandline options:\n" + e.getMessage());
+            return;
+        }
+
+        String connString = cmd.getOptionValue("url");
+        String user = cmd.getOptionValue("user");
+        String pass = cmd.getOptionValue("pass");
+
+        String outputPath = cmd.getOptionValue("out",".");
+
+        int numThreads = NUM_DEFAULT_THREADS;
+        String numThreadsStr = cmd.getOptionValue("num_threads", "" + numThreads);
+        try
+        {
+            numThreads = Integer.parseInt(numThreadsStr);
+        }
+        catch(Exception e)
+        {
+            System.out.println("Invalid option value for num_threads: "
+                              + numThreadsStr + ", must be an integer. Defaulting to: " + NUM_DEFAULT_THREADS + " threads.");
+        }
+
+        String formatStr = cmd.getOptionValue("format");
+        if (formatStr == null)
+        {
+            formatStr = "THOR";
+        }
+
+        FileFormat format = FileFormat.THOR;
+        switch (formatStr.toUpperCase())
+        {
+            case "THOR":
+                format = FileFormat.THOR;
+                break;
+            case "PARQUET":
+                format = FileFormat.PARQUET;
+                break;
+            default:
+                System.out.println("Error unsupported format specified: " + format);
+                return;
+        }
+
+        String datasetName = cmd.getOptionValue("read_test");
+        context.startOperation("Read Test " + datasetName);
+
+        HPCCFile file = null;
+        try
+        {
+            file = new HPCCFile(datasetName, connString, user, pass);
+        }
+        catch (Exception e)
+        {
+            System.out.println("Error while attempting to open file: '" + datasetName + "': " + e.getMessage());
+            return;
+        }
+
+        DataPartition[] fileParts = null;
+        FieldDef recordDef = null;
+        try
+        {
+            fileParts = file.getFileParts();
+            recordDef = file.getRecordDefinition();
+        }
+        catch (Exception e)
+        {
+            System.out.println("Error while retrieving file parts for: '" + datasetName + "': " + e.getMessage());
+            return;
+        }
+
+        String[] filePartsStrs = cmd.getOptionValues("file_parts");
+        if (filePartsStrs != null && filePartsStrs.length > 0)
+        {
+            ArrayList<DataPartition> filePartList = new ArrayList<DataPartition>();
+            for (int i = 0; i < filePartsStrs.length; i++)
+            {
+                try
+                {
+                    int filePartIndex = Integer.parseInt(filePartsStrs[i]) - 1;
+                    if (filePartIndex < 0 || filePartIndex >= fileParts.length)
+                    {
+                        System.out.println("Skipping invalid file part index: " + filePartsStrs[i]
+                                        + " outside of range: [0," + fileParts.length + "]");
+                        continue;
+                    }
+
+                    filePartList.add(fileParts[filePartIndex]);
+                }
+                catch (NumberFormatException e)
+                {
+                    System.out.println("Skipping invalid file part index: " + filePartsStrs[i]);
+                }
+            }
+        }
+
+        Runnable[] tasks = null;
+        try
+        {
+            switch (format)
+            {
+                case THOR:
+                    tasks = createReadTestTasks(fileParts, recordDef, context);
+                    break;
+                case PARQUET:
+                default:
+                    throw new Exception("Error unsupported format specified: " + format);
+            };
+        }
+        catch (Exception e)
+        {
+            context.addError("Error while attempting to create read tasks for file: '" + datasetName + "': " + e.getMessage());
+            return;
+        }
+
+        try
+        {
+            executeTasks(tasks, numThreads);
+        }
+        catch (Exception e)
+        {
+            context.addError("Error while attempting to execute read tasks for file: '" + datasetName + "': " + e.getMessage());
+            return;
+        }
+
+        if (context.hasError())
+        {
+            return;
+        }
+
+        try
+        {
+            String fileName = file.getFileName().replace(":","_");
+            String filePath = outputPath + File.separator + fileName + ".meta";
+            FileOutputStream metaFile = new FileOutputStream(filePath);
+
+            String metaStr = RecordDefinitionTranslator.toJsonRecord(file.getRecordDefinition()).toString();
+            metaFile.write(metaStr.getBytes());
+            metaFile.close();
+        }
+        catch (Exception e)
+        {
+            context.addError("Error while attempting to write meta-data for file: '" + datasetName + "': " + e.getMessage());
+            return;
+        }
+
+        context.endOperation();
+    }
+
     private static void performCopy(String[] args, TaskContext context)
     {
         Options options = getCopyOptions();
@@ -1576,6 +1792,10 @@ public static JSONArray run(String[] args)
         {
             performRead(args, context);
         }
+        else if (cmd.hasOption("read_test"))
+        {
+            performReadTest(args, context);
+        }
         else if (cmd.hasOption("copy"))
         {
             performCopy(args, context);