bitflicker64 commented on code in PR #3193:
URL: https://github.com/apache/hugegraph/pull/3193#discussion_r3955851303
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java:
##########
@@ -2009,45 +1981,13 @@ private boolean rightResultFromIndexQuery(Query query,
HugeElement elem) {
return false;
}
- protected static boolean queryNeedsPostFilter(Query query) {
- while (query != null) {
- if (query instanceof ConditionQuery) {
- ConditionQuery cq = (ConditionQuery) query;
- /*
- * Search conditions need post-filtering even before query
- * optimization marks the query with an optimized type.
- */
- boolean edgeIndexWithLabel =
- cq.resultType().isEdge() &&
- cq.optimized() == OptimizedType.INDEX &&
- cq.condition(HugeKeys.LABEL) != null;
- if (cq.hasSearchCondition() ||
- (conditionQueryNeedsPostFilter(cq) &&
- !edgeIndexWithLabel)) {
- return true;
- }
- }
- query = query.originQuery();
- }
- return false;
- }
-
- private static boolean conditionQueryNeedsPostFilter(ConditionQuery query)
{
- return query.optimized() != OptimizedType.NONE;
- }
-
- private <T extends HugeElement> Iterator<T> filterExpiredResultFromBackend(
- Query query, Iterator<T> results) {
- if (this.store().features().supportsTtl() || query.showExpired()) {
- return results;
- }
- // Filter expired vertices/edges with TTL
- return new FilterIterator<>(results, elem -> {
- if (elem.expired()) {
- DeleteExpiredJob.asyncDeleteExpiredObject(this.graph(), elem);
- return false;
+ protected <T extends HugeElement> QueryResults<T>
filterExpiredBatches(QueryResults<T> batches) {
Review Comment:
🧹 The TTL stage is now unconditional.
Base `filterExpiredResultFromBackend` returned the source iterator untouched
when `store().features().supportsTtl()` or `query.showExpired()` held, so a
TTL-capable store carried no wrapper at all. This version always adds a
`filter` stage, which is one of the four counted in the `QueryBatch` comment,
and moves the store-feature test into the per-element predicate.
Requested change: read `storeFeatures().supportsTtl()` once and return
`batches` unchanged when it is true, leaving `context.showExpired()` and
`elem.expired()` in the predicate since those do vary.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryBatch.java:
##########
@@ -0,0 +1,262 @@
+/*
+ * 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.backend.query;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.function.BiPredicate;
+import java.util.function.Function;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.iterator.CIter;
+import org.apache.hugegraph.iterator.Metadatable;
+import org.apache.hugegraph.type.Idfiable;
+import org.apache.hugegraph.util.InsertionOrderUtil;
+
+/** A single query's results, with exclusive ownership of its iterator chain.
*/
+public final class QueryBatch<R> implements AutoCloseable {
+
+ private final BatchIterator<R> results;
+ private final QueryResultContext context;
+
+ public QueryBatch(Iterator<R> results, QueryResultContext context) {
+ this.context = context;
+ this.results = new BatchIterator<R>() {
+ @Override
+ protected R fetch() {
+ return results.hasNext() ? results.next() : null;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ closeAll(results);
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return metadataOf(results, meta, args);
+ }
+ };
+ }
+
+ public Iterator<R> results() {
+ return this.results;
+ }
+
+ public QueryResultContext context() {
+ return this.context;
+ }
+
+ public <T> QueryBatch<T> map(Function<R, T> mapper) {
+ return this.flatMap(value -> {
+ T mapped = mapper.apply(value);
+ return mapped == null ? Collections.emptyIterator() :
+ Collections.singleton(mapped).iterator();
Review Comment:
⚠️ Each `map` or `filter` stage allocates a set and an iterator per
surviving element.
`map` routes through `flatMap` and returns
`Collections.singleton(mapped).iterator()` for every value that passes, so each
surviving element costs one `SingletonSet` plus one iterator, per stage.
`filter` at line 117 is built on `map`, so it pays the same. Rejected elements
are free, since they get the shared `Collections.emptyIterator()`.
A plain vertex read stacks four such stages: `.map(this::parseEntry)` and
`filterExpiredBatches` in `GraphTransaction.fetchVertexBatch:883`,
`filterUnmatchedRecord` in `processBatches:892`, and `filterInvalidRecord` in
`queryValidVerticesFromBackend:888`. Edges stack three on top of the one real
expansion at `fetchEdgeBatch:1133`. `MapperIterator.fetch()`, which this
replaces, loops on a null result in place and allocates nothing.
Requested change: give `map` its own `BatchIterator` that applies the
function and loops while the result is null; `filter` can keep delegating to it.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java:
##########
@@ -316,135 +318,143 @@ private boolean needCacheVertex(HugeVertex vertex) {
@Override
@Watched(prefix = "graphcache")
- protected Iterator<HugeVertex> queryVerticesFromBackend(Query query) {
- if (this.enableCacheVertex() &&
- query.idsSize() > 0 && query.conditionsSize() == 0 &&
- !queryNeedsPostFilter(query)) {
- return this.queryVerticesByIds((IdQuery) query);
- } else {
- return super.queryVerticesFromBackend(query);
+ protected QueryResults<HugeVertex> fetchVertexBatch(Query query) {
+ if (!this.enableCacheVertex() || query.paging() ||
+ query.idsSize() == 0 || query.conditionsSize() != 0) {
+ return super.fetchVertexBatch(query);
}
- }
-
- @Watched(prefix = "graphcache")
- private Iterator<HugeVertex> queryVerticesByIds(IdQuery query) {
- if (query.idsSize() == 1) {
- Id vertexId = query.ids().iterator().next();
- HugeVertex vertex = (HugeVertex) this.verticesCache.get(vertexId);
- if (vertex != null) {
- if (!vertex.expired()) {
- return QueryResults.iterator(vertex);
- }
- this.verticesCache.invalidate(vertexId);
- }
- Iterator<HugeVertex> rs = super.queryVerticesFromBackend(query);
- vertex = QueryResults.one(rs);
- if (vertex == null) {
- return QueryResults.emptyIterator();
- }
- if (needCacheVertex(vertex)) {
- this.verticesCache.update(vertex.id(), vertex);
- }
- return QueryResults.iterator(vertex);
- }
-
- IdQuery newQuery = new IdQuery(HugeType.VERTEX, query);
+ QueryResultContext context = new QueryResultContext(query);
+ IdQuery missing = new IdQuery(query.resultType(), query);
List<HugeVertex> vertices = new ArrayList<>();
- for (Id vertexId : query.ids()) {
- HugeVertex vertex = (HugeVertex) this.verticesCache.get(vertexId);
- if (vertex == null) {
- newQuery.query(vertexId);
- } else if (vertex.expired()) {
- newQuery.query(vertexId);
- this.verticesCache.invalidate(vertexId);
+ for (Id id : query.ids()) {
+ HugeVertex vertex = (HugeVertex) this.verticesCache.get(id);
+ if (vertex == null || vertex.expired()) {
+ missing.query(id);
+ if (vertex != null) {
+ this.verticesCache.invalidate(id);
+ }
} else {
vertices.add(vertex);
}
}
-
- // Join results from cache and backend
- ExtendableIterator<HugeVertex> results = new ExtendableIterator<>();
- if (!vertices.isEmpty()) {
- results.extend(vertices.iterator());
- } else {
- // Just use the origin query if find none from the cache
- newQuery = query;
- }
-
- if (!newQuery.empty()) {
- Iterator<HugeVertex> rs = super.queryVerticesFromBackend(newQuery);
- // Generally there are not too much data with id query
- ListIterator<HugeVertex> listIterator = QueryResults.toList(rs);
- for (HugeVertex vertex : listIterator.list()) {
- // Skip large vertex
- if (needCacheVertex(vertex)) {
+ if (!missing.empty()) {
+ QueryResults<HugeVertex> fetched =
super.fetchVertexBatch(vertices.isEmpty() ? query : missing);
+ if (vertices.isEmpty() && !fetched.batches().hasNext()) {
+ return fetched;
+ }
+ ListIterator<HugeVertex> candidates =
QueryResults.toList(fetched.iterator());
+ for (HugeVertex vertex : candidates.list()) {
+ if (this.needCacheVertex(vertex)) {
this.verticesCache.update(vertex.id(), vertex);
}
+ vertices.add(vertex);
}
- results.extend(listIterator);
}
-
- return results;
+ // Keep hits and misses in one logical batch for filtering and ID
ordering.
+ return this.filterExpiredBatches(new
QueryResults<>(vertices.iterator(), context));
}
@Override
- @Watched(prefix = "graphcache")
- protected Iterator<HugeEdge> queryEdgesFromBackend(Query query) {
+ protected QueryResults<HugeEdge> queryEdgesFromMemory(Query query) {
RamTable ramtable = this.params().ramtable();
if (ramtable != null && ramtable.matched(query)) {
- return ramtable.query(query);
+ return new QueryResults<>(ramtable.query(query), query);
}
+ return null;
+ }
- if (!this.enableCacheEdge() || query.empty() || query.paging() ||
- query.bigCapacity() || queryNeedsPostFilter(query)) {
- // Don't cache all-edge, paging, large, or post-filtered queries
- return super.queryEdgesFromBackend(query);
+ @Override
+ @Watched(prefix = "graphcache")
+ protected QueryResults<HugeEdge> fetchEdgeBatch(Query query) {
+ QueryResultContext context = new QueryResultContext(query);
+ List<Query> chain = context.queries();
+ Query request = chain.get(chain.size() - 1);
+ if (!this.enableCacheEdge() || request.empty() || request.paging() ||
request.bigCapacity()) {
+ return super.fetchEdgeBatch(query);
}
-
- Id cacheKey = new QueryId(query);
- Object value = this.edgesCache.get(cacheKey);
- @SuppressWarnings("unchecked")
- Collection<HugeEdge> edges = (Collection<HugeEdge>) value;
- if (value != null) {
- for (HugeEdge edge : edges) {
+ Id cacheKey = new QueryId(request);
+ Id batchKey = new QueryId(query);
+ CachedEdgeQuery group = new
CachedEdgeQuery(this.edgesCache.get(cacheKey));
Review Comment:
🧹 The read path shallow-copies the group's index list for nothing.
Line 377 builds a `CachedEdgeQuery` on every edge fetch that reaches the
cache, and the constructor at 427-430 does `new ArrayList<>((List<Object>)
cached)`. Nothing on the read path mutates it: `get()` only scans.
`cacheEdgeBatch:413` already makes its own copy inside `synchronized
(this.edgesCache)`, which is where the copy belongs.
Requested change: scan the cached list directly for the lookup and keep the
copy in `cacheEdgeBatch`.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java:
##########
@@ -316,135 +318,143 @@ private boolean needCacheVertex(HugeVertex vertex) {
@Override
@Watched(prefix = "graphcache")
- protected Iterator<HugeVertex> queryVerticesFromBackend(Query query) {
- if (this.enableCacheVertex() &&
- query.idsSize() > 0 && query.conditionsSize() == 0 &&
- !queryNeedsPostFilter(query)) {
- return this.queryVerticesByIds((IdQuery) query);
- } else {
- return super.queryVerticesFromBackend(query);
+ protected QueryResults<HugeVertex> fetchVertexBatch(Query query) {
+ if (!this.enableCacheVertex() || query.paging() ||
+ query.idsSize() == 0 || query.conditionsSize() != 0) {
+ return super.fetchVertexBatch(query);
}
- }
-
- @Watched(prefix = "graphcache")
- private Iterator<HugeVertex> queryVerticesByIds(IdQuery query) {
- if (query.idsSize() == 1) {
- Id vertexId = query.ids().iterator().next();
- HugeVertex vertex = (HugeVertex) this.verticesCache.get(vertexId);
- if (vertex != null) {
- if (!vertex.expired()) {
- return QueryResults.iterator(vertex);
- }
- this.verticesCache.invalidate(vertexId);
- }
- Iterator<HugeVertex> rs = super.queryVerticesFromBackend(query);
- vertex = QueryResults.one(rs);
- if (vertex == null) {
- return QueryResults.emptyIterator();
- }
- if (needCacheVertex(vertex)) {
- this.verticesCache.update(vertex.id(), vertex);
- }
- return QueryResults.iterator(vertex);
- }
-
- IdQuery newQuery = new IdQuery(HugeType.VERTEX, query);
+ QueryResultContext context = new QueryResultContext(query);
+ IdQuery missing = new IdQuery(query.resultType(), query);
List<HugeVertex> vertices = new ArrayList<>();
- for (Id vertexId : query.ids()) {
- HugeVertex vertex = (HugeVertex) this.verticesCache.get(vertexId);
- if (vertex == null) {
- newQuery.query(vertexId);
- } else if (vertex.expired()) {
- newQuery.query(vertexId);
- this.verticesCache.invalidate(vertexId);
+ for (Id id : query.ids()) {
+ HugeVertex vertex = (HugeVertex) this.verticesCache.get(id);
+ if (vertex == null || vertex.expired()) {
+ missing.query(id);
+ if (vertex != null) {
+ this.verticesCache.invalidate(id);
+ }
} else {
vertices.add(vertex);
}
}
-
- // Join results from cache and backend
- ExtendableIterator<HugeVertex> results = new ExtendableIterator<>();
- if (!vertices.isEmpty()) {
- results.extend(vertices.iterator());
- } else {
- // Just use the origin query if find none from the cache
- newQuery = query;
- }
-
- if (!newQuery.empty()) {
- Iterator<HugeVertex> rs = super.queryVerticesFromBackend(newQuery);
- // Generally there are not too much data with id query
- ListIterator<HugeVertex> listIterator = QueryResults.toList(rs);
- for (HugeVertex vertex : listIterator.list()) {
- // Skip large vertex
- if (needCacheVertex(vertex)) {
+ if (!missing.empty()) {
+ QueryResults<HugeVertex> fetched =
super.fetchVertexBatch(vertices.isEmpty() ? query : missing);
+ if (vertices.isEmpty() && !fetched.batches().hasNext()) {
+ return fetched;
+ }
+ ListIterator<HugeVertex> candidates =
QueryResults.toList(fetched.iterator());
+ for (HugeVertex vertex : candidates.list()) {
+ if (this.needCacheVertex(vertex)) {
this.verticesCache.update(vertex.id(), vertex);
}
+ vertices.add(vertex);
}
- results.extend(listIterator);
}
-
- return results;
+ // Keep hits and misses in one logical batch for filtering and ID
ordering.
+ return this.filterExpiredBatches(new
QueryResults<>(vertices.iterator(), context));
}
@Override
- @Watched(prefix = "graphcache")
- protected Iterator<HugeEdge> queryEdgesFromBackend(Query query) {
+ protected QueryResults<HugeEdge> queryEdgesFromMemory(Query query) {
RamTable ramtable = this.params().ramtable();
if (ramtable != null && ramtable.matched(query)) {
- return ramtable.query(query);
+ return new QueryResults<>(ramtable.query(query), query);
}
+ return null;
+ }
- if (!this.enableCacheEdge() || query.empty() || query.paging() ||
- query.bigCapacity() || queryNeedsPostFilter(query)) {
- // Don't cache all-edge, paging, large, or post-filtered queries
- return super.queryEdgesFromBackend(query);
+ @Override
+ @Watched(prefix = "graphcache")
+ protected QueryResults<HugeEdge> fetchEdgeBatch(Query query) {
+ QueryResultContext context = new QueryResultContext(query);
+ List<Query> chain = context.queries();
+ Query request = chain.get(chain.size() - 1);
+ if (!this.enableCacheEdge() || request.empty() || request.paging() ||
request.bigCapacity()) {
+ return super.fetchEdgeBatch(query);
}
-
- Id cacheKey = new QueryId(query);
- Object value = this.edgesCache.get(cacheKey);
- @SuppressWarnings("unchecked")
- Collection<HugeEdge> edges = (Collection<HugeEdge>) value;
- if (value != null) {
- for (HugeEdge edge : edges) {
+ Id cacheKey = new QueryId(request);
+ Id batchKey = new QueryId(query);
Review Comment:
🧹 The batch key stored in the value can be larger than the value it labels.
`QueryId.asString()` is `Query.toString()`, and `Query.toString():610`
appends `id in <ids>`. `put` stores that string next to the candidates, so an
index-derived leaf writes up to `query.batch_size` ids of rendered text
(`CoreOptions.java:466-472`, default 1000) beside a candidate list capped at
`MAX_CACHE_EDGES_PER_QUERY`, which is 100 (line 63).
In the common adjacency shape the leaf is also the chain root, so `batchKey`
and `cacheKey` render the same string and the value holds a verbatim copy of
its own cache key.
Requested change: skip storing the label when `batchKey.equals(cacheKey)`
and treat a group of one unlabelled entry as that case, which keeps the stored
value a nested list and leaves off-heap serialization unchanged.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/PageEntryIterator.java:
##########
@@ -17,112 +17,77 @@
package org.apache.hugegraph.backend.page;
-import java.util.NoSuchElementException;
+import java.util.Iterator;
import org.apache.hugegraph.backend.query.Query;
-import org.apache.hugegraph.backend.query.QueryResults;
+import org.apache.hugegraph.backend.query.QueryBatch.BatchIterator;
+import org.apache.hugegraph.backend.query.QueryBatch;
import org.apache.hugegraph.exception.NotSupportException;
-import org.apache.hugegraph.iterator.CIter;
import org.apache.hugegraph.util.E;
-import org.apache.tinkerpop.gremlin.structure.util.CloseableIterator;
-public class PageEntryIterator<R> implements CIter<R> {
+/** Produces pages without probing the following page to delimit the current
one. */
+public class PageEntryIterator<R> extends BatchIterator<QueryBatch<R>> {
private final QueryList<R> queries;
private final long pageSize;
private final PageInfo pageInfo;
- private final QueryResults<R> queryResults; // for upper layer
-
- private QueryList.PageResults<R> pageResults;
+ private Iterator<QueryBatch<R>> pageBatches;
private long remaining;
public PageEntryIterator(QueryList<R> queries, long pageSize) {
this.queries = queries;
this.pageSize = pageSize;
- this.pageInfo = this.parsePageInfo();
- this.queryResults = new QueryResults<>(this, queries.parent());
-
- this.pageResults = QueryList.PageResults.emptyIterator();
+ this.pageInfo =
PageInfo.fromString(queries.parent().pageWithoutCheck());
+ E.checkState(this.pageInfo.offset() < queries.total(),
+ "Invalid page offset '%s' exceeds the size of
IdHolderList",
Review Comment:
🧹 The offset check dropped the page token from its message.
Base `parsePageInfo` reported `"Invalid page '%s' with an offset '%s'
exceeds the size of IdHolderList"` with both the raw page string and the
offset. This check keeps only the offset, and it fires exactly when a caller
supplies a stale or hand-written `~page` value, which is the case where the
token itself is the thing worth seeing in the log.
Requested change: put `queries.parent().pageWithoutCheck()` back in the
message alongside the offset.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java:
##########
@@ -45,14 +44,16 @@
import org.apache.hugegraph.backend.page.IdHolderList;
import org.apache.hugegraph.backend.page.PageInfo;
import org.apache.hugegraph.backend.page.QueryList;
-import org.apache.hugegraph.backend.query.Aggregate;
import org.apache.hugegraph.backend.query.Aggregate.AggregateFunc;
+import org.apache.hugegraph.backend.query.Aggregate;
import org.apache.hugegraph.backend.query.Condition;
-import org.apache.hugegraph.backend.query.ConditionQuery;
import org.apache.hugegraph.backend.query.ConditionQuery.OptimizedType;
+import org.apache.hugegraph.backend.query.ConditionQuery;
import org.apache.hugegraph.backend.query.ConditionQueryFlatten;
import org.apache.hugegraph.backend.query.IdQuery;
import org.apache.hugegraph.backend.query.Query;
+import org.apache.hugegraph.backend.query.QueryBatch;
Review Comment:
🧹 Four imports are left unused, and two were reordered for no reason.
Anchored here because lines 23, 68, 70 and 105 fall outside every hunk in this
file.
`java.util.Collections` (23),
`org.apache.hugegraph.iterator.FlatMapperIterator` (68),
`org.apache.hugegraph.iterator.ListIterator` (70) and
`com.google.common.collect.Iterators` (105) each appear exactly once in the
file at this head, on their own import line, and all four are used at base.
`style/checkstyle.xml:59` enables `UnusedImports`, inheriting the `info`
severity set at line 23, so the `validate`-bound execution logs them without
failing the build.
Lines 47-48 and 50-51 also moved `Aggregate.AggregateFunc` above `Aggregate`
and `ConditionQuery.OptimizedType` above `ConditionQuery`, which is neither
alphabetical nor related to the change. `PageEntryIterator.java:23-24` has the
same swap.
Requested change: delete the four import lines and restore the original
order of the two pairs.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java:
##########
@@ -1051,87 +1069,77 @@ public Iterator<Edge> queryEdges(Query query) {
}
protected Iterator<HugeEdge> queryEdgesFromBackend(Query query) {
- assert query.resultType().isEdge();
+ return this.queryEdgeBatchesFromBackend(query).iterator();
+ }
+
+ protected QueryResults<HugeEdge> queryEdgesFromMemory(Query query) {
+ return null;
+ }
+ private QueryResults<HugeEdge> queryEdgeBatchesFromBackend(Query query) {
+ assert query.resultType().isEdge();
+ QueryResults<HugeEdge> memory = this.queryEdgesFromMemory(query);
+ if (memory != null) {
+ return memory;
+ }
if (query instanceof ConditionQuery && !query.paging()) {
- // TODO: support: paging + parent label
boolean supportIn =
this.storeFeatures().supportsQueryWithInCondition();
- // consider multi labels + properties,
- // see
org.apache.hugegraph.core.EdgeCoreTest.testQueryInEdgesOfVertexByLabels
- Stream<ConditionQuery> flattenedQueries =
- ConditionQueryFlatten.flatten((ConditionQuery) query,
supportIn).stream();
-
- Stream<Iterator<HugeEdge>> edgeIterators = flattenedQueries.map(cq
-> {
+ List<ConditionQuery> flattened = ConditionQueryFlatten.flatten(
+ (ConditionQuery) query, supportIn);
+ Function<ConditionQuery, QueryResults<HugeEdge>> fetcher = cq -> {
Id label = cq.condition(HugeKeys.LABEL);
if (this.storeFeatures().supportsFatherAndSubEdgeLabel() &&
- label != null &&
- graph().edgeLabel(label).isFather() &&
+ label != null && graph().edgeLabel(label).isFather() &&
cq.condition(HugeKeys.SUB_LABEL) == null &&
cq.condition(HugeKeys.OWNER_VERTEX) != null &&
cq.condition(HugeKeys.DIRECTION) != null &&
matchEdgeSortKeys(cq, false, this.graph())) {
- // g.V("V.id").outE("parentLabel").has("sortKey","value")
- return parentElQueryWithSortKeys(
- graph().edgeLabel(label), graph().edgeLabels(),
cq);
- } else {
- return queryEdgesFromBackendInternal(cq);
+ EdgeLabel parent = graph().edgeLabel(label);
+ Iterator<EdgeLabel> children =
graph().edgeLabels().stream()
+ .filter(el -> el.edgeLabelType().sub() &&
+
el.fatherId().equals(parent.id())).iterator();
+ return QueryResults.flatMap(children, child -> {
+ ConditionQuery subQuery = cq.copy();
+ subQuery.eq(HugeKeys.SUB_LABEL, child.id());
+ return this.queryEdgeBatchesFromBackend(subQuery);
+ });
}
- });
-
- return edgeIterators.reduce(ExtendableIterator::concat)
- .orElse(Collections.emptyIterator());
+ return this.queryEdgeBatchesFromBackendInternal(cq);
+ };
+ // Preserve immediate validation for a single query without
activating a sibling.
+ if (flattened.size() == 1) {
+ return fetcher.apply(flattened.get(0));
+ }
+ return QueryResults.flatMap(flattened.iterator(), fetcher);
Review Comment:
🧹 Multi-branch edge queries lost their eager validation, and only the
single-query case says so.
Base built a `Stream<Iterator<HugeEdge>>` and ended with
`reduce(ExtendableIterator::concat)`. That is a terminal operation, so every
`queryEdgesFromBackendInternal(cq)` ran at call time and any validation failure
surfaced from the call itself. `QueryResults.flatMap` here runs each branch
only when iteration reaches it, so a failure in the second or later flattened
query now surfaces mid-iteration. Line 1109 documents the choice for
`flattened.size() == 1` and nothing covers the multi-branch case.
Requested change: say in the PR body which failures are expected to move
from call time to iteration time for multi-branch edge queries, or apply the
same eager first step there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
