Skip to content

Commit

Permalink
Support null values (#5342)
Browse files Browse the repository at this point in the history 
Signed-off-by: Hai Yan <[email protected]>
  • Loading branch information
@oeyh
oeyh authored Jan 17, 2025
1 parent 41e3227 commit 3b12ca1
Show file tree
Hide file tree
Showing 2 changed files with 72 additions and 8 deletions.
14 changes: 6 additions & 8 deletions ...ain/java/org/opensearch/dataprepper/plugins/processor/mutateevent/MapToListProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
package org.opensearch.dataprepper.plugins.processor.mutateevent;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.opensearch.dataprepper.expression.ExpressionEvaluator;
import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
Expand All @@ -22,6 +21,7 @@
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
Expand All @@ -33,7 +33,6 @@
@DataPrepperPlugin(name = "map_to_list", pluginType = Processor.class, pluginConfigurationType = MapToListProcessorConfig.class)
public class MapToListProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
private static final Logger LOG = LoggerFactory.getLogger(MapToListProcessor.class);
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private final MapToListProcessorConfig config;
private final ExpressionEvaluator expressionEvaluator;
private final Set<String> excludeKeySet = new HashSet<>();
Expand Down Expand Up @@ -72,20 +71,19 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor

for (final Map.Entry<String, Object> entry : sourceMap.entrySet()) {
if (!excludeKeySet.contains(entry.getKey())) {
targetNestedList.add(List.of(entry.getKey(), entry.getValue()));
targetNestedList.add(Arrays.asList(entry.getKey(), entry.getValue()));
}

}
removeProcessedFields(sourceMap, recordEvent);
recordEvent.put(config.getTarget(), targetNestedList);
} else {
final List<Map<String, Object>> targetList = new ArrayList<>();
for (final Map.Entry<String, Object> entry : sourceMap.entrySet()) {
if (!excludeKeySet.contains(entry.getKey())) {
targetList.add(Map.of(
config.getKeyName(), entry.getKey(),
config.getValueName(), entry.getValue()
));
final Map<String, Object> listItem = new HashMap<>();
listItem.put(config.getKeyName(), entry.getKey());
listItem.put(config.getValueName(), entry.getValue());
targetList.add(listItem);
}
}
removeProcessedFields(sourceMap, recordEvent);
Expand Down
66 changes: 66 additions & 0 deletions ...java/org/opensearch/dataprepper/plugins/processor/mutateevent/MapToListProcessorTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@
import org.opensearch.dataprepper.model.record.Record;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
Expand All @@ -27,6 +29,7 @@
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.when;
Expand Down Expand Up @@ -357,6 +360,51 @@ void testFailureTagsAreAddedWhenException() {
assertThat(resultEvent.getMetadata().getTags(), is(new HashSet<>(testTags)));
}

@Test
void testMapToListSuccessWithNullValuesInMap() {

final MapToListProcessor processor = createObjectUnderTest();
final Record<Event> testRecord = createTestRecordWithNullValues();
final List<Record<Event>> resultRecord = (List<Record<Event>>) processor.doExecute(Collections.singletonList(testRecord));

assertThat(resultRecord.size(), is(1));

final Event resultEvent = resultRecord.get(0).getData();
List<Map<String, Object>> resultList = resultEvent.get("my-list", List.class);

assertThat(resultList.size(), is(2));
Map<String, Object> resultMapWithNullValue = new HashMap<>();
resultMapWithNullValue.put("key", "key2");
resultMapWithNullValue.put("value", null);
assertThat(resultList, containsInAnyOrder(
Map.of("key", "key1", "value", "value1"),
resultMapWithNullValue
));
assertThat(resultEvent.containsKey("my-map"), is(true));
assertSourceMapUnchangedWithNullValues(resultEvent);
}

@Test
public void testConvertFieldToListSuccessWithNullValuesInMap() {
when(mockConfig.getConvertFieldToList()).thenReturn(true);

final MapToListProcessor processor = createObjectUnderTest();
final Record<Event> testRecord = createTestRecordWithNullValues();
final List<Record<Event>> resultRecord = (List<Record<Event>>) processor.doExecute(Collections.singletonList(testRecord));

assertThat(resultRecord.size(), is(1));

final Event resultEvent = resultRecord.get(0).getData();
List<List<Object>> resultList = resultEvent.get("my-list", List.class);

assertThat(resultList.size(), is(2));
assertThat(resultList, containsInAnyOrder(
Arrays.asList("key1", "value1"),
Arrays.asList("key2", null)
));
assertSourceMapUnchangedWithNullValues(resultEvent);
}

private MapToListProcessor createObjectUnderTest() {
return new MapToListProcessor(pluginMetrics, mockConfig, expressionEvaluator);
}
Expand Down Expand Up @@ -396,6 +444,18 @@ private Record<Event> createTestRecordWithNestedMap() {
return new Record<>(event);
}

private Record<Event> createTestRecordWithNullValues() {
final Map<String, Object> mapData = new HashMap<>();
mapData.put("key1", "value1");
mapData.put("key2", null);
final Map<String, Map<String, Object>> data = Map.of("my-map", mapData);
final Event event = JacksonEvent.builder()
.withData(data)
.withEventType("event")
.build();
return new Record<>(event);
}

private void assertSourceMapUnchanged(final Event resultEvent) {
assertThat(resultEvent.containsKey("my-map"), is(true));
assertThat(resultEvent.get("my-map/key1", String.class), is("value1"));
Expand All @@ -408,4 +468,10 @@ private void assertSourceMapUnchangedForFlatRecord(final Event resultEvent) {
assertThat(resultEvent.get("key2", String.class), is("value2"));
assertThat(resultEvent.get("key3", String.class), is("value3"));
}

private void assertSourceMapUnchangedWithNullValues(final Event resultEvent) {
assertThat(resultEvent.containsKey("my-map"), is(true));
assertThat(resultEvent.get("my-map/key1", String.class), is("value1"));
assertThat(resultEvent.get("my-map/key2", String.class), nullValue());
}
}

0 comments on commit 3b12ca1

Please sign in to comment.