fix dynamic schema data can't rollup correctly (#3949)

* fix dynamic schema data can't rollup correctly

* add ut

processing/src/main/java/io/druid/segment/incremental/IncrementalIndex.java

@@ -963,13 +963,28 @@ public abstract class IncrementalIndex<AggregatorType> implements Iterable<Row>,
       }
 
       if (retVal == 0) {
-        return Ints.compare(lhs.dims.length, rhs.dims.length);
+        int lengthDiff = Ints.compare(lhs.dims.length, rhs.dims.length);
+        if (lengthDiff == 0) {
+          return 0;
+        }
+        Object[] largerDims = lengthDiff > 0 ? lhs.dims : rhs.dims;
+        return allNull(largerDims, numComparisons) ? 0 : lengthDiff;
       }
 
       return retVal;
     }
   }
 
+  private static boolean allNull(Object[] dims, int startPosition)
+  {
+    for (int i = startPosition; i < dims.length; i++) {
+      if (dims[i] != null) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   public static class FactsEntry implements Map.Entry<TimeAndDims, Integer>
   {
     TimeAndDims key = null;

processing/src/test/java/io/druid/segment/data/IncrementalIndexTest.java

@@ -781,4 +781,38 @@ public class IncrementalIndexTest
 
     Assert.assertEquals(Arrays.asList("dim0", "dim1"), incrementalIndex.getDimensionNames());
   }
+
+  @Test
+  public void testDynamicSchemaRollup() throws IndexSizeExceededException
+  {
+    IncrementalIndex<Aggregator> index = new OnheapIncrementalIndex(
+        new IncrementalIndexSchema.Builder().withQueryGranularity(QueryGranularities.NONE).build(),
+        true,
+        10
+    );
+    closer.closeLater(index);
+    index.add(
+        new MapBasedInputRow(
+            1481871600000L,
+            Arrays.asList("name", "host"),
+            ImmutableMap.<String, Object>of("name", "name1", "host", "host")
+        )
+    );
+    index.add(
+        new MapBasedInputRow(
+            1481871670000L,
+            Arrays.asList("name", "table"),
+            ImmutableMap.<String, Object>of("name", "name2", "table", "table")
+        )
+    );
+    index.add(
+        new MapBasedInputRow(
+            1481871600000L,
+            Arrays.asList("name", "host"),
+            ImmutableMap.<String, Object>of("name", "name1", "host", "host")
+        )
+    );
+
+    Assert.assertEquals(2, index.size());
+  }
 }