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 8 commits into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 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
00e365d
refactor(server): remove unused query result diagnostics
contrueCT Sep 13, 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
210 changes: 114 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,151 @@ 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) {
// Test the leaf: direct paging needs backend metadata, while index paging
// takes its cursor from the IdHolder and can cache the ID lookup.
if (!this.enableCacheVertex() || query.paging() ||
Comment thread
contrueCT marked this conversation as resolved.
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);
Comment thread
contrueCT marked this conversation as resolved.
// 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));
Comment thread
contrueCT marked this conversation as resolved.
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