more robust bulk file processing

java/core/src/main/java/co/worklytics/psoxy/storage/BulkDataSanitizer.java

@@ -2,8 +2,8 @@
 
 import co.worklytics.psoxy.Pseudonymizer;
 
+import java.io.BufferedReader;
 import java.io.IOException;
-import java.io.Reader;
 import java.io.Writer;
 
 /**
@@ -22,7 +22,7 @@ public interface BulkDataSanitizer {
      * @param pseudonymizer The pseudonymizer to use
      * @throws IOException  IO problem reading or writing
      */
-    void sanitize(Reader reader,
+    void sanitize(BufferedReader reader,
                   Writer writer,
                   Pseudonymizer pseudonymizer) throws IOException;
 }

java/core/src/main/java/co/worklytics/psoxy/storage/StorageHandler.java

@@ -379,7 +379,7 @@ void process(StorageEventRequest request,
 
         try (
             InputStream inputStream = readInputStream(request, bufferSize, inputStreamSupplier);
-            Reader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8), bufferSize);
+            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8), bufferSize);
             OutputStream outputStream = writeOutputStream(request, bufferSize, outputStreamSupplier);
             OutputStreamWriter writer = new OutputStreamWriter(outputStream)
         ) {

java/core/src/main/java/co/worklytics/psoxy/storage/impl/ColumnarBulkDataSanitizerImpl.java

@@ -32,14 +32,13 @@
 import org.apache.commons.lang3.tuple.Pair;
 
 import javax.inject.Inject;
-import java.io.IOException;
-import java.io.Reader;
-import java.io.Writer;
+import java.io.*;
 import java.util.*;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.UnaryOperator;
+import java.util.logging.Level;
 import java.util.regex.Matcher;
 import java.util.stream.Collectors;
 
@@ -74,8 +73,9 @@ public ColumnarBulkDataSanitizerImpl(@Assisted ColumnarRules rules) {
         this.rules = rules;
     }
 
+
     @Override
-    public void sanitize(@NonNull Reader reader,
+    public void sanitize(@NonNull BufferedReader reader,
                          @NonNull Writer writer,
                          @NonNull Pseudonymizer pseudonymizer) throws IOException {
 
@@ -87,8 +87,9 @@ public void sanitize(@NonNull Reader reader,
             .setTrim(true)
             .build();
 
+        ParsedFirstLine parsedFirstLine = parseFirstLine(reader);
 
-        CSVParser records = inputCSVFormat.parse(reader);
+        CSVParser records = inputCSVFormat.parse(new StringReader(parsedFirstLine.contents));
 
         Preconditions.checkArgument(records.getHeaderMap() != null, "Failed to parse header from file");
 
@@ -275,44 +276,85 @@ public void sanitize(@NonNull Reader reader,
         List<String> columnNamesForOutputFile = columnTransforms.keySet()
             .stream()
             .sorted(originalHeadersOrRenamedFirst.thenComparing(byColumnName))
-            .collect(Collectors.toList())
+            .toList()
             .stream()
             // leave original casing for headers
             .map( h -> headers.stream().filter(h::equalsIgnoreCase).findFirst().orElse(h))
             .collect(Collectors.toList());
 
-        CSVFormat csvFormat = CSVFormat.Builder.create()
+        CSVFormat inputCsvFormat = CSVFormat.DEFAULT.builder()
+            .setHeader(records.getHeaderNames().toArray(new String[0]))
+            .build();
+
+
+        CSVFormat outputFormat = CSVFormat.Builder.create()
             .setHeader(columnNamesForOutputFile.toArray(new String[0]))
-            .setRecordSeparator(records.getFirstEndOfLine())
+            .setRecordSeparator(parsedFirstLine.getEndOfLine()) // use the same EOL as in the input file
             .setNullString("")
             .build();
 
-        // create an empty record to fill with the transformed values, ensuring all rows have
-        // the same columns
-        // linked hash map: need predictable iteration order and null values
 
-        LinkedHashMap<String, String> newRecord = new LinkedHashMap<>();
-        columnNamesForOutputFile.forEach(h -> newRecord.put(h, null));
+        try (CSVPrinter printer = new CSVPrinter(writer, outputFormat)) {
+            ProcessingBuffer<ProcessedRecord> buffer = getRecordsProcessingBuffer(printer);
+            processRecords(columnNamesForOutputFile, inputCsvFormat, columnTransforms, reader, buffer);
 
-        try (CSVPrinter printer = new CSVPrinter(writer, csvFormat)) {
+            if (buffer.flush()) {
+                log.info(String.format("Processed records: %d", buffer.getProcessed()));
+            }
+        }
+    }
 
-            ProcessingBuffer<ProcessedRecord> buffer = getRecordsProcessingBuffer(printer);
+    /**
+     * processes records into outputBuffer, applying the transformations to each
+     * @param columnNamesForOutput to render in the output buffer; so columns which DON'T have values in particular record will be filled
+     * @param columnTransformsToApply to apply to each column in the record
+     * @param bodyRecords to process; should be a CSVParser with headers
+     * @param outputBuffer to write the processed records to
+     */
+    void processRecords(
+                                List<String> columnNamesForOutput,
+                                CSVFormat inputCsvFormat,
+                                Map<String, Pair<String, List<Function<String, Optional<String>>>>> columnTransformsToApply,
+                                BufferedReader bodyRecords,
+                                ProcessingBuffer<ProcessedRecord> outputBuffer) throws IOException {
+
+        // collect transforms that are MISSING mappings in the records
+        Set<String> transformsWithoutMappings = new HashSet<>();
+
+        // we're going to read line-by-line, so we need to ensure that the CSVFormat is set to not skip the header record
+        CSVFormat perLineFormat = CSVFormat.DEFAULT.builder()
+            .setDelimiter(inputCsvFormat.getDelimiter())
+            .setSkipHeaderRecord(false).build();
+
+        Map<String, Integer> headerMap = new HashMap<>(inputCsvFormat.getHeader().length);
+        for (int i = 0; i < inputCsvFormat.getHeader().length; i++) {
+            headerMap.put(inputCsvFormat.getHeader()[i].toLowerCase(), i);
+        }
 
-            Set<String> transformsWithoutMappings = new HashSet<>();
+        // linked hash map: need predictable iteration order and null values
+        LinkedHashMap<String, String> newRecord = new LinkedHashMap<>();
+        // create an empty record to fill with the transformed values, ensuring all rows have
+        // the same columns
+        columnNamesForOutput.forEach(h -> newRecord.put(h, null));
 
-            for (CSVRecord record : records) {
-                // clean up the record prior to use
+        String line;
+        while ((line = bodyRecords.readLine()) != null) {
+            try (CSVParser parser = CSVParser.parse(line, perLineFormat)) {
+                CSVRecord record = parser.getRecords().get(0);
                 newRecord.replaceAll((k, v) -> null);
                 newRecord.keySet().forEach(h -> {
-
-                    Pair<String, List<Function<String, Optional<String>>>> transforms = columnTransforms.getOrDefault(h, null);
+                    Pair<String, List<Function<String, Optional<String>>>> transforms = columnTransformsToApply.getOrDefault(h, null);
                     if (transforms == null) {
                         newRecord.put(h, null);
                     } else {
                         // apply all transformations in insertion order
                         // key holds the original column
-                        if (record.isMapped(transforms.getKey())) {
-                            String v = record.get(transforms.getKey());
+
+                        //TODO: build this into columnTransformsToApply, so don't need to call lowerCase on every loop
+                        String lcKey = transforms.getKey().toLowerCase();
+
+                        if (headerMap.containsKey(lcKey)) {
+                            String v = StringUtils.trim(record.get(headerMap.get(lcKey)));
                             if (StringUtils.isNotBlank(v)) {
                                 for (Function<String, Optional<String>> transform : transforms.getValue()) {
                                     v = transform.apply(v).orElse(null);
@@ -328,19 +370,18 @@ public void sanitize(@NonNull Reader reader,
                             }
                         }
                     }
-
                 });
 
-                if (buffer.addAndAttemptFlush(ProcessedRecord.of(Lists.newArrayList(newRecord.values())))) {
-                    log.info(String.format("Processed records: %d", buffer.getProcessed()));
-                };
-            }
-            if (buffer.flush()) {
-                log.info(String.format("Processed records: %d", buffer.getProcessed()));
+                if (outputBuffer.addAndAttemptFlush(ProcessedRecord.of(Lists.newArrayList(newRecord.values())))) {
+                    log.info(String.format("Processed records: %d", outputBuffer.getProcessed()));
+                }
+            } catch (IndexOutOfBoundsException | UncheckedIOException | IOException e) {
+                log.log(Level.WARNING, "Failed to process record", e);
             }
         }
     }
 
+
     @VisibleForTesting
     Set<String> determineMissingColumnsToPseudonymize(Set<String> columnsToPseudonymize, Set<String> outputColumnsCI) {
         Function<Set<String>, Set<String>> asLowercase = (Set<String> set) -> set.stream().map(String::toLowerCase).collect(Collectors.toSet());
@@ -375,7 +416,7 @@ private ProcessingBuffer<ProcessedRecord> getRecordsProcessingBuffer(final CSVPr
 
     @Value
     @RequiredArgsConstructor(staticName = "of")
-    private static class ProcessedRecord {
+   static class ProcessedRecord {
         Collection<String> values;
     }
 
@@ -436,4 +477,65 @@ TriFunction<String, String, Pseudonymizer, String> buildPseudonymizationFunction
         };
     }
 
+
+
+    @Value
+    static class ParsedFirstLine {
+        String contents;
+        /**
+         * one of the following:
+         * - "\n" (LF)
+         * - "\r\n" (CRLF)
+         * - "\r" (CR)
+         *
+         */
+        String endOfLine;
+    }
+
+    /**
+     * helper to parse the first line from a BufferedReader
+     * @param reader to parse
+     * @return ParsedFirstLine with the contents of the first line and the end of line character(s) used
+     */
+    ParsedFirstLine parseFirstLine(BufferedReader reader) {
+        String line = null, endOfLine = null;
+        try {
+            StringBuilder sb = new StringBuilder();
+            int ch;
+            boolean foundEOL = false;
+            while ((ch = reader.read()) != -1) {
+                char c = (char) ch;
+                sb.append(c);
+                if (c == '\r') {
+                    reader.mark(1);
+                    int next = reader.read();
+                    if (next == '\n') {
+                        sb.append('\n');
+                        endOfLine = "\r\n";
+                    } else {
+                        if (next != -1) {
+                            reader.reset();
+                        }
+                        endOfLine = "\r";
+                    }
+                    foundEOL = true;
+                    break;
+                } else if (c == '\n') {
+                    endOfLine = "\n";
+                    foundEOL = true;
+                    break;
+                }
+            }
+            if (sb.isEmpty() && !foundEOL) {
+                throw new IllegalArgumentException("Empty file, no header found");
+            }
+            line = sb.toString().replaceAll("(\r\n|\r|\n)$", "");
+            if (endOfLine == null) {
+                endOfLine = "\n"; // default
+            }
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+        return new ParsedFirstLine(line, endOfLine);
+    }
 }

java/core/src/main/java/co/worklytics/psoxy/storage/impl/RecordBulkDataSanitizerImpl.java

@@ -58,7 +58,7 @@ public RecordBulkDataSanitizerImpl(@Assisted RecordRules rules) {
 
 
     @Override
-    public void sanitize(@NonNull Reader reader,
+    public void sanitize(BufferedReader reader,
                          @NonNull Writer writer,
                          @NonNull Pseudonymizer pseudonymizer) throws IOException {