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);
+    }
+
 }

hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/traversal/optimize/TraversalUtilOptimizeTest.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.traversal.optimize;
+
+import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.exception.NotFoundException;
+import org.apache.hugegraph.schema.PropertyKey;
+import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.type.define.DataType;
+import org.apache.tinkerpop.gremlin.process.traversal.P;
+import org.apache.tinkerpop.gremlin.process.traversal.step.util.HasContainer;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class TraversalUtilOptimizeTest {
+
+    @Test
+    public void testCanExtractHasContainerWithoutGraph() {
+        Assert.assertTrue(TraversalUtil.canExtractHasContainer(
+                null, new HasContainer("~label", P.eq("person"))));
+        Assert.assertTrue(TraversalUtil.canExtractHasContainer(
+                null, new HasContainer("~id", P.eq("1"))));
+        Assert.assertFalse(TraversalUtil.canExtractHasContainer(
+                null, new HasContainer("name", P.eq("marko"))));
+    }
+
+    @Test
+    public void testCanExtractHasContainerWithMissingPropertyKey() {
+        HugeGraph graph = Mockito.mock(HugeGraph.class);
+        Mockito.when(graph.propertyKey("missing"))
+               .thenThrow(new NotFoundException("missing"));
+
+        Assert.assertFalse(TraversalUtil.canExtractHasContainer(
+                graph, new HasContainer("missing", P.eq("marko"))));
+    }
+
+    @Test
+    public void testCanExtractHasContainerWithNonTextProperty() {
+        HugeGraph graph = Mockito.mock(HugeGraph.class);
+        PropertyKey age = propertyKey(1L, "age", DataType.INT);
+        Mockito.when(graph.propertyKey("age")).thenReturn(age);
+
+        Assert.assertTrue(TraversalUtil.canExtractHasContainer(
+                graph, new HasContainer("age", P.eq(1))));
+    }
+
+    @Test
+    public void testCanExtractHasContainerWithTextRangePredicate() {
+        HugeGraph graph = Mockito.mock(HugeGraph.class);
+        PropertyKey name = propertyKey(1L, "name", DataType.TEXT);
+        Mockito.when(graph.propertyKey("name")).thenReturn(name);
+
+        Assert.assertFalse(TraversalUtil.canExtractHasContainer(
+                graph, new HasContainer("name", P.lt(""))));
+        Assert.assertTrue(TraversalUtil.canExtractHasContainer(
+                graph, new HasContainer("name", P.eq("marko"))));
+    }
+
+    private static PropertyKey propertyKey(long id, String name,
+                                           DataType dataType) {
+        Id keyId = IdGenerator.of(id);
+        PropertyKey key = new PropertyKey(null, keyId, name);
+        key.dataType(dataType);
+        return key;
+    }
+}