[improve][core]优化相关udf，修改判空方式，防止出现字段存在但是值为null导致的异常

21 files changed, 93 insertions, 184 deletions

diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/AsnLookup.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/AsnLookup.java
index bdb3698..6d8373e 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/AsnLookup.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/AsnLookup.java
@@ -54,8 +54,9 @@ public class AsnLookup implements ScalarFunction {
     @Override
     public Event evaluate(Event event) {
 
-        if (event.getExtractedFields().containsKey(lookupFieldName)) {
-            String asn = AsnKnowledgeBaseHandler.lookUp(kbName, option, event.getExtractedFields().get(lookupFieldName).toString());
+        Object valueObj = event.getExtractedFields().get(lookupFieldName);
+        if (valueObj!=null) {
+            String asn = AsnKnowledgeBaseHandler.lookUp(kbName, option, valueObj.toString());
             if (!asn.isEmpty()) {
                 event.getExtractedFields().put(outputFieldName, asn);
             }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/DecodeBase64.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/DecodeBase64.java
index 3581d5c..663e626 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/DecodeBase64.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/DecodeBase64.java
@@ -41,15 +41,11 @@ public class DecodeBase64 implements ScalarFunction {
 
     @Override
     public Event evaluate(Event event) {
-        if (event.getExtractedFields().containsKey(valueField)) {
-
+        Object valueObj = event.getExtractedFields().get(valueField);
+        if (valueObj!=null) {
             String decodeResult = "";
-            String message =
-                    (String)
-                            event.getExtractedFields()
-                                    .get(valueField);
-            Object charset =
-                    event.getExtractedFields().getOrDefault(charsetField,"");
+            String message = (String)valueObj;
+            Object charset = event.getExtractedFields().getOrDefault(charsetField,"UTF-8");
             try {
                 if (StringUtil.isNotBlank(message)) {
                     byte[] base64decodedBytes = Base64.getDecoder().decode(message);
@@ -62,14 +58,9 @@ public class DecodeBase64 implements ScalarFunction {
             } catch (RuntimeException e) {
                 log.error("Resolve Base64 exception, exception information:" + e.getMessage());
             } catch (UnsupportedEncodingException e) {
-                log.error(
-                        "The Character Encoding ["
-                                + charset.toString()
-                                + "] is not supported.exception information:"
-                                + e.getMessage());
+                log.error("The Character Encoding [" + charset.toString() + "] is not supported.exception information:" + e.getMessage());
             }
-            event.getExtractedFields()
-                    .put(outputFieldName, decodeResult);
+            event.getExtractedFields().put(outputFieldName, decodeResult);
         }
         return event;
     }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java
index 77b3246..7bc293e 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java
@@ -54,12 +54,10 @@ public class Domain implements ScalarFunction {
         switch (option) {
             case "TOP_LEVEL_DOMAIN":
                 for (String lookupField : lookupFields){
-                    if(event.getExtractedFields().containsKey(lookupField)) {
-                        domain =
-                                getTopPrivateDomain(
-                                        (String)
-                                                event.getExtractedFields()
-                                                        .get(lookupField));
+
+                    Object valueObj = event.getExtractedFields().get(lookupField);
+                    if (valueObj!=null) {
+                        domain = getTopPrivateDomain((String) valueObj);
                         if (domain.contains(".")) {
                             domain = domain.substring(domain.indexOf(".") + 1);
                         }
@@ -71,12 +69,9 @@ public class Domain implements ScalarFunction {
                 break;
             case "FIRST_SIGNIFICANT_SUBDOMAIN":
                 for (String lookupField : lookupFields){
-                    if(event.getExtractedFields().containsKey(lookupField)) {
-                        domain =
-                                getTopPrivateDomain(
-                                        (String)
-                                                event.getExtractedFields()
-                                                        .get(lookupField));
+                    Object valueObj = event.getExtractedFields().get(lookupField);
+                    if (valueObj!=null) {
+                        domain = getTopPrivateDomain((String)valueObj);
                     }
                     if(!domain.isEmpty()){
                         break;
@@ -85,9 +80,9 @@ public class Domain implements ScalarFunction {
                 break;
             case "FQDN":
                 for (String lookupField : lookupFields) {
-                    if (event.getExtractedFields().containsKey(lookupField)) {
-                        domain = (String) event.getExtractedFields()
-                                .get(lookupField);
+                    Object valueObj = event.getExtractedFields().get(lookupField);
+                    if (valueObj!=null) {
+                        domain = (String)valueObj;
                     }
                     if(!domain.isEmpty()){
                         break;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/GeoIpLookup.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/GeoIpLookup.java
index afe1bfb..559748e 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/GeoIpLookup.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/GeoIpLookup.java
@@ -61,10 +61,10 @@ public class GeoIpLookup implements ScalarFunction {
 
     @Override
     public Event evaluate(Event event) {
-        if (event.getExtractedFields().containsKey(lookupFieldName)) {
-
+        Object valueObj = event.getExtractedFields().get(lookupFieldName);
+        if (valueObj!=null) {
             if ("IP_TO_OBJECT".equals(option)) {
-                LocationResponse response  = GeoIpKnowledgeBaseHandler.lookUpObject(kbName,event.getExtractedFields().get(lookupFieldName).toString());
+                LocationResponse response  = GeoIpKnowledgeBaseHandler.lookUpObject(kbName,valueObj.toString());
                 for (Map.Entry<String, String> entry : geoLocationFieldMapping.entrySet()) {
                     String result = "";
                     if (response!=null) {
@@ -97,8 +97,7 @@ public class GeoIpLookup implements ScalarFunction {
                     }
                 }
             } else {
-                event.getExtractedFields().put(outputFieldName, GeoIpKnowledgeBaseHandler
-                        .lookUp(kbName, option, event.getExtractedFields().get(lookupFieldName).toString()));
+                event.getExtractedFields().put(outputFieldName, GeoIpKnowledgeBaseHandler.lookUp(kbName, option, valueObj.toString()));
             }
         }
         return event;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/JsonExtract.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/JsonExtract.java
index f78b952..36c6cea 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/JsonExtract.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/JsonExtract.java
@@ -39,14 +39,9 @@ public class JsonExtract implements ScalarFunction {
 
     @Override
     public Event evaluate(Event event) {
-        if (event.getExtractedFields().containsKey(lookupFieldName)) {
-
-            String result =
-                    (String)
-                            JsonPathUtil.analysis(
-                                    event.getExtractedFields()
-                                            .get(lookupFieldName)
-                                            .toString(),expression);
+        Object valueObj = event.getExtractedFields().get(lookupFieldName);
+        if (valueObj!=null) {
+            String result = (String) JsonPathUtil.analysis(valueObj.toString(),expression);
             event.getExtractedFields().put(outputFieldName, result);
         }
         return event;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/PathCombine.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/PathCombine.java
index 874735d..65617ba 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/PathCombine.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/PathCombine.java
@@ -65,9 +65,9 @@ public class PathCombine implements ScalarFunction {
         List<String> pathBuilder = new LinkedList<>();
         for (Map.Entry<String, String> entry : pathParameters.entrySet()) {
             if (entry.getValue().isEmpty()) {
-
-                if(event.getExtractedFields().containsKey(entry.getKey())) {
-                    pathBuilder.add(event.getExtractedFields().get(entry.getKey()).toString());
+                Object valueObj = event.getExtractedFields().get(entry.getKey());
+                if (valueObj!=null) {
+                    pathBuilder.add(valueObj.toString());
                 }
                 else {
                     return event;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/UnixTimestampConverter.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/UnixTimestampConverter.java
index a601944..bff8f18 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/UnixTimestampConverter.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/UnixTimestampConverter.java
@@ -59,9 +59,9 @@ public class UnixTimestampConverter implements ScalarFunction {
 
     @Override
     public Event evaluate(Event event) {
-
-        if(event.getExtractedFields().containsKey(lookupFieldName)) {
-            Long timestamp = Long.parseLong(event.getExtractedFields().get(lookupFieldName).toString());
+        Object valueObj = event.getExtractedFields().get(lookupFieldName);
+        if (valueObj!=null) {
+            Long timestamp = Long.parseLong(valueObj.toString());
             Instant instant ;
             if (String.valueOf(timestamp).length() > 10) {
                 instant = Instant.ofEpochMilli(timestamp);
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/CollectList.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/CollectList.java
index 3921ee2..230a3ca 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/CollectList.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/CollectList.java
@@ -56,8 +56,9 @@ public class CollectList implements AggregateFunction {
 
     @Override
     public Accumulator add(Event event, Accumulator acc) {
-        if (event.getExtractedFields().containsKey(lookupField)) {
-            Object object = event.getExtractedFields().get(lookupField);
+        Object valueObj = event.getExtractedFields().get(lookupField);
+        if (valueObj != null) {
+            Object object = valueObj;
             List<Object> aggregate = (List<Object>) acc.getMetricsFields().get(outputField);
             aggregate.add(object);
             acc.getMetricsFields().put(outputField, aggregate);
@@ -77,12 +78,14 @@ public class CollectList implements AggregateFunction {
 
     @Override
     public Accumulator merge(Accumulator firstAcc, Accumulator secondAcc) {
-        if (firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
-            List<Object> firstValue = (List<Object>) firstAcc.getMetricsFields().get(outputField);
-            List<Object> secondValue = (List<Object>) secondAcc.getMetricsFields().get(outputField);
+        Object firstValueObj = firstAcc.getMetricsFields().get(outputField);
+        Object secondValueObj = secondAcc.getMetricsFields().get(outputField);
+        if (firstValueObj != null && secondValueObj != null) {
+            List<Object> firstValue = (List<Object>) firstValueObj;
+            List<Object> secondValue = (List<Object>) secondValueObj;
             firstValue.addAll(secondValue);
-        } else if (!firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
-            List<Object> secondValue = (List<Object>) secondAcc.getMetricsFields().get(outputField);
+        } else if (firstValueObj == null && secondValueObj != null) {
+            List<Object> secondValue = (List<Object>) secondValueObj;
             firstAcc.getMetricsFields().put(outputField, secondValue);
         }
         return firstAcc;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/CollectSet.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/CollectSet.java
index 9ec9b09..708a8a6 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/CollectSet.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/CollectSet.java
@@ -41,8 +41,9 @@ public class CollectSet implements AggregateFunction {
 
     @Override
     public Accumulator add(Event event, Accumulator acc) {
-        if (event.getExtractedFields().containsKey(lookupField)) {
-            Object object = event.getExtractedFields().get(lookupField);
+        Object valueObj = event.getExtractedFields().get(lookupField);
+        if (valueObj != null) {
+            Object object = valueObj;
             Set<Object> aggregate = (Set<Object>) acc.getMetricsFields().get(outputField);
             aggregate.add(object);
             acc.getMetricsFields().put(outputField, aggregate);
@@ -62,12 +63,14 @@ public class CollectSet implements AggregateFunction {
 
     @Override
     public Accumulator merge(Accumulator firstAcc, Accumulator secondAcc) {
-        if (firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
-            Set<Object> firstValue = (Set<Object>) firstAcc.getMetricsFields().get(outputField);
-            Set<Object> secondValue = (Set<Object>) secondAcc.getMetricsFields().get(outputField);
+        Object firstValueObj = firstAcc.getMetricsFields().get(outputField);
+        Object secondValueObj = secondAcc.getMetricsFields().get(outputField);
+        if (firstValueObj != null && secondValueObj != null) {
+            Set<Object> firstValue = (Set<Object>) firstValueObj;
+            Set<Object> secondValue = (Set<Object>) secondValueObj;
             firstValue.addAll(secondValue);
-        } else if (!firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
-            Set<Object> secondValue = (Set<Object>) secondAcc.getMetricsFields().get(outputField);
+        } else if (firstValueObj == null && secondValueObj !=null) {
+            Set<Object> secondValue = (Set<Object>)secondValueObj;
             firstAcc.getMetricsFields().put(outputField, secondValue);
         }
         return firstAcc;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/FirstValue.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/FirstValue.java
index a1a35be..c6ba61b 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/FirstValue.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/FirstValue.java
@@ -52,7 +52,8 @@ public class FirstValue implements AggregateFunction {
 
     @Override
     public Accumulator add(Event event, Accumulator acc) {
-        if (!acc.getMetricsFields().containsKey(outputField) && event.getExtractedFields().containsKey(lookupField)) {
+        Object valueObj = event.getExtractedFields().get(lookupField);
+        if (!acc.getMetricsFields().containsKey(outputField) && valueObj != null) {
             acc.getMetricsFields().put(outputField, event.getExtractedFields().get(lookupField));
         }
         return acc;
@@ -70,8 +71,10 @@ public class FirstValue implements AggregateFunction {
 
     @Override
     public Accumulator merge(Accumulator firstAcc, Accumulator secondAcc) {
-        if (!firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
-            firstAcc.getMetricsFields().put(outputField, secondAcc.getMetricsFields().get(outputField));
+        Object firstValueObj = firstAcc.getMetricsFields().get(outputField);
+        Object secondValueObj = secondAcc.getMetricsFields().get(outputField);
+        if (firstValueObj == null  && secondValueObj != null ) {
+            firstAcc.getMetricsFields().put(outputField, secondValueObj);
         }
         return firstAcc;
     }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/LastValue.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/LastValue.java
index 44b374e..e7435eb 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/LastValue.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/LastValue.java
@@ -70,8 +70,10 @@ public class LastValue implements AggregateFunction {
 
     @Override
     public Accumulator merge(Accumulator firstAcc, Accumulator secondAcc) {
-        if (secondAcc.getMetricsFields().containsKey(outputField)) {
-            firstAcc.getMetricsFields().put(outputField, secondAcc.getMetricsFields().get(outputField));
+
+        Object secondValueObj = secondAcc.getMetricsFields().get(outputField);
+        if (secondValueObj != null) {
+            firstAcc.getMetricsFields().put(outputField, secondValueObj);
         }
         return firstAcc;
     }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/LongCount.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/LongCount.java
index 05de38c..bf5260e 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/LongCount.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/LongCount.java
@@ -44,14 +44,16 @@ public class LongCount implements AggregateFunction {
 
     @Override
     public Accumulator merge(Accumulator firstAcc, Accumulator secondAcc) {
-        if (firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
 
-            long firstValue = (long) firstAcc.getMetricsFields().get(outputField);
-            long secondValue = (long) secondAcc.getMetricsFields().get(outputField);
+        Object firstValueObj = firstAcc.getMetricsFields().get(outputField);
+        Object secondValueObj = secondAcc.getMetricsFields().get(outputField);
+        if (firstValueObj!=null && secondValueObj!=null) {
+            long firstValue = (long) firstValueObj;
+            long secondValue = (long) secondValueObj;
             firstValue = firstValue + secondValue;
             firstAcc.getMetricsFields().put(outputField, firstValue);
-        } else if (!firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
-            Number secondValue = (Number) secondAcc.getMetricsFields().get(outputField);
+        } else if (firstValueObj==null && secondValueObj!=null) {
+            Number secondValue = (Number) secondValueObj;
             firstAcc.getMetricsFields().put(outputField, secondValue);
         }
         return firstAcc;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/NumberSum.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/NumberSum.java
index e972133..031d161 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/NumberSum.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/udaf/NumberSum.java
@@ -33,9 +33,9 @@ public class NumberSum implements AggregateFunction {
 
     @Override
     public Accumulator add(Event event, Accumulator acc) {
-
-        if (event.getExtractedFields().containsKey(lookupField)) {
-            Number val = (Number) event.getExtractedFields().get(lookupField);
+        Object valueObj = event.getExtractedFields().get(lookupField);
+        if (valueObj!=null) {
+            Number val = (Number) valueObj;
             Number aggregate = (Number) acc.getMetricsFields().getOrDefault(outputField, 0L);
             if (aggregate instanceof Long && (val instanceof Integer || val instanceof Long)) {
                 aggregate = aggregate.longValue() + val.longValue();
@@ -66,10 +66,14 @@ public class NumberSum implements AggregateFunction {
 
     @Override
     public Accumulator merge(Accumulator firstAcc, Accumulator secondAcc) {
-        if (firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
 
-            Number firstValue = (Number) firstAcc.getMetricsFields().get(outputField);
-            Number secondValue = (Number) secondAcc.getMetricsFields().get(outputField);
+
+        Object firstValueObj = firstAcc.getMetricsFields().get(outputField);
+        Object secondValueObj = secondAcc.getMetricsFields().get(outputField);
+        if (firstValueObj != null && secondValueObj != null) {
+
+            Number firstValue = (Number) firstValueObj;
+            Number secondValue = (Number) secondValueObj;
             if (firstValue instanceof Long && (secondValue instanceof Integer || secondValue instanceof Long)) {
                 firstValue = firstValue.longValue() + secondValue.longValue();
             } else if (firstValue instanceof Float || secondValue instanceof Float) {
@@ -78,8 +82,8 @@ public class NumberSum implements AggregateFunction {
                 firstValue = firstValue.doubleValue() + secondValue.doubleValue();
             }
             firstAcc.getMetricsFields().put(outputField, firstValue);
-        } else if (!firstAcc.getMetricsFields().containsKey(outputField) && secondAcc.getMetricsFields().containsKey(outputField)) {
-            Number secondValue = (Number) secondAcc.getMetricsFields().get(outputField);
+        } else if (firstValueObj == null && secondValueObj != null) {
+            Number secondValue = (Number) secondValueObj;
             firstAcc.getMetricsFields().put(outputField, secondValue);
         }
         return firstAcc;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/udtf/JsonUnroll.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/udtf/JsonUnroll.java
index 2e8eb7e..012c2df 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/udtf/JsonUnroll.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/udtf/JsonUnroll.java
@@ -57,13 +57,14 @@ public class JsonUnroll implements TableFunction {
     @Override
     public List<Event> evaluate(Event event) {
         try {
-            if(event.getExtractedFields().containsKey(lookupFieldName) ){
+            Object valueObj = event.getExtractedFields().get(lookupFieldName);
+            if (valueObj!=null) {
                 try {
                     if(path.isEmpty()){
-                        JSONArray jsonArray = JSONArray.parseArray(event.getExtractedFields().get(lookupFieldName).toString());
+                        JSONArray jsonArray = JSONArray.parseArray(valueObj.toString());
                         return parseList(jsonArray,event);
                     }else {
-                        JSONObject jsonObject = JSONObject.parseObject(event.getExtractedFields().get(lookupFieldName).toString());
+                        JSONObject jsonObject = JSONObject.parseObject(valueObj.toString());
                         Object obj = JsonPathUtil.get(jsonObject,path);
                         if(obj instanceof List || obj instanceof Array) {
                             List list = (List) obj;
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/udtf/Unroll.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/udtf/Unroll.java
index 5becb8e..9fd5a6e 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/udtf/Unroll.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/udtf/Unroll.java
@@ -42,20 +42,20 @@ public class Unroll implements TableFunction {
     @Override
     public List<Event> evaluate(Event event) {
         try {
-            if(event.getExtractedFields().containsKey(lookupFieldName)) {
-
+            Object valueObj = event.getExtractedFields().get(lookupFieldName);
+            if (valueObj!=null) {
                 if(regex.isEmpty()){
-                    if (event.getExtractedFields().get(lookupFieldName) instanceof List  ) {
-                        return parseList(event.getExtractedFields().get(lookupFieldName), event);
-                    } else if(event.getExtractedFields().get(lookupFieldName) instanceof Object[]){
-                        return parseArray(event.getExtractedFields().get(lookupFieldName), event);
+                    if (valueObj instanceof List  ) {
+                        return parseList(valueObj, event);
+                    } else if(valueObj instanceof Object[]){
+                        return parseArray(valueObj, event);
                     }else {
                         log.error("Invalid unroll ! Object is not instance of list or array. expression=" + regex);
                     }
                 }
                 else {
-                    if (event.getExtractedFields().get(lookupFieldName) instanceof String) {
-                        String[] array =((String) event.getExtractedFields().get(lookupFieldName)).split(regex);
+                    if (valueObj instanceof String) {
+                        String[] array =((String) valueObj).split(regex);
                         return parseArray(array, event);
                     }else {
                         log.error("Invalid unroll ! Object is not instance of String. expression=" + regex);
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/CollectListTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/CollectListTest.java
index 2bf13a5..cc96993 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/CollectListTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/CollectListTest.java
@@ -1,18 +1,3 @@
-/**
-  * Copyright 2017 Hortonworks.
-  *
-  * Licensed under the Apache License, Version 2.0 (the "License");
-  * you may not use this file except in compliance with the License.
-  * You may obtain a copy of the License at
-
-  *   http://www.apache.org/licenses/LICENSE-2.0
-
-  * Unless required by applicable law or agreed to in writing, software
-  * distributed under the License is distributed on an "AS IS" BASIS,
-  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  * See the License for the specific language governing permissions and
-  * limitations under the License.
- **/
 package com.geedgenetworks.core.udf.test.aggregate;
 
 
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/CollectSetTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/CollectSetTest.java
index 8e992f6..cf897f6 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/CollectSetTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/CollectSetTest.java
@@ -1,18 +1,3 @@
-/**
-  * Copyright 2017 Hortonworks.
-  *
-  * Licensed under the Apache License, Version 2.0 (the "License");
-  * you may not use this file except in compliance with the License.
-  * You may obtain a copy of the License at
-
-  *   http://www.apache.org/licenses/LICENSE-2.0
-
-  * Unless required by applicable law or agreed to in writing, software
-  * distributed under the License is distributed on an "AS IS" BASIS,
-  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  * See the License for the specific language governing permissions and
-  * limitations under the License.
- **/
 package com.geedgenetworks.core.udf.test.aggregate;
 
 
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/FirstValueTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/FirstValueTest.java
index 43a9732..d114a09 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/FirstValueTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/FirstValueTest.java
@@ -1,18 +1,3 @@
-/**
-  * Copyright 2017 Hortonworks.
-  *
-  * Licensed under the Apache License, Version 2.0 (the "License");
-  * you may not use this file except in compliance with the License.
-  * You may obtain a copy of the License at
-
-  *   http://www.apache.org/licenses/LICENSE-2.0
-
-  * Unless required by applicable law or agreed to in writing, software
-  * distributed under the License is distributed on an "AS IS" BASIS,
-  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  * See the License for the specific language governing permissions and
-  * limitations under the License.
- **/
 package com.geedgenetworks.core.udf.test.aggregate;
 
 
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/LastValueTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/LastValueTest.java
index e952908..3f5432f 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/LastValueTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/LastValueTest.java
@@ -1,18 +1,3 @@
-/**
-  * Copyright 2017 Hortonworks.
-  *
-  * Licensed under the Apache License, Version 2.0 (the "License");
-  * you may not use this file except in compliance with the License.
-  * You may obtain a copy of the License at
-
-  *   http://www.apache.org/licenses/LICENSE-2.0
-
-  * Unless required by applicable law or agreed to in writing, software
-  * distributed under the License is distributed on an "AS IS" BASIS,
-  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  * See the License for the specific language governing permissions and
-  * limitations under the License.
- **/
 package com.geedgenetworks.core.udf.test.aggregate;
 
 
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/LongCountTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/LongCountTest.java
index c1dfb9e..e747c42 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/LongCountTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/LongCountTest.java
@@ -1,18 +1,3 @@
-/**
-  * Copyright 2017 Hortonworks.
-  *
-  * Licensed under the Apache License, Version 2.0 (the "License");
-  * you may not use this file except in compliance with the License.
-  * You may obtain a copy of the License at
-
-  *   http://www.apache.org/licenses/LICENSE-2.0
-
-  * Unless required by applicable law or agreed to in writing, software
-  * distributed under the License is distributed on an "AS IS" BASIS,
-  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  * See the License for the specific language governing permissions and
-  * limitations under the License.
- **/
 package com.geedgenetworks.core.udf.test.aggregate;
 
 
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/MeanTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/MeanTest.java
index cc4eaf0..8255cb1 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/MeanTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/aggregate/MeanTest.java
@@ -1,18 +1,3 @@
-/**
-  * Copyright 2017 Hortonworks.
-  *
-  * Licensed under the Apache License, Version 2.0 (the "License");
-  * you may not use this file except in compliance with the License.
-  * You may obtain a copy of the License at
-
-  *   http://www.apache.org/licenses/LICENSE-2.0
-
-  * Unless required by applicable law or agreed to in writing, software
-  * distributed under the License is distributed on an "AS IS" BASIS,
-  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  * See the License for the specific language governing permissions and
-  * limitations under the License.
- **/
 package com.geedgenetworks.core.udf.test.aggregate;