Fix #1438 - optionally gather fields after ingest exception

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

@@ -11,6 +11,7 @@
 import datawave.data.type.OneToManyNormalizerType;
 import datawave.ingest.config.IngestConfiguration;
 import datawave.ingest.config.IngestConfigurationFactory;
+import datawave.ingest.data.RawRecordContainer;
 import datawave.ingest.data.Type;
 import datawave.ingest.data.TypeRegistry;
 import datawave.ingest.data.config.DataTypeHelperImpl;
@@ -697,6 +698,12 @@ protected NormalizedContentInterface normalizeFieldValue(NormalizedContentInterf
         return copy;
     }
 
+    @Override
+    public void getEventFields(RawRecordContainer value, Multimap<String,NormalizedContentInterface> fields) {
+        // default implementation calls legacy method
+        fields.putAll(this.getEventFields(value));
+    }
+
     /**
      * This is a helper routine that will create a normalize a field out of a base normalized field.
      * 

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

@@ -43,13 +43,19 @@ default boolean isShardExcluded(String fieldName) {
     }
 
     /**
-     * Fully parse the raw record and return a map of field names and values.
-     * 
-     * @param value
-     * @return
+     * Deprecated. Use #getEventFields(value, fields)
      */
+    @Deprecated
     Multimap<String,NormalizedContentInterface> getEventFields(RawRecordContainer value);
 
+    /**
+     * Fully parse the raw record and update the provided multimap of field names and values, with a partial update in the event of an exception.
+     *
+     * @param value
+     * @param fields
+     */
+    void getEventFields(RawRecordContainer value, Multimap<String,NormalizedContentInterface> fields);
+
     Multimap<String,NormalizedContentInterface> normalizeMap(Multimap<String,NormalizedContentInterface> fields);
 
     Multimap<String,NormalizedContentInterface> normalize(Multimap<String,String> fields);

warehouse/ingest-core/src/main/java/datawave/ingest/mapreduce/FieldHarvester.java

@@ -0,0 +1,245 @@
+package datawave.ingest.mapreduce;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+import datawave.data.normalizer.DateNormalizer;
+import datawave.ingest.data.RawRecordContainer;
+import datawave.ingest.data.config.NormalizedContentInterface;
+import datawave.ingest.data.config.NormalizedFieldAndValue;
+import datawave.ingest.data.config.ingest.CompositeIngest;
+import datawave.ingest.data.config.ingest.FilterIngest;
+import datawave.ingest.data.config.ingest.IngestHelperInterface;
+import datawave.ingest.data.config.ingest.VirtualIngest;
+import datawave.ingest.time.Now;
+import datawave.util.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.log4j.Logger;
+import org.apache.log4j.NDC;
+
+import java.util.Date;
+import java.util.Map;
+
+/**
+ * Encapsulates the logic for extracting fields from a record, making use of a provided IngestHelperInterface. Generates virtual fields, composite fields, and
+ * supplemental fields (like LOAD_DATE, ORIG_FILE, and RAW_FILE). Some logic for handling errors is also included here: extracting salvagable fields if any
+ * exception occurs, and detecting if there were field errors (indicating a normalization failure).
+ */
+public class FieldHarvester {
+    private static final Logger log = Logger.getLogger(FieldHarvester.class);
+
+    private static Now now = Now.getInstance();
+
+    public static final String LOAD_DATE_FIELDNAME = "LOAD_DATE";
+    public static final String SEQUENCE_FILE_FIELDNAME = "ORIG_FILE";
+    public static final String RAW_FILE_FIELDNAME = "RAW_FILE";
+    public static final String LOAD_SEQUENCE_FILE_NAME = "ingest.event.mapper.load.seq.filename";
+    public static final String TRIM_SEQUENCE_FILE_NAME = "ingest.event.mapper.trim.sequence.filename";
+    public static final String LOAD_RAW_FILE_NAME = "ingest.event.mapper.load.raw.filename";
+
+    private boolean createSequenceFileName;
+    private boolean trimSequenceFileName;
+    private boolean createRawFileName;
+    private final DateNormalizer dateNormalizer = new DateNormalizer();
+
+    private static final String SRC_FILE_DEL = "|";
+    private Exception originalException;
+
+    public FieldHarvester(Configuration configuration) {
+        this.createSequenceFileName = configuration.getBoolean(LOAD_SEQUENCE_FILE_NAME, true);
+        this.trimSequenceFileName = configuration.getBoolean(TRIM_SEQUENCE_FILE_NAME, true);
+        this.createRawFileName = configuration.getBoolean(LOAD_RAW_FILE_NAME, true);
+    }
+
+    /**
+     * Updates "fields" with extracted, derived, and automatically generated fields. Will capture exception along the way and attempt to add salvaged fields
+     * before rethrowing the exception.
+     *
+     * @param fields
+     *            the Multimap to modify with extracted and generated fields
+     * @param ingestHelper
+     *            interface to use for field extraction
+     * @param value
+     *            the record from which the fields will be extracted
+     * @param offset
+     *            record offset within the source file
+     * @param splitStart
+     *            the splitStart for the record
+     */
+    public void extractFields(Multimap<String,NormalizedContentInterface> fields, IngestHelperInterface ingestHelper, RawRecordContainer value, long offset,
+                    String splitStart) throws Exception {
+        // reset exception-in-extraction tracking
+        this.originalException = null;
+
+        // "candidateFields" holds the fields that will eventually be added to "fields"
+        Multimap<String,NormalizedContentInterface> candidateFields = HashMultimap.create();
+
+        try {
+            // parse the record into its candidate field names and values using the IngestHelperInterface.
+            ingestHelper.getEventFields(value, candidateFields);
+        } catch (Exception exception) {
+            // delay throwing the exception to attempt salvaging
+            this.originalException = exception;
+        }
+
+        try {
+            // try adding supplemental fields to candidateFields, whether or not they were complete
+            addSupplementalFields(value, offset, splitStart, ingestHelper, candidateFields);
+        } catch (Exception exception) {
+            if (null == this.originalException) {
+                this.originalException = exception;
+            } else {
+                // preserve original exception and log the latest exception
+                log.error("A secondary exception occurred while adding supplemental fields", exception);
+            }
+        }
+
+        // add candidateFields to fields, even if there was an error
+        // identify if any individual fields contain an error
+        addFieldsAndDetectFieldErrors(fields, candidateFields);
+
+        if (null != this.originalException) {
+            log.error("Rethrowing original exception after completing field extraction.");
+            throw originalException;
+        }
+    }
+
+    @VisibleForTesting
+    boolean hasError() {
+        return null != this.originalException;
+    }
+
+    @VisibleForTesting
+    Exception getOriginalException() {
+        return this.originalException;
+    }
+
+    private void addSupplementalFields(RawRecordContainer value, long offset, String splitStart, IngestHelperInterface ingestHelper,
+                    Multimap<String,NormalizedContentInterface> fields) {
+        addVirtualFields(ingestHelper, fields);
+        addCompositeFields(ingestHelper, fields);
+        addLoadDateField(fields);
+        addFileNameFields(value, offset, splitStart, fields);
+        applyFieldFilters(ingestHelper, fields);
+    }
+
+    private void addVirtualFields(IngestHelperInterface ingestHelper, Multimap<String,NormalizedContentInterface> newFields) {
+        // Also get the virtual fields, if applicable.
+        if (null != newFields && ingestHelper instanceof VirtualIngest) {
+            VirtualIngest vHelper = (VirtualIngest) ingestHelper;
+            Multimap<String,NormalizedContentInterface> virtualFields = vHelper.getVirtualFields(newFields);
+            for (Map.Entry<String,NormalizedContentInterface> v : virtualFields.entries())
+                newFields.put(v.getKey(), v.getValue());
+        }
+    }
+
+    private void addCompositeFields(IngestHelperInterface ingestHelper, Multimap<String,NormalizedContentInterface> newFields) {
+        // Also get the composite fields, if applicable
+        if (null != newFields && ingestHelper instanceof CompositeIngest) {
+            CompositeIngest vHelper = (CompositeIngest) ingestHelper;
+            Multimap<String,NormalizedContentInterface> compositeFields = vHelper.getCompositeFields(newFields);
+            for (String fieldName : compositeFields.keySet()) {
+                // if this is an overloaded composite field, we are replacing the existing field data
+                if (vHelper.isOverloadedCompositeField(fieldName)) {
+                    newFields.removeAll(fieldName);
+                }
+                newFields.putAll(fieldName, compositeFields.get(fieldName));
+            }
+        }
+    }
+
+    private void addLoadDateField(Multimap<String,NormalizedContentInterface> newFields) {
+        // Create a LOAD_DATE parameter, which is the current time in milliseconds, for all datatypes
+        long loadDate = now.get();
+        NormalizedFieldAndValue loadDateValue = new NormalizedFieldAndValue(LOAD_DATE_FIELDNAME, Long.toString(loadDate));
+        // set an indexed field value for use by the date index data type handler
+        loadDateValue.setIndexedFieldValue(dateNormalizer.normalizeDelegateType(new Date(loadDate)));
+        newFields.put(LOAD_DATE_FIELDNAME, loadDateValue);
+    }
+
+    private void addRawFileField(RawRecordContainer value, Multimap<String,NormalizedContentInterface> newFields, String seqFileName) {
+        if (createRawFileName && !value.getRawFileName().isEmpty() && !value.getRawFileName().equals(seqFileName)) {
+            newFields.put(RAW_FILE_FIELDNAME, new NormalizedFieldAndValue(RAW_FILE_FIELDNAME, value.getRawFileName()));
+        }
+    }
+
+    private void addOrigFileField(Multimap<String,NormalizedContentInterface> newFields, long offset, String splitStart, String seqFileName) {
+        if (null != seqFileName) {
+            StringBuilder seqFile = new StringBuilder(seqFileName);
+
+            seqFile.append(SRC_FILE_DEL).append(offset);
+
+            if (null != splitStart) {
+                seqFile.append(SRC_FILE_DEL).append(splitStart);
+            }
+
+            newFields.put(SEQUENCE_FILE_FIELDNAME, new NormalizedFieldAndValue(SEQUENCE_FILE_FIELDNAME, seqFile.toString()));
+        }
+    }
+
+    private String getSeqFileName() {
+        String seqFileName;
+        seqFileName = NDC.peek();
+
+        if (trimSequenceFileName) {
+            seqFileName = StringUtils.substringAfterLast(seqFileName, "/");
+        }
+        return seqFileName;
+    }
+
+    private void addFileNameFields(RawRecordContainer value, long offset, String splitStart, Multimap<String,NormalizedContentInterface> newFields) {
+        String seqFileName = null;
+
+        if (createSequenceFileName) {
+            seqFileName = getSeqFileName();
+
+            // place the sequence filename into the event
+            addOrigFileField(newFields, offset, splitStart, seqFileName);
+        }
+
+        addRawFileField(value, newFields, seqFileName);
+    }
+
+    private void applyFieldFilters(IngestHelperInterface ingestHelper, Multimap<String,NormalizedContentInterface> newFields) {
+        // Also if this helper needs to filter the fields before returning, apply now
+        if (ingestHelper instanceof FilterIngest) {
+            FilterIngest fHelper = (FilterIngest) ingestHelper;
+            fHelper.filter(newFields);
+        }
+    }
+
+    /**
+     * Adds candidateFields to fields. Looks at each of the candidate fields, inspection for field errors. Sets the field harvester's exception field if any
+     * field errors were found.
+     */
+    private void addFieldsAndDetectFieldErrors(Multimap<String,NormalizedContentInterface> fields, Multimap<String,NormalizedContentInterface> candidateFields) {
+        if (null == candidateFields) {
+            return;
+        }
+        Throwable fieldError = null;
+        for (Map.Entry<String,NormalizedContentInterface> entry : candidateFields.entries()) {
+            // noinspection ThrowableResultOfMethodCallIgnored
+            if (null != entry.getValue().getError()) {
+                fieldError = entry.getValue().getError();
+            }
+            fields.put(entry.getKey(), entry.getValue());
+        }
+        if (null != fieldError) {
+            if (null == this.originalException) {
+                this.originalException = new FieldNormalizationError("Failed getting all fields", fieldError);
+            } else {
+                // preserve original exception
+                log.error("A field exception was observed while adding fields", fieldError);
+            }
+        }
+    }
+
+    public static class FieldNormalizationError extends Exception {
+
+        private static final long serialVersionUID = 1L;
+
+        public FieldNormalizationError(String message, Throwable cause) {
+            super(message, cause);
+        }
+    }
+}

warehouse/ingest-core/src/main/java/datawave/ingest/mapreduce/handler/edge/ProtobufEdgeDataTypeHandler.java

@@ -20,7 +20,7 @@
 import datawave.ingest.data.config.GroupedNormalizedContentInterface;
 import datawave.ingest.data.config.NormalizedContentInterface;
 import datawave.ingest.data.config.ingest.IngestHelperInterface;
-import datawave.ingest.mapreduce.EventMapper;
+import datawave.ingest.mapreduce.FieldHarvester;
 import datawave.ingest.mapreduce.handler.ExtendedDataTypeHandler;
 import datawave.ingest.mapreduce.handler.edge.define.EdgeDataBundle;
 import datawave.ingest.mapreduce.handler.edge.define.EdgeDefinition;
@@ -500,7 +500,7 @@ public long process(KEYIN key, RawRecordContainer event, Multimap<String,Normali
         }
 
         // Get the load date of the event from the fields map
-        Collection<NormalizedContentInterface> loadDates = fields.get(EventMapper.LOAD_DATE_FIELDNAME);
+        Collection<NormalizedContentInterface> loadDates = fields.get(FieldHarvester.LOAD_DATE_FIELDNAME);
         if (!loadDates.isEmpty()) {
             NormalizedContentInterface nci = loadDates.iterator().next();
             Date date = new Date(Long.parseLong(nci.getEventFieldValue()));

warehouse/ingest-core/src/test/java/datawave/ingest/data/RawRecordContainerImplTest.java

@@ -19,6 +19,7 @@
 import datawave.ingest.config.RawRecordContainerImpl;
 import datawave.ingest.data.config.MarkingsHelper;
 
+import datawave.util.TypeRegistryTestSetup;
 import org.apache.hadoop.conf.Configuration;
 import org.junit.Before;
 import org.junit.Test;
@@ -53,8 +54,7 @@ public void setUp() throws Exception {
         conf.set("samplecsv" + TypeRegistry.INGEST_HELPER, TestCSVIngestHelper.class.getName());
         conf.set("samplecsv.reader.class", TestCSVReader.class.getName());
         conf.set("samplecsv" + MarkingsHelper.DEFAULT_MARKING, "PUBLIC|PRIVATE");
-        TypeRegistry.reset();
-        TypeRegistry.getInstance(conf);
+        TypeRegistryTestSetup.resetTypeRegistry(conf);
         dataType = TypeRegistry.getType("samplecsv");
     }
 

warehouse/ingest-core/src/test/java/datawave/ingest/data/config/DataTypeHelperImplTest.java

@@ -1,8 +1,7 @@
 package datawave.ingest.data.config;
 
-import datawave.ingest.data.TypeRegistry;
-
 import datawave.policy.IngestPolicyEnforcer;
+import datawave.util.TypeRegistryTestSetup;
 import org.apache.hadoop.conf.Configuration;
 import org.junit.Assert;
 import org.junit.Before;
@@ -25,8 +24,7 @@ public void setup() {
     @Test(expected = IllegalArgumentException.class)
     public void testInvalidConfig() {
         DataTypeHelperImpl helper = new DataTypeHelperImpl();
-        TypeRegistry.reset();
-        TypeRegistry.getInstance(conf);
+        TypeRegistryTestSetup.resetTypeRegistry(conf);
         helper.setup(conf);
     }
 
@@ -36,8 +34,7 @@ public void testValidConfig() throws Exception {
         Assert.assertNotNull(configStream);
         conf.addResource(configStream);
         Assert.assertThat(conf.get("data.name"), is("fake"));
-        TypeRegistry.reset();
-        TypeRegistry.getInstance(conf);
+        TypeRegistryTestSetup.resetTypeRegistry(conf);
         DataTypeHelperImpl helper = new DataTypeHelperImpl();
         helper.setup(conf);
 
@@ -54,8 +51,7 @@ public void testDowncaseFields() throws Exception {
         Assert.assertNotNull(configStream);
         conf.addResource(configStream);
         conf.set("fake" + DataTypeHelper.Properties.DOWNCASE_FIELDS, "one,two,three,FOUR");
-        TypeRegistry.reset();
-        TypeRegistry.getInstance(conf);
+        TypeRegistryTestSetup.resetTypeRegistry(conf);
         DataTypeHelperImpl helper = new DataTypeHelperImpl();
         helper.setup(conf);
 

warehouse/ingest-core/src/test/java/datawave/ingest/data/config/NormalizedFieldAndValueTest.java

@@ -21,7 +21,7 @@ public static class NonGroupedInstance implements NormalizedContentInterface {
         private Map<String,String> _markings;
         private Throwable _error;
 
-        protected NonGroupedInstance() {
+        public NonGroupedInstance() {
             _fieldName = "TestNonGroupedInstance";
 
             _indexedFieldName = "TestIndexedField";
@@ -141,7 +141,7 @@ public static class GroupedInstance implements GroupedNormalizedContentInterface
         private String _group;
         private String _subGroup;
 
-        protected GroupedInstance() {
+        public GroupedInstance() {
             _fieldName = "TestNonGroupedInstance";
 
             _indexedFieldName = "TestIndexedField";