[ML] Refactor delimited file structure detection (#33233)

1. Use the term "delimited" rather than "separated values"
2. Use a single factory class with arguments to specify the
   delimiter and identification constraints

This change makes it easier to add support for other
delimiter characters.

x-pack/plugin/ml/log-structure-finder/src/main/java/org/elasticsearch/xpack/ml/logstructurefinder/SeparatedValuesLogStructureFinder.java → x-pack/plugin/ml/log-structure-finder/src/main/java/org/elasticsearch/xpack/ml/logstructurefinder/DelimitedLogStructureFinder.java

@@ -29,17 +29,16 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-public class SeparatedValuesLogStructureFinder implements LogStructureFinder {
+public class DelimitedLogStructureFinder implements LogStructureFinder {
 
     private static final int MAX_LEVENSHTEIN_COMPARISONS = 100;
 
     private final List<String> sampleMessages;
     private final LogStructure structure;
 
-    static SeparatedValuesLogStructureFinder makeSeparatedValuesLogStructureFinder(List<String> explanation, String sample,
-                                                                                   String charsetName, Boolean hasByteOrderMarker,
-                                                                                   CsvPreference csvPreference, boolean trimFields)
-        throws IOException {
+    static DelimitedLogStructureFinder makeDelimitedLogStructureFinder(List<String> explanation, String sample, String charsetName,
+                                                                       Boolean hasByteOrderMarker, CsvPreference csvPreference,
+                                                                       boolean trimFields) throws IOException {
 
         Tuple<List<List<String>>, List<Integer>> parsed = readRows(sample, csvPreference);
         List<List<String>> rows = parsed.v1();
@@ -73,13 +72,14 @@ static SeparatedValuesLogStructureFinder makeSeparatedValuesLogStructureFinder(L
         String preamble = Pattern.compile("\n").splitAsStream(sample).limit(lineNumbers.get(1)).collect(Collectors.joining("\n", "", "\n"));
 
         char delimiter = (char) csvPreference.getDelimiterChar();
-        LogStructure.Builder structureBuilder = new LogStructure.Builder(LogStructure.Format.fromSeparator(delimiter))
+        LogStructure.Builder structureBuilder = new LogStructure.Builder(LogStructure.Format.DELIMITED)
             .setCharset(charsetName)
             .setHasByteOrderMarker(hasByteOrderMarker)
             .setSampleStart(preamble)
             .setNumLinesAnalyzed(lineNumbers.get(lineNumbers.size() - 1))
             .setNumMessagesAnalyzed(sampleRecords.size())
             .setHasHeaderRow(isHeaderInFile)
+            .setDelimiter(delimiter)
             .setInputFields(Arrays.stream(headerWithNamedBlanks).collect(Collectors.toList()));
 
         if (trimFields) {
@@ -131,10 +131,10 @@ static SeparatedValuesLogStructureFinder makeSeparatedValuesLogStructureFinder(L
             .setExplanation(explanation)
             .build();
 
-        return new SeparatedValuesLogStructureFinder(sampleMessages, structure);
+        return new DelimitedLogStructureFinder(sampleMessages, structure);
     }
 
-    private SeparatedValuesLogStructureFinder(List<String> sampleMessages, LogStructure structure) {
+    private DelimitedLogStructureFinder(List<String> sampleMessages, LogStructure structure) {
         this.sampleMessages = Collections.unmodifiableList(sampleMessages);
         this.structure = structure;
     }

x-pack/plugin/ml/log-structure-finder/src/main/java/org/elasticsearch/xpack/ml/logstructurefinder/DelimitedLogStructureFinderFactory.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ml.logstructurefinder;
+
+import org.supercsv.prefs.CsvPreference;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+
+public class DelimitedLogStructureFinderFactory implements LogStructureFinderFactory {
+
+    private final CsvPreference csvPreference;
+    private final int minFieldsPerRow;
+    private final boolean trimFields;
+
+    DelimitedLogStructureFinderFactory(char delimiter, int minFieldsPerRow, boolean trimFields) {
+        csvPreference = new CsvPreference.Builder('"', delimiter, "\n").build();
+        this.minFieldsPerRow = minFieldsPerRow;
+        this.trimFields = trimFields;
+    }
+
+    /**
+     * Rules are:
+     * - It must contain at least two complete records
+     * - There must be a minimum number of fields per record (otherwise files with no commas could be treated as CSV!)
+     * - Every record except the last must have the same number of fields
+     * The reason the last record is allowed to have fewer fields than the others is that
+     * it could have been truncated when the file was sampled.
+     */
+    @Override
+    public boolean canCreateFromSample(List<String> explanation, String sample) {
+        String formatName;
+        switch ((char) csvPreference.getDelimiterChar()) {
+            case ',':
+                formatName = "CSV";
+                break;
+            case '\t':
+                formatName = "TSV";
+                break;
+            default:
+                formatName = Character.getName(csvPreference.getDelimiterChar()).toLowerCase(Locale.ROOT) + " delimited values";
+                break;
+        }
+        return DelimitedLogStructureFinder.canCreateFromSample(explanation, sample, minFieldsPerRow, csvPreference, formatName);
+    }
+
+    @Override
+    public LogStructureFinder createFromSample(List<String> explanation, String sample, String charsetName, Boolean hasByteOrderMarker)
+        throws IOException {
+        return DelimitedLogStructureFinder.makeDelimitedLogStructureFinder(explanation, sample, charsetName, hasByteOrderMarker,
+            csvPreference, trimFields);
+    }
+}