You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hugegraph.apache.org by "contrueCT (via GitHub)" <gi...@apache.org> on 2026/09/05 04:13:35 UTC
[PR] fix(server): make query result batch boundaries explicit [hugegraph]
contrueCT opened a new pull request, #3193:
URL: https://github.com/apache/hugegraph/pull/3193
## Purpose of the PR
Closes #3190.
`QueryResults` currently detects a query boundary by probing `hasNext()` and then checking whether the active query changed. That probe can activate the next index query or backend page before the current batch finishes processing. Ordering is also selected from the first active segment, so an unsorted batch `[1]` followed by a batch with input IDs `[3, 2]` can incorrectly produce `[1, 2, 3]`.
This change gives each batch its own results and captured processing context. Parsing, TTL checks, residual filtering and input-order restoration finish within that batch before the caller consumes a flattened stream.
## Main Changes
- Introduce `QueryBatch` and `QueryResultContext` to carry the input IDs, ordering decision, visibility flags and effective results filter with the results that produced them.
- Make `QueryResults`, `QueryList` and `PageEntryIterator` compose explicit batches. Remove `queryVersion`/`currentQueries` boundary inference and shared results-filter propagation.
- Apply vertex and edge processing within each batch, preserving public/internal visibility, diagnostics and index-cleanup ownership. Keep native scans on their original single-batch path so backend page tokens and capacity checks remain intact.
- Cache candidates before residual matching and reapply the current batch context on hits. Preserve mixed hit/miss input ordering, the original edge-cache request key, off-heap serialization, TASK/SERVER query types and the RamTable fast path.
- Make batch wrappers close idempotently and preserve the primary exception when cleanup also fails. Native iterators that close themselves still rely on their existing backend close contract.
The lifecycle is:
```mermaid
flowchart LR
A[Activate batch A] --> C[Capture IDs, order and filter context]
C --> P[Parse and check TTL]
P --> F[Apply residual filter]
F --> O[Restore input order when needed]
O --> V[Apply public visibility when applicable]
V --> R[Return A results]
R --> E[A exhausted and closed]
E --> B[Activate batch B or next page]
```
The flattened consumer can advance after A is exhausted; an iterator operating inside A never probes B to discover A's boundary. HStore partition merging, `ORDER_BY_KEY` and physical-key cursors are outside this change.
## Verifying these changes
- [ ] Trivial rework / code cleanup without any test coverage. (No Need)
- [x] Already covered by existing tests, including `VertexCoreTest`, `EdgeCoreTest` and cache regressions.
- [x] Need tests and can be verified as follows:
Three focused tests first failed on the old implementation: mixed ordering returned `[1, 2, 3]` instead of `[1, 3, 2]`, and the next-query and next-page fetch counters reached 2 when only 1 batch should have been activated. All now pass.
Final verification on JDK 17 and Maven 3.9.16:
| Scope | Selected | Skipped | Failures / errors |
| --- | ---: | ---: | ---: |
| Seven focused unit-test classes | 58 | 0 | 0 / 0 |
| Memory: `VertexCoreTest`, `EdgeCoreTest` | 436 | 69 | 0 / 0 |
| RocksDB: `VertexCoreTest`, `EdgeCoreTest` | 436 | 29 | 0 / 0 |
| HStore: search/joint-index, input-order and paging regressions | 26 | 0 | 0 / 0 |
Coverage includes empty batches/holders, null and expanded mapper results, mixed ordering in both directions, real page-fetch counters and cursor metadata, limits, cross-batch materialization capacity, early close and suppressed exceptions, residual filtering on warm vertex caches, and an off-heap edge-cache hit with zero backend reads. The HStore selection also passed on the pre-change baseline using the same isolated PD/Store setup.
<details>
<summary>Reproduction commands</summary>
```bash
mvn test -pl hugegraph-server/hugegraph-test -am -P unit-test \
-Dtest=QueryResultsTest,QueryListTest,GraphTransactionTest,CachedGraphTransactionTest,IdHolderTest,QueryTest,PageStateTest \
-DfailIfNoTests=false -Drat.skip=true
# Run once with memory, then with rocksdb
mvn test -pl hugegraph-server/hugegraph-test -am -P core-test,memory \
-Dtest=VertexCoreTest,EdgeCoreTest -DfailIfNoTests=false -Drat.skip=true
# Requires an isolated, initialized PD/Store and an HStore properties file
mvn test -pl hugegraph-server/hugegraph-test -am -P core-test,hstore \
-Dbackend=hstore -Dconfig_path=/path/to/hstore-test.properties \
'-Dtest=VertexCoreTest#testQueryByJointIndexesWithSearch*+testQueryByTextContainsPropertyOrderByMatchedCount*+testQueryByRangeIndexKeeps*+testQueryByPage*+testQueryByMultiLabelInPage,EdgeCoreTest#testQueryEdgeByPage*+testQuery*EdgesOfVertexInPaging' \
-DfailIfNoTests=false -Drat.skip=true
mvn editorconfig:format
mvn clean compile -Dmaven.javadoc.skip=true -Drat.skip=true
git diff --check
```
</details>
Formatting and full reactor compilation passed. Verification used `-Drat.skip=true`; the complete UnitTestSuite, API and TinkerPop suites were not run. The tested Java files match this commit byte-for-byte.
## Does this PR potentially affect the following parts?
- [ ] Dependencies
- [ ] Modify configurations
- [x] The public API (`QueryResults` Java helper methods; Gremlin/REST result semantics are preserved)
- [x] Other affects (internal query iteration, paging and cache lifecycle)
- [ ] Nope
## Documentation Status
- [ ] `Doc - TODO`
- [ ] `Doc - Done`
- [x] `Doc - No Need`
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(server): make query result batch boundaries explicit [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(server): make query result batch boundaries explicit [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(server): make query result batch boundaries explicit [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
bitflicker64 commented on code in PR #3193:
URL: https://github.com/apache/hugegraph/pull/3193#discussion_r3972183962
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResults.java:
##########
@@ -18,234 +18,362 @@
package org.apache.hugegraph.backend.query;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
-import java.util.Set;
+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.backend.query.QueryBatch.BatchIterator;
import org.apache.hugegraph.iterator.CIter;
-import org.apache.hugegraph.iterator.FlatMapperIterator;
import org.apache.hugegraph.iterator.ListIterator;
-import org.apache.hugegraph.iterator.MapperIterator;
-import org.apache.hugegraph.iterator.WrappedIterator;
import org.apache.hugegraph.perf.PerfUtil.Watched;
import org.apache.hugegraph.type.Idfiable;
-import org.apache.hugegraph.util.E;
-import org.apache.hugegraph.util.InsertionOrderUtil;
-import org.apache.tinkerpop.gremlin.structure.util.CloseableIterator;
+/** A lazy stream of query batches. Only the final consumer flattens the stream. */
public class QueryResults<R> {
- private static final Iterator<?> EMPTY_ITERATOR = new EmptyIterator<>();
-
- private static final QueryResults<?> EMPTY = new QueryResults<>(
- emptyIterator(), Query.NONE);
-
- private final Iterator<R> results;
+ private final BatchIterator<QueryBatch<R>> batches;
private final List<Query> queries;
- private List<Query> currentQueries;
- private long queryVersion;
+ private final Object metadata;
+ private Iterator<R> results;
public QueryResults(Iterator<R> results, Query query) {
- this(results);
- this.addQuery(query);
+ this(results, new QueryResultContext(query));
}
- private QueryResults(Iterator<R> results) {
- this.results = results;
- this.queries = InsertionOrderUtil.newList();
- this.currentQueries = Collections.emptyList();
- this.queryVersion = 0L;
+ public QueryResults(Iterator<R> results, QueryResultContext context) {
+ QueryBatch<R> batch = new QueryBatch<>(results, context);
+ this.queries = new ArrayList<>(Collections.singletonList(context.queries().get(0)));
+ this.batches = this.trackBatches(new BatchIterator<QueryBatch<R>>() {
+ private boolean fetched;
+
+ @Override
+ protected QueryBatch<R> fetch() {
+ if (this.fetched) {
+ return null;
+ }
+ this.fetched = true;
+ return batch;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ batch.close();
+ }
+ });
+ this.metadata = batch.results();
}
- public void setQuery(Query query) {
- if (!this.queries.isEmpty()) {
- this.queries.clear();
- }
- this.addQuery(query);
+ private QueryResults(Iterator<QueryBatch<R>> batches, Object metadata) {
+ this.queries = new ArrayList<>();
+ this.batches = this.trackBatches(batches);
+ this.metadata = metadata;
}
- private void addQuery(Query query) {
- E.checkNotNull(query, "query");
- this.addQueries(Collections.singletonList(query));
+ private BatchIterator<QueryBatch<R>> trackBatches(Iterator<QueryBatch<R>> origin) {
+ return new BatchIterator<QueryBatch<R>>() {
+ private QueryBatch<R> active;
+
+ @Override
+ protected QueryBatch<R> fetch() throws Exception {
+ QueryBatch<R> previous = this.active;
+ this.active = null;
+ QueryBatch.closeAll(previous);
+ if (!origin.hasNext()) {
+ return null;
+ }
+ this.active = origin.next();
+ queries.clear();
+ queries.add(this.active.context().queries().get(0));
+ return this.active;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ QueryBatch<R> previous = this.active;
+ this.active = null;
+ queries.clear();
+ QueryBatch.closeAll(previous, origin);
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return QueryBatch.metadataOf(origin, meta, args);
+ }
+ };
}
- private void addQueries(List<Query> queries) {
- assert !queries.isEmpty();
- for (Query query : queries) {
- E.checkNotNull(query, "query");
- this.queries.add(query);
- }
- this.currentQueries = new ArrayList<>(queries);
- this.queryVersion++;
+ public static <R> QueryResults<R> fromBatches(Iterator<QueryBatch<R>> batches) {
+ return new QueryResults<>(batches, batches);
+ }
+
+ public Iterator<QueryBatch<R>> batches() {
+ return this.batches;
}
public Iterator<R> iterator() {
+ if (this.results == null) {
+ this.results = new CIter<R>() {
+ private boolean closed;
+
+ @Override
+ public boolean hasNext() {
+ if (this.closed) {
+ return false;
+ }
+ try {
+ while (batches.hasNext()) {
+ // Leave the batch and its prefetched element in the shared cursor.
+ if (batches.peek().results().hasNext()) {
+ return true;
+ }
+ batches.next().close();
+ }
+ this.close();
+ return false;
+ } catch (Throwable failure) {
+ QueryResults.close(this, failure);
+ throw QueryBatch.propagate(failure);
+ }
+ }
+
+ @Override
+ public R next() {
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ }
+ try {
+ return batches.peek().results().next();
+ } catch (Throwable failure) {
+ QueryResults.close(this, failure);
+ throw QueryBatch.propagate(failure);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (!this.closed) {
+ this.closed = true;
+ batches.close();
+ }
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return QueryBatch.metadataOf(metadata, meta, args);
+ }
+ };
+ }
return this.results;
}
- public R one() {
- return one(this.results);
+ public <T> QueryResults<T> mapBatches(Function<QueryBatch<R>, QueryBatch<T>> mapper) {
+ Iterator<QueryBatch<R>> origin = this.batches;
+ return new QueryResults<>(new BatchIterator<QueryBatch<T>>() {
+ private QueryBatch<?> active;
+
+ @Override
+ protected QueryBatch<T> fetch() throws Exception {
+ QueryBatch.closeAll(this.active);
+ this.active = null;
+ if (!origin.hasNext()) {
+ return null;
+ }
+ QueryBatch<R> batch = origin.next();
+ this.active = batch;
+ QueryBatch<T> mapped = mapper.apply(batch);
+ this.active = mapped;
+ return mapped;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ QueryBatch.closeAll(this.active, origin);
+ }
+ }, this.metadata);
}
- public QueryResults<R> toList() {
- QueryResults<R> fetched = new QueryResults<>(toList(this.results));
- fetched.addQueries(this.queries);
- return fetched;
+ public <T> QueryResults<T> map(Function<R, T> mapper) {
+ return this.mapBatches(batch -> batch.map(mapper));
}
- public List<Query> queries() {
- return Collections.unmodifiableList(this.queries);
+ public <T> QueryResults<T> flatMap(Function<R, Iterator<T>> mapper) {
+ return this.mapBatches(batch -> batch.flatMap(mapper));
}
- public <T extends Idfiable> Iterator<T> keepInputOrderIfNeeded(
- Iterator<T> origin) {
- if (!origin.hasNext()) {
- // None result found
- return origin;
- }
- if (!mustSortByInputIds(this.currentQueries)) {
- return origin;
- }
- return new InputOrderIterator<>(this, origin);
+ public QueryResults<R> filter(BiPredicate<QueryResultContext, R> predicate) {
+ return this.mapBatches(batch -> batch.filter(predicate));
}
- private static boolean mustSortByInputIds(List<Query> queries) {
- assert !queries.isEmpty() : queries;
- for (Query query : queries) {
- if (query instanceof IdQuery &&
- ((IdQuery) query).mustSortByInput()) {
- return true;
- }
- }
- return false;
+ public <T extends Idfiable> QueryResults<T> keepInputOrderIfNeeded() {
+ return this.mapBatches(QueryBatch::keepInputOrder);
}
- @SuppressWarnings("unused")
- private boolean bigCapacity() {
- assert !this.queries.isEmpty();
- for (Query query : this.queries) {
- if (query.bigCapacity()) {
- return true;
+ public R one() {
+ return one(this.iterator());
+ }
+
+ /**
+ * Source query of the current batch. A known single source is available before
+ * activation; composed streams start empty. Closing clears the diagnostics.
+ */
+ public List<Query> queries() {
Review Comment:
🧹 `queries()` has no main-code caller left after this change.
At base it had exactly one: `QueryList.PageResults.query()` (base `QueryList.java:347`), which `cf403002` deleted. Ordering now comes from `QueryResultContext`, and base `PageEntryIterator.java:79` (`queryResults.setQuery(pageResults.query())`) is gone. Over this head, `git grep '\.queries()'` finds only `context.queries()` calls plus assertions in `QueryResultsTest.java:161,164,166` and `QueryListTest.java:129,131,135`.
The bookkeeping still runs on the hot path: an `ArrayList` per `QueryResults` (line 51), `queries.clear()` plus `queries.add(...)` on every batch activation (lines 91-92), a clear on close (line 100), and a re-seed in `toList()` (line 250). Same shape as the dead `PageResults.query()`/`get()` pair you removed last round.
Requested change: drop the `queries` field and this accessor, and rework the assertions onto `results.batches()` where the test structure allows. Note that `java.util.Collections` (line 21) is then unused, since lines 51 and 226 are its only uses. If it should stay, name its reader in the Javadoc at lines 221-224, because no main code reads it today.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java:
##########
@@ -316,135 +318,149 @@ 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() ||
Review Comment:
🧹 This `paging()` test is on the leaf, while `fetchEdgeBatch` tests the chain root.
Separate from my earlier thread that asked for this guard: now that it exists, the two overrides disagree about what "paged" means. Line 322 tests `query.paging()` on the `QueryList` fetcher's leaf; `fetchEdgeBatch` builds the same context and tests `request.paging()` on `chain.get(chain.size() - 1)`, the chain root (lines 369-372).
The difference is observable. `QueryList.IndexQuery.iterator(int, String, long)` builds its leaf through `indexIdQuery` (`QueryList.java:326-332`) as `new IdQuery(parent().resultType(), bindQuery)`, and `Query(HugeType, Query)` sets `this.page = null` (`Query.java:95`), so that leaf reaches line 322 with `paging() == false`, `idsSize() > 0` and `conditionsSize() == 0`. The vertex cache therefore still runs inside a paged index query, and is correct only for the reason I traced on the earlier thread: that path takes its cursor from `pageIds.pageState()`, not from result metadata.
Requested change: state in a comment that the leaf test is deliberate and that the index-paging leaf is safe because its cursor comes from the `IdHolder`. Matching `fetchEdgeBatch` and testing the root instead would also work, but it bypasses the vertex cache for every paged index query, so it is the more expensive option.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/QueryList.java:
##########
@@ -25,15 +25,17 @@
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.page.IdHolder.BatchIdHolder;
import org.apache.hugegraph.backend.page.IdHolder.FixedIdHolder;
-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.IdQuery;
import org.apache.hugegraph.backend.query.Query;
+import org.apache.hugegraph.backend.query.QueryBatch.BatchIterator;
+import org.apache.hugegraph.backend.query.QueryBatch;
Review Comment:
🧹 These two import pairs kept the ordering that `9ad4a27e` reverted in the other two files.
This PR moved `ConditionQuery` below `ConditionQuery.OptimizedType` (lines 28-29) and added `QueryBatch.BatchIterator` above `QueryBatch` (lines 32-33). `9ad4a27e` restored outer-class-first for exactly these two patterns elsewhere: `Aggregate`/`Aggregate.AggregateFunc` and `ConditionQuery`/`ConditionQuery.OptimizedType` in `GraphTransaction.java`, and `QueryBatch`/`QueryBatch.BatchIterator` in `PageEntryIterator.java`. This file was missed, and nothing flags it because `style/checkstyle.xml:43` has `ImportOrder` commented out.
Requested change: swap both pairs so this file matches the ordering the same commit restored in the other two.
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(server): make query result batch boundaries explicit [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin merged PR #3193:
URL: https://github.com/apache/hugegraph/pull/3193
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(server): make query result batch boundaries explicit [hugegraph]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3193:
URL: https://github.com/apache/hugegraph/pull/3193#issuecomment-5549295196
## [Codecov](https://app.codecov.io/gh/apache/hugegraph/pull/3193?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 93.25%. Comparing base ([`3681148`](https://app.codecov.io/gh/apache/hugegraph/commit/36811483a2040f70ca6923a288a8c28cad0086c0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`6b0469b`](https://app.codecov.io/gh/apache/hugegraph/commit/6b0469bcc90094fdb68417f13360c88a9e4eb4e9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3193 +/- ##
=============================================
+ Coverage 37.77% 93.25% +55.48%
+ Complexity 6560 65 -6495
=============================================
Files 800 9 -791
Lines 68960 267 -68693
Branches 9166 22 -9144
=============================================
- Hits 26052 249 -25803
+ Misses 39841 8 -39833
+ Partials 3067 10 -3057
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/hugegraph/pull/3193?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(server): make query result batch boundaries explicit [hugegraph]
Posted by "contrueCT (via GitHub)" <gi...@apache.org>.
contrueCT commented on PR #3193:
URL: https://github.com/apache/hugegraph/pull/3193#issuecomment-5708677492
@imbajin This PR may be ready to merge ?
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] fix(server): make query result batch boundaries explicit [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
bitflicker64 commented on code in PR #3193:
URL: https://github.com/apache/hugegraph/pull/3193#discussion_r3949905190
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/QueryList.java:
##########
@@ -344,10 +354,11 @@ public boolean hasNextPage() {
}
public Query query() {
- List<Query> queries = this.results.queries();
- E.checkState(queries.size() == 1,
- "Expect query size 1, but got: %s", queries);
- return queries.get(0);
+ return this.query;
Review Comment:
🧹 `query()` and `get()` have no callers left after this change.
`query()` existed at base, deriving from `this.results.queries()` behind an `E.checkState(queries.size() == 1)`. This commit reworks it onto a new `query` field (line 339) and a third constructor parameter (line 341), with `emptyIterator()` passing `null` for it (line 374).
But the rewritten `PageEntryIterator` is the only `PageResults` consumer in main code (`PageEntryIterator.java:62`), and it reads `results()`, `hasNextPage()`, `page()` and `total()` only. Base `PageEntryIterator:79` called `pageResults.query()`, and `:60/81/97/106` called `pageResults.get()`; both are gone. `git grep` over this head finds no call to either accessor, and `QueryListTest:49-50` uses `.results()`.
Suggest dropping `get()`, `query()`, the field and the constructor parameter, returning `PageResults` to `(results, pageState)`, which also removes the `null` argument in `emptyIterator()`.
##########
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()) {
Review Comment:
🧹 Was dropping the `queryNeedsPostFilter` exclusion from this guard intentional?
Base `:399` had it in the same condition, plus a re-check after the backend call at `:424` ("re-check before caching"). Base `queryNeedsPostFilter` returned true for `hasSearchCondition()`, and for `optimized() != NONE` unless the edge query also carried a LABEL condition. So at base `g.E().has(<indexed prop>, v)` and primary-key/sort-key optimized edge queries were never cached, while `g.E().hasLabel(L).has(p, v)` was.
Here only `empty`/`paging`/`bigCapacity` remain. Since `fetchEdgeBatch` is now the `QueryList` fetcher rather than a `queryEdgesFromBackend` override, the index-derived `IdQuery` leaf reaches it with `request` (line 371, the chain root) equal to the user `ConditionQuery`: non-empty, non-paging, default capacity. Line 404 then caches it.
Results stay correct, because `processBatches` re-applies `filterUnmatchedRecord` on hits as well as misses. The cost is key count: `edge.cache_capacity` defaults to 1,000,000 entries (`CoreOptions.java:566-572`) on a per-graph cache, so high-cardinality indexed lookups now mint one guaranteed-miss key per predicate value and evict the adjacency entries the cache exists to serve. `fetchVertexBatch` at `:322` dropped the same guard, which looks harmless there since `verticesCache` is id-keyed.
If the widening is deliberate, a line in the PR body would save the next reader this trace. If not, `context.conditionFilterRequired()` (the context is already built on line 369) is the ready-made predicate, and the base's LABEL carve-out is worth keeping.
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org