fix(interactive): Fix Bugs of Group Returning Results from Compiler (#…

…3533)

<!--
Thanks for your contribution! please review
https://github.com/alibaba/GraphScope/blob/main/CONTRIBUTING.md before
opening an issue.
-->

## What do these changes do?
Group Returning Results are split into multiple maps in user scenarios,
for the query which results size is larger than 64 can trigger the
issue, i.e.
```
gremlin> g.V().hasLabel('PERSON').limit(100).group().by('firstName').by(count())
==>{Dặng Dinh=1, Carey=1, Michel=1, Wojciech=1, Ayesha=2, Anand=2, Franz=1, Volodymyr=1, Maria=1, Fritz=1, Jie=1, Kunal=1, Aditya=1, Mikhail=1, Milan=1, Serghei=1, Neil=1, Shweta=2, Walter=1, Arturo=1, Rahul=2, Barney=1, Peter=2, Evgeny=1, Charlie=1, Bing=1, Bob=1, Pope=1, Wilson=1, Arjun=1, Antonio=3, Carlos=2, Dania=1, André=1, K.=2, Crown Prince=1, Li=1, Karl=1, Gabriel=1, Javed=1, Michael=1, Asim=1, James=1, Alexander=2, Bela=1, Chipo=1, Philippe=1, Jun=2, Andry=1, Wei=2, Bingbing=1, Andrew=1, Yang=2, Abdala=1, Hao=1, Aa Ngurah=1, Mohammad Reza=1, Eddie=1, Luis=1, Lei=1, Zheng=1, Giuseppe=1, Dhafer=1, David=1}
==>{John=3, Ricky=1, Peng=1, A.=1, Ivan=1, Abdel Wahab=1, Chen=4, Sam=1, Boris=1, Mario=1, Lin=1, Paresh=1, Jose=1, Fernando=1, Almira=1, Abraham=1, R.=1, Ai=1}
```
This pr fixes the issue aforementioned.

Co-authored-by: Longbin Lai <[email protected]>

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/pattern/PatternQueryTest.java

@@ -25,6 +25,8 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.Map;
+
 public abstract class PatternQueryTest extends AbstractGremlinProcessTest {
     public abstract Traversal<Vertex, Long> get_pattern_1_test();
 
@@ -60,6 +62,8 @@ public abstract class PatternQueryTest extends AbstractGremlinProcessTest {
 
     public abstract Traversal<Vertex, Long> get_pattern_17_test();
 
+    public abstract Traversal<Vertex, Map<Object, Object>> get_g_V_limit_100_group_test();
+
     @Test
     public void run_pattern_1_test() {
         Traversal<Vertex, Long> traversal = this.get_pattern_1_test();
@@ -179,6 +183,15 @@ public void run_pattern_17_test() {
         Assert.assertEquals(17367L, traversal.next().longValue());
     }
 
+    @Test
+    public void run_g_V_limit_100_group_test() {
+        Traversal<Vertex, Map<Object, Object>> traversal = this.get_g_V_limit_100_group_test();
+        this.printTraversalForm(traversal);
+        Map<Object, Object> map = traversal.next();
+        Assert.assertEquals(100, map.size());
+        Assert.assertFalse(traversal.hasNext());
+    }
+
     public static class Traversals extends PatternQueryTest {
 
         // PM1
@@ -386,5 +399,10 @@ public Traversal<Vertex, Long> get_pattern_17_test() {
                                     .as("b"))
                     .count();
         }
+
+        @Override
+        public Traversal<Vertex, Map<Object, Object>> get_g_V_limit_100_group_test() {
+            return g.V().hasLabel("PERSON").limit(100).group().by("id").by(__.count());
+        }
     }
 }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/result/processor/AbstractResultProcessor.java

@@ -19,6 +19,7 @@
 import com.alibaba.graphscope.common.config.QueryTimeoutConfig;
 import com.alibaba.graphscope.common.result.ResultParser;
 import com.alibaba.graphscope.gremlin.plugin.QueryStatusCallback;
+import com.alibaba.graphscope.gremlin.result.GroupResultParser;
 import com.alibaba.pegasus.intf.ResultProcessor;
 import com.alibaba.pegasus.service.protocol.PegasusClient;
 
@@ -82,7 +83,8 @@ public synchronized void process(PegasusClient.JobResponse response) {
             if (isContextWritable) {
                 // send back a page of results if batch size is met and then reset the
                 // resultCollectors
-                if (this.resultCollectors.size() >= this.resultCollectorsBatchSize) {
+                if (this.resultCollectors.size() >= this.resultCollectorsBatchSize
+                        && !(resultParser instanceof GroupResultParser)) {
                     aggregateResults();
                     writeResultList(
                             writeResult, resultCollectors, ResponseStatusCode.PARTIAL_CONTENT);