bitflicker64 commented on code in PR #3193:
URL: https://github.com/apache/hugegraph/pull/3193#discussion_r3940228184
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java:
##########
@@ -316,135 +318,142 @@ 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.idsSize() == 0 ||
query.conditionsSize() != 0) {
Review Comment:
‼️ This bypass needs `|| query.paging()`.
Unlike the `queryVerticesFromBackend` override it replaces, this method is
the QueryList *fetcher*, so it runs on the inner query.
`GraphTransaction.optimizeQuery:1564` rewrites `label + primary-key` into `new
IdQuery(query, id)`, and `Query(HugeType, Query originQuery)`
(`Query.java:84-109`) copies neither page nor conditions, so that leaf reaches
line 322 with `idsSize()==1` and `conditionsSize()==0` even after
`QueryList.java:187-188` sets its page.
Line 353 then returns a `QueryResults` whose `metadata` is the batch's
`BatchIterator` delegating to a plain `ArrayList` iterator, so
`metadata(PageInfo.PAGE)` is null, and `mapBatches` passes that null through
`filterExpiredBatches`. `QueryList.java:198` calls
`PageInfo.pageState(results.iterator())`, which is `E.checkState(page
instanceof PageState, "Invalid PageState '%s'", page)`. So `g.V().hasLabel(<pk
label>).has(<pk>, v).has("~page", "")` fails with `Invalid PageState 'null'`
wherever `supportsQueryByPage()` is true. The only escape is a cache miss that
returns nothing, via the early return at 341-343.
The green paging tests do not contradict this. Index-paged queries go
through `IndexQuery.iterator`, which takes its cursor from
`pageIds.pageState()` (`QueryList.java:297`) rather than from result metadata,
and no `VertexCoreTest` case pairs a primary-key label with its own primary-key
property under `~page`. `InMemoryDBStore.supportsQueryByPage()` is false, so
the unit tests cannot reach it either.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java:
##########
@@ -316,135 +318,142 @@ 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.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));
+ Collection<HugeEdge> cached = group.get(batchKey);
+ if (cached != null) {
+ for (HugeEdge edge : cached) {
if (edge.expired()) {
this.edgesCache.invalidate(cacheKey);
- value = null;
+ cached = null;
break;
}
}
}
-
- if (value != null) {
- // Not cached or the cache expired
- return edges.iterator();
+ if (cached != null) {
+ return this.filterExpiredBatches(new
QueryResults<>(cached.iterator(), context));
}
+ QueryResults<HugeEdge> fetched = super.fetchEdgeBatch(query);
+ if (!fetched.batches().hasNext()) {
+ this.cacheEdgeBatch(cacheKey, batchKey, Collections.emptyList());
+ return fetched;
+ }
+ return fetched.mapBatches(batch -> {
+ Iterator<HugeEdge> source = batch.results();
+ List<HugeEdge> candidates = new
ArrayList<>(MAX_CACHE_EDGES_PER_QUERY + 1);
+ // Limit probing to this batch; never request another batch to
fill the cache.
+ while (candidates.size() <= MAX_CACHE_EDGES_PER_QUERY &&
source.hasNext()) {
+ candidates.add(source.next());
+ }
+ if (candidates.size() <= MAX_CACHE_EDGES_PER_QUERY) {
+ this.cacheEdgeBatch(cacheKey, batchKey, candidates);
+ }
+ return new QueryBatch<>(
+ new ExtendableIterator<>(candidates.iterator(), source),
batch.context());
+ });
+ }
- Iterator<HugeEdge> rs = super.queryEdgesFromBackend(query);
- if (queryNeedsPostFilter(query)) {
- // The backend query may promote query.optimized() through origin-
- // query propagation, so re-check before caching
- return rs;
+ private void cacheEdgeBatch(Id cacheKey, Id batchKey, List<HugeEdge>
candidates) {
+ synchronized (this.edgesCache) {
Review Comment:
🧹 This monitor is shared by every transaction on the graph.
`edgesCache` comes from `CacheManager` (lines 104-108, 122-135).
`edge.cache_type` defaults to `l2` (`CoreOptions.java:559-565`), so it is
`levelCache(...)`, which like `cache(...)` returns one instance per name out of
the static `INSTANCE.caches` map (`CacheManager.java:37, 59, 136-152`). The
name is `edge-<graph>`, so all concurrent `CachedGraphTransaction`s for that
graph serialise here, where the previous code did a lock-free
`edgesCache.update(cacheKey, edges)`.
It is only reached on a miss (lines 392, 403) and the critical section is
bounded, since the group caps at 100 entries (449-450), so the cost is small.
Still, striping the lock per `cacheKey`, or just tolerating a lost update from
a concurrent sibling batch, would keep the miss path lock-free.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResultContext.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.List;
+
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.query.ConditionQuery.OptimizedType;
+import org.apache.hugegraph.backend.query.ConditionQuery.ResultsFilter;
+
+/** Decisions captured for one batch; queries retain shared index cleanup
data. */
+public final class QueryResultContext {
+
+ private final List<Query> queries;
+ private final List<Id> inputIds;
+ private final boolean mustSortByInputIds;
+ private final ConditionQuery matchQuery;
+ private final ResultsFilter resultsFilter;
+ private final OptimizedType optimizedType;
+ private final boolean showExpired;
+ private final boolean showHidden;
+ private final boolean showDeleting;
+
+ public QueryResultContext(Query query) {
+ this(query, false);
+ }
+
+ public QueryResultContext(Query query, boolean inputOrderSatisfied) {
+ List<Query> chain = new ArrayList<>();
+ ConditionQuery match = null;
+ ResultsFilter filter = null;
+ OptimizedType optimized = OptimizedType.NONE;
+ Query visibility = query;
+ for (Query current = query; current != null; current =
current.originQuery()) {
+ chain.add(current);
+ visibility = current;
+ if (current instanceof ConditionQuery) {
+ ConditionQuery condition = (ConditionQuery) current;
+ if (optimized == OptimizedType.NONE) {
+ optimized = condition.optimized();
+ }
+ if (filter == null) {
+ filter = condition.resultsFilter();
+ }
+ if (current.resultType().isGraph()) {
Review Comment:
🧹 `match` is last-wins here while `optimized` and `filter` above are
first-wins.
Lines 56-61 keep the innermost non-`NONE` `optimized()` and the innermost
non-null `resultsFilter()`, but line 63 reassigns `match` on every graph-typed
`ConditionQuery` in the chain, so `matchQuery()` ends up as the outermost one.
`rightResultFromIndexQuery` then runs `cq.test(elem, context.resultsFilter())`
(`GraphTransaction.java:1951`) with a filter and an optimized type that need
not belong to `cq`. The old code read all three off a single query: `query`, or
`query.originQuery()` when that was the `ConditionQuery`.
The chains are also longer now that `QueryList.indexIdQuery` roots the
`IdQuery` at `bindQuery` instead of `parent()`. I could not build a chain where
the two directions disagree, so this may be deliberate, but a line of comment
saying why they differ would save the next reader the same trace.
##########
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",
+ this.pageInfo.offset());
this.remaining = queries.parent().limit();
}
- private PageInfo parsePageInfo() {
- String page = this.queries.parent().pageWithoutCheck();
- PageInfo pageInfo = PageInfo.fromString(page);
- E.checkState(pageInfo.offset() < this.queries.total(),
- "Invalid page '%s' with an offset '%s' exceeds " +
- "the size of IdHolderList", page, pageInfo.offset());
- return pageInfo;
- }
-
@Override
- public boolean hasNext() {
- if (this.pageResults.get().hasNext()) {
- return true;
- }
- return this.fetch();
- }
-
- private boolean fetch() {
- if ((this.remaining != Query.NO_LIMIT && this.remaining <= 0L) ||
- this.pageInfo.offset() >= this.queries.total()) {
- return false;
- }
-
- long pageSize = this.pageSize;
- if (this.remaining != Query.NO_LIMIT && this.remaining < pageSize) {
- pageSize = this.remaining;
- }
- this.closePageResults();
- this.pageResults = this.queries.fetchNext(this.pageInfo, pageSize);
- assert this.pageResults != null;
- this.queryResults.setQuery(this.pageResults.query());
-
- if (this.pageResults.get().hasNext()) {
- if (!this.pageResults.hasNextPage()) {
+ protected QueryBatch<R> fetch() throws Exception {
+ while (true) {
+ if (this.pageBatches != null && this.pageBatches.hasNext()) {
+ return this.pageBatches.next();
+ }
+ Iterator<QueryBatch<R>> previous = this.pageBatches;
+ this.pageBatches = null;
+ QueryBatch.closeAll(previous);
+ if ((this.remaining != Query.NO_LIMIT && this.remaining <= 0L) ||
+ this.pageInfo.offset() >= this.queries.total()) {
+ return null;
+ }
+ long size = this.remaining == Query.NO_LIMIT ? this.pageSize :
+ Math.min(this.pageSize, this.remaining);
+ QueryList.PageResults<R> page =
this.queries.fetchNext(this.pageInfo, size);
+ this.pageBatches = page.results().batches();
+ if (!this.pageBatches.hasNext()) {
Review Comment:
🧹 This `hasNext()` consumes and closes the static `PageResults.EMPTY`.
`IndexQuery.iterator` returns the shared `PageResults.emptyIterator()` when
a holder yields no ids (`QueryList.java:292`; the singleton is declared at
337-339). `page.results().batches()` on it is that singleton's one-shot
`BatchIterator`, so this call flips its `fetched` flag and, on the following
turn, closes it and clears its `queries` list, process-wide and unsynchronised.
Consequence: the first empty index page in the JVM takes the "one empty
batch" path (lines 70-77) and every later one takes the "no batch" path
(65-68). Both advance the offset by one and emit nothing, so it is benign
today, but the base code guarded against exactly this (`if (this.pageResults !=
QueryList.PageResults.EMPTY)`, base `PageEntryIterator.java:95-99`). Returning
a fresh `PageResults` from `emptyIterator()` would drop the coupling.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java:
##########
@@ -853,26 +854,43 @@ public Iterator<Vertex> queryVertices(Query query) {
}
protected Iterator<HugeVertex> queryVerticesFromBackend(Query query) {
- assert query.resultType().isVertex();
-
- QueryResults<BackendEntry> results = this.query(query);
- Iterator<BackendEntry> entries = results.iterator();
+ return this.queryVertexBatchesFromBackend(query).iterator();
+ }
- Iterator<HugeVertex> vertices = new MapperIterator<>(entries,
- this::parseEntry);
- vertices = this.filterExpiredResultFromBackend(query, vertices);
- vertices = this.filterUnmatchedRecords(vertices, query);
+ private QueryResults<HugeVertex> queryVertexBatchesFromBackend(Query
query) {
+ assert query.resultType().isVertex();
+ if (!(query instanceof ConditionQuery)) {
+ return this.processBatches(this.fetchVertexBatch(query));
+ }
+ QueryList<HugeVertex> queries = this.optimizeQueries(query,
this::fetchVertexBatch);
+ return this.processBatches(queries.empty() ? QueryResults.empty() :
+ queries.fetch(this.pageSize));
+ }
- if (!this.store().features().supportsQuerySortByInputIds()) {
- // There is no id in BackendEntry, so sort after deserialization
- vertices = results.keepInputOrderIfNeeded(vertices);
+ private QueryResults<BackendEntry> backendBatches(Query query) {
+ QueryResults<BackendEntry> results = super.query(query);
+ if (!results.iterator().hasNext()) {
+ // No raw records means no batch. Filtering a nonempty source may
+ // still produce an empty batch, which must keep its page cursor.
+ return results;
}
- return vertices;
+ QueryResultContext context = new QueryResultContext(
+ query, this.storeFeatures().supportsQuerySortByInputIds());
+ return results.mapBatches(batch -> new QueryBatch<>(batch.results(),
context));
+ }
+
+ protected QueryResults<HugeVertex> fetchVertexBatch(Query query) {
Review Comment:
🧹 Optional follow-up: `HugeFactoryAuthProxy.registerPrivateActions()` no
longer matches this class.
That list hides sensitive members from the Gremlin sandbox
(`HugeFactoryAuthProxy.java:239-281`). `filterInvalidRecords`,
`filterUnmatchedRecords`, `queryNeedsPostFilter` and
`conditionQueryNeedsPostFilter` are still listed but gone at this head, while
none of the internals introduced here are registered:
`queryVertexBatchesFromBackend`, `backendBatches`, `fetchVertexBatch`,
`processBatches`, `queryEdgeBatchesFromBackend`,
`queryEdgeBatchesFromBackendInternal`, `queryEdgesFromMemory`,
`fetchEdgeBatch`, `filterInvalidRecord`, `filterUnmatchedRecord`,
`filterExpiredBatches` (declared at 860, 870, 882, 891, 1075, 1079, 1123, 1132,
1893, 1898, 1984).
That file is outside this diff and the list was already imperfect
(`filterExpiredResultFromFromBackend` is a typo matching nothing, and line 96
carries a `TODO: add some test to ensure the effect`), so this is a follow-up
rather than something to fix here. Flagging it because the renames are what
made it stale.
--
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]
