Fix field split in Top K sketch (#111)

Author: 0aix

src/main/java/com/yahoo/bullet/querying/aggregations/SketchingStrategy.java

@@ -12,11 +12,10 @@
 import com.yahoo.bullet.result.Clip;
 import com.yahoo.bullet.result.Meta;
 
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -111,7 +110,19 @@ String composeField(Stream<String> fields) {
      * @return A {@link List} of the fields that this field was made of.
      */
     List<String> decomposeField(String field) {
-        return Arrays.asList(field.split(Pattern.quote(separator)));
+        List<String> fields = new ArrayList<>();
+        int index = 0;
+        while (true) {
+            int nextIndex = field.indexOf(separator, index);
+            if (nextIndex >= 0) {
+                fields.add(field.substring(index, nextIndex));
+                index = nextIndex + separator.length();
+            } else {
+                fields.add(field.substring(index));
+                break;
+            }
+        }
+        return fields;
     }
 
     private String getMetaKey() {

src/test/java/com/yahoo/bullet/querying/aggregations/FrequentItemsSketchingStrategyTest.java

@@ -367,4 +367,27 @@ public void testResetting() {
         Assert.assertEquals(topK.getRecords(), records);
         Assert.assertEquals(topK.getMetadata().asMap(), result.getMeta().asMap());
     }
+
+    @Test
+    public void testEmptyValues() {
+        FrequentItemsSketchingStrategy topK = makeTopK(asList("A", "B"), 64, 20);
+
+        topK.consume(RecordBox.get().add("A", String.valueOf("")).add("B", String.valueOf("_")).getRecord());
+        topK.consume(RecordBox.get().add("A", String.valueOf("_")).add("B", String.valueOf("")).getRecord());
+        topK.consume(RecordBox.get().add("A", String.valueOf("")).add("B", String.valueOf("")).getRecord());
+
+        Clip result = topK.getResult();
+
+        List<BulletRecord> records = result.getRecords();
+        Assert.assertEquals(records.size(), 3);
+        for (BulletRecord actual : records) {
+            Assert.assertEquals(actual.fieldCount(), 3);
+            Assert.assertNotNull(actual.typedGet("A").getValue());
+            Assert.assertNotNull(actual.typedGet("B").getValue());
+            Assert.assertEquals(actual.typedGet(COUNT_NAME).getValue(), 1L);
+        }
+
+        Assert.assertEquals(topK.getRecords(), records);
+        Assert.assertEquals(topK.getMetadata().asMap(), result.getMeta().asMap());
+    }
 }