Make CsvTokenizer independent from LineDecoder

Author: Dai MIKURUBE

embulk-guess-csv/src/main/java/org/embulk/guess/csv/CsvGuessPlugin.java

@@ -307,28 +307,26 @@ private static List<List<String>> splitLines(
             final CsvTokenizer tokenizer = buildCsvTokenizer(parserTask, sample);
 
             final ArrayList<List<String>> rows = new ArrayList<>();
-            while (tokenizer.nextFile()) {
-                while (tokenizer.nextRecord(skipEmptyLines)) {
-                    try {
-                        final ArrayList<String> columns = new ArrayList<>();
-                        while (true) {
-                            try {
-                                final String column = tokenizer.nextColumn();
-                                final boolean quoted = tokenizer.wasQuotedColumn();
-                                if (nullString != null && !quoted && nullString.equals(column)) {
-                                    columns.add(null);
-                                } else {
-                                    columns.add(column);
-                                }
-                            } catch (final TooFewColumnsException ex) {
-                                rows.add(Collections.unmodifiableList(columns));
-                                break;
+            while (tokenizer.nextRecord(skipEmptyLines)) {
+                try {
+                    final ArrayList<String> columns = new ArrayList<>();
+                    while (true) {
+                        try {
+                            final String column = tokenizer.nextColumn();
+                            final boolean quoted = tokenizer.wasQuotedColumn();
+                            if (nullString != null && !quoted && nullString.equals(column)) {
+                                columns.add(null);
+                            } else {
+                                columns.add(column);
                             }
+                        } catch (final TooFewColumnsException ex) {
+                            rows.add(Collections.unmodifiableList(columns));
+                            break;
                         }
-                    } catch (final InvalidValueException ex) {
-                        // TODO warning
-                        tokenizer.skipCurrentLine();
                     }
+                } catch (final InvalidValueException ex) {
+                    // TODO warning
+                    tokenizer.skipCurrentLine();
                 }
             }
             return Collections.unmodifiableList(rows);
@@ -366,7 +364,8 @@ private static CsvTokenizer buildCsvTokenizer(final CsvParserPlugin.PluginTask p
         final LineDecoder decoder = LineDecoder.of(
                 new ListFileInput(listListBuffer), parserTask.getCharset(), parserTask.getLineDelimiterRecognized().orElse(null));
 
-        return builder.build(decoder);
+        decoder.nextFile();
+        return builder.build(decoder.iterator());
     }
 
     private String guessDelimiter(final List<String> sampleLines) {

embulk-parser-csv/src/main/java/org/embulk/parser/csv/CsvParserPlugin.java

@@ -314,14 +314,17 @@ public void run(TaskSource taskSource, final Schema schema,
         final PluginTask task = CONFIG_MAPPER_FACTORY.createTaskMapper().map(taskSource, PluginTask.class);
         final TimestampFormatter[] timestampFormatters = newTimestampColumnFormatters(task, task.getSchemaConfig());
         final JsonParser jsonParser = new JsonParser();
-        final CsvTokenizer tokenizer = buildCsvTokenizer(task, input);
+        final CsvTokenizer.Builder tokenizerBuilder = buildCsvTokenizerBuilder(task);
         final boolean allowOptionalColumns = task.getAllowOptionalColumns();
         final boolean allowExtraColumns = task.getAllowExtraColumns();
         final boolean stopOnInvalidRecord = task.getStopOnInvalidRecord();
         final int skipHeaderLines = task.getSkipHeaderLines();
 
         try (final PageBuilder pageBuilder = getPageBuilder(Exec.getBufferAllocator(), schema, output)) {
-            while (tokenizer.nextFile()) {
+            while (input.nextFile()) {
+                final CsvTokenizer tokenizer = tokenizerBuilder.build(
+                        LineDecoder.of(input, task.getCharset(), task.getLineDelimiterRecognized().orElse(null)).iterator());
+
                 final String fileName = input.hintOfCurrentInputFileNameForLogging().orElse("-");
 
                 // skip the header lines for each file
@@ -468,7 +471,7 @@ static class CsvRecordValidateException extends DataException {
         }
     }
 
-    private static CsvTokenizer buildCsvTokenizer(final PluginTask task, final FileInput input) {
+    private static CsvTokenizer.Builder buildCsvTokenizerBuilder(final PluginTask task) {
         final CsvTokenizer.Builder builder = CsvTokenizer.builder(task.getDelimiter());
         task.getQuoteChar().ifPresent(q -> builder.setQuote(q.getCharacter()));
         task.getEscapeChar().ifPresent(e -> builder.setEscape(e.getCharacter()));
@@ -482,7 +485,7 @@ private static CsvTokenizer buildCsvTokenizer(final PluginTask task, final FileI
         builder.setMaxQuotedFieldLength(task.getMaxQuotedSizeLimit());
         task.getCommentLineMarker().ifPresent(m -> builder.setCommentLineMarker(m));
         task.getNullString().ifPresent(n -> builder.setNullString(n));
-        return builder.build(LineDecoder.of(input, task.getCharset(), task.getLineDelimiterRecognized().orElse(null)));
+        return builder;
     }
 
     @SuppressWarnings("deprecation")  // For the use of new PageBuilder().

embulk-parser-csv/src/main/java/org/embulk/parser/csv/CsvTokenizer.java

@@ -19,13 +19,13 @@
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Deque;
+import java.util.Iterator;
 import java.util.List;
 import java.util.stream.Collectors;
-import org.embulk.util.text.LineDecoder;
 
 public class CsvTokenizer {
     private CsvTokenizer(
-            final LineDecoder input,
+            final Iterator<String> iterator,
             final char delimiterChar,
             final String delimiterFollowingString,
             final char quote,
@@ -49,7 +49,7 @@ private CsvTokenizer(
 
         this.quotedValueLines = new ArrayList<>();
         this.unreadLines = new ArrayDeque<>();
-        this.input = input;
+        this.iterator = iterator;
 
         this.recordState = RecordState.END;  // initial state is end of a record. nextRecord() must be called first
         this.lineNumber = 0;
@@ -142,15 +142,15 @@ public Builder setNullString(final String nullString) {
             return this;
         }
 
-        public CsvTokenizer build(final LineDecoder input) {
+        public CsvTokenizer build(final Iterator<String> iterator) {
             if (this.trimIfNotQuoted && this.quotesInQuotedFields != QuotesInQuotedFields.ACCEPT_ONLY_RFC4180_ESCAPED) {
                 // The combination makes some syntax very ambiguous such as:
                 //     val1,  \"\"val2\"\"  ,val3
                 throw new IllegalStateException(
                         "[quotes_in_quoted_fields != ACCEPT_ONLY_RFC4180_ESCAPED] is not allowed to specify with [trim_if_not_quoted = true]");
             }
             return new CsvTokenizer(
-                    input,
+                    iterator,
                     delimiterChar,
                     delimiterFollowingString,
                     quote,
@@ -185,11 +185,12 @@ public long getCurrentLineNumber() {
     }
 
     public boolean skipHeaderLine() {
-        final boolean skipped = this.input.poll() != null;
-        if (skipped) {
-            this.lineNumber++;
+        if (!this.iterator.hasNext()) {
+            return false;
         }
-        return skipped;
+        this.iterator.next();
+        this.lineNumber++;
+        return true;
     }
 
     // returns skipped line
@@ -212,14 +213,6 @@ public String skipCurrentLine() {
         return skippedLine;
     }
 
-    public boolean nextFile() {
-        final boolean next = this.input.nextFile();
-        if (next) {
-            this.lineNumber = 0;
-        }
-        return next;
-    }
-
     // used by guess-csv
     public boolean nextRecord() {
         return this.nextRecord(true);
@@ -245,10 +238,10 @@ private boolean nextLine(final boolean skipEmptyLine) {
             if (!this.unreadLines.isEmpty()) {
                 this.line = this.unreadLines.removeFirst();
             } else {
-                this.line = this.input.poll();
-                if (this.line == null) {
+                if (!this.iterator.hasNext()) {
                     return false;
                 }
+                this.line = this.iterator.next();
             }
             this.linePos = 0;
             this.lineNumber++;
@@ -625,6 +618,8 @@ private static String escapeControl(final String from) {
 
     private static final char END_OF_LINE = '\0';
 
+    private final Iterator<String> iterator;
+
     private final char delimiterChar;
     private final String delimiterFollowingString;
     private final char quote;
@@ -634,7 +629,6 @@ private static String escapeControl(final String from) {
     private final QuotesInQuotedFields quotesInQuotedFields;
     private final long maxQuotedFieldLength;
     private final String commentLineMarker;
-    private final LineDecoder input;
     private final String nullString;
 
     private final List<String> quotedValueLines;

embulk-parser-csv/src/test/java/org/embulk/parser/csv/TestCsvTokenizer.java

@@ -61,9 +61,8 @@ private static List<List<String>> parse(final CsvTokenizer.Builder builder, fina
 
     private static List<List<String>> parse(final CsvTokenizer.Builder builder, final int columns, final FileInput input) {
         LineDecoder decoder = LineDecoder.of(input, StandardCharsets.UTF_8, null);
-        final CsvTokenizer tokenizer = builder.build(decoder);
-
-        tokenizer.nextFile();
+        decoder.nextFile();
+        final CsvTokenizer tokenizer = builder.build(decoder.iterator());
 
         List<List<String>> records = new ArrayList<>();
         while (tokenizer.nextRecord()) {
@@ -376,9 +375,8 @@ public void recoverFromQuotedSizeLimitExceededException() throws Exception {
         };
         final FileInput input = newFileInputFromLines("\n", lines);
         final LineDecoder decoder = LineDecoder.of(input, StandardCharsets.UTF_8, null);
-        final CsvTokenizer tokenizer = builder.build(decoder);
-
-        tokenizer.nextFile();
+        decoder.nextFile();
+        final CsvTokenizer tokenizer = builder.build(decoder.iterator());
 
         assertTrue(tokenizer.nextRecord());
         assertEquals("v1", tokenizer.nextColumn());