Fix DPP enabled some sort and aggregate input may be empty issue (#59)

Signed-off-by: Chendi Xue <chendi.xue@intel.com>
oap-project · Jan 26, 2021 · 9ce64d7 · 9ce64d7
1 parent b47116d
commit 9ce64d7
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 3 deletions.
diff --git a/core/src/main/scala/com/intel/oap/expression/ColumnarAggregation.scala b/core/src/main/scala/com/intel/oap/expression/ColumnarAggregation.scala
@@ -511,6 +511,12 @@ class ColumnarAggregation(
             }
             numInputBatches += 1
           }
+          if (processedNumRows == 0) {
+            data_loaded = true
+            aggrTime += NANOSECONDS.toMillis(eval_elapse)
+            nextBatch = false
+            return false
+          }
           if (groupingFieldList.size > 0) {
             val beforeFinish = System.nanoTime()
             result_iterator = aggregator.finishByIterator()

diff --git a/core/src/main/scala/com/intel/oap/expression/ColumnarSorter.scala b/core/src/main/scala/com/intel/oap/expression/ColumnarSorter.scala
@@ -137,7 +137,7 @@ class ColumnarSorter(
       var has_next: Boolean = true
 
       override def hasNext: Boolean = {
-        if (sort_iterator == null) {
+        if (has_next && sort_iterator == null) {
           while (cbIterator.hasNext) {
             cb = cbIterator.next()
 
@@ -148,14 +148,18 @@ class ColumnarSorter(
           }
 
           val beforeSort = System.nanoTime()
-          sort_iterator = sorter.finishByIterator();
+          if (processedNumRows > 0) {
+            sort_iterator = sorter.finishByIterator();
+          }
           sort_elapse += System.nanoTime() - beforeSort
           total_elapse += System.nanoTime() - beforeSort
         }
-        if (sort_iterator.hasNext()) {
+        if (sort_iterator != null && sort_iterator.hasNext()) {
           return true
         } else {
+          has_next = false
           inputBatchHolder.foreach(cb => cb.close())
+          inputBatchHolder.clear
           return false
         }
       }