Add delete_source parameter to the csv processor (opensearch-project#…

…4828) Signed-off-by: Taylor Gray <tylgry@amazon.com>
dlvenable · Aug 12, 2024 · 2f21a43 · 2f21a43
1 parent aa50a1d
commit 2f21a43
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 0 deletions.
diff --git a/...rocessor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessor.java b/...rocessor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessor.java
@@ -99,6 +99,10 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
                 if (thisEventHasHeaderSource && Boolean.TRUE.equals(config.isDeleteHeader())) {
                     event.delete(config.getColumnNamesSourceKey());
                 }
+
+                if (config.isDeleteSource()) {
+                    event.delete(config.getSource());
+                }
             } catch (final IOException e) {
                 csvInvalidEventsCounter.increment();
                 LOG.error(EVENT, "An exception occurred while reading event [{}]", event, e);

diff --git a/...or/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorConfig.java b/...or/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorConfig.java
@@ -62,6 +62,10 @@ public class CsvProcessorConfig {
             "the processor should be applied to the event.")
     private String csvWhen;
 
+    @JsonPropertyDescription("If true, the configured source field will be deleted after the CSV data is parsed into separate fields.")
+    @JsonProperty
+    private boolean deleteSource = false;
+
     /**
      * The field of the Event that contains the CSV data to be processed.
      *
@@ -120,6 +124,8 @@ public List<String> getColumnNames() {
 
     public String getCsvWhen() { return csvWhen; }
 
+    public Boolean isDeleteSource() { return deleteSource; }
+
     @AssertTrue(message = "delimiter must be exactly one character.")
     boolean isValidDelimiter() {
         return delimiter.length() == 1;

diff --git a/...ssor/src/test/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorTest.java b/...ssor/src/test/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorTest.java
@@ -56,6 +56,7 @@ void setup() {
         lenient().when(processorConfig.getQuoteCharacter()).thenReturn(defaultConfig.getQuoteCharacter());
         lenient().when(processorConfig.getColumnNamesSourceKey()).thenReturn(defaultConfig.getColumnNamesSourceKey());
         lenient().when(processorConfig.getColumnNames()).thenReturn(defaultConfig.getColumnNames());
+        lenient().when(processorConfig.isDeleteSource()).thenReturn(false);
 
         lenient().when(pluginMetrics.counter(CsvProcessor.CSV_INVALID_EVENTS)).thenReturn(csvInvalidEventsCounter);
 
@@ -66,6 +67,24 @@ private CsvProcessor createObjectUnderTest() {
         return new CsvProcessor(pluginMetrics, processorConfig, expressionEvaluator);
     }
 
+    @Test
+    void delete_source_true_deletes_the_source() {
+        when(processorConfig.isDeleteSource()).thenReturn(true);
+
+        when(processorConfig.getSource()).thenReturn("different_source");
+
+        final Map<String, Object> eventData = new HashMap<>();
+        eventData.put("different_source","1,2,3");
+        final Record<Event> eventUnderTest = buildRecordWithEvent(eventData);
+
+        final List<Record<Event>> editedEvents = (List<Record<Event>>) csvProcessor.doExecute(Collections.singletonList(eventUnderTest));
+        final Event parsedEvent = getSingleEvent(editedEvents);
+        assertThat(parsedEvent.containsKey("different_source"), equalTo(false));
+        assertThatKeyEquals(parsedEvent, "column1", "1");
+        assertThatKeyEquals(parsedEvent, "column2", "2");
+        assertThatKeyEquals(parsedEvent, "column3", "3");
+    }
+
     @Test
     void do_nothing_when_source_is_null_value_or_does_not_exist_in_the_Event() {