refactor(server): preserve core gremlin semantics (2/3) (#3262)

Part 2 of upgrade series. Preserve transaction count and predicate semantics while retaining TinkerPop 3.5.1

Stabilize query behavior on TinkerPop 3.5.1: pending removals take precedence in transaction-visible counts, unsupported mixed filters remain in one local HasStep, and count barriers, input multiplicity and iterator cleanup are preserved.

---------

Co-authored-by: imbajin <jin@apache.org>

Author: contrueCT

docs/query-semantics.md

@@ -0,0 +1,41 @@
+# Query count and filter behavior
+
+Counts made inside a transaction include that transaction's uncommitted vertex
+and edge changes. Only `count()` supports this fallback; other aggregates with
+uncommitted changes are rejected. Queries that combine uncommitted changes with
+pagination, a limit, or an offset remain unsupported.
+
+Queries and counts select the current transaction version before applying filters.
+When an element is removed and re-added with the same ID, the new element shadows
+the old updated record, even if only the old record matches the query's label.
+
+A count query that enters an index query path remains unsupported when that index
+transaction has uncommitted changes. It reports `Can't do index query when there
+are changes in transaction`. This includes indexed-property conditions and
+label-only queries on backends that need a label index. Graph-wide and explicit-ID
+counts can still use transaction merging without entering that index path.
+
+Unsupported text predicates remain traversal filters. For ordinary GraphStep and
+VertexStep extraction, if any condition in a `HasStep` cannot be converted, the
+whole step remains local, including its sibling label, ID and property conditions.
+Count optimization preserves that filter step. Existing special handling around
+`match()` and connective label filters is inherited from master.
+
+New generalized selective pushdown, predicate-specific local ID/SEARCH matching and
+candidate-index coverage are outside this upgrade series. They are coordinated
+through [PR #2994](https://github.com/apache/hugegraph/pull/2994),
+[issue #3201](https://github.com/apache/hugegraph/issues/3201) and
+[PR #3243](https://github.com/apache/hugegraph/pull/3243).
+
+A self-loop contributes two occurrences to a vertex's `bothE()` traversal and
+one to each of `outE()`, `inE()` and the graph-wide `E()` traversal. Counts retain
+these multiplicities before and after transaction commit, including edge updates.
+
+Count optimization preserves steps that can filter candidates. Resetting an
+optimized count traversal permits it to execute again. Query-step equality does
+not depend on the result iterator from a previous execution.
+
+A scan in the middle of a traversal is counted once per incoming traverser,
+including its bulk. It is not replaced by a single backend count. For example,
+with three vertices, `g.V().V().count()` returns `9L`, while `g.V().count()`
+returns `3L` and remains eligible for count optimization.

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java

@@ -64,6 +64,7 @@
 import org.apache.hugegraph.iterator.BatchMapperIterator;
 import org.apache.hugegraph.iterator.ExtendableIterator;
 import org.apache.hugegraph.iterator.FilterIterator;
+import org.apache.hugegraph.iterator.FlatMapperIterator;
 import org.apache.hugegraph.iterator.LimitIterator;
 import org.apache.hugegraph.iterator.MapperIterator;
 import org.apache.hugegraph.job.system.DeleteExpiredJob;
@@ -548,24 +549,27 @@ public Number queryNumber(Query query) {
         boolean hasUpdate = this.hasUpdate();
         Aggregate aggregate = query.aggregateNotNull();
 
-        // TODO: we can concat index-query results and tx uncommitted records.
-        if (hasUpdate) {
-            E.checkArgument(!isConditionQuery,
-                            "It's not allowed to query by index when " +
-                            "there are uncommitted records.");
+        if (hasUpdate && (query.resultType() == HugeType.VERTEX ||
+                          query.resultType() == HugeType.EDGE)) {
+            E.checkArgument(aggregate.func() == AggregateFunc.COUNT,
+                            "The %s operator with uncommitted records " +
+                            "is not supported",
+                            aggregate.func().string());
+            Query queryWithoutAggregate = query.copy();
+            queryWithoutAggregate.aggregate(null);
+            Iterator<?> results = queryWithoutAggregate.resultType().isVertex() ?
+                                  this.queryVertices(queryWithoutAggregate) :
+                                  this.queryEdges(queryWithoutAggregate);
+            return countAndClose(results);
         }
 
         QueryList<Number> queries = this.optimizeQueries(query, q -> {
             boolean isIndexQuery = q instanceof IdQuery;
             assert isIndexQuery || isConditionQuery || q == query;
-            // Need to fall back if there are uncommitted records
-            boolean fallback = hasUpdate;
+            boolean fallback = false;
             Number result;
 
-            if (fallback) {
-                // Here just ignore it, and do fall back later
-                result = null;
-            } else if (!isIndexQuery || !isConditionQuery) {
+            if (!isIndexQuery || !isConditionQuery) {
                 // It's a sysprop-query, let parent tx do it
                 assert !fallback;
                 result = super.queryNumber(q);
@@ -591,9 +595,9 @@ public Number queryNumber(Query query) {
                 assert q.resultType().isVertex() || q.resultType().isEdge();
                 // Reset aggregate to fallback and scan
                 q.aggregate(null);
-                result = IteratorUtils.count(q.resultType().isVertex() ?
-                                             this.queryVertices(q) :
-                                             this.queryEdges(q));
+                result = countAndClose(q.resultType().isVertex() ?
+                                       this.queryVertices(q) :
+                                       this.queryEdges(q));
             }
 
             return new QueryResults<>(IteratorUtils.of(result), q);
@@ -605,6 +609,19 @@ public Number queryNumber(Query query) {
         return aggregate.reduce(results.iterator());
     }
 
+    private static long countAndClose(Iterator<?> results) {
+        try {
+            long count = 0L;
+            while (results.hasNext()) {
+                results.next();
+                count++;
+            }
+            return count;
+        } finally {
+            CloseableIterator.closeIterator(results);
+        }
+    }
+
     @Watched(prefix = "graph")
     public HugeVertex addVertex(Object... keyValues) {
         return this.addVertex(this.constructVertex(true, keyValues));
@@ -819,7 +836,7 @@ public Iterator<Vertex> queryVertices() {
     public Iterator<Vertex> queryVertices(Query query) {
         if (this.hasUpdate()) {
             E.checkArgument(query.noLimitAndOffset(),
-                            "It's not allowed to query with offser/limit " +
+                            "It's not allowed to query with offset/limit " +
                             "when there are uncommitted records.");
             // TODO: also add check: no SCAN, no OLAP
             E.checkArgument(!query.paging(),
@@ -1008,7 +1025,7 @@ public Iterator<Edge> queryEdges() {
     public Iterator<Edge> queryEdges(Query query) {
         if (this.hasUpdate()) {
             E.checkArgument(query.noLimitAndOffset(),
-                            "It's not allowed to query with offser/limit " +
+                            "It's not allowed to query with offset/limit " +
                             "when there are uncommitted records.");
             // TODO: also add check: no SCAN, no OLAP
             E.checkArgument(!query.paging(),
@@ -2037,6 +2054,22 @@ private Iterator<?> joinTxEdges(Query query, Iterator<HugeEdge> edges,
         edges = this.joinTxRecords(query, edges, matchTxFunc,
                                    this.addedEdges, this.removedEdges,
                                    this.updatedEdges);
+        if (query instanceof ConditionQuery &&
+            ((ConditionQuery) query).containsCondition(HugeKeys.OWNER_VERTEX) &&
+            !(this.addedEdges.isEmpty() && this.updatedEdges.isEmpty())) {
+            edges = new FlatMapperIterator<>(edges, edge -> {
+                // Rehydrated self-loops may hold distinct vertex objects.
+                if (edge.sourceVertex().id().equals(edge.targetVertex().id()) &&
+                    (this.addedEdges.containsKey(edge.id()) ||
+                     this.updatedEdges.containsKey(edge.id()))) {
+                    HugeEdge opposite = edge.switchOwner();
+                    if (query.test(opposite)) {
+                        return ImmutableList.of(edge, opposite).iterator();
+                    }
+                }
+                return ImmutableList.of(edge).iterator();
+            });
+        }
         if (removingVertices.isEmpty()) {
             return edges;
         }
@@ -2073,6 +2106,9 @@ private <V extends HugeElement> Iterator<V> joinTxRecords(
          * Records in memory have higher priority than a query from backend store
          */
         for (V elem : addedTxRecords.values()) {
+            if (removedTxRecords.containsKey(elem.id())) {
+                continue;
+            }
             if (query.reachLimit(txResults.size())) {
                 break;
             }
@@ -2081,6 +2117,9 @@ private <V extends HugeElement> Iterator<V> joinTxRecords(
             }
         }
         for (V elem : updatedTxRecords.values()) {
+            if (removedTxRecords.containsKey(elem.id()) || addedTxRecords.containsKey(elem.id())) {
+                continue;
+            }
             if (query.reachLimit(txResults.size())) {
                 break;
             }

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

@@ -55,12 +55,18 @@ public boolean equals(Object obj) {
 
         HugeCountStep other = (HugeCountStep) obj;
         return Objects.equals(this.originGraphStep,
-                              other.originGraphStep) && this.done == other.done;
+                              other.originGraphStep);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(super.hashCode(), this.originGraphStep, this.done);
+        return Objects.hash(super.hashCode(), this.originGraphStep);
+    }
+
+    @Override
+    public void reset() {
+        super.reset();
+        this.done = false;
     }
 
     @Override

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

@@ -34,7 +34,6 @@
 import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.AggregateLocalStep;
 import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.IdentityStep;
 import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.SideEffectStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.util.CollectingBarrierStep;
 import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
 import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
 import org.apache.tinkerpop.gremlin.structure.Element;
@@ -73,11 +72,11 @@ public void apply(Traversal.Admin<?, ?> traversal) {
         HugeGraphStep<?, ? extends Element> graphStep = null;
         Step<?, ?> step = originStep;
         do {
+            // Only skip steps that preserve the number of traversers.
             if (!(step instanceof CountGlobalStep ||
                   step instanceof GraphStep ||
                   step instanceof IdentityStep ||
-                  step instanceof NoOpBarrierStep ||
-                  step instanceof CollectingBarrierStep) ||
+                  step instanceof NoOpBarrierStep) ||
                 (step instanceof TraversalParent &&
                  TraversalHelper.anyStepRecursively(s -> {
                      return s instanceof SideEffectStep ||
@@ -94,7 +93,9 @@ public void apply(Traversal.Admin<?, ?> traversal) {
             step = step.getPreviousStep();
         } while (step != null);
 
-        if (graphStep == null) {
+        // A mid-traversal scan runs once per incoming traverser (including bulk).
+        if (graphStep == null || !graphStep.isStartStep() ||
+            traversal.getStartStep() != graphStep) {
             return;
         }
 

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

@@ -210,8 +210,7 @@ public boolean equals(Object obj) {
 
         HugeGraphStep other = (HugeGraphStep) obj;
         return this.hasContainers.equals(other.hasContainers) &&
-               this.queryInfo.equals(other.queryInfo) &&
-               this.lastTimeResults.equals(other.lastTimeResults);
+               this.queryInfo.equals(other.queryInfo);
     }
 
     @Override

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

@@ -60,6 +60,7 @@
 import org.apache.tinkerpop.gremlin.process.traversal.Order;
 import org.apache.tinkerpop.gremlin.process.traversal.P;
 import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.TextP;
 import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
 import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
 import org.apache.tinkerpop.gremlin.process.traversal.step.HasContainerHolder;
@@ -177,7 +178,8 @@ public static void extractHasContainer(HugeGraphStep<?, ?> newStep,
         while (step instanceof HasStep || step instanceof NoOpBarrierStep) {
             Step<?, ?> nextStep = step.getNextStep();
             if (step instanceof HasStep) {
-                HasContainerHolder holder = (HasContainerHolder) step;
+                HasContainerHolder holder =
+                        (HasContainerHolder) step;
                 boolean connectiveLabelStep =
                         removeConnectiveLabelStep(step);
                 /*
@@ -385,7 +387,7 @@ private static boolean isLabelContainer(HasContainer has) {
     }
 
     static boolean isPositiveLabelContainer(HasContainer has) {
-        if (!isLabelContainer(has)) {
+        if (!isLabelContainer(has) || hasNullLabelValue(has)) {
             return false;
         }
 
@@ -541,6 +543,29 @@ private static boolean hasNullPredicate(HasContainer has) {
         return false;
     }
 
+    private static boolean hasNullLabelValue(HasContainer has) {
+        if (!isLabelContainer(has)) {
+            return false;
+        }
+
+        List<P<Object>> predicates = new ArrayList<>();
+        collectPredicates(predicates, ImmutableList.of(has.getPredicate()));
+        for (P<Object> pred : predicates) {
+            Object value = pred.getValue();
+            if (value == null) {
+                return true;
+            }
+            if (value instanceof Collection) {
+                for (Object label : (Collection<?>) value) {
+                    if (label == null) {
+                        return true;
+                    }
+                }
+            }
+        }
+        return false;
+    }
+
     private static boolean hasBooleanIndex(HugeGraph graph,
                                            SchemaLabel schemaLabel,
                                            PropertyKey pkey) {
@@ -607,7 +632,8 @@ public static void extractHasContainer(HugeVertexStep<?> newStep,
             Step<?, ?> nextStep = step.getNextStep();
             if (step instanceof HasStep) {
                 removeConnectiveLabelStep(step);
-                HasContainerHolder holder = (HasContainerHolder) step;
+                HasContainerHolder holder =
+                        (HasContainerHolder) step;
                 if (extractHasContainers(newStep, holder)) {
                     TraversalHelper.copyLabels(step, step.getPreviousStep(), false);
                     traversal.removeStep(step);
@@ -655,6 +681,11 @@ private static boolean canExtractHasContainers(HugeGraph graph,
 
     static boolean canExtractHasContainer(HugeGraph graph,
                                           HasContainer has) {
+        if (has.getKey() == null || has.getPredicate() == null ||
+            hasNullLabelValue(has) ||
+            hasTextPredicate(has)) {
+            return false;
+        }
         if (isSysProp(has.getKey())) {
             return true;
         }
@@ -687,6 +718,17 @@ static boolean canExtractHasContainer(HugeGraph graph,
         return true;
     }
 
+    private static boolean hasTextPredicate(HasContainer has) {
+        List<P<Object>> predicates = new ArrayList<>();
+        collectPredicates(predicates, ImmutableList.of(has.getPredicate()));
+        for (P<Object> predicate : predicates) {
+            if (TextP.class.isInstance(predicate)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     public static void extractOrder(Step<?, ?> newStep,
                                     Traversal.Admin<?, ?> traversal) {
         Step<?, ?> step = newStep;
@@ -1097,6 +1139,9 @@ public static HugeKeys string2HugeKey(String key) {
     }
 
     public static HugeKeys token2HugeKey(String key) {
+        if (key == null) {
+            return null;
+        }
         if (key.equals(T.label.getAccessor())) {
             return HugeKeys.LABEL;
         } else if (key.equals(T.id.getAccessor())) {
@@ -1187,7 +1232,7 @@ public static void convHasStep(HugeGraph graph, HasStep<?> step) {
     private static void convPredicateValue(HugeGraph graph,
                                            HasContainer has) {
         // No need to convert if key is sys-prop
-        if (isSysProp(has.getKey())) {
+        if (has.getKey() == null || isSysProp(has.getKey())) {
             return;
         }
         PropertyKey pkey = graph.propertyKey(has.getKey());
@@ -1198,15 +1243,24 @@ private static void updatePredicateValue(P<?> predicate, PropertyKey pkey) {
         List<P<Object>> leafPredicates = new ArrayList<>();
         collectPredicates(leafPredicates, ImmutableList.of(predicate));
         for (P<Object> pred : leafPredicates) {
-            if (pred.getBiPredicate() == Compare.neq &&
-                pred.getValue() == null) {
+            if (isNullInequalityPredicate(pred)) {
                 continue;
             }
             Object value = validPropertyValue(pred.getValue(), pkey);
             pred.setValue(value);
         }
     }
 
+    private static boolean isNullInequalityPredicate(P<?> predicate) {
+        if (predicate.getValue() != null) {
+            return false;
+        }
+        if (predicate.getBiPredicate() == Compare.neq) {
+            return true;
+        }
+        return false;
+    }
+
     private static boolean isSysProp(String key) {
         if (QueryHolder.SYSPROP_PAGE.equals(key)) {
             return true;