fix(server): avoid extracting text range filters

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

@@ -39,6 +39,7 @@
 import org.apache.hugegraph.backend.query.Condition;
 import org.apache.hugegraph.backend.query.ConditionQuery;
 import org.apache.hugegraph.backend.query.Query;
+import org.apache.hugegraph.exception.NotFoundException;
 import org.apache.hugegraph.exception.NotSupportException;
 import org.apache.hugegraph.iterator.FilterIterator;
 import org.apache.hugegraph.schema.PropertyKey;
@@ -166,38 +167,103 @@ public static void trySetGraph(Step<?, ?> step, HugeGraph graph) {
 
     public static void extractHasContainer(HugeGraphStep<?, ?> newStep,
                                            Traversal.Admin<?, ?> traversal) {
-        Step<?, ?> step = newStep;
-        do {
-            step = step.getNextStep();
+        Step<?, ?> step = newStep.getNextStep();
+        while (step instanceof HasStep || step instanceof NoOpBarrierStep) {
+            Step<?, ?> nextStep = step.getNextStep();
             if (step instanceof HasStep) {
                 HasContainerHolder holder = (HasContainerHolder) step;
-                for (HasContainer has : holder.getHasContainers()) {
-                    if (!GraphStep.processHasContainerIds(newStep, has)) {
-                        newStep.addHasContainer(has);
-                    }
+                if (extractHasContainers(newStep, holder)) {
+                    TraversalHelper.copyLabels(step, step.getPreviousStep(), false);
+                    traversal.removeStep(step);
                 }
-                TraversalHelper.copyLabels(step, step.getPreviousStep(), false);
-                traversal.removeStep(step);
             }
-        } while (step instanceof HasStep || step instanceof NoOpBarrierStep);
+            step = nextStep;
+        }
     }
 
     public static void extractHasContainer(HugeVertexStep<?> newStep,
                                            Traversal.Admin<?, ?> traversal) {
         Step<?, ?> step = newStep;
         do {
+            Step<?, ?> nextStep = step.getNextStep();
             if (step instanceof HasStep) {
                 HasContainerHolder holder = (HasContainerHolder) step;
-                for (HasContainer has : holder.getHasContainers()) {
-                    newStep.addHasContainer(has);
+                if (extractHasContainers(newStep, holder)) {
+                    TraversalHelper.copyLabels(step, step.getPreviousStep(), false);
+                    traversal.removeStep(step);
                 }
-                TraversalHelper.copyLabels(step, step.getPreviousStep(), false);
-                traversal.removeStep(step);
             }
-            step = step.getNextStep();
+            step = nextStep;
         } while (step instanceof HasStep || step instanceof NoOpBarrierStep);
     }
 
+    private static boolean extractHasContainers(HugeGraphStep<?, ?> newStep,
+                                                HasContainerHolder holder) {
+        HugeGraph graph = TraversalUtil.tryGetGraph(newStep);
+        if (!canExtractHasContainers(graph, holder)) {
+            return false;
+        }
+        for (HasContainer has : holder.getHasContainers()) {
+            if (!GraphStep.processHasContainerIds(newStep, has)) {
+                newStep.addHasContainer(has);
+            }
+        }
+        return true;
+    }
+
+    private static boolean extractHasContainers(HugeVertexStep<?> newStep,
+                                                HasContainerHolder holder) {
+        HugeGraph graph = TraversalUtil.tryGetGraph(newStep);
+        if (!canExtractHasContainers(graph, holder)) {
+            return false;
+        }
+        for (HasContainer has : holder.getHasContainers()) {
+            newStep.addHasContainer(has);
+        }
+        return true;
+    }
+
+    private static boolean canExtractHasContainers(HugeGraph graph,
+                                                   HasContainerHolder holder) {
+        for (HasContainer has : holder.getHasContainers()) {
+            if (!canExtractHasContainer(graph, has)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    static boolean canExtractHasContainer(HugeGraph graph,
+                                          HasContainer has) {
+        if (isSysProp(has.getKey())) {
+            return true;
+        }
+        if (graph == null) {
+            return false;
+        }
+
+        PropertyKey pkey;
+        try {
+            pkey = graph.propertyKey(has.getKey());
+        } catch (NotFoundException e) {
+            return false;
+        }
+        if (!pkey.dataType().isText()) {
+            return true;
+        }
+
+        List<P<Object>> predicates = new ArrayList<>();
+        collectPredicates(predicates, ImmutableList.of(has.getPredicate()));
+        for (P<Object> pred : predicates) {
+            BiPredicate<?, ?> bp = pred.getBiPredicate();
+            if (bp == Compare.gt || bp == Compare.gte ||
+                bp == Compare.lt || bp == Compare.lte) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     public static void extractOrder(Step<?, ?> newStep,
                                     Traversal.Admin<?, ?> traversal) {
         Step<?, ?> step = newStep;

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

@@ -48,6 +48,17 @@ private void initGraph() {
         commitTx();
     }
 
+    private void initTextRangeSchema(boolean withEdge) {
+        SchemaManager schema = graph().schema();
+        schema.propertyKey("vp4").asText().create();
+        schema.propertyKey("age").asInt().create();
+        schema.vertexLabel("vl1").properties("vp4", "age")
+              .nullableKeys("vp4", "age").create();
+        if (withEdge) {
+            schema.edgeLabel("el2").link("vl1", "vl1").create();
+        }
+    }
+
     @Test
     public void testWhereCountLtNegativeIsAlwaysFalse() {
         this.initSchema();
@@ -82,7 +93,7 @@ public void testWhereCountOutsideNegativeKeepsOriginalSemantics() {
                              .count().next();
 
         Assert.assertEquals(1L, direct);
-        Assert.assertEquals(viaMatch, direct);
+        Assert.assertEquals(direct, viaMatch);
     }
 
     @Test
@@ -125,4 +136,73 @@ public void testWhereCountGteNegativeDoesNotBuildInvalidRange() {
 
         Assert.assertEquals(4L, count);
     }
+
+    @Test
+    public void testRepeatAfterTextRangeFilterWithEmptyResult() {
+        this.initTextRangeSchema(true);
+
+        Vertex v1 = graph().addVertex(T.label, "vl1", "vp4", "a", "age", 1);
+        Vertex v2 = graph().addVertex(T.label, "vl1", "vp4", "b", "age", 2);
+        v1.addEdge("el2", v2);
+        commitTx();
+
+        long direct = graph().traversal().V().has("vp4", P.lt(""))
+                           .repeat(__.out("el2")).emit().times(1)
+                           .count().next();
+        long viaMatch = graph().traversal().V()
+                             .match(__.as("start").has("vp4", P.lt(""))
+                                      .out("el2").as("m"))
+                             .select("m").count().next();
+
+        Assert.assertEquals(0L, direct);
+        Assert.assertEquals(direct, viaMatch);
+    }
+
+    @Test
+    public void testTextRangeFilterKeepsMixedGraphHasStep() {
+        this.initTextRangeSchema(false);
+
+        graph().addVertex(T.label, "vl1", "vp4", "a", "age", 1);
+        graph().addVertex(T.label, "vl1", "vp4", "b", "age", 2);
+        commitTx();
+
+        long direct = graph().traversal().V()
+                           .hasLabel("vl1")
+                           .has("vp4", P.lt(""))
+                           .has("age", 1)
+                           .count().next();
+        long viaMatch = graph().traversal().V()
+                             .match(__.as("v").hasLabel("vl1")
+                                      .has("vp4", P.lt(""))
+                                      .has("age", 1))
+                             .select("v").count().next();
+
+        Assert.assertEquals(0L, direct);
+        Assert.assertEquals(direct, viaMatch);
+    }
+
+    @Test
+    public void testTextRangeFilterKeepsMixedVertexHasStep() {
+        this.initTextRangeSchema(true);
+
+        Vertex v1 = graph().addVertex(T.label, "vl1", "vp4", "a", "age", 1);
+        Vertex v2 = graph().addVertex(T.label, "vl1", "vp4", "b", "age", 2);
+        v1.addEdge("el2", v2);
+        commitTx();
+
+        long direct = graph().traversal().V(v1.id()).out("el2")
+                           .hasLabel("vl1")
+                           .has("vp4", P.lt(""))
+                           .has("age", 2)
+                           .count().next();
+        long viaMatch = graph().traversal().V(v1.id()).out("el2")
+                             .match(__.as("v").hasLabel("vl1")
+                                      .has("vp4", P.lt(""))
+                                      .has("age", 2))
+                             .select("v").count().next();
+
+        Assert.assertEquals(0L, direct);
+        Assert.assertEquals(direct, viaMatch);
+    }
+
 }