From 2f21a437091b2a41cfe6a53df9978ec0b456fd2c Mon Sep 17 00:00:00 2001
From: Taylor Gray <tylgry@amazon.com>
Date: Mon, 12 Aug 2024 16:21:05 -0500
Subject: [PATCH] Add delete_source parameter to the csv processor (#4828)

Signed-off-by: Taylor Gray <tylgry@amazon.com>
---
 .../plugins/processor/csv/CsvProcessor.java   |  4 ++++
 .../processor/csv/CsvProcessorConfig.java     |  6 ++++++
 .../processor/csv/CsvProcessorTest.java       | 19 +++++++++++++++++++
 3 files changed, 29 insertions(+)

diff --git a/data-prepper-plugins/csv-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessor.java b/data-prepper-plugins/csv-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessor.java
index a9f99e5862..3e8780a417 100644
--- a/data-prepper-plugins/csv-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessor.java
+++ b/data-prepper-plugins/csv-processor/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/data-prepper-plugins/csv-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorConfig.java b/data-prepper-plugins/csv-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorConfig.java
index 8c770b597a..fb803798b2 100644
--- a/data-prepper-plugins/csv-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorConfig.java
+++ b/data-prepper-plugins/csv-processor/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/data-prepper-plugins/csv-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorTest.java b/data-prepper-plugins/csv-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorTest.java
index 5239679fab..cacfedb609 100644
--- a/data-prepper-plugins/csv-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorTest.java
+++ b/data-prepper-plugins/csv-processor/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() {