Skip to content

fix(server): make query result batch boundaries explicit #3193

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
contrueCT wants to merge 7 commits into
base: master
Choose a base branch
from
+2,410 −784
Open

fix(server): make query result batch boundaries explicit #3193

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6b0469b
fix(server): preserve query result batch boundaries
contrueCT Sep 5, 2026
e22126d
fix(server): preserve remaining query batches and bound diagnostics
contrueCT Sep 5, 2026
c694bb0
fix(server): preserve empty backend page boundaries
contrueCT Sep 5, 2026
17788a4
fix(server): preserve paging metadata and isolate empty pages
contrueCT Sep 5, 2026
05186a0
fix(server): cover batch lifecycle and register query helpers
contrueCT Sep 5, 2026
cf40300
refactor(server): remove unused page result accessors
contrueCT Sep 8, 2026
9ad4a27
refactor(server): reduce query batch allocation overhead
contrueCT Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions ...ph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.StandardHugeGraph;
import org.apache.hugegraph.backend.cache.CacheManager;
import org.apache.hugegraph.backend.cache.CachedGraphTransaction;
import org.apache.hugegraph.backend.tx.AbstractTransaction;
import org.apache.hugegraph.backend.tx.GraphTransaction;
import org.apache.hugegraph.backend.tx.IndexableTransaction;
Expand Down Expand Up @@ -272,13 +273,19 @@ private static void registerPrivateActions() {
"verifyEdgesConditionQuery", "indexQuery",
"joinTxRecords", "propertyUpdated", "parseEntry",
"traverseByLabel", "reset", "queryVerticesByIds",
"filterInvalidRecords", "filterUnmatchedRecords",
"filterInvalidRecord", "filterUnmatchedRecord",
"invalidRecord", "warnLeftRecord",
"skipOffsetOrStopLimit",
"filterExpiredResultFromFromBackend", "queryEdgesByIds",
"filterExpiredBatches", "queryEdgesByIds",
"matchEdgeSortKeys", "rightResultFromIndexQuery",
"queryNeedsPostFilter",
"conditionQueryNeedsPostFilter");
"queryVertexBatchesFromBackend", "backendBatches",
"fetchVertexBatch", "processBatches",
"queryEdgeBatchesFromBackend",
"queryEdgeBatchesFromBackendInternal",
"queryEdgesFromMemory", "fetchEdgeBatch");
Reflection.registerMethodsToFilter(CachedGraphTransaction.class,
"fetchVertexBatch", "fetchEdgeBatch",
"queryEdgesFromMemory", "cacheEdgeBatch");
Reflection.registerFieldsToFilter(IndexableTransaction.class, "$assertionsDisabled");
Reflection.registerMethodsToFilter(IndexableTransaction.class, "indexTransaction",
"commit2Backend", "reset");
Expand Down
208 changes: 112 additions & 96 deletions ...gegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.query.IdQuery;
import org.apache.hugegraph.backend.query.Query;
import org.apache.hugegraph.backend.query.QueryBatch;
import org.apache.hugegraph.backend.query.QueryResultContext;
import org.apache.hugegraph.backend.query.QueryResults;
import org.apache.hugegraph.backend.store.BackendMutation;
import org.apache.hugegraph.backend.store.BackendStore;
Expand Down Expand Up @@ -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() ||

@bitflicker64 bitflicker64 Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 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.

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()) {
Comment thread
contrueCT marked this conversation as resolved.
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);

@bitflicker64 bitflicker64 Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 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.

// An empty label denotes the outer request itself without duplicating its text.
String batchLabel = batchKey.equals(cacheKey) ? "" : batchKey.asString();
CachedEdgeQuery group = new CachedEdgeQuery(this.edgesCache.get(cacheKey));

@bitflicker64 bitflicker64 Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 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.

Collection<HugeEdge> cached = group.get(batchLabel);
if (cached != null) {
for (HugeEdge edge : cached) {
if (edge.expired()) {
this.edgesCache.invalidate(cacheKey);
value = null;
cached = null;
break;
}
}
}
if (cached != null) {
return this.filterExpiredBatches(new QueryResults<>(cached.iterator(), context));
}
QueryResults<HugeEdge> fetched = super.fetchEdgeBatch(query);
if (!fetched.batches().hasNext()) {
this.cacheEdgeBatch(cacheKey, batchLabel, 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, batchLabel, candidates);
}
return new QueryBatch<>(
new ExtendableIterator<>(candidates.iterator(), source), batch.context());
});
}

if (value != null) {
// Not cached or the cache expired
return edges.iterator();
private void cacheEdgeBatch(Id cacheKey, String batchLabel, List<HugeEdge> candidates) {
synchronized (this.edgesCache) {
Comment thread
contrueCT marked this conversation as resolved.
CachedEdgeQuery existing = new CachedEdgeQuery(this.edgesCache.get(cacheKey)).copy();
if (existing.put(batchLabel, candidates)) {
this.edgesCache.update(cacheKey, existing.values);
}
}
}

/** Nested lists retain the existing off-heap cache's serialization support. */
private static final class CachedEdgeQuery {

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;
// Alternating batch query strings and raw candidate lists; never store filter closures.
private final List<Object> values;

@SuppressWarnings("unchecked")
private CachedEdgeQuery(Object cached) {
this.values = cached == null ? new ArrayList<>() :
(List<Object>) cached;
}

/*
* Iterator can't be cached, caching list instead
* there may be super node and too many edges in a query,
* try fetch a few of the head results and determine whether to cache.
*/
final int tryMax = 1 + MAX_CACHE_EDGES_PER_QUERY;
edges = new ArrayList<>(tryMax);
for (int i = 0; rs.hasNext() && i < tryMax; i++) {
edges.add(rs.next());
private CachedEdgeQuery copy() {
return new CachedEdgeQuery(new ArrayList<>(this.values));
}

if (edges.isEmpty()) {
this.edgesCache.update(cacheKey, Collections.emptyList());
} else if (edges.size() <= MAX_CACHE_EDGES_PER_QUERY) {
this.edgesCache.update(cacheKey, edges);
@SuppressWarnings("unchecked")
public Collection<HugeEdge> get(String batch) {
for (int i = 0; i < this.values.size(); i += 2) {
if (this.values.get(i).equals(batch)) {
return (List<HugeEdge>) this.values.get(i + 1);
}
}
return null;
}

return new ExtendableIterator<>(edges.iterator(), rs);
public boolean put(String batch, List<HugeEdge> candidates) {
if (this.get(batch) != null) {
return false;
}
int size = candidates.size();
for (int i = 1; i < this.values.size(); i += 2) {
size += ((List<?>) this.values.get(i)).size();
}
if (size > MAX_CACHE_EDGES_PER_QUERY ||
this.values.size() / 2 >= MAX_CACHE_EDGES_PER_QUERY) {
return false;
}
this.values.add(batch);
this.values.add(new ArrayList<>(candidates));
return true;
}
}

@Override
Expand Down
Loading
Loading