Fix CSVReaderBase validation, split, timestamp, and progress bugs

warehouse/ingest-core/src/main/java/datawave/ingest/data/config/CSVHelper.java

@@ -1,6 +1,7 @@
 package datawave.ingest.data.config;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -266,6 +267,10 @@ public boolean isFieldRequired(final String fieldName) {
         return _hasReqFields && _requiredFields.contains(fieldName);
     }
 
+    public Set<String> getRequiredFields() {
+        return _hasReqFields ? Collections.unmodifiableSet(_requiredFields) : Collections.emptySet();
+    }
+
     /**
      * @return datatype specific field header
      */

warehouse/ingest-csv/src/main/java/datawave/ingest/csv/mr/input/CSVReaderBase.java

@@ -1,6 +1,8 @@
 package datawave.ingest.csv.mr.input;
 
 import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
 
 import org.apache.commons.lang.StringEscapeUtils;
 import org.apache.commons.lang.StringUtils;
@@ -48,11 +50,13 @@ public class CSVReaderBase extends LongLineEventRecordReader implements EventRec
      * The size of the InputSplit * 4.
      */
     private long totalSize;
+    private boolean firstSplit = true;
 
     /**
      * Primary DataTypeHelper for CSV records.
      */
     private CSVHelper csvHelper;
+    private Set<String> processedFieldNames;
 
     /**
      * Splits raw input records Strings according to the configured separator.
@@ -73,7 +77,7 @@ public LongWritable getCurrentKey() {
      */
     @Override
     public boolean nextKeyValue() throws IOException {
-        if (counter == 0 && csvHelper.skipHeaderRow())
+        if (counter == 0 && csvHelper.skipHeaderRow() && firstSplit)
             super.nextKeyValue();
         counter++;
 
@@ -88,11 +92,16 @@ public void setInputDate(final long time) {
     @Override
     public void initialize(final InputSplit genericSplit, final TaskAttemptContext context) throws IOException {
         super.initialize(genericSplit, context);
-        setInputDate(System.currentTimeMillis());
+        initializeFirstSplit(genericSplit);
+        setInputDate(getRawInputFileTimestamp() > 0 ? getRawInputFileTimestamp() : System.currentTimeMillis());
         initializeRawFileName(genericSplit);
         initializeTotalSize(genericSplit);
     }
 
+    private void initializeFirstSplit(final InputSplit genericSplit) {
+        firstSplit = !(genericSplit instanceof FileSplit) || ((FileSplit) genericSplit).getStart() == 0;
+    }
+
     public void initializeRawFileName(final InputSplit genericSplit) {
         if (genericSplit instanceof FileSplit) {
             final FileSplit fs = (FileSplit) genericSplit;
@@ -156,6 +165,7 @@ public RawRecordContainer getEvent() {
 
         final String[] rawEventFields = _tokenizer.getTokenArray();
         final String[] header = csvHelper.getHeader();
+        processedFieldNames = csvHelper.hasRequiredFields() ? new HashSet<>() : null;
 
         // If the event date field name is not specified in the configuration, then set the event date to the file modification time.
         if (StringUtils.isEmpty(eventDateFieldName))
@@ -183,6 +193,8 @@ public RawRecordContainer getEvent() {
             }
         }
 
+        checkMissingRequiredFields();
+
         // decorate with additional data (used by overriding classes)
         decorateEvent();
 
@@ -217,6 +229,10 @@ protected void decorateEvent() { /* default is noop */}
 
     @Override
     protected void checkField(final String name, final String value) {
+        if (processedFieldNames != null) {
+            processedFieldNames.add(name);
+        }
+
         super.checkField(name, value);
 
         if (csvHelper.isFieldRequired(name) && StringUtils.isEmpty(value)) {
@@ -225,6 +241,17 @@ protected void checkField(final String name, final String value) {
         }
     }
 
+    private void checkMissingRequiredFields() {
+        if (processedFieldNames != null) {
+            for (String requiredField : csvHelper.getRequiredFields()) {
+                if (!processedFieldNames.contains(requiredField)) {
+                    event.addError(RawDataErrorNames.MISSING_DATA_ERROR);
+                    log.error("Missing required field: {}", requiredField);
+                }
+            }
+        }
+    }
+
     /**
      * Creates a new StrTokenizer based on the configuration.
      *
@@ -314,6 +341,8 @@ public void setFileModificationTime(long fileModificationTime) {
 
     @Override
     public float getProgress() {
+        if (totalSize <= 0)
+            return 0f;
         return Math.min(1f, (float) processedSize / (float) totalSize);
     }
 

warehouse/ingest-csv/src/test/java/datawave/ingest/csv/CSVReaderBaseTest.java

@@ -0,0 +1,107 @@
+package datawave.ingest.csv;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import datawave.ingest.csv.mr.input.CSVReaderBase;
+import datawave.ingest.data.RawRecordContainer;
+import datawave.ingest.data.TypeRegistry;
+import datawave.ingest.data.config.CSVHelper;
+import datawave.ingest.data.config.DataTypeHelper;
+import datawave.ingest.input.reader.EventRecordReader;
+import datawave.policy.IngestPolicyEnforcer;
+
+public class CSVReaderBaseTest {
+    private static final String DATA_TYPE = "mycsv";
+    private static final String ALL_CONFIG_FILE = "config/ingest/all-config.xml";
+    private static final String CSV_CONFIG_FILE = "config/ingest/csv-ingest-config.xml";
+
+    @Rule
+    public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+    private Configuration conf;
+
+    @Before
+    public void setup() {
+        conf = new Configuration();
+        conf.addResource(ClassLoader.getSystemResource(ALL_CONFIG_FILE));
+        conf.addResource(ClassLoader.getSystemResource(CSV_CONFIG_FILE));
+        conf.set("all" + DataTypeHelper.Properties.INGEST_POLICY_ENFORCER_CLASS, IngestPolicyEnforcer.NoOpIngestPolicyEnforcer.class.getName());
+        conf.set(DATA_TYPE + EventRecordReader.Properties.EVENT_DATE_FIELD_NAME, "");
+        conf.set(DATA_TYPE + CSVHelper.DATA_HEADER, "VALUE");
+
+        TypeRegistry.reset();
+        TypeRegistry.getInstance(conf);
+    }
+
+    @After
+    public void tearDown() {
+        TypeRegistry.reset();
+    }
+
+    @Test
+    public void testSkipHeaderRowDoesNotSkipFirstRecordInNonFirstSplit() throws Exception {
+        conf.set(DATA_TYPE + CSVHelper.SKIP_CSV_HEADER_ROW, "true");
+
+        String data = "VALUE\nfirst\nsecond\n";
+        File file = writeFile(data);
+        long splitStart = data.indexOf("first") + 2;
+        CSVReaderBase reader = initializeReader(file, splitStart, data.length() - splitStart);
+
+        assertTrue(reader.nextKeyValue());
+        assertEquals("second", reader.getCurrentValue().toString());
+        assertFalse(reader.nextKeyValue());
+    }
+
+    @Test
+    public void testDefaultTimestampUsesRawFileModificationTime() throws Exception {
+        String data = "value\n";
+        File file = writeFile(data);
+        long fileModificationTime = 1_700_000_000_000L;
+        assertTrue(file.setLastModified(fileModificationTime));
+
+        CSVReaderBase reader = initializeReader(file, 0, data.length());
+        assertTrue(reader.nextKeyValue());
+        RawRecordContainer event = reader.getEvent();
+
+        assertEquals(file.lastModified(), event.getTimestamp());
+    }
+
+    @Test
+    public void testProgressIsZeroWhenTotalSizeIsZero() {
+        CSVReaderBase reader = new CSVReaderBase();
+        reader.setProcessedSize(1);
+        reader.setTotalSize(0);
+
+        assertEquals(0f, reader.getProgress(), 0f);
+    }
+
+    private File writeFile(String data) throws Exception {
+        File file = temporaryFolder.newFile("input.csv");
+        Files.write(file.toPath(), data.getBytes(StandardCharsets.UTF_8));
+        return file;
+    }
+
+    private CSVReaderBase initializeReader(File file, long start, long length) throws Exception {
+        CSVReaderBase reader = new CSVReaderBase();
+        FileSplit split = new FileSplit(new Path(file.toURI().toString()), start, length, new String[0]);
+        reader.initialize(split, new TaskAttemptContextImpl(conf, new TaskAttemptID()));
+        return reader;
+    }
+}

warehouse/ingest-csv/src/test/java/datawave/ingest/csv/CSVRecordReaderRequiredFieldsTest.java

@@ -0,0 +1,71 @@
+package datawave.ingest.csv;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import datawave.ingest.csv.mr.input.CSVRecordReader;
+import datawave.ingest.data.RawDataErrorNames;
+import datawave.ingest.data.RawRecordContainer;
+import datawave.ingest.data.TypeRegistry;
+import datawave.ingest.data.config.CSVHelper;
+import datawave.ingest.data.config.DataTypeHelper;
+import datawave.ingest.input.reader.EventRecordReader;
+import datawave.policy.IngestPolicyEnforcer;
+
+public class CSVRecordReaderRequiredFieldsTest {
+    private static final String DATA_TYPE = "mycsv";
+    private static final String ALL_CONFIG_FILE = "config/ingest/all-config.xml";
+    private static final String CSV_CONFIG_FILE = "config/ingest/csv-ingest-config.xml";
+
+    private Configuration conf;
+    private CSVRecordReader reader;
+
+    @Before
+    public void setup() {
+        conf = new Configuration();
+        conf.addResource(ClassLoader.getSystemResource(ALL_CONFIG_FILE));
+        conf.addResource(ClassLoader.getSystemResource(CSV_CONFIG_FILE));
+        conf.set("all" + DataTypeHelper.Properties.INGEST_POLICY_ENFORCER_CLASS, IngestPolicyEnforcer.NoOpIngestPolicyEnforcer.class.getName());
+        conf.set(DATA_TYPE + EventRecordReader.Properties.EVENT_DATE_FIELD_NAME, "");
+        conf.set(DATA_TYPE + CSVHelper.DATA_HEADER_ENABLED, "false");
+        conf.set(DATA_TYPE + CSVHelper.PROCESS_EXTRA_FIELDS, "true");
+        conf.setStrings(DATA_TYPE + CSVHelper.REQUIRED_FIELDS, "REQUIRED");
+
+        TypeRegistry.reset();
+        TypeRegistry.getInstance(conf);
+        reader = new CSVRecordReader();
+        reader.setup(conf);
+    }
+
+    @After
+    public void tearDown() {
+        TypeRegistry.reset();
+    }
+
+    @Test
+    public void testMissingRequiredFieldInNameValueModeAddsError() {
+        // Verify required fields are checked even when the field is absent from name=value input.
+        RawRecordContainer event = read("EVENT_ID=test.a2025001,OPTIONAL=value");
+
+        assertTrue(event.getErrors().toString(), event.getErrors().contains(RawDataErrorNames.MISSING_DATA_ERROR));
+    }
+
+    @Test
+    public void testPresentRequiredFieldInNameValueModeDoesNotAddError() {
+        // Verify present required name=value fields satisfy the required-field check.
+        RawRecordContainer event = read("EVENT_ID=test.a2025001,OPTIONAL=value,REQUIRED=value");
+
+        assertFalse(event.getErrors().toString(), event.getErrors().contains(RawDataErrorNames.MISSING_DATA_ERROR));
+    }
+
+    private RawRecordContainer read(String rawData) {
+        reader.setCurrentValue(new Text(rawData));
+        return reader.getEvent();
+    }
+}