From d5d170f866f9512ed859f643b84fc11ff341fba4 Mon Sep 17 00:00:00 2001
From: Himanshu <g.himanshu@gmail.com>
Date: Tue, 27 Aug 2019 15:01:07 -0700
Subject: [PATCH] skip unnecessary aggregate(..) calls with
 LimitedBufferHashGrouper (#8412)

* skip unnecessary aggregate(..) calls with LimitedBufferHashGrouper

* remove unused bucketWasUsed arg from canSkipAggregate(..)
---
 .../AbstractBufferHashGrouper.java            |  5 ++-
 .../epinephelinae/BufferHashGrouper.java      |  2 +-
 .../LimitedBufferHashGrouper.java             | 34 +++++++++----------
 3 files changed, 19 insertions(+), 22 deletions(-)

diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/AbstractBufferHashGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/AbstractBufferHashGrouper.java
index aae2e4b3a3c..0fe22b8f433 100644
--- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/AbstractBufferHashGrouper.java
+++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/AbstractBufferHashGrouper.java
@@ -76,13 +76,12 @@ public abstract class AbstractBufferHashGrouper<KeyType> implements Grouper<KeyT
   /**
    * Called to check if it's possible to skip aggregation for a row.
    *
-   * @param bucketWasUsed Was the row a new entry in the hash table?
    * @param bucketOffset  Offset of the bucket containing this row's entry in the hash table,
    *                      within the buffer returned by hashTable.getTableBuffer()
    *
    * @return true if aggregation can be skipped, false otherwise.
    */
-  public abstract boolean canSkipAggregate(boolean bucketWasUsed, int bucketOffset);
+  public abstract boolean canSkipAggregate(int bucketOffset);
 
   /**
    * Called after a row is aggregated. An implementing BufferHashGrouper class can use this to update
@@ -154,7 +153,7 @@ public abstract class AbstractBufferHashGrouper<KeyType> implements Grouper<KeyT
       newBucketHook(bucketStartOffset);
     }
 
-    if (canSkipAggregate(bucketWasUsed, bucketStartOffset)) {
+    if (canSkipAggregate(bucketStartOffset)) {
       return AggregateResult.ok();
     }
 
diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java
index a695066b499..9c46cd63fe2 100644
--- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java
+++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java
@@ -244,7 +244,7 @@ public class BufferHashGrouper<KeyType> extends AbstractBufferHashGrouper<KeyTyp
   }
 
   @Override
-  public boolean canSkipAggregate(boolean bucketWasUsed, int bucketOffset)
+  public boolean canSkipAggregate(int bucketOffset)
   {
     return false;
   }
diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java
index 4c71c31167a..92ad9c43e5f 100644
--- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java
+++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java
@@ -146,33 +146,31 @@ public class LimitedBufferHashGrouper<KeyType> extends AbstractBufferHashGrouper
   public void newBucketHook(int bucketOffset)
   {
     heapIndexUpdater.updateHeapIndexForOffset(bucketOffset, -1);
+    if (!sortHasNonGroupingFields) {
+      offsetHeap.addOffset(bucketOffset);
+    }
   }
 
   @Override
-  public boolean canSkipAggregate(boolean bucketWasUsed, int bucketOffset)
+  public boolean canSkipAggregate(int bucketOffset)
   {
-    if (bucketWasUsed) {
-      if (!sortHasNonGroupingFields) {
-        if (heapIndexUpdater.getHeapIndexForOffset(bucketOffset) < 0) {
-          return true;
-        }
-      }
-    }
-    return false;
+    return !sortHasNonGroupingFields && heapIndexUpdater.getHeapIndexForOffset(bucketOffset) < 0;
   }
 
   @Override
   public void afterAggregateHook(int bucketOffset)
   {
-    int heapIndex = heapIndexUpdater.getHeapIndexForOffset(bucketOffset);
-    if (heapIndex < 0) {
-      // not in the heap, add it
-      offsetHeap.addOffset(bucketOffset);
-    } else if (sortHasNonGroupingFields) {
-      // Since the sorting columns contain at least one aggregator, we need to remove and reinsert
-      // the entries after aggregating to maintain proper ordering
-      offsetHeap.removeAt(heapIndex);
-      offsetHeap.addOffset(bucketOffset);
+    if (sortHasNonGroupingFields) {
+      int heapIndex = heapIndexUpdater.getHeapIndexForOffset(bucketOffset);
+
+      if (heapIndex < 0) {
+        offsetHeap.addOffset(bucketOffset);
+      } else {
+        // Since the sorting columns contain at least one aggregator, we need to remove and reinsert
+        // the entries after aggregating to maintain proper ordering
+        offsetHeap.removeAt(heapIndex);
+        offsetHeap.addOffset(bucketOffset);
+      }
     }
   }