optimizes SpillingGrouper

Author: maytasm

processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java

@@ -485,12 +485,20 @@ private Iterator<Entry<KeyType>> combineByKey(
   {
     return new AbstractIterator<>()
     {
-      private Entry<KeyType> current = sortedIterator.hasNext() ? sortedIterator.next() : null;
+      private final ReusableEntry<KeyType> current = ReusableEntry.create(keySerde, aggregatorFactories.length);
+      private boolean hasCurrent = false;
+
+      {
+        if (sortedIterator.hasNext()) {
+          copyEntry(sortedIterator.next(), current);
+          hasCurrent = true;
+        }
+      }
 
       @Override
       protected Entry<KeyType> computeNext()
       {
-        if (current == null) {
+        if (!hasCurrent) {
           return endOfData();
         }
 
@@ -501,18 +509,24 @@ protected Entry<KeyType> computeNext()
 
         while (sortedIterator.hasNext()) {
           final Entry<KeyType> next = sortedIterator.next();
-          if (comparator.compare(current, next) != 0) {
-            current = next;
+          if (comparator.compare(combined, next) != 0) {
+            copyEntry(next, current);
             return combined;
           }
           for (int i = 0; i < combinedValues.length; i++) {
             combinedValues[i] = aggregatorFactories[i].combine(combinedValues[i], next.getValues()[i]);
           }
         }
 
-        current = null;
+        hasCurrent = false;
         return combined;
       }
+
+      private void copyEntry(Entry<KeyType> from, ReusableEntry<KeyType> to)
+      {
+        to.setKey(from.getKey());
+        System.arraycopy(from.getValues(), 0, to.getValues(), 0, aggregatorFactories.length);
+      }
     };
   }