Fix case of process empty block in Streaming(Hash)AggregationOperator

Author: Wei-hao-Li

integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java

@@ -5246,4 +5246,26 @@ public void exceptionTest2() {
 
     tableAssertTestFail("select * from table1 where (s1,s2) is not null", errMsg, DATABASE_NAME);
   }
+
+  @Test
+  public void emptyBlockInStreamOperatorTest() {
+    String[] expectedHeader = new String[] {"_col0"};
+    String[] retArray = new String[] {};
+
+    // the sub-query produces empty block
+
+    // test StreamingHashAggregationOperator
+    tableResultSetEqualTest(
+        "select count(1) from (select * from table1 where s1 + 1 < 1) group by device_id,s1",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+
+    // test StreamingAggregationOperator
+    tableResultSetEqualTest(
+        "select count(1) from (select * from table1 where s1 + 1 < 1) group by device_id",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java

@@ -122,7 +122,7 @@ public TsBlock next() throws Exception {
     TsBlock block;
     if (child.hasNextWithTimer()) {
       block = child.nextWithTimer();
-      if (block == null) {
+      if (block == null || block.isEmpty()) {
         return null;
       }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingHashAggregationOperator.java

@@ -159,7 +159,7 @@ public TsBlock next() throws Exception {
     TsBlock block;
     if (child.hasNextWithTimer()) {
       block = child.nextWithTimer();
-      if (block == null) {
+      if (block == null || block.isEmpty()) {
         return null;
       }