skip unsafe count optimization for nested predicates

Author: lokidundun

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/optimize/HugeCountStrategy.java

@@ -272,10 +272,28 @@ private boolean doStrategy(final Step step) {
             return false;
         }
 
+        final P<?> predicate = ((IsStep<?>) step.getNextStep()).getPredicate();
+        if (this.hasNestedConnectivePredicate(predicate)) {
+            return false;
+        }
+
         final Step parent = step.getTraversal().getParent().asStep();
         return (parent instanceof FilterStep || parent.getLabels().isEmpty()) &&
                !(parent.getNextStep() instanceof MatchStep.MatchEndStep &&
                  ((MatchStep.MatchEndStep) parent.getNextStep())
                          .getMatchKey().isPresent());
     }
+
+    private boolean hasNestedConnectivePredicate(P<?> predicate) {
+        if (!(predicate instanceof ConnectiveP)) {
+            return false;
+        }
+
+        for (P<?> child : ((ConnectiveP<?>) predicate).getPredicates()) {
+            if (child instanceof ConnectiveP) {
+                return true;
+            }
+        }
+        return false;
+    }
 }

hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CountStrategyCoreTest.java

@@ -209,6 +209,40 @@ public void testWhereCountGteNegativeDoesNotBuildInvalidRange() {
         Assert.assertEquals(4L, count);
     }
 
+    @Test
+    public void testWhereCountNestedConnectivePredicate() {
+        this.initSchema();
+        Vertex source = graph().addVertex(T.label, "person", "name", "source");
+        Vertex target = graph().addVertex(T.label, "person", "name", "target");
+        source.addEdge("knows", target);
+        commitTx();
+
+        long count = graph().traversal().V(source.id())
+                            .where(__.both("knows").count()
+                                     .is(P.<Long>outside(1L, 18L)
+                                          .and(P.gte(0L))))
+                            .count().next();
+
+        Assert.assertEquals(0L, count);
+    }
+
+    @Test
+    public void testWhereCountNegatedNestedConnectivePredicate() {
+        this.initSchema();
+        Vertex source = graph().addVertex(T.label, "person", "name", "source");
+        Vertex target = graph().addVertex(T.label, "person", "name", "target");
+        source.addEdge("knows", target);
+        commitTx();
+
+        long count = graph().traversal().V(source.id())
+                            .where(__.both("knows").count()
+                                     .is(P.not(P.<Long>outside(1L, 18L)
+                                                 .and(P.gte(0L)))))
+                            .count().next();
+
+        Assert.assertEquals(1L, count);
+    }
+
     @Test
     public void testRepeatAfterTextRangeFilterWithEmptyResult() {
         this.initTextRangeSchema(true);