remove null assignments and check that the map is not empty check (#479)

* remove null assignments and check that the map is not empty check

* return an empty string if mode receives an empty map, add test for empty dataset

* run spotless fix typo

Author: elliVM

src/main/java/com/teragrep/pth10/ast/commands/aggregate/UDAFs/BufferClasses/ModeBuffer.java

@@ -45,6 +45,8 @@
  */
 package com.teragrep.pth10.ast.commands.aggregate.UDAFs.BufferClasses;
 
+import com.teragrep.pth10.ast.NullValue;
+
 import java.io.Serializable;
 import java.util.Map;
 
@@ -93,18 +95,12 @@ public void mergeMap(Map<String, Long> another) {
      * @return most frequent entry as a string
      */
     public String mode() {
-        Map.Entry<String, Long> mostFrequentEntry = null;
-
-        for (Map.Entry<String, Long> entry : this.map.entrySet()) {
-            if (mostFrequentEntry == null) {
-                mostFrequentEntry = entry;
-            }
-            else if (mostFrequentEntry != null && entry.getValue() > mostFrequentEntry.getValue()) {
-                mostFrequentEntry = entry;
-            }
-        }
-
-        return mostFrequentEntry.getKey();
+        return map
+                .entrySet()
+                .stream()
+                .max(Map.Entry.comparingByValue()) // find max value
+                .map(Map.Entry::getKey) // select key
+                .orElse(new NullValue(NullValue.Type.EMPTY_STRING).value()); // if an empty map returns an empty string
     }
 
 }

src/test/java/com/teragrep/pth10/statsTransformationTest.java

@@ -358,6 +358,29 @@ void statsTransform_AggMode_Test() {
         });
     }
 
+    @Test
+    @DisabledIfSystemProperty(
+            named = "skipSparkTest",
+            matches = "true"
+    )
+    void statsTransformAggModeTestEmptyDataset() {
+        streamingTestUtil
+                .performDPLTest(
+                        "index=index_A earliest=2020-10-10T11:20:10.100+03:00 | stats mode(offset)", testFile, ds -> {
+                            Assertions.assertEquals("[mode(offset)]", Arrays.toString(ds.columns()));
+
+                            List<Object> destAsList = ds
+                                    .select("mode(offset)")
+                                    .collectAsList()
+                                    .stream()
+                                    .map(r -> r.getAs(0).toString())
+                                    .collect(Collectors.toList());
+
+                            Assertions.assertEquals(Collections.singletonList(""), destAsList);
+                        }
+                );
+    }
+
     // Test min()
     @Test
     @DisabledIfSystemProperty(