You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by ji...@apache.org on 2026/09/17 11:01:47 UTC
(hugegraph) branch master updated: fix(server): make query result batch boundaries explicit (#3193)
This is an automated email from the ASF dual-hosted git repository.
imbajin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hugegraph.git
The following commit(s) were added to refs/heads/master by this push:
new ee22dd7ef fix(server): make query result batch boundaries explicit (#3193)
ee22dd7ef is described below
commit ee22dd7ef0ceaa5f356c90fea0414ff13c05a932
Author: contrueCT <co...@outlook.com>
AuthorDate: Thu Sep 17 19:01:42 2026 +0800
fix(server): make query result batch boundaries explicit (#3193)
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.
---
.../hugegraph/auth/HugeFactoryAuthProxy.java | 15 +-
.../backend/cache/CachedGraphTransaction.java | 210 +++++----
.../hugegraph/backend/page/PageEntryIterator.java | 115 ++---
.../apache/hugegraph/backend/page/QueryList.java | 85 ++--
.../hugegraph/backend/query/ConditionQuery.java | 29 +-
.../apache/hugegraph/backend/query/QueryBatch.java | 279 ++++++++++++
.../backend/query/QueryResultContext.java | 120 +++++
.../hugegraph/backend/query/QueryResults.java | 485 ++++++++++-----------
.../hugegraph/backend/tx/GraphTransaction.java | 291 +++++--------
.../hugegraph/backend/page/QueryListTest.java | 409 +++++++++++++++++
.../hugegraph/backend/tx/GraphTransactionTest.java | 244 ++++++++---
.../org/apache/hugegraph/core/VertexCoreTest.java | 38 ++
.../org/apache/hugegraph/unit/UnitTestSuite.java | 8 +-
.../unit/cache/CachedGraphTransactionTest.java | 417 +++++++++++++++++-
.../hugegraph/unit/core/QueryResultsTest.java | 440 ++++++++++++++++---
15 files changed, 2401 insertions(+), 784 deletions(-)
diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java
index c9ce7cca4..5ff2925e6 100644
--- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java
+++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java
@@ -32,6 +32,7 @@ import org.apache.hugegraph.HugeFactory;
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;
@@ -272,13 +273,19 @@ public final class HugeFactoryAuthProxy {
"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");
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java
index 1543d7315..a6085d32c 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java
@@ -32,6 +32,8 @@ import org.apache.hugegraph.backend.cache.CachedBackendStore.QueryId;
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;
@@ -316,135 +318,151 @@ public final class CachedGraphTransaction extends GraphTransaction {
@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() ||
+ 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);
+ // 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));
+ 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) {
+ 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
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/PageEntryIterator.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/PageEntryIterator.java
index bbc93c79b..17996bae9 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/PageEntryIterator.java
+++ b/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;
+import org.apache.hugegraph.backend.query.QueryBatch.BatchIterator;
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 '%s' with an offset '%s' exceeds the size of IdHolderList",
+ queries.parent().pageWithoutCheck(), 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()) {
+ // Preserve the raw-empty-page boundary. An existing batch
+ // emptied by parsing or TTL filtering still follows its cursor.
this.pageInfo.increase();
+ continue;
+ }
+ if (page.hasNextPage()) {
+ this.pageInfo.page(page.page());
} else {
- this.pageInfo.page(this.pageResults.page());
+ this.pageInfo.increase();
+ }
+ if (this.remaining != Query.NO_LIMIT) {
+ this.remaining -= page.total();
}
- this.remaining -= this.pageResults.total();
- return true;
- } else {
- this.pageInfo.increase();
- return this.fetch();
- }
- }
-
- private void closePageResults() {
- if (this.pageResults != QueryList.PageResults.EMPTY) {
- CloseableIterator.closeIterator(this.pageResults.get());
}
}
@Override
- public R next() {
- if (!this.hasNext()) {
- throw new NoSuchElementException();
- }
- return this.pageResults.get().next();
+ protected void closeResources() throws Exception {
+ QueryBatch.closeAll(this.pageBatches, this.queries);
}
@Override
public Object metadata(String meta, Object... args) {
if (PageInfo.PAGE.equals(meta)) {
- if (this.pageInfo.offset() >= this.queries.total()) {
- return null;
- }
- return this.pageInfo;
+ return this.pageInfo.offset() >= this.queries.total() ? null : this.pageInfo;
}
throw new NotSupportException("Invalid meta '%s'", meta);
}
-
- @Override
- public void close() throws Exception {
- this.closePageResults();
- }
-
- public QueryResults<R> results() {
- return this.queryResults;
- }
}
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/QueryList.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/QueryList.java
index 327c4aab4..d3d8e6960 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/QueryList.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/QueryList.java
@@ -29,11 +29,13 @@ import org.apache.hugegraph.backend.query.ConditionQuery;
import org.apache.hugegraph.backend.query.ConditionQuery.OptimizedType;
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.QueryBatch.BatchIterator;
import org.apache.hugegraph.backend.query.QueryResults;
import org.apache.hugegraph.util.Bytes;
import org.apache.hugegraph.util.E;
-public final class QueryList<R> {
+public final class QueryList<R> implements AutoCloseable {
private final Query parent;
// The size of each page fetched by the inner page
@@ -87,16 +89,32 @@ public final class QueryList<R> {
if (this.parent.paging()) {
@SuppressWarnings("resource") // closed by QueryResults
PageEntryIterator<R> iter = new PageEntryIterator<>(this, pageSize);
- /*
- * NOTE: PageEntryIterator query will change every fetch time.
- * QueryResults tracks this change and restores input-id order
- * within each page without fetching later pages eagerly.
- */
- return iter.results();
+ return QueryResults.fromBatches(iter);
}
- // Fetch all results once
- return QueryResults.flatMap(this.queries.iterator(), FlattenQuery::iterator);
+ Iterator<FlattenQuery<R>> source = this.queries.iterator();
+ return QueryResults.flatMap(new BatchIterator<FlattenQuery<R>>() {
+ @Override
+ protected FlattenQuery<R> fetch() {
+ return source.hasNext() ? source.next() : null;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ QueryList.this.close();
+ }
+ }, FlattenQuery::iterator);
+ }
+
+ @Override
+ public void close() throws Exception {
+ List<Object> resources = new ArrayList<>();
+ for (FlattenQuery<R> query : this.queries) {
+ if (query instanceof QueryList.IndexQuery) {
+ resources.addAll(((IndexQuery) query).holders);
+ }
+ }
+ QueryBatch.closeAll(resources.toArray());
}
PageResults<R> fetchNext(PageInfo pageInfo, long pageSize) {
@@ -220,7 +238,6 @@ public final class QueryList<R> {
private QueryResults<R> each(IdHolder holder) {
assert !holder.paging();
Query bindQuery = holder.query();
- this.updateResultsFilter(bindQuery);
this.updateOffsetIfNeeded(bindQuery);
// Iterate by all
@@ -237,7 +254,7 @@ public final class QueryList<R> {
* in order by ids weight. In addition all the ids (IdQuery)
* can be collected by upper layer.
*/
- return this.queryByIndexIds(ids, holder.keepOrder());
+ return this.queryByIndexIds(bindQuery, ids, holder.keepOrder());
}
// Iterate by batch
@@ -259,7 +276,7 @@ public final class QueryList<R> {
return null;
}
- return this.queryByIndexIds(ids, holder.keepOrder());
+ return this.queryByIndexIds(bindQuery, ids, holder.keepOrder());
});
}
@@ -270,15 +287,13 @@ public final class QueryList<R> {
"Invalid page index %s", index);
IdHolder holder = this.holders.get(index);
Query bindQuery = holder.query();
- this.updateResultsFilter(bindQuery);
PageIds pageIds = holder.fetchNext(page, pageSize);
if (pageIds.empty()) {
return PageResults.emptyIterator();
}
- QueryResults<R> results = this.queryByIndexIds(pageIds.ids(),
- holder.keepOrder());
-
+ IdQuery query = this.indexIdQuery(bindQuery, pageIds.ids(), holder.keepOrder());
+ QueryResults<R> results = fetcher().apply(query);
return new PageResults<>(results, pageIds.pageState());
}
@@ -303,29 +318,22 @@ public final class QueryList<R> {
query.copyOffset(parent);
}
- private void updateResultsFilter(Query query) {
- while (query != null) {
- if (query instanceof ConditionQuery) {
- ((ConditionQuery) query).updateResultsFilter();
- return;
- }
- query = query.originQuery();
- }
+ private QueryResults<R> queryByIndexIds(Query bindQuery, Set<Id> ids,
+ boolean inOrder) {
+ return fetcher().apply(this.indexIdQuery(bindQuery, ids, inOrder));
}
- private QueryResults<R> queryByIndexIds(Set<Id> ids, boolean inOrder) {
- IdQuery query = new IdQuery(parent(), ids);
+ private IdQuery indexIdQuery(Query bindQuery, Set<Id> ids, boolean inOrder) {
+ // The holder can query an index table; fetch graph elements by ID.
+ IdQuery query = new IdQuery(parent().resultType(), bindQuery);
+ query.query(ids);
query.mustSortByInput(inOrder);
- return fetcher().apply(query);
+ return query;
}
}
public static class PageResults<R> {
- public static final PageResults<?> EMPTY = new PageResults<>(
- QueryResults.empty(),
- PageState.EMPTY);
-
private final QueryResults<R> results;
private final PageState pageState;
@@ -334,20 +342,13 @@ public final class QueryList<R> {
this.pageState = pageState;
}
- public Iterator<R> get() {
- return this.results.iterator();
- }
-
public boolean hasNextPage() {
return !Bytes.equals(this.pageState.position(),
PageState.EMPTY_BYTES);
}
- 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);
+ public QueryResults<R> results() {
+ return this.results;
}
public String page() {
@@ -358,9 +359,9 @@ public final class QueryList<R> {
return this.pageState.total();
}
- @SuppressWarnings("unchecked")
public static <R> PageResults<R> emptyIterator() {
- return (PageResults<R>) EMPTY;
+ // Batch cursors are single-use, including those for empty pages.
+ return new PageResults<>(QueryResults.empty(), PageState.EMPTY);
}
}
}
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/ConditionQuery.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/ConditionQuery.java
index 097e98df1..6da8da23d 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/ConditionQuery.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/ConditionQuery.java
@@ -611,6 +611,10 @@ public class ConditionQuery extends IdQuery {
@Override
public boolean test(HugeElement element) {
+ return this.test(element, this.resultsFilter);
+ }
+
+ public boolean test(HugeElement element, ResultsFilter filter) {
if (!this.ids().isEmpty() && !super.test(element)) {
return false;
}
@@ -622,8 +626,8 @@ public class ConditionQuery extends IdQuery {
* We can't use sub-query results-filter here for fresh element which is
* not committed to backend store, because it's not from a sub-query.
*/
- if (this.resultsFilter != null && !element.fresh()) {
- return this.resultsFilter.test(element);
+ if (filter != null && !element.fresh()) {
+ return filter.test(element);
}
/*
@@ -705,25 +709,8 @@ public class ConditionQuery extends IdQuery {
this.resultsFilter = filter;
}
- public void updateResultsFilter() {
- Query originQuery = this.originQuery();
- if (originQuery instanceof ConditionQuery) {
- ConditionQuery originCQ = ((ConditionQuery) originQuery);
- if (this.resultsFilter != null) {
- originCQ.updateResultsFilter(this.resultsFilter);
- } else {
- originCQ.updateResultsFilter();
- }
- }
- }
-
- protected void updateResultsFilter(ResultsFilter filter) {
- this.resultsFilter = filter;
- Query originQuery = this.originQuery();
- if (originQuery instanceof ConditionQuery) {
- ConditionQuery originCQ = ((ConditionQuery) originQuery);
- originCQ.updateResultsFilter(filter);
- }
+ ResultsFilter resultsFilter() {
+ return this.resultsFilter;
}
public ConditionQuery originConditionQuery() {
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryBatch.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryBatch.java
new file mode 100644
index 000000000..f210a503a
--- /dev/null
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryBatch.java
@@ -0,0 +1,279 @@
+/*
+ * 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.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) {
+ Iterator<R> origin = this.results;
+ return new QueryBatch<>(new BatchIterator<T>() {
+ @Override
+ protected T fetch() {
+ while (origin.hasNext()) {
+ T mapped = mapper.apply(origin.next());
+ if (mapped != null) {
+ return mapped;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ closeAll(origin);
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return metadataOf(origin, meta, args);
+ }
+ }, this.context);
+ }
+
+ public <T> QueryBatch<T> flatMap(Function<R, Iterator<T>> mapper) {
+ Iterator<R> origin = this.results;
+ return new QueryBatch<>(new BatchIterator<T>() {
+ private Iterator<T> child;
+
+ @Override
+ protected T fetch() throws Exception {
+ while (true) {
+ if (this.child != null && this.child.hasNext()) {
+ T value = this.child.next();
+ if (value != null) {
+ return value;
+ }
+ continue;
+ }
+ Iterator<T> previous = this.child;
+ this.child = null;
+ closeAll(previous);
+ if (!origin.hasNext()) {
+ return null;
+ }
+ this.child = mapper.apply(origin.next());
+ }
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ Iterator<T> previous = this.child;
+ this.child = null;
+ closeAll(previous, origin);
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return metadataOf(origin, meta, args);
+ }
+ }, this.context);
+ }
+
+ public QueryBatch<R> filter(BiPredicate<QueryResultContext, R> predicate) {
+ return this.map(value -> predicate.test(this.context, value) ? value : null);
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T extends Idfiable> QueryBatch<T> keepInputOrder() {
+ if (!this.context.mustSortByInputIds()) {
+ return (QueryBatch<T>) this;
+ }
+ List<T> values = new ArrayList<>();
+ QueryResults.fillList((Iterator<T>) this.results, values);
+ List<Id> ids = this.context.inputIds();
+ if (ids.size() <= 1) {
+ return new QueryBatch<>(values.iterator(), this.context);
+ }
+ Map<Id, T> byId = InsertionOrderUtil.newMap();
+ for (T value : values) {
+ byId.put(value.id(), value);
+ Query.checkForceCapacity(byId.size());
+ }
+ if (byId.size() > ids.size()) {
+ // A partial ID description cannot order every returned element.
+ return new QueryBatch<>(values.iterator(), this.context);
+ }
+ List<T> ordered = new ArrayList<>(values.size());
+ for (Id id : ids) {
+ T value = byId.remove(id);
+ if (value != null) {
+ ordered.add(value);
+ }
+ }
+ ordered.addAll(byId.values());
+ return new QueryBatch<>(ordered.iterator(), this.context);
+ }
+
+ @Override
+ public void close() throws Exception {
+ this.results.close();
+ }
+
+ /** Internal iterators use null as exhaustion, like the existing mappers. */
+ public abstract static class BatchIterator<T> implements CIter<T> {
+
+ private T current;
+ private boolean closed;
+
+ protected abstract T fetch() throws Exception;
+
+ protected abstract void closeResources() throws Exception;
+
+ final T peek() {
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ }
+ return this.current;
+ }
+
+ @Override
+ public final boolean hasNext() {
+ if (this.closed) {
+ return false;
+ }
+ if (this.current != null) {
+ return true;
+ }
+ try {
+ this.current = this.fetch();
+ if (this.current == null) {
+ this.close();
+ return false;
+ }
+ return true;
+ } catch (Throwable failure) {
+ try {
+ this.close();
+ } catch (Throwable closing) {
+ if (closing != failure) {
+ failure.addSuppressed(closing);
+ }
+ }
+ throw propagate(failure);
+ }
+ }
+
+ @Override
+ public final T next() {
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ }
+ T value = this.current;
+ this.current = null;
+ return value;
+ }
+
+ @Override
+ public final void close() throws Exception {
+ if (!this.closed) {
+ this.closed = true;
+ this.current = null;
+ this.closeResources();
+ }
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return null;
+ }
+ }
+
+ public static Object metadataOf(Object iterator, String meta, Object... args) {
+ return iterator instanceof Metadatable ?
+ ((Metadatable) iterator).metadata(meta, args) : null;
+ }
+
+ public static void closeAll(Object... resources) throws Exception {
+ Throwable failure = null;
+ for (Object resource : resources) {
+ if (!(resource instanceof AutoCloseable)) {
+ continue;
+ }
+ try {
+ ((AutoCloseable) resource).close();
+ } catch (Throwable closing) {
+ if (failure == null) {
+ failure = closing;
+ } else if (failure != closing) {
+ failure.addSuppressed(closing);
+ }
+ }
+ }
+ if (failure instanceof Exception) {
+ throw (Exception) failure;
+ }
+ if (failure != null) {
+ throw (Error) failure;
+ }
+ }
+
+ public static RuntimeException propagate(Throwable failure) {
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ return failure instanceof RuntimeException ? (RuntimeException) failure :
+ new HugeException("Failed to iterate query results", failure);
+ }
+}
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResultContext.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResultContext.java
new file mode 100644
index 000000000..27e6ec67a
--- /dev/null
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResultContext.java
@@ -0,0 +1,120 @@
+/*
+ * 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;
+ // The nearest filter and optimization describe how this batch was fetched;
+ // the outermost graph condition retains the complete request to match.
+ 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()) {
+ match = condition;
+ }
+ }
+ }
+ this.queries = Collections.unmodifiableList(chain);
+ this.inputIds = Collections.unmodifiableList(new ArrayList<>(query.ids()));
+ this.mustSortByInputIds = !inputOrderSatisfied && query instanceof IdQuery &&
+ ((IdQuery) query).mustSortByInput();
+ this.matchQuery = match;
+ this.resultsFilter = filter;
+ this.optimizedType = optimized;
+ this.showExpired = visibility.showExpired();
+ this.showHidden = visibility.showHidden();
+ this.showDeleting = visibility.showDeleting();
+ }
+
+ public List<Query> queries() {
+ return this.queries;
+ }
+
+ public List<Id> inputIds() {
+ return this.inputIds;
+ }
+
+ public boolean mustSortByInputIds() {
+ return this.mustSortByInputIds;
+ }
+
+ public ConditionQuery matchQuery() {
+ return this.matchQuery;
+ }
+
+ public ResultsFilter resultsFilter() {
+ return this.resultsFilter;
+ }
+
+ public OptimizedType optimizedType() {
+ return this.optimizedType;
+ }
+
+ public boolean conditionFilterRequired() {
+ return this.optimizedType != OptimizedType.NONE;
+ }
+
+ public boolean showExpired() {
+ return this.showExpired;
+ }
+
+ public boolean showHidden() {
+ return this.showHidden;
+ }
+
+ public boolean showDeleting() {
+ return this.showDeleting;
+ }
+}
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResults.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResults.java
index 01c2024e4..35675bcf3 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResults.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResults.java
@@ -18,234 +18,343 @@
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 final BatchIterator<QueryBatch<R>> batches;
+ private final Object metadata;
+ private Iterator<R> results;
- private static final QueryResults<?> EMPTY = new QueryResults<>(
- emptyIterator(), Query.NONE);
+ public QueryResults(Iterator<R> results, Query query) {
+ this(results, new QueryResultContext(query));
+ }
- private final Iterator<R> results;
- private final List<Query> queries;
- private List<Query> currentQueries;
- private long queryVersion;
+ public QueryResults(Iterator<R> results, QueryResultContext context) {
+ QueryBatch<R> batch = new QueryBatch<>(results, context);
+ this.batches = this.trackBatches(new BatchIterator<QueryBatch<R>>() {
+ private boolean fetched;
- public QueryResults(Iterator<R> results, Query query) {
- this(results);
- this.addQuery(query);
+ @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();
}
- private QueryResults(Iterator<R> results) {
- this.results = results;
- this.queries = InsertionOrderUtil.newList();
- this.currentQueries = Collections.emptyList();
- this.queryVersion = 0L;
+ private QueryResults(Iterator<QueryBatch<R>> batches, Object metadata) {
+ this.batches = this.trackBatches(batches);
+ this.metadata = metadata;
}
- public void setQuery(Query query) {
- if (!this.queries.isEmpty()) {
- this.queries.clear();
- }
- this.addQuery(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();
+ return this.active;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ QueryBatch<R> previous = this.active;
+ this.active = null;
+ QueryBatch.closeAll(previous, origin);
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return QueryBatch.metadataOf(origin, meta, args);
+ }
+ };
}
- private void addQuery(Query query) {
- E.checkNotNull(query, "query");
- this.addQueries(Collections.singletonList(query));
+ public static <R> QueryResults<R> fromBatches(Iterator<QueryBatch<R>> batches) {
+ return new QueryResults<>(batches, batches);
}
- 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 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);
+ }
+
+ public R one() {
+ return one(this.iterator());
}
- @SuppressWarnings("unused")
- private boolean bigCapacity() {
- assert !this.queries.isEmpty();
- for (Query query : this.queries) {
- if (query.bigCapacity()) {
- return true;
+ public QueryResults<R> toList() {
+ List<QueryBatch<R>> fetched = new ArrayList<>();
+ long count = 0L;
+ Throwable failure = null;
+ try {
+ while (this.batches.hasNext()) {
+ QueryBatch<R> batch = this.batches.next();
+ ListIterator<R> values = toList(batch.results());
+ count += values.list().size();
+ Query.checkForceCapacity(count);
+ fetched.add(new QueryBatch<>(values, batch.context()));
+ Query.checkForceCapacity(fetched.size());
}
+ } catch (Throwable e) {
+ failure = e;
+ throw QueryBatch.propagate(e);
+ } finally {
+ close(this.batches, failure);
}
- return false;
+ return new QueryResults<>(fetched.iterator(), this.metadata);
}
- private static Collection<Id> queryIds(List<Query> queries) {
- assert !queries.isEmpty();
- if (queries.size() == 1) {
- return queries.get(0).ids();
- }
+ public static <T, R> QueryResults<R> flatMap(
+ Iterator<T> inputs, Function<T, QueryResults<R>> mapper) {
+ return fromBatches(new BatchIterator<QueryBatch<R>>() {
+ private QueryResults<R> child;
+
+ @Override
+ protected QueryBatch<R> fetch() throws Exception {
+ while (true) {
+ if (this.child != null && this.child.batches.hasNext()) {
+ return this.child.batches.next();
+ }
+ QueryResults<R> previous = this.child;
+ this.child = null;
+ QueryBatch.closeAll(previous == null ? null : previous.batches);
+ if (!inputs.hasNext()) {
+ return null;
+ }
+ this.child = mapper.apply(inputs.next());
+ }
+ }
- Set<Id> ids = InsertionOrderUtil.newSet();
- for (Query query : queries) {
- ids.addAll(query.ids());
- }
- return ids;
+ @Override
+ protected void closeResources() throws Exception {
+ QueryBatch.closeAll(this.child == null ? null : this.child.batches, inputs);
+ }
+ });
}
@Watched
public static <T> ListIterator<T> toList(Iterator<T> iterator) {
+ Throwable failure = null;
try {
return new ListIterator<>(Query.DEFAULT_CAPACITY, iterator);
+ } catch (Throwable e) {
+ failure = e;
+ throw QueryBatch.propagate(e);
} finally {
- CloseableIterator.closeIterator(iterator);
+ close(iterator, failure);
}
}
@Watched
public static <T> void fillList(Iterator<T> iterator, List<T> list) {
+ Throwable failure = null;
try {
while (iterator.hasNext()) {
- T result = iterator.next();
- list.add(result);
+ list.add(iterator.next());
Query.checkForceCapacity(list.size());
}
+ } catch (Throwable e) {
+ failure = e;
+ throw QueryBatch.propagate(e);
} finally {
- CloseableIterator.closeIterator(iterator);
+ close(iterator, failure);
}
}
@Watched
- public static <T extends Idfiable> void fillMap(Iterator<T> iterator,
- Map<Id, T> map) {
+ public static <T extends Idfiable> void fillMap(Iterator<T> iterator, Map<Id, T> map) {
+ Throwable failure = null;
try {
while (iterator.hasNext()) {
- T result = iterator.next();
- assert result.id() != null;
- map.put(result.id(), result);
+ T value = iterator.next();
+ map.put(value.id(), value);
Query.checkForceCapacity(map.size());
}
+ } catch (Throwable e) {
+ failure = e;
+ throw QueryBatch.propagate(e);
} finally {
- CloseableIterator.closeIterator(iterator);
+ close(iterator, failure);
}
}
- public static <T, R> QueryResults<R> flatMap(
- Iterator<T> iterator, Function<T, QueryResults<R>> func) {
- @SuppressWarnings("unchecked")
- QueryResults<R>[] qr = new QueryResults[1];
- qr[0] = new QueryResults<>(new FlatMapperIterator<>(iterator, i -> {
- QueryResults<R> results = func.apply(i);
- if (results == null || !results.iterator().hasNext()) {
- return null;
- }
- return new QueryTrackingIterator<>(qr[0], results);
- }));
- return qr[0];
- }
-
- private long queryVersion() {
- return this.queryVersion;
- }
-
- private List<Query> currentQueries() {
- return new ArrayList<>(this.currentQueries);
- }
-
@Watched
public static <T> T one(Iterator<T> iterator) {
+ Throwable failure = null;
try {
if (iterator.hasNext()) {
- T result = iterator.next();
+ T value = iterator.next();
if (iterator.hasNext()) {
- throw new HugeException("Expect just one result, " +
- "but got at least two: [%s, %s]",
- result, iterator.next());
+ throw new HugeException("Expect just one result, but got at least two: [%s, %s]",
+ value, iterator.next());
}
- return result;
+ return value;
}
+ return null;
+ } catch (Throwable e) {
+ failure = e;
+ throw QueryBatch.propagate(e);
} finally {
- CloseableIterator.closeIterator(iterator);
+ close(iterator, failure);
}
- return null;
}
- public static <T> Iterator<T> iterator(T elem) {
- return new OneIterator<>(elem);
+ private static void close(Object iterator, Throwable failure) {
+ try {
+ QueryBatch.closeAll(iterator);
+ } catch (Throwable closing) {
+ if (failure == null) {
+ throw QueryBatch.propagate(closing);
+ }
+ if (closing != failure) {
+ failure.addSuppressed(closing);
+ }
+ }
+ }
+
+ public static <T> Iterator<T> iterator(T value) {
+ return new OneIterator<>(value);
}
- @SuppressWarnings("unchecked")
public static <T> QueryResults<T> empty() {
- return (QueryResults<T>) EMPTY;
+ return new QueryResults<T>(QueryResults.<T>emptyIterator(), Query.NONE);
}
- @SuppressWarnings("unchecked")
public static <T> Iterator<T> emptyIterator() {
- return (Iterator<T>) EMPTY_ITERATOR;
+ return new EmptyIterator<>();
}
public interface Fetcher<R> extends Function<Query, QueryResults<R>> {
-
}
-
private static class EmptyIterator<T> implements CIter<T> {
@Override
@@ -304,118 +413,4 @@ public class QueryResults<R> {
}
}
- private static class QueryTrackingIterator<R>
- extends WrappedIterator<R> {
-
- private final QueryResults<R> parent;
- private final QueryResults<R> child;
- private long childQueryVersion;
-
- public QueryTrackingIterator(QueryResults<R> parent,
- QueryResults<R> child) {
- this.parent = parent;
- this.child = child;
- this.childQueryVersion = -1L;
- }
-
- @Override
- protected Iterator<R> originIterator() {
- return this.child.iterator();
- }
-
- @Override
- protected boolean fetch() {
- Iterator<R> origin = this.child.iterator();
- if (!origin.hasNext()) {
- return false;
- }
- R result = origin.next();
- long queryVersion = this.child.queryVersion();
- if (this.childQueryVersion != queryVersion) {
- this.parent.addQueries(this.child.currentQueries());
- this.childQueryVersion = queryVersion;
- }
- assert this.current == none();
- this.current = result;
- return true;
- }
- }
-
- private static class InputOrderIterator<T extends Idfiable>
- extends WrappedIterator<T> {
-
- private final QueryResults<?> queryResults;
- private final Iterator<T> origin;
- private Iterator<T> currentBatch;
-
- public InputOrderIterator(QueryResults<?> queryResults,
- Iterator<T> origin) {
- this.queryResults = queryResults;
- this.origin = origin;
- this.currentBatch = Collections.emptyIterator();
- }
-
- @Override
- protected Iterator<T> originIterator() {
- return this.origin;
- }
-
- @Override
- protected boolean fetch() {
- while (true) {
- if (this.currentBatch.hasNext()) {
- assert this.current == none();
- this.current = this.currentBatch.next();
- return true;
- }
- if (!this.origin.hasNext()) {
- return false;
- }
- this.currentBatch = this.fetchBatch();
- }
- }
-
- private Iterator<T> fetchBatch() {
- long queryVersion = this.queryResults.queryVersion();
- List<Query> queries = this.queryResults.currentQueries();
- List<T> results = InsertionOrderUtil.newList();
- do {
- results.add(this.origin.next());
- Query.checkForceCapacity(results.size());
- } while (this.origin.hasNext() &&
- queryVersion == this.queryResults.queryVersion());
-
- if (!mustSortByInputIds(queries)) {
- return results.iterator();
- }
- Collection<Id> ids = queryIds(queries);
- if (ids.size() <= 1) {
- return results.iterator();
- }
-
- Map<Id, T> byId = InsertionOrderUtil.newMap();
- for (T result : results) {
- assert result.id() != null;
- byId.put(result.id(), result);
- }
- if (byId.size() > ids.size()) {
- /*
- * The current query only describes part of this segment.
- * Preserve backend order because it can't fully define the
- * order of every returned result.
- */
- return results.iterator();
- }
-
- List<T> ordered = new ArrayList<>(results.size());
- for (Id id : ids) {
- T result = byId.remove(id);
- if (result != null) {
- ordered.add(result);
- }
- }
- ordered.addAll(byId.values());
- return ordered.iterator();
- }
- }
}
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java
index d47b2825c..3999b3d0c 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java
@@ -20,7 +20,6 @@ package org.apache.hugegraph.backend.tx;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -32,7 +31,6 @@ import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
import org.apache.commons.collections.CollectionUtils;
import org.apache.hugegraph.HugeException;
@@ -53,6 +51,8 @@ import org.apache.hugegraph.backend.query.ConditionQuery.OptimizedType;
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;
+import org.apache.hugegraph.backend.query.QueryResultContext;
import org.apache.hugegraph.backend.query.QueryResults;
import org.apache.hugegraph.backend.store.BackendEntry;
import org.apache.hugegraph.backend.store.BackendMutation;
@@ -64,9 +64,7 @@ import org.apache.hugegraph.exception.NotFoundException;
import org.apache.hugegraph.iterator.BatchMapperIterator;
import org.apache.hugegraph.iterator.ExtendableIterator;
import org.apache.hugegraph.iterator.FilterIterator;
-import org.apache.hugegraph.iterator.FlatMapperIterator;
import org.apache.hugegraph.iterator.LimitIterator;
-import org.apache.hugegraph.iterator.ListIterator;
import org.apache.hugegraph.iterator.MapperIterator;
import org.apache.hugegraph.job.system.DeleteExpiredJob;
import org.apache.hugegraph.perf.PerfUtil.Watched;
@@ -101,7 +99,6 @@ import org.apache.tinkerpop.gremlin.structure.util.ElementHelper;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterators;
import jakarta.ws.rs.ForbiddenException;
@@ -853,26 +850,43 @@ public class GraphTransaction extends IndexableTransaction {
}
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) {
+ return this.filterExpiredBatches(this.backendBatches(query).map(this::parseEntry));
}
private Iterator<HugeVertex> queryValidVerticesFromBackend(Query query) {
- Iterator<HugeVertex> results = this.queryVerticesFromBackend(query);
- return this.filterInvalidRecords(results, query);
+ return this.queryVertexBatchesFromBackend(query)
+ .filter(this::filterInvalidRecord).iterator();
+ }
+
+ private <T extends HugeElement> QueryResults<T> processBatches(QueryResults<T> batches) {
+ batches = batches.filter(this::filterUnmatchedRecord);
+ return batches.keepInputOrderIfNeeded();
}
@Watched(prefix = "graph")
@@ -1051,87 +1065,79 @@ public class GraphTransaction extends IndexableTransaction {
}
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));
+ }
+ // Each branch is validated when activated, so later branches may fail
+ // during iteration after earlier branches have already returned results.
+ return QueryResults.flatMap(flattened.iterator(), fetcher);
}
-
- return queryEdgesFromBackendInternal(query);
+ return this.queryEdgeBatchesFromBackendInternal(query);
}
private Iterator<HugeEdge> queryValidEdgesFromBackend(Query query) {
- Iterator<HugeEdge> results = this.queryEdgesFromBackend(query);
- return this.filterInvalidRecords(results, query);
+ return this.queryEdgeBatchesFromBackend(query)
+ .filter(this::filterInvalidRecord).iterator();
}
- private Iterator<HugeEdge> queryEdgesFromBackendInternal(Query query) {
- assert query.resultType().isEdge();
-
- QueryResults<BackendEntry> results = this.query(query);
- Iterator<BackendEntry> entries = results.iterator();
+ private QueryResults<HugeEdge> queryEdgeBatchesFromBackendInternal(Query query) {
+ if (!(query instanceof ConditionQuery)) {
+ return this.processBatches(this.fetchEdgeBatch(query));
+ }
+ QueryList<HugeEdge> queries = this.optimizeQueries(query, this::fetchEdgeBatch);
+ return this.processBatches(queries.empty() ? QueryResults.empty() :
+ queries.fetch(this.pageSize));
+ }
- Iterator<HugeEdge> edges = new FlatMapperIterator<>(entries, entry -> {
- // Edges are in a vertex
+ protected QueryResults<HugeEdge> fetchEdgeBatch(Query query) {
+ QueryResults<HugeEdge> edges = this.backendBatches(query).flatMap(entry -> {
HugeVertex vertex = this.parseEntry(entry);
if (vertex == null) {
return null;
}
assert query.idsSize() != 1 || vertex.getEdges().size() == 1;
- /*
- * Copy to avoid ConcurrentModificationException when removing edge
- * because HugeEdge.remove() will update edges in owner vertex
- */
- return new ListIterator<>(ImmutableList.copyOf(vertex.getEdges()));
+ // Removing an edge may change its owner's collection during iteration.
+ return ImmutableList.copyOf(vertex.getEdges()).iterator();
});
-
- edges = this.filterExpiredResultFromBackend(query, edges);
- edges = this.filterUnmatchedRecords(edges, query);
-
- if (!this.store().features().supportsQuerySortByInputIds()) {
- // There is no id in BackendEntry, so sort after deserialization
- edges = results.keepInputOrderIfNeeded(edges);
- }
- return edges;
- }
-
- private Iterator<HugeEdge> parentElQueryWithSortKeys(EdgeLabel label,
- Collection<EdgeLabel> allEls,
- ConditionQuery cq) {
- return allEls.stream()
- .filter(el -> el.edgeLabelType().sub() && el.fatherId().equals(label.id()))
- .map(el -> {
- ConditionQuery tempQuery = cq.copy();
- tempQuery.eq(HugeKeys.SUB_LABEL, el.id());
- return this.queryEdgesFromBackend(tempQuery);
- })
- .reduce(Iterators::concat)
- .orElse(Collections.emptyIterator());
+ return this.filterExpiredBatches(edges);
}
@Watched(prefix = "graph")
@@ -1882,56 +1888,29 @@ public class GraphTransaction extends IndexableTransaction {
}
}
- private <T extends HugeElement> Iterator<T> filterInvalidRecords(
- Iterator<T> results,
- Query query) {
- // Filter unused records
- return new FilterIterator<>(results, elem -> {
- warnLeftRecord(elem);
- return !invalidRecord(elem, query);
- });
+ private boolean filterInvalidRecord(QueryResultContext context, HugeElement elem) {
+ warnLeftRecord(elem);
+ return !invalidRecord(elem, context);
}
- private <T extends HugeElement> Iterator<T> filterUnmatchedRecords(
- Iterator<T> results,
- Query query) {
- /*
- * Filter against the current index sub-query before restoring input
- * order, since the order iterator may prefetch the next sub-query and
- * update the results filter of the origin query.
- * FilterIterator tests and buffers each element before its upstream
- * iterator is advanced again, so the element is always tested with
- * the results filter of the sub-query that produced it.
- */
- return new FilterIterator<>(results, elem -> {
- /*
- * Preserve the original predicate order: hidden records and
- * records of deleting labels must be handled by the downstream
- * invalid-record filter without triggering left-index cleanup.
- */
- if (invalidRecord(elem, query)) {
- return true;
- }
- // Process results that query from left index or primary-key
- // Only index query will come here
- boolean matched =
- query.resultType().isVertex() != elem.type().isVertex() ||
- rightResultFromIndexQuery(query, elem);
- if (!matched) {
- warnLeftRecord(elem);
- }
- return matched;
- });
+ private boolean filterUnmatchedRecord(QueryResultContext context, HugeElement elem) {
+ // Invisible records pass to the public visibility stage, without index cleanup.
+ if (invalidRecord(elem, context)) {
+ return true;
+ }
+ boolean matched = this.rightResultFromIndexQuery(context, elem);
+ if (!matched) {
+ // Rejected records never reach the public stage; raw callers keep this diagnostic too.
+ warnLeftRecord(elem);
+ }
+ return matched;
}
- private static boolean invalidRecord(HugeElement elem, Query query) {
- // Filter hidden results
- if (!query.showHidden() && Graph.Hidden.isHidden(elem.label())) {
+ private static boolean invalidRecord(HugeElement elem, QueryResultContext context) {
+ if (!context.showHidden() && Graph.Hidden.isHidden(elem.label())) {
return true;
}
- // Filter vertices/edges of deleting label
- return elem.schemaLabel().status().deleting() &&
- !query.showDeleting();
+ return elem.schemaLabel().status().deleting() && !context.showDeleting();
}
private static void warnLeftRecord(HugeElement elem) {
@@ -1943,26 +1922,17 @@ public class GraphTransaction extends IndexableTransaction {
}
}
- private boolean rightResultFromIndexQuery(Query query, HugeElement elem) {
- /*
- * If query is ConditionQuery or query.originQuery() is ConditionQuery
- * means it's index query
- */
- if (!(query instanceof ConditionQuery)) {
- if (query.originQuery() instanceof ConditionQuery) {
- query = query.originQuery();
- } else {
- return true;
- }
+ private boolean rightResultFromIndexQuery(QueryResultContext context, HugeElement elem) {
+ ConditionQuery cq = context.matchQuery();
+ if (cq == null || cq.resultType().isVertex() != elem.type().isVertex()) {
+ return true;
}
-
- ConditionQuery cq = (ConditionQuery) query;
if (cq.condition(HugeKeys.LABEL) != null && cq.resultType().isEdge()) {
if (cq.conditions().size() == 1) {
// g.E().hasLabel(xxx)
return true;
}
- if (cq.optimized() == OptimizedType.INDEX) {
+ if (context.optimizedType() == OptimizedType.INDEX) {
// g.E().hasLabel(xxx).has(yyy)
// consider OptimizedType.INDEX_FILTER occurred in org.apache.hugegraph.core
// .EdgeCoreTest.testQueryCount
@@ -1976,7 +1946,7 @@ public class GraphTransaction extends IndexableTransaction {
}
}
- if (!conditionQueryNeedsPostFilter(cq) || cq.test(elem)) {
+ if (!context.conditionFilterRequired() || cq.test(elem, context.resultsFilter())) {
if (cq.existLeftIndex(elem.id())) {
/*
* Both have correct and left index, wo should return true
@@ -1992,13 +1962,13 @@ public class GraphTransaction extends IndexableTransaction {
/* Return true if:
* 1.not query by index or by primary-key/sort-key
- * (cq.optimized() == 0 means query just by sysprop)
+ * (context.optimizedType() == 0 means query just by sysprop)
* 2.the result match all conditions
*/
return true;
}
- if (cq.optimized() == OptimizedType.INDEX) {
+ if (context.optimizedType() == OptimizedType.INDEX) {
try {
this.indexTx.asyncRemoveIndexLeft(cq, elem);
} catch (Throwable e) {
@@ -2009,45 +1979,16 @@ public class GraphTransaction extends IndexableTransaction {
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;
+ protected <T extends HugeElement> QueryResults<T> filterExpiredBatches(QueryResults<T> batches) {
+ if (this.storeFeatures().supportsTtl()) {
+ return batches;
}
- // Filter expired vertices/edges with TTL
- return new FilterIterator<>(results, elem -> {
- if (elem.expired()) {
- DeleteExpiredJob.asyncDeleteExpiredObject(this.graph(), elem);
- return false;
+ return batches.filter((context, elem) -> {
+ if (context.showExpired() || !elem.expired()) {
+ return true;
}
- return true;
+ DeleteExpiredJob.asyncDeleteExpiredObject(this.graph(), elem);
+ return false;
});
}
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/backend/page/QueryListTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/backend/page/QueryListTest.java
new file mode 100644
index 000000000..7fc020889
--- /dev/null
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/backend/page/QueryListTest.java
@@ -0,0 +1,409 @@
+/*
+ * 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.page;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.page.IdHolder.BatchIdHolder;
+import org.apache.hugegraph.backend.page.IdHolder.FixedIdHolder;
+import org.apache.hugegraph.backend.page.IdHolder.PagingIdHolder;
+import org.apache.hugegraph.backend.query.ConditionQuery.OptimizedType;
+import org.apache.hugegraph.backend.query.ConditionQuery;
+import org.apache.hugegraph.backend.query.QueryBatch;
+import org.apache.hugegraph.backend.query.QueryResults;
+import org.apache.hugegraph.backend.serializer.TextBackendEntry;
+import org.apache.hugegraph.backend.store.BackendEntry;
+import org.apache.hugegraph.iterator.CIter;
+import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.type.Idfiable;
+import org.apache.hugegraph.util.InsertionOrderUtil;
+import org.junit.Test;
+
+public class QueryListTest {
+
+ @Test
+ public void testEmptyPagesHaveIndependentBatchLifecycles() {
+ Iterator<QueryBatch<Object>> first = QueryList.PageResults.emptyIterator().results().batches();
+ Iterator<QueryBatch<Object>> second = QueryList.PageResults.emptyIterator().results().batches();
+ Assert.assertTrue(first.hasNext());
+ Assert.assertTrue(second.hasNext());
+ Assert.assertFalse(second.next().results().hasNext());
+ Assert.assertFalse(second.hasNext());
+ Assert.assertFalse(first.next().results().hasNext());
+ Assert.assertFalse(first.hasNext());
+ }
+
+ @Test
+ public void testPageWithoutBackendBatchesEndsHolder() {
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ query.page("");
+ query.limit(2L);
+ int[] firstFetches = {0};
+ IdHolderList holders = new IdHolderList(true);
+ holders.add(new PagingIdHolder(query, page -> {
+ firstFetches[0]++;
+ return new PageIds(ids(1L), new PageState(new byte[]{1}, 0, 1));
+ }));
+ holders.add(new PagingIdHolder(query, page -> new PageIds(ids(2L), PageState.EMPTY)));
+ QueryList<Item> list = new QueryList<>(query, batch -> {
+ if (batch.ids().contains(IdGenerator.of(1L))) {
+ return QueryResults.fromBatches(Collections.emptyIterator());
+ }
+ return new QueryResults<>(Collections.singletonList(new Item(IdGenerator.of(2L)))
+ .iterator(), batch);
+ });
+ list.add(holders, 1L);
+ Iterator<Item> results = list.fetch(1).iterator();
+ Assert.assertEquals(IdGenerator.of(2L), results.next().id());
+ Assert.assertFalse(results.hasNext());
+ Assert.assertEquals(1, firstFetches[0]);
+ Assert.assertNull(PageInfo.pageInfo(results));
+ }
+
+ @Test
+ public void testFilteredEmptyBatchKeepsFollowingPage() {
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ query.page("");
+ query.limit(2L);
+ int[] pages = {0};
+ IdHolderList holders = new IdHolderList(true);
+ holders.add(new PagingIdHolder(query, page -> {
+ long id = ++pages[0];
+ return new PageIds(ids(id), new PageState(id == 1L ? new byte[]{1} : new byte[0], 0, 1));
+ }));
+ QueryList<Item> list = new QueryList<>(query, batch -> {
+ Item item = new Item(batch.ids().iterator().next());
+ return new QueryResults<>(Collections.singletonList(item).iterator(), batch)
+ .filter((context, value) -> value.id().asLong() == 2L);
+ });
+ list.add(holders, 1L);
+ Iterator<Item> results = list.fetch(1).iterator();
+ Assert.assertEquals(IdGenerator.of(2L), results.next().id());
+ Assert.assertFalse(results.hasNext());
+ Assert.assertEquals(2, pages[0]);
+ Assert.assertNull(PageInfo.pageInfo(results));
+ }
+
+ @Test
+ public void testPagingKeepsEachBatchContext() throws Exception {
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ query.page("");
+ query.limit(10000L);
+ int[] pages = {0};
+ IdHolderList holders = new IdHolderList(true);
+ holders.add(new PagingIdHolder(query, page -> {
+ long id = ++pages[0];
+ return new PageIds(ids(id), new PageState(new byte[]{1}, 0, 1));
+ }));
+ QueryList<Item> list = new QueryList<>(query, batch -> new QueryResults<>(
+ Collections.singletonList(new Item(batch.ids().iterator().next())).iterator(),
+ batch));
+ list.add(holders, 1L);
+ QueryResults<Item> results = list.fetch(1);
+ Iterator<QueryBatch<Item>> batches = results.batches();
+ for (long id = 1L; id <= 10000L; id++) {
+ QueryBatch<Item> batch = batches.next();
+ Assert.assertEquals(Collections.singletonList(IdGenerator.of(id)),
+ batch.context().inputIds());
+ Assert.assertEquals(IdGenerator.of(id), batch.results().next().id());
+ Assert.assertFalse(batch.results().hasNext());
+ Assert.assertEquals(id, (long) pages[0]);
+ }
+ Assert.assertEquals(10000, pages[0]);
+ ((AutoCloseable) batches).close();
+ Assert.assertFalse(batches.hasNext());
+ }
+
+ @Test
+ public void testOrderingDoesNotFetchNextIndexPage() {
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ query.page("");
+ query.limit(4);
+ int[] pageFetches = {0};
+ int[] backendFetches = {0};
+ IdHolderList holders = new IdHolderList(true);
+ holders.add(new PagingIdHolder(query, page -> {
+ Assert.assertEquals(pageFetches[0] == 0 ? "" :
+ new PageState(new byte[]{1}, 0, 2).toString(), page.page());
+ Assert.assertEquals(2L, page.limit());
+ int offset = pageFetches[0]++ * 2;
+ Set<Id> ids = InsertionOrderUtil.newSet();
+ ids.add(IdGenerator.of(offset + 2L));
+ ids.add(IdGenerator.of(offset + 1L));
+ return new PageIds(ids, new PageState(
+ pageFetches[0] == 1 ? new byte[]{1} : new byte[0], 0, 2));
+ }, true));
+ QueryList<Item> queries = new QueryList<>(query, batch -> {
+ backendFetches[0]++;
+ List<Item> items = new ArrayList<>();
+ batch.ids().stream().sorted().forEach(id -> items.add(new Item(id)));
+ return new QueryResults<>(items.iterator(), batch);
+ });
+ queries.add(holders, 2);
+ QueryResults<Item> results = queries.fetch(2);
+ Iterator<Item> ordered = results.<Item>keepInputOrderIfNeeded().iterator();
+ Assert.assertEquals(IdGenerator.of(2L), ordered.next().id());
+ Assert.assertEquals(IdGenerator.of(1L), ordered.next().id());
+ Assert.assertEquals(1, pageFetches[0]);
+ Assert.assertEquals(1, backendFetches[0]);
+ PageInfo nextPage = PageInfo.fromString(PageInfo.pageInfo(ordered));
+ Assert.assertEquals(0, nextPage.offset());
+ Assert.assertEquals(new PageState(new byte[]{1}, 0, 2).toString(), nextPage.page());
+ Assert.assertEquals(1, pageFetches[0]);
+ Assert.assertEquals(IdGenerator.of(4L), ordered.next().id());
+ Assert.assertEquals(IdGenerator.of(3L), ordered.next().id());
+ Assert.assertFalse(ordered.hasNext());
+ Assert.assertEquals(2, pageFetches[0]);
+ Assert.assertEquals(2, backendFetches[0]);
+ Assert.assertNull(PageInfo.pageInfo(ordered));
+ }
+
+ @Test
+ public void testLimitStopsAtPageBoundaryAndRetainsCursor() {
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ query.page("");
+ query.limit(1L);
+ int[] fetches = {0};
+ PageState next = new PageState(new byte[]{7}, 0, 1);
+ IdHolderList holders = new IdHolderList(true);
+ holders.add(new PagingIdHolder(query, page -> {
+ Assert.assertEquals("", page.page());
+ Assert.assertEquals(1L, page.limit());
+ fetches[0]++;
+ return new PageIds(ids(1L), next);
+ }));
+ QueryList<Item> list = new QueryList<>(query, batch ->
+ new QueryResults<>(Collections.singletonList(new Item(IdGenerator.of(1L)))
+ .iterator(), batch));
+ list.add(holders, 2L);
+ Iterator<Item> values = list.fetch(2).iterator();
+ Assert.assertEquals(IdGenerator.of(1L), values.next().id());
+ Assert.assertFalse(values.hasNext());
+ Assert.assertEquals(1, fetches[0]);
+ Assert.assertEquals(next.toString(), PageInfo.fromString(PageInfo.pageInfo(values)).page());
+ }
+
+ @Test
+ public void testEmptyHolderAdvancesToNextHolder() {
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ query.page("");
+ query.limit(2L);
+ int[] fetches = {0};
+ IdHolderList holders = new IdHolderList(true);
+ holders.add(new PagingIdHolder(query, page -> PageIds.EMPTY));
+ holders.add(new PagingIdHolder(query, page -> {
+ Assert.assertEquals("", page.page());
+ fetches[0]++;
+ return new PageIds(ids(1L), PageState.EMPTY);
+ }));
+ QueryList<Item> list = new QueryList<>(query, batch ->
+ new QueryResults<>(Collections.singletonList(new Item(IdGenerator.of(1L)))
+ .iterator(), batch));
+ list.add(holders, 2L);
+ Iterator<Item> values = list.fetch(2).iterator();
+ Assert.assertEquals(IdGenerator.of(1L), values.next().id());
+ Assert.assertFalse(values.hasNext());
+ Assert.assertNull(PageInfo.pageInfo(values));
+ Assert.assertEquals(1, fetches[0]);
+ }
+
+ @Test
+ public void testCopiedSearchFiltersAndEmptyHoldersStayLocal() {
+ ConditionQuery root = new ConditionQuery(HugeType.VERTEX);
+ root.optimized(OptimizedType.INDEX_FILTER);
+ ConditionQuery first = root.copy();
+ ConditionQuery second = root.copy();
+ ConditionQuery.ResultsFilter firstFilter = element -> true;
+ ConditionQuery.ResultsFilter secondFilter = element -> false;
+ first.registerResultsFilter(firstFilter);
+ second.registerResultsFilter(secondFilter);
+ IdHolderList holders = new IdHolderList(false);
+ holders.add(new FixedIdHolder(first.copy(), ids(2L, 1L)));
+ holders.add(new FixedIdHolder(first.copy(), Collections.emptySet()));
+ holders.add(new FixedIdHolder(second.copy(), ids(4L, 3L)));
+ int[] fetched = {0};
+ QueryList<Item> list = new QueryList<>(root, query -> {
+ Assert.assertEquals(HugeType.VERTEX, query.resultType());
+ fetched[0]++;
+ List<Item> values = new ArrayList<>();
+ query.ids().forEach(id -> values.add(new Item(id)));
+ return new QueryResults<>(values.iterator(), query);
+ });
+ list.add(holders, 2L);
+ List<Id> accepted = new ArrayList<>();
+ QueryResults<Item> results = list.fetch(2).filter((context, item) -> {
+ boolean firstBatch = item.id().asLong() <= 2L;
+ Assert.assertSame(firstBatch ? firstFilter : secondFilter, context.resultsFilter());
+ Assert.assertEquals(firstBatch ? 1 : 2, fetched[0]);
+ Assert.assertSame(root, context.matchQuery());
+ return item.id().asLong() % 2 == 0;
+ });
+ results.iterator().forEachRemaining(item -> accepted.add(item.id()));
+ Assert.assertEquals(Arrays.asList(IdGenerator.of(2L), IdGenerator.of(4L)), accepted);
+ Assert.assertEquals(2, fetched[0]);
+ }
+
+ @Test
+ public void testMultipleIndexHoldersFetchOneIdBatchAtATime() {
+ ConditionQuery root = new ConditionQuery(HugeType.VERTEX);
+ IdHolderList holders = new IdHolderList(false);
+ int[] idFetches = {0};
+ for (long start : new long[]{1L, 5L}) {
+ List<BackendEntry> entries = new ArrayList<>();
+ for (long id = start; id < start + 4L; id++) {
+ entries.add(new TextBackendEntry(HugeType.VERTEX, IdGenerator.of(id)));
+ }
+ Iterator<BackendEntry> source = entries.iterator();
+ ConditionQuery index = new ConditionQuery(HugeType.SECONDARY_INDEX, root);
+ holders.add(new BatchIdHolder(index, source, size -> {
+ idFetches[0]++;
+ Set<Id> ids = InsertionOrderUtil.newSet();
+ while (ids.size() < size && source.hasNext()) {
+ ids.add(source.next().id());
+ }
+ return ids;
+ }, true));
+ }
+ QueryList<Item> list = new QueryList<>(root, query -> {
+ Assert.assertEquals(HugeType.VERTEX, query.resultType());
+ List<Item> values = new ArrayList<>();
+ query.ids().forEach(id -> values.add(0, new Item(id)));
+ return new QueryResults<>(values.iterator(), query);
+ });
+ list.add(holders, 2L);
+ Iterator<Item> values = list.fetch(2).<Item>keepInputOrderIfNeeded().iterator();
+ for (long id = 1L; id <= 8L; id++) {
+ Assert.assertEquals(IdGenerator.of(id), values.next().id());
+ Assert.assertEquals((int) ((id + 1L) / 2L), idFetches[0]);
+ }
+ Assert.assertFalse(values.hasNext());
+ }
+
+ @Test
+ public void testClosingQueryListReleasesUnconsumedHolders() throws Exception {
+ this.assertHoldersClosed(false);
+ }
+
+ @Test
+ public void testFetcherFailureReleasesAllHolders() throws Exception {
+ this.assertHoldersClosed(true);
+ }
+
+ private void assertHoldersClosed(boolean failFetcher) throws Exception {
+ ConditionQuery root = new ConditionQuery(HugeType.VERTEX);
+ IdHolderList holders = new IdHolderList(false);
+ List<TrackedSource> sources = new ArrayList<>();
+ for (long id = 1L; id <= 2L; id++) {
+ TrackedSource source = new TrackedSource(id);
+ sources.add(source);
+ holders.add(new BatchIdHolder(new ConditionQuery(HugeType.SECONDARY_INDEX, root),
+ source, size -> ids(source.next().id().asLong())));
+ }
+ RuntimeException failure = new IllegalStateException("fetcher");
+ QueryList<Item> list = new QueryList<>(root, query -> {
+ if (failFetcher) {
+ throw failure;
+ }
+ return new QueryResults<>(Collections.singletonList(new Item(query.ids().iterator().next()))
+ .iterator(), query);
+ });
+ list.add(holders, 1L);
+ Iterator<Item> results = list.fetch(1).iterator();
+ if (failFetcher) {
+ try {
+ results.hasNext();
+ Assert.fail("Expected fetcher failure");
+ } catch (IllegalStateException actual) {
+ Assert.assertSame(failure, actual);
+ }
+ Assert.assertEquals(1, sources.get(0).closed);
+ Assert.assertEquals(1, sources.get(1).closed);
+ } else {
+ Assert.assertTrue(results.hasNext());
+ }
+ ((AutoCloseable) results).close();
+ ((AutoCloseable) results).close();
+ Assert.assertEquals(1, sources.get(0).read);
+ Assert.assertEquals(0, sources.get(1).read);
+ Assert.assertEquals(1, sources.get(0).closed);
+ Assert.assertEquals(1, sources.get(1).closed);
+ Assert.assertFalse(results.hasNext());
+ }
+
+ private static final class TrackedSource implements CIter<BackendEntry> {
+
+ private final BackendEntry entry;
+ private int read;
+ private int closed;
+
+ private TrackedSource(long id) {
+ this.entry = new TextBackendEntry(HugeType.VERTEX, IdGenerator.of(id));
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.read == 0;
+ }
+
+ @Override
+ public BackendEntry next() {
+ this.read++;
+ return this.entry;
+ }
+
+ @Override
+ public void close() {
+ this.closed++;
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return null;
+ }
+ }
+
+ private static Set<Id> ids(Long... values) {
+ Set<Id> ids = InsertionOrderUtil.newSet();
+ for (long value : values) {
+ ids.add(IdGenerator.of(value));
+ }
+ return ids;
+ }
+
+ private static final class Item implements Idfiable {
+
+ private final Id id;
+
+ private Item(Id id) {
+ this.id = id;
+ }
+
+ @Override
+ public Id id() {
+ return this.id;
+ }
+ }
+}
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/backend/tx/GraphTransactionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/backend/tx/GraphTransactionTest.java
index 85b3c0e76..3cb79e368 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/backend/tx/GraphTransactionTest.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/backend/tx/GraphTransactionTest.java
@@ -17,74 +17,206 @@
package org.apache.hugegraph.backend.tx;
-import java.util.Collections;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hugegraph.HugeFactory;
+import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.backend.cache.Cache;
+import org.apache.hugegraph.backend.cache.CachedGraphTransaction;
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.id.IdGenerator;
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.IdQuery;
-import org.apache.hugegraph.backend.query.Query;
+import org.apache.hugegraph.backend.query.QueryResultContext;
+import org.apache.hugegraph.backend.tx.GraphIndexTransaction.RemoveLeftIndexJob;
+import org.apache.hugegraph.job.EphemeralJob;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.testutil.Whitebox;
import org.apache.hugegraph.type.HugeType;
-import org.apache.hugegraph.type.define.HugeKeys;
+import org.apache.hugegraph.type.define.SchemaStatus;
+import org.apache.hugegraph.unit.FakeObjects;
+import org.apache.tinkerpop.gremlin.structure.T;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.junit.Test;
+import org.mockito.Mockito;
public class GraphTransactionTest {
@Test
- public void testQueryNeedsPostFilter() {
- Id key = IdGenerator.of(1);
- ConditionQuery search = new ConditionQuery(HugeType.EDGE);
- search.query(Condition.textContains(key, "word"));
-
- Assert.assertTrue(GraphTransaction.queryNeedsPostFilter(search));
- IdQuery searchIds = new IdQuery(search, IdGenerator.of(2));
- Assert.assertTrue(GraphTransaction.queryNeedsPostFilter(searchIds));
-
- ConditionQuery searchAny = new ConditionQuery(HugeType.EDGE);
- searchAny.query(Condition.textContainsAny(
- key, Collections.singleton("word")));
- Assert.assertTrue(GraphTransaction.queryNeedsPostFilter(searchAny));
-
- ConditionQuery exact = new ConditionQuery(HugeType.EDGE);
- exact.query(Condition.eq(key, "word"));
- Assert.assertFalse(GraphTransaction.queryNeedsPostFilter(exact));
- exact.optimized(OptimizedType.INDEX_FILTER);
- Assert.assertTrue(GraphTransaction.queryNeedsPostFilter(exact));
-
- ConditionQuery index = new ConditionQuery(HugeType.EDGE);
- index.query(Condition.eq(key, "word"));
- index.optimized(OptimizedType.INDEX);
- Assert.assertTrue(GraphTransaction.queryNeedsPostFilter(index));
-
- ConditionQuery labelIndex = new ConditionQuery(HugeType.EDGE);
- labelIndex.query(Condition.eq(HugeKeys.LABEL, IdGenerator.of(2)));
- labelIndex.query(Condition.eq(key, "word"));
- labelIndex.optimized(OptimizedType.INDEX);
- Assert.assertFalse(GraphTransaction.queryNeedsPostFilter(labelIndex));
- IdQuery labelIndexIds = new IdQuery(labelIndex, IdGenerator.of(2));
- Assert.assertFalse(GraphTransaction.queryNeedsPostFilter(labelIndexIds));
-
- ConditionQuery vertexLabelIndex =
- new ConditionQuery(HugeType.VERTEX);
- vertexLabelIndex.query(Condition.eq(HugeKeys.LABEL,
- IdGenerator.of(2)));
- vertexLabelIndex.query(Condition.eq(key, "word"));
- vertexLabelIndex.optimized(OptimizedType.INDEX);
- Assert.assertTrue(GraphTransaction.queryNeedsPostFilter(
- vertexLabelIndex));
-
- ConditionQuery primaryKey = new ConditionQuery(HugeType.VERTEX);
- primaryKey.optimized(OptimizedType.PRIMARY_KEY);
- Assert.assertTrue(GraphTransaction.queryNeedsPostFilter(primaryKey));
-
- ConditionQuery sortKeys = new ConditionQuery(HugeType.EDGE);
- sortKeys.query(Condition.eq(key, "word"));
- sortKeys.optimized(OptimizedType.SORT_KEYS);
- Assert.assertTrue(GraphTransaction.queryNeedsPostFilter(sortKeys));
- Assert.assertFalse(GraphTransaction.queryNeedsPostFilter(
- new Query(HugeType.EDGE)));
+ public void testBatchDecisionsRemainFixedAfterOriginChanges() {
+ ConditionQuery root = new ConditionQuery(HugeType.VERTEX);
+ root.optimized(OptimizedType.PRIMARY_KEY);
+ root.showHidden(true);
+ root.showDeleting(true);
+ root.showExpired(true);
+ IdQuery query = new IdQuery(root, IdGenerator.of(1L));
+ QueryResultContext context = new QueryResultContext(query);
+ root.optimized(OptimizedType.INDEX_FILTER);
+ root.showHidden(false);
+ root.showDeleting(false);
+ root.showExpired(false);
+ query.resetIds();
+ query.mustSortByInput(false);
+ Assert.assertEquals(OptimizedType.PRIMARY_KEY, context.optimizedType());
+ Assert.assertTrue(context.conditionFilterRequired());
+ Assert.assertTrue(context.showHidden());
+ Assert.assertTrue(context.showDeleting());
+ Assert.assertTrue(context.showExpired());
+ Assert.assertTrue(context.mustSortByInputIds());
+ Assert.assertEquals(IdGenerator.of(1L), context.inputIds().get(0));
+ Assert.assertEquals(1, context.inputIds().size());
+ Assert.assertSame(root, context.matchQuery());
+ }
+
+ @Test
+ public void testSiblingOptimizationDoesNotReplaceBatchDecision() {
+ ConditionQuery root = new ConditionQuery(HugeType.EDGE);
+ ConditionQuery first = root.copy();
+ first.optimized(OptimizedType.INDEX_FILTER);
+ ConditionQuery second = root.copy();
+ second.optimized(OptimizedType.INDEX);
+ Assert.assertEquals(OptimizedType.INDEX_FILTER, root.optimized());
+ QueryResultContext context = new QueryResultContext(
+ new IdQuery(second, IdGenerator.of(1L)));
+ Assert.assertEquals(OptimizedType.INDEX, context.optimizedType());
+ Assert.assertSame(root, context.matchQuery());
+ }
+
+ @Test
+ public void testDirectIdsHaveNoConditionFilter() {
+ IdQuery query = new IdQuery(HugeType.EDGE, IdGenerator.of(1L));
+ QueryResultContext context = new QueryResultContext(query);
+ Assert.assertNull(context.matchQuery());
+ Assert.assertFalse(context.conditionFilterRequired());
+ Assert.assertNull(context.resultsFilter());
+ Assert.assertFalse(new QueryResultContext(query, true).mustSortByInputIds());
+ }
+ @Test
+ public void testInvisibleCandidatesDoNotScheduleIndexCleanup() throws Exception {
+ try (FilterFixture fixture = new FilterFixture()) {
+ for (boolean hidden : new boolean[]{true, false}) {
+ VertexLabel label = new VertexLabel(fixture.graph, fixture.vertex.schemaLabel().id(),
+ hidden ? "~hidden" : "person");
+ if (!hidden) {
+ label.status(SchemaStatus.DELETING);
+ }
+ HugeVertex candidate = new HugeVertex(fixture.graph, fixture.vertex.id(), label);
+ candidate.addProperty(fixture.graph.propertyKey("name"), "actual");
+ fixture.vertices.update(candidate.id(), candidate);
+ ConditionQuery query = fixture.query("absent", OptimizedType.INDEX);
+ Assert.assertTrue(fixture.fetch(query).isEmpty());
+ Assert.assertTrue(fixture.jobs.isEmpty());
+
+ // Making the same candidate visible must reach residual filtering and cleanup.
+ query.showHidden(hidden);
+ query.showDeleting(!hidden);
+ Assert.assertTrue(fixture.fetch(query).isEmpty());
+ fixture.assertCleanup(query);
+ fixture.jobs.clear();
+ }
+ }
+ }
+
+ @Test
+ public void testResidualMismatchCleansOnlyIndexOptimizedQueries() throws Exception {
+ try (FilterFixture fixture = new FilterFixture()) {
+ for (OptimizedType type : Arrays.asList(OptimizedType.PRIMARY_KEY,
+ OptimizedType.INDEX_FILTER,
+ OptimizedType.INDEX)) {
+ ConditionQuery query = fixture.query("absent", type);
+ Assert.assertTrue(fixture.fetch(query).isEmpty());
+ if (type == OptimizedType.INDEX) {
+ fixture.assertCleanup(query);
+ } else {
+ Assert.assertTrue(fixture.jobs.isEmpty());
+ }
+ fixture.jobs.clear();
+ }
+ }
+ }
+
+ @Test
+ public void testMatchingCandidateStillSchedulesItsStaleIndexCleanup() throws Exception {
+ try (FilterFixture fixture = new FilterFixture()) {
+ ConditionQuery query = fixture.query("actual", OptimizedType.INDEX);
+ Id name = fixture.graph.propertyKey("name").id();
+ query.recordIndexValue(name, fixture.vertex.id(), "actual");
+ query.recordIndexValue(name, fixture.vertex.id(), "stale");
+ query.selectedIndexField(name);
+ List<Vertex> found = fixture.fetch(query);
+ Assert.assertEquals(1, found.size());
+ Assert.assertEquals(fixture.vertex.id(), found.get(0).id());
+ Assert.assertTrue(query.existLeftIndex(fixture.vertex.id()));
+ fixture.assertCleanup(query);
+ }
}
+
+ private static final class FilterFixture implements AutoCloseable {
+
+ private final HugeGraph graph;
+ private final CachedGraphTransaction transaction;
+ private final HugeVertex vertex;
+ private final Cache<Id, Object> vertices;
+ private final List<EphemeralJob<?>> jobs;
+
+ private FilterFixture() {
+ this.graph = HugeFactory.open(FakeObjects.newConfig());
+ this.graph.schema().propertyKey("name").asText().create();
+ this.graph.schema().vertexLabel("person").useCustomizeNumberId()
+ .properties("name").create();
+ this.vertex = (HugeVertex) this.graph.addVertex(T.label, "person", T.id, 1L,
+ "name", "actual");
+ this.graph.tx().commit();
+ HugeGraphParams params = Whitebox.getInternalState(this.graph, "params");
+ HugeGraphParams observed = Mockito.spy(params);
+ this.jobs = new ArrayList<>();
+ Mockito.doAnswer(invocation -> {
+ this.jobs.add(invocation.getArgument(0));
+ return null;
+ }).when(observed).submitEphemeralJob(Mockito.any());
+ this.transaction = new CachedGraphTransaction(observed, params.loadGraphStore());
+ this.transaction.clearCache(null, false);
+ this.vertices = Whitebox.getInternalState(this.transaction, "verticesCache");
+ }
+
+ private ConditionQuery query(String value, OptimizedType type) {
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ query.query(Condition.eq(this.graph.propertyKey("name").id(), value));
+ query.optimized(type);
+ return query;
+ }
+
+ private List<Vertex> fetch(ConditionQuery query) {
+ List<Vertex> found = new ArrayList<>();
+ this.transaction.queryVertices(new IdQuery(query, this.vertex.id()))
+ .forEachRemaining(found::add);
+ return found;
+ }
+
+ private void assertCleanup(ConditionQuery query) {
+ Assert.assertEquals(1, this.jobs.size());
+ Assert.assertTrue(this.jobs.get(0) instanceof RemoveLeftIndexJob);
+ Assert.assertSame(query, Whitebox.getInternalState(this.jobs.get(0), "query"));
+ HugeVertex cleaned = Whitebox.getInternalState(this.jobs.get(0), "element");
+ Assert.assertEquals(this.vertex.id(), cleaned.id());
+ }
+
+ @Override
+ public void close() throws Exception {
+ try {
+ this.transaction.close();
+ } finally {
+ this.graph.clearBackend();
+ this.graph.close();
+ }
+ }
+ }
+
}
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java
index 80c4aef50..335130ede 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java
@@ -36,6 +36,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hugegraph.HugeException;
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.backend.BackendException;
+import org.apache.hugegraph.backend.cache.CachedGraphTransaction;
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.id.Id.IdType;
import org.apache.hugegraph.backend.id.IdGenerator;
@@ -3348,6 +3349,43 @@ public class VertexCoreTest extends BaseCoreTest {
Assert.assertEquals(0, vertices.size());
}
+ @Test
+ public void testQueryByPrimaryValuesInPageWithVertexCache() {
+ Assume.assumeTrue("Not support paging", storeFeatures().supportsQueryByPage());
+ HugeGraph graph = graph();
+ Vertex vertex = graph.addVertex(T.label, "person", "name", "marko",
+ "age", 29, "city", "Beijing");
+ this.commitTx();
+ CachedGraphTransaction cache = (CachedGraphTransaction) this.params().graphTransaction();
+ cache.clearCache(HugeType.VERTEX, false);
+
+ for (int i = 0; i < 2; i++) {
+ if (i == 1) {
+ Assert.assertEquals(vertex.id(), graph.vertices(vertex.id()).next().id());
+ }
+ GraphTraversal<Vertex, Vertex> results = graph.traversal().V()
+ .hasLabel("person").has("name", "marko").has("~page", "").limit(2);
+ List<Vertex> vertices = results.toList();
+ Assert.assertEquals(1, vertices.size());
+ Assert.assertEquals(vertex.id(), vertices.get(0).id());
+ Assert.assertNull(TraversalUtil.page(results));
+ CloseableIterator.closeIterator(results);
+ }
+
+ GraphTraversal<Vertex, Vertex> filtered = graph.traversal().V()
+ .hasLabel("person").has("name", "marko").has("age", 30)
+ .has("~page", "").limit(2);
+ Assert.assertFalse(filtered.hasNext());
+ Assert.assertNull(TraversalUtil.page(filtered));
+ CloseableIterator.closeIterator(filtered);
+
+ GraphTraversal<Vertex, Vertex> missing = graph.traversal().V()
+ .hasLabel("person").has("name", "missing").has("~page", "").limit(2);
+ Assert.assertFalse(missing.hasNext());
+ Assert.assertNull(TraversalUtil.page(missing));
+ CloseableIterator.closeIterator(missing);
+ }
+
@Test
public void testQueryByPrimaryValuesAndPropsWithCachedVertex() {
HugeGraph graph = graph();
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java
index d48738b84..85b33b466 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java
@@ -17,12 +17,13 @@
package org.apache.hugegraph.unit;
-import org.apache.hugegraph.backend.tx.GraphIndexTransactionTest;
-import org.apache.hugegraph.backend.tx.GraphTransactionTest;
import org.apache.hugegraph.api.auth.GraphSpaceAuthPayloadTest;
import org.apache.hugegraph.api.auth.GraphSpaceGroupAPITest;
import org.apache.hugegraph.auth.StandardAuthManagerV2Test;
import org.apache.hugegraph.auth.WsAndHttpBasicAuthHandlerTest;
+import org.apache.hugegraph.backend.page.QueryListTest;
+import org.apache.hugegraph.backend.tx.GraphIndexTransactionTest;
+import org.apache.hugegraph.backend.tx.GraphTransactionTest;
import org.apache.hugegraph.core.RoleElectionStateMachineTest;
import org.apache.hugegraph.meta.EtcdMetaDriverTest;
import org.apache.hugegraph.meta.MetaManagerSchemaCacheClearEventTest;
@@ -58,7 +59,6 @@ import org.apache.hugegraph.unit.core.LocksTableTest;
import org.apache.hugegraph.unit.core.PageStateTest;
import org.apache.hugegraph.unit.core.QueryResultsTest;
import org.apache.hugegraph.unit.core.QueryTest;
-import org.apache.hugegraph.unit.core.StandardHugeGraphClearBackendTest;
import org.apache.hugegraph.unit.core.RangeTest;
import org.apache.hugegraph.unit.core.RolePermissionTest;
import org.apache.hugegraph.unit.core.RowLockTest;
@@ -66,6 +66,7 @@ import org.apache.hugegraph.unit.core.SchemaElementTest;
import org.apache.hugegraph.unit.core.SecurityManagerTest;
import org.apache.hugegraph.unit.core.SerialEnumTest;
import org.apache.hugegraph.unit.core.ServerInfoManagerTest;
+import org.apache.hugegraph.unit.core.StandardHugeGraphClearBackendTest;
import org.apache.hugegraph.unit.core.SystemSchemaStoreTest;
import org.apache.hugegraph.unit.core.TaskSchedulerServerInfoTest;
import org.apache.hugegraph.unit.core.TraversalUtilTest;
@@ -156,6 +157,7 @@ import org.junit.runners.Suite;
GraphTransactionTest.class,
QueryTest.class,
QueryResultsTest.class,
+ QueryListTest.class,
RangeTest.class,
SecurityManagerTest.class,
RolePermissionTest.class,
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/cache/CachedGraphTransactionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/cache/CachedGraphTransactionTest.java
index e01a8d799..ae0321ee0 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/cache/CachedGraphTransactionTest.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/cache/CachedGraphTransactionTest.java
@@ -18,8 +18,12 @@
package org.apache.hugegraph.unit.cache;
import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -29,31 +33,54 @@ import org.apache.hugegraph.HugeFactory;
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.HugeGraphParams;
import org.apache.hugegraph.backend.cache.Cache;
+import org.apache.hugegraph.backend.cache.CachedBackendStore.QueryId;
import org.apache.hugegraph.backend.cache.CachedGraphTransaction;
+import org.apache.hugegraph.backend.cache.OffheapCache;
+import org.apache.hugegraph.backend.id.EdgeId;
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.id.IdGenerator;
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.IdQuery;
import org.apache.hugegraph.backend.query.Query;
+import org.apache.hugegraph.backend.query.QueryResultContext;
+import org.apache.hugegraph.backend.query.QueryResults;
+import org.apache.hugegraph.backend.store.BackendStore;
import org.apache.hugegraph.backend.store.BackendStoreProvider;
+import org.apache.hugegraph.backend.store.ram.RamTable;
import org.apache.hugegraph.event.EventHub;
import org.apache.hugegraph.event.EventListener;
import org.apache.hugegraph.schema.VertexLabel;
import org.apache.hugegraph.structure.HugeEdge;
+import org.apache.hugegraph.structure.HugeElement;
import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.structure.HugeVertexProperty;
import org.apache.hugegraph.testutil.Assert;
import org.apache.hugegraph.testutil.Whitebox;
import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.type.define.Directions;
+import org.apache.hugegraph.type.define.HugeKeys;
import org.apache.hugegraph.type.define.IdStrategy;
import org.apache.hugegraph.type.define.SchemaStatus;
import org.apache.hugegraph.unit.BaseUnitTest;
import org.apache.hugegraph.unit.FakeObjects;
import org.apache.hugegraph.util.Events;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.core.config.Property;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.T;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
public class CachedGraphTransactionTest extends BaseUnitTest {
@@ -146,6 +173,34 @@ public class CachedGraphTransactionTest extends BaseUnitTest {
return out.addEdge("person_know_person", in);
}
+ @Test
+ public void testMissingVertexDoesNotCreateAnEmptyBatch() {
+ Query query = new IdQuery(HugeType.VERTEX, IdGenerator.of(999L));
+ QueryResults<HugeVertex> results = Whitebox.invoke(
+ CachedGraphTransaction.class, new Class<?>[]{Query.class},
+ "fetchVertexBatch", this.cache, query);
+ Assert.assertFalse(results.batches().hasNext());
+ }
+
+ @Test
+ public void testExpiredVertexKeepsItsBatchAfterCacheMaterialization() throws InterruptedException {
+ this.graph.schema().vertexLabel("expiring").useCustomizeNumberId()
+ .ttl(1000L).create();
+ Vertex vertex = this.graph.addVertex(T.label, "expiring", T.id, 1L);
+ this.graph.tx().commit();
+ Thread.sleep(1100L);
+ // TTL is evaluated at transaction opening time, including internal fetches.
+ this.graph.tx().open();
+ Assert.assertTrue(((HugeVertex) vertex).expired());
+ Query query = new IdQuery(HugeType.VERTEX, (Id) vertex.id());
+ QueryResults<HugeVertex> results = Whitebox.invoke(
+ CachedGraphTransaction.class, new Class<?>[]{Query.class},
+ "fetchVertexBatch", this.cache, query);
+ Assert.assertTrue(results.batches().hasNext());
+ Assert.assertFalse(results.batches().next().results().hasNext());
+ Assert.assertFalse(results.batches().hasNext());
+ }
+
@Test
public void testEventClear() throws Exception {
CachedGraphTransaction cache = this.cache();
@@ -528,6 +583,348 @@ public class CachedGraphTransactionTest extends BaseUnitTest {
Assert.assertFalse(cache.queryEdgesByVertex(IdGenerator.of(2)).hasNext());
}
+ @Test
+ public void testWarmPrimaryKeyCandidatesAreFilteredForEachQuery() {
+ this.graph.schema().propertyKey("key").asText().create();
+ this.graph.schema().propertyKey("age").asInt().create();
+ this.graph.schema().vertexLabel("primary").properties("key", "age")
+ .primaryKeys("key").create();
+ Vertex vertex = this.graph.addVertex(T.label, "primary", "key", "marko", "age", 20);
+ this.graph.tx().commit();
+ this.cache.clearCache(null, false);
+ Cache<Id, Object> vertices = Whitebox.getInternalState(this.cache, "verticesCache");
+ ConditionQuery miss = this.primaryQuery(21);
+ Assert.assertFalse(this.cache.queryVertices(miss).hasNext());
+ Assert.assertEquals(1L, vertices.size());
+ long hits = vertices.hits();
+ Vertex found = QueryResults.one(this.cache.queryVertices(this.primaryQuery(20)));
+ Assert.assertEquals(vertex.id(), found.id());
+ Assert.assertEquals(hits + 1L, vertices.hits());
+ Assert.assertFalse(this.cache.queryVertices(this.primaryQuery(21)).hasNext());
+ Assert.assertEquals(hits + 2L, vertices.hits());
+ }
+
+ private ConditionQuery primaryQuery(int age) {
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ query.eq(HugeKeys.LABEL, this.graph.vertexLabel("primary").id());
+ query.query(Condition.eq(this.graph.propertyKey("key").id(), "marko"));
+ query.query(Condition.eq(this.graph.propertyKey("age").id(), age));
+ return query;
+ }
+
+ @Test
+ public void testMixedVertexCacheHitsRetainInputOrder() {
+ HugeVertex first = this.newVertex(IdGenerator.of(1L));
+ HugeVertex second = this.newVertex(IdGenerator.of(2L));
+ HugeVertex third = this.newVertex(IdGenerator.of(3L));
+ for (HugeVertex vertex : Arrays.asList(first, second, third)) {
+ this.cache.addVertex(vertex);
+ }
+ this.cache.commit();
+ this.cache.clearCache(null, false);
+ QueryResults.one(this.cache.queryVertices(second.id()));
+ Cache<Id, Object> vertices = Whitebox.getInternalState(this.cache, "verticesCache");
+ long hits = vertices.hits();
+ IdQuery query = new IdQuery(HugeType.VERTEX);
+ query.query(third.id()).query(second.id()).query(first.id());
+ List<Object> ids = new ArrayList<>();
+ this.cache.queryVertices(query).forEachRemaining(vertex -> ids.add(vertex.id()));
+ Assert.assertEquals(Arrays.asList(third.id(), second.id(), first.id()), ids);
+ Assert.assertEquals(hits + 1L, vertices.hits());
+ }
+
+ @Test
+ public void testColdSpecialVertexQueryRetainsResultType() {
+ BackendStore store = Mockito.spy(this.params.loadGraphStore());
+ CachedGraphTransaction transaction = new CachedGraphTransaction(this.params, store);
+ try {
+ transaction.clearCache(null, false);
+ for (HugeType type : Arrays.asList(HugeType.TASK, HugeType.SERVER)) {
+ IdQuery query = new IdQuery(type);
+ query.query(IdGenerator.of(999L));
+ Mockito.doAnswer(invocation -> {
+ Query fetched = invocation.getArgument(0);
+ Assert.assertEquals(type, fetched.resultType());
+ Assert.assertEquals(1, fetched.idsSize());
+ return Collections.emptyIterator();
+ }).when(store).query(Mockito.any(Query.class));
+ Assert.assertFalse(transaction.queryVertices(query).hasNext());
+ }
+ Mockito.verify(store, Mockito.times(2)).query(Mockito.any(Query.class));
+ } finally {
+ transaction.close();
+ }
+ }
+
+ @Test
+ public void testEdgeBatchCacheRoundTripsThroughOffheap() throws Exception {
+ HugeVertex first = this.newVertex(IdGenerator.of(1L));
+ HugeVertex second = this.newVertex(IdGenerator.of(2L));
+ this.cache.addVertex(first);
+ this.cache.addVertex(second);
+ this.cache.commit();
+ HugeEdge edge = this.newEdge(first, second);
+ this.cache.addEdge(edge);
+ this.cache.commit();
+ BackendStore store = Mockito.spy(this.params.loadGraphStore());
+ CachedGraphTransaction transaction = new CachedGraphTransaction(this.params, store);
+ OffheapCache offheap = new OffheapCache(this.graph, 100, 1024, 1);
+ offheap.enableMetrics(true);
+ Whitebox.setInternalState(transaction, "edgesCache", offheap);
+ try {
+ Edge cold = QueryResults.one(transaction.queryEdgesByVertex(first.id()));
+ Assert.assertEquals(edge.id(), cold.id());
+ Assert.assertEquals(1L, offheap.size());
+ long hits = offheap.hits();
+ Mockito.clearInvocations(store);
+ Edge warm = QueryResults.one(transaction.queryEdgesByVertex(first.id()));
+ Assert.assertEquals(edge.id(), warm.id());
+ Assert.assertTrue(offheap.hits() > hits);
+ Mockito.verify(store, Mockito.never()).query(Mockito.any(Query.class));
+ } finally {
+ transaction.close();
+ AutoCloseable nativeCache = Whitebox.getInternalState(offheap, "cache");
+ nativeCache.close();
+ }
+ }
+
+ @Test
+ public void testRootEdgeBatchUsesCompactLabelAndCopyOnWrite() {
+ List<Id> ids = this.persistEdges(2);
+ IdQuery root = new IdQuery(HugeType.EDGE, new LinkedHashSet<>(ids.subList(0, 1)));
+ BackendStore store = Mockito.spy(this.params.loadGraphStore());
+ CachedGraphTransaction transaction = new CachedGraphTransaction(this.params, store);
+ try {
+ transaction.clearCache(null, false);
+ this.assertEdgeIds(ids.subList(0, 1), this.fetchEdgeIds(transaction, root));
+ Cache<Id, Object> cache = Whitebox.getInternalState(transaction, "edgesCache");
+ Id key = new QueryId(root);
+ List<?> original = (List<?>) cache.get(key);
+ Assert.assertEquals("", original.get(0));
+ Assert.assertEquals(2, original.size());
+ IdQuery sibling = new IdQuery(root, new LinkedHashSet<>(ids.subList(1, 2)));
+ this.assertEdgeIds(ids.subList(1, 2), this.fetchEdgeIds(transaction, sibling));
+ Assert.assertEquals(2, original.size());
+ Assert.assertEquals(4, ((List<?>) cache.get(key)).size());
+ Mockito.clearInvocations(store);
+ this.assertEdgeIds(ids.subList(0, 1), this.fetchEdgeIds(transaction, root));
+ this.assertEdgeIds(ids.subList(1, 2), this.fetchEdgeIds(transaction, sibling));
+ Mockito.verify(store, Mockito.never()).query(Mockito.any(Query.class));
+ } finally {
+ transaction.close();
+ }
+ }
+
+ @Test
+ public void testEdgeCacheKeepsDistinctBatchesWithinTotalCandidateLimit() {
+ List<Id> ids = this.persistEdges(101);
+ ConditionQuery root = this.edgeCacheRequest();
+ BackendStore store = Mockito.spy(this.params.loadGraphStore());
+ CachedGraphTransaction transaction = new CachedGraphTransaction(this.params, store);
+ try {
+ transaction.clearCache(null, false);
+ for (int start : new int[]{0, 50}) {
+ List<Id> batch = ids.subList(start, start + 50);
+ IdQuery query = new IdQuery(root, new LinkedHashSet<>(batch));
+ this.assertEdgeIds(batch, this.fetchEdgeIds(transaction, query));
+ }
+ Mockito.clearInvocations(store);
+ for (int start : new int[]{50, 0}) {
+ List<Id> batch = ids.subList(start, start + 50);
+ IdQuery query = new IdQuery(root, new LinkedHashSet<>(batch));
+ this.assertEdgeIds(batch, this.fetchEdgeIds(transaction, query));
+ }
+ Mockito.verify(store, Mockito.never()).query(Mockito.any(Query.class));
+
+ List<Id> overflow = ids.subList(100, 101);
+ for (int i = 0; i < 2; i++) {
+ IdQuery query = new IdQuery(root, new LinkedHashSet<>(overflow));
+ this.assertEdgeIds(overflow, this.fetchEdgeIds(transaction, query));
+ }
+ Mockito.verify(store, Mockito.times(2)).query(Mockito.any(Query.class));
+ Mockito.clearInvocations(store);
+ IdQuery cached = new IdQuery(root, new LinkedHashSet<>(ids.subList(0, 50)));
+ this.assertEdgeIds(ids.subList(0, 50), this.fetchEdgeIds(transaction, cached));
+ Mockito.verify(store, Mockito.never()).query(Mockito.any(Query.class));
+ } finally {
+ transaction.close();
+ }
+ }
+
+ @Test
+ public void testOversizedEdgeBatchReturnsEveryCandidateWithoutCaching() {
+ List<Id> all = this.persistEdges(102);
+ BackendStore store = Mockito.spy(this.params.loadGraphStore());
+ CachedGraphTransaction transaction = new CachedGraphTransaction(this.params, store);
+ try {
+ transaction.clearCache(null, false);
+ for (int size : new int[]{101, 102}) {
+ List<Id> ids = all.subList(0, size);
+ IdQuery query = new IdQuery(this.edgeCacheRequest(), new LinkedHashSet<>(ids));
+ this.assertEdgeIds(ids, this.fetchEdgeIds(transaction, query));
+ Mockito.clearInvocations(store);
+ this.assertEdgeIds(ids, this.fetchEdgeIds(transaction, query));
+ Mockito.verify(store, Mockito.times(1)).query(Mockito.any(Query.class));
+ }
+ } finally {
+ transaction.close();
+ }
+ }
+
+ @Test
+ public void testEmptyEdgeBatchesRespectBatchCountLimit() {
+ this.persistEdges(1);
+ ConditionQuery root = this.edgeCacheRequest();
+ Id label = this.graph.edgeLabel("person_know_person").id();
+ List<IdQuery> batches = new ArrayList<>();
+ for (long id = 1000L; id <= 1100L; id++) {
+ Id edge = new EdgeId(IdGenerator.of(1L), Directions.OUT, label, label,
+ "", IdGenerator.of(id));
+ batches.add(new IdQuery(root, edge));
+ }
+ BackendStore store = Mockito.spy(this.params.loadGraphStore());
+ CachedGraphTransaction transaction = new CachedGraphTransaction(this.params, store);
+ try {
+ transaction.clearCache(null, false);
+ for (IdQuery batch : batches) {
+ Assert.assertTrue(this.fetchEdgeIds(transaction, batch).isEmpty());
+ }
+ Mockito.clearInvocations(store);
+ for (int i = 0; i < 100; i++) {
+ Assert.assertTrue(this.fetchEdgeIds(transaction, batches.get(i)).isEmpty());
+ }
+ Mockito.verify(store, Mockito.never()).query(Mockito.any(Query.class));
+ Assert.assertTrue(this.fetchEdgeIds(transaction, batches.get(100)).isEmpty());
+ Mockito.verify(store, Mockito.times(1)).query(Mockito.any(Query.class));
+ } finally {
+ transaction.close();
+ }
+ }
+
+ private List<Id> persistEdges(int count) {
+ HugeVertex first = this.newVertex(IdGenerator.of(1L));
+ this.cache.addVertex(first);
+ List<HugeVertex> targets = new ArrayList<>();
+ for (long id = 2L; id < count + 2L; id++) {
+ HugeVertex vertex = new HugeVertex(this.graph, IdGenerator.of(id), first.schemaLabel());
+ this.cache.addVertex(vertex);
+ targets.add(vertex);
+ }
+ this.cache.commit();
+ List<Id> ids = new ArrayList<>();
+ for (HugeVertex target : targets) {
+ HugeEdge edge = this.newEdge(first, target);
+ this.cache.addEdge(edge);
+ ids.add(edge.id());
+ }
+ this.cache.commit();
+ return ids;
+ }
+
+ private ConditionQuery edgeCacheRequest() {
+ ConditionQuery query = new ConditionQuery(HugeType.EDGE);
+ query.eq(HugeKeys.OWNER_VERTEX, IdGenerator.of(1L));
+ query.eq(HugeKeys.DIRECTION, Directions.OUT);
+ return query;
+ }
+
+ private void assertEdgeIds(List<Id> expected, List<Id> actual) {
+ // Candidate-cache tests protect membership and cardinality, independent of backend ordering.
+ Assert.assertEquals(expected.size(), actual.size());
+ Assert.assertEquals(new LinkedHashSet<>(expected), new LinkedHashSet<>(actual));
+ }
+
+ private List<Id> fetchEdgeIds(CachedGraphTransaction transaction, Query query) {
+ QueryResults<HugeEdge> results = Whitebox.invoke(
+ CachedGraphTransaction.class, new Class<?>[]{Query.class},
+ "fetchEdgeBatch", transaction, query);
+ List<Id> ids = new ArrayList<>();
+ results.iterator().forEachRemaining(edge -> ids.add(edge.id()));
+ return ids;
+ }
+
+ @Test
+ public void testRamTableHitDoesNotReadBackendBeforeOptimization() {
+ HugeVertex first = this.newVertex(IdGenerator.of(1L));
+ HugeVertex second = this.newVertex(IdGenerator.of(2L));
+ HugeEdge edge = this.newEdge(first, second);
+ RamTable table = new RamTable(this.graph, 16, 16);
+ table.addEdge(true, 1L, 2L, Directions.OUT, (int) edge.schemaLabel().id().asLong());
+ HugeGraphParams params = Mockito.spy(this.params);
+ Mockito.doReturn(table).when(params).ramtable();
+ BackendStore store = Mockito.spy(this.params.loadGraphStore());
+ CachedGraphTransaction transaction = new CachedGraphTransaction(params, store);
+ try {
+ transaction.clearCache(null, false);
+ ConditionQuery query = new ConditionQuery(HugeType.EDGE);
+ query.eq(HugeKeys.OWNER_VERTEX, first.id());
+ query.eq(HugeKeys.DIRECTION, Directions.OUT);
+ query.eq(HugeKeys.LABEL, edge.schemaLabel().id());
+ Mockito.clearInvocations(store);
+ Edge found =
+ QueryResults.one(transaction.queryEdges(query));
+ Assert.assertEquals(edge.id(), found.id());
+ Mockito.verify(store, Mockito.never()).query(Mockito.any(Query.class));
+ } finally {
+ transaction.close();
+ }
+ }
+
+ @Test
+ public void testRejectedUndefinedRecordWarnsOnceForPublicAndRawQueries() {
+ this.newVertex(IdGenerator.of(1L));
+ Id id = IdGenerator.of(999L);
+ Cache<Id, Object> vertices = Whitebox.getInternalState(this.cache, "verticesCache");
+ vertices.update(id, HugeVertex.undefined(this.graph, id));
+ ConditionQuery origin = new ConditionQuery(HugeType.VERTEX);
+ origin.query(Condition.eq(this.graph.propertyKey("name").id(), "absent"));
+ origin.optimized(OptimizedType.PRIMARY_KEY);
+ origin.showHidden(true);
+ String loggerName = "org.apache.hugegraph.backend.tx.AbstractTransaction";
+ LoggerContext logging = (LoggerContext) LogManager.getContext(false);
+ Configuration configuration = logging.getConfiguration();
+ LoggerConfig previous = configuration.getLoggers().get(loggerName);
+ WarningCounter counter = new WarningCounter();
+ counter.start();
+ LoggerConfig logger = new LoggerConfig(loggerName, Level.WARN, false);
+ logger.addAppender(counter, Level.WARN, null);
+ configuration.removeLogger(loggerName);
+ configuration.addLogger(loggerName, logger);
+ logging.updateLoggers();
+ try {
+ Assert.assertFalse(this.cache.queryVertices(new IdQuery(origin, id)).hasNext());
+ Assert.assertEquals(1, counter.count);
+ Iterator<HugeVertex> raw = Whitebox.invoke(
+ CachedGraphTransaction.class, new Class<?>[]{Query.class},
+ "queryVerticesFromBackend", this.cache, new IdQuery(origin, id));
+ Assert.assertFalse(raw.hasNext());
+ Assert.assertEquals(2, counter.count);
+ } finally {
+ configuration.removeLogger(loggerName);
+ if (previous != null) {
+ configuration.addLogger(loggerName, previous);
+ }
+ logging.updateLoggers();
+ counter.stop();
+ }
+ }
+
+ private static final class WarningCounter extends AbstractAppender {
+
+ private int count;
+
+ private WarningCounter() {
+ super("BatchWarningCounter", null, null, false, Property.EMPTY_ARRAY);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ if (event.getMessage().getFormattedMessage().startsWith("Left record is found:")) {
+ this.count++;
+ }
+ }
+ }
+
@Test
public void testPostFilterDefersDeletingLabelToInvalidFilter() {
HugeVertex vertex = this.newVertex(IdGenerator.of(1));
@@ -538,18 +935,16 @@ public class CachedGraphTransactionTest extends BaseUnitTest {
query.query(Condition.eq(name, "marko"));
query.optimized(OptimizedType.INDEX);
- Class<?>[] classes = new Class<?>[]{Iterator.class, Query.class};
- Iterator<HugeVertex> unmatched = Whitebox.invoke(
+ Class<?>[] classes = new Class<?>[]{QueryResultContext.class, HugeElement.class};
+ QueryResultContext context = new QueryResultContext(query);
+ boolean unmatched = Whitebox.invoke(
CachedGraphTransaction.class, classes,
- "filterUnmatchedRecords", this.cache,
- Collections.singleton(vertex).iterator(), query);
- Assert.assertTrue(unmatched.hasNext());
-
- Iterator<HugeVertex> invalid = Whitebox.invoke(
+ "filterUnmatchedRecord", this.cache, context, vertex);
+ Assert.assertTrue(unmatched);
+ boolean invalid = Whitebox.invoke(
CachedGraphTransaction.class, classes,
- "filterInvalidRecords", this.cache,
- Collections.singleton(vertex).iterator(), query);
- Assert.assertFalse(invalid.hasNext());
+ "filterInvalidRecord", this.cache, context, vertex);
+ Assert.assertFalse(invalid);
}
@Test
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/QueryResultsTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/QueryResultsTest.java
index 0ee2cf167..64c675f64 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/QueryResultsTest.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/QueryResultsTest.java
@@ -19,17 +19,23 @@ package org.apache.hugegraph.unit.core;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Iterator;
import java.util.List;
-import java.util.NoSuchElementException;
import java.util.Set;
-import java.util.function.Consumer;
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.page.PageInfo;
+import org.apache.hugegraph.backend.page.PageState;
+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;
+import org.apache.hugegraph.backend.query.QueryResultContext;
import org.apache.hugegraph.backend.query.QueryResults;
+import org.apache.hugegraph.exception.LimitExceedException;
+import org.apache.hugegraph.iterator.CIter;
import org.apache.hugegraph.testutil.Assert;
import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.type.Idfiable;
@@ -40,6 +46,143 @@ import com.google.common.collect.ImmutableList;
public class QueryResultsTest {
+ @Test
+ public void testLaterBranchFailureClosesEarlierBranch() {
+ CountingIterator source = new CountingIterator(1L);
+ RuntimeException failure = new IllegalArgumentException("later branch");
+ int[] activated = {0};
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ ImmutableList.of(0, 1).iterator(), index -> {
+ activated[0]++;
+ if (index == 1) {
+ throw failure;
+ }
+ return new QueryResults<>(source, queryOf(1L));
+ });
+ Assert.assertEquals(0, activated[0]);
+ Iterator<TestIdfiable> values = results.iterator();
+ Assert.assertEquals(IdGenerator.of(1L), values.next().id());
+ Assert.assertEquals(1, activated[0]);
+ try {
+ values.hasNext();
+ Assert.fail("Expected later branch failure");
+ } catch (IllegalArgumentException actual) {
+ Assert.assertSame(failure, actual);
+ }
+ Assert.assertEquals(1, source.closed);
+ Assert.assertEquals(2, activated[0]);
+ Assert.assertFalse(values.hasNext());
+ }
+
+ @Test
+ public void testMaterializationKeepsPageMetadataAfterClosingSource() {
+ PageState page = new PageState(new byte[]{1, 2}, 0, 2);
+ CountingIterator source = new CountingIterator(1L, 2L);
+ CIter<TestIdfiable> paged = new CIter<TestIdfiable>() {
+ @Override
+ public boolean hasNext() {
+ return source.hasNext();
+ }
+
+ @Override
+ public TestIdfiable next() {
+ return source.next();
+ }
+
+ @Override
+ public void close() {
+ source.close();
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ Assert.assertEquals(PageInfo.PAGE, meta);
+ return page;
+ }
+ };
+ QueryResults<TestIdfiable> results = new QueryResults<>(paged, queryOf(1L, 2L));
+ Assert.assertTrue(results.iterator().hasNext());
+ QueryResults<TestIdfiable> fetched = results.toList();
+ Assert.assertEquals(1, source.closed);
+ Assert.assertSame(page, PageInfo.pageState(results.iterator()));
+ Assert.assertSame(page, PageInfo.pageState(fetched.iterator()));
+ List<Id> actual = new ArrayList<>();
+ fetched.iterator().forEachRemaining(item -> actual.add(item.id()));
+ Assert.assertEquals(Arrays.asList(IdGenerator.of(1L), IdGenerator.of(2L)), actual);
+ Assert.assertSame(page, PageInfo.pageState(fetched.iterator()));
+ }
+
+ @Test
+ public void testMaterializeAfterPeekAndPartialConsumption() {
+ for (int consumed = 0; consumed <= 2; consumed++) {
+ CountingIterator first = new CountingIterator(1L, 2L);
+ CountingIterator second = new CountingIterator(3L, 4L);
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ Arrays.asList(0, 1).iterator(), index -> new QueryResults<>(
+ index == 0 ? first : second,
+ index == 0 ? queryOf(1L, 2L) : queryOf(3L, 4L)));
+ Iterator<TestIdfiable> iterator = results.iterator();
+ for (int i = 0; i < consumed; i++) {
+ Assert.assertEquals(IdGenerator.of(i + 1L), iterator.next().id());
+ }
+ Assert.assertTrue(iterator.hasNext());
+ QueryResults<TestIdfiable> fetched = results.toList();
+ List<Id> actual = new ArrayList<>();
+ fetched.filter((context, item) -> {
+ Assert.assertTrue(context.inputIds().contains(item.id()));
+ return true;
+ }).iterator().forEachRemaining(item -> actual.add(item.id()));
+ List<Id> expected = new ArrayList<>();
+ for (long id = consumed + 1L; id <= 4L; id++) {
+ expected.add(IdGenerator.of(id));
+ }
+ Assert.assertEquals(expected, actual);
+ Assert.assertFalse(iterator.hasNext());
+ Assert.assertEquals(1, first.closed);
+ Assert.assertEquals(1, second.closed);
+ }
+ }
+
+ @Test
+ public void testMapAfterPeekKeepsCurrentBatch() {
+ CountingIterator source = new CountingIterator(1L, 2L);
+ QueryResults<TestIdfiable> results = new QueryResults<>(source, queryOf(1L, 2L));
+ Assert.assertTrue(results.iterator().hasNext());
+ List<Id> actual = new ArrayList<>();
+ results.map(TestIdfiable::id).iterator().forEachRemaining(actual::add);
+ Assert.assertEquals(Arrays.asList(IdGenerator.of(1L), IdGenerator.of(2L)), actual);
+ Assert.assertEquals(1, source.closed);
+ }
+
+ @Test
+ public void testBatchContextAndSourceClose() throws Exception {
+ IdQuery query = queryOf(1L, 2L);
+ CountingIterator source = new CountingIterator(1L, 2L);
+ QueryResults<TestIdfiable> results = new QueryResults<>(source, query);
+ Iterator<QueryBatch<TestIdfiable>> batches = results.batches();
+ Assert.assertTrue(batches.hasNext());
+ Assert.assertTrue(batches.hasNext());
+ QueryBatch<TestIdfiable> batch = batches.next();
+ Assert.assertSame(query, batch.context().queries().get(0));
+ Assert.assertEquals(2, batch.context().inputIds().size());
+ Assert.assertTrue(batch.results().hasNext());
+ ((AutoCloseable) batches).close();
+ Assert.assertEquals(1, source.closed);
+ Assert.assertFalse(batch.results().hasNext());
+ Assert.assertFalse(batches.hasNext());
+ }
+
+ @Test
+ public void testMaterializationCapacityAppliesAcrossBatches() {
+ Query query = new Query(HugeType.VERTEX);
+ List<Integer> values = Collections.nCopies((int) Query.DEFAULT_CAPACITY / 2 + 1, 1);
+ QueryResults<Integer> results = QueryResults.flatMap(Arrays.asList(1, 2).iterator(),
+ ignored -> new QueryResults<>(values.iterator(), query));
+ Assert.assertThrows(LimitExceedException.class, () -> {
+ results.toList();
+ });
+ }
+
@Test
public void testKeepInputOrderForPagingIdQuery() {
Id id1 = IdGenerator.of(1L);
@@ -60,9 +203,7 @@ public class QueryResultsTest {
idQuery);
List<Id> orderedIds = new ArrayList<>();
- results.keepInputOrderIfNeeded(
- Arrays.asList(new TestIdfiable(id1),
- new TestIdfiable(id2)).iterator())
+ results.<TestIdfiable>keepInputOrderIfNeeded().iterator()
.forEachRemaining(item -> orderedIds.add(item.id()));
Assert.assertEquals(ImmutableList.of(id2, id1), orderedIds);
@@ -88,7 +229,7 @@ public class QueryResultsTest {
ImmutableList.of(first, second).iterator(), result -> result);
List<Id> orderedIds = new ArrayList<>();
- results.keepInputOrderIfNeeded(results.iterator())
+ results.<TestIdfiable>keepInputOrderIfNeeded().iterator()
.forEachRemaining(item -> orderedIds.add(item.id()));
List<Id> expected = new ArrayList<>();
@@ -105,7 +246,7 @@ public class QueryResultsTest {
ImmutableList.of(1L, 2L, 3L));
List<Id> orderedIds = new ArrayList<>();
- results.keepInputOrderIfNeeded(results.iterator())
+ results.<TestIdfiable>keepInputOrderIfNeeded().iterator()
.forEachRemaining(item -> orderedIds.add(item.id()));
Assert.assertEquals(ImmutableList.of(IdGenerator.of(1L),
@@ -116,29 +257,224 @@ public class QueryResultsTest {
@Test
public void testKeepInputOrderDoesNotDrainFollowingPages() {
- IdQuery firstQuery = queryOf(2L, 1L);
- IdQuery secondQuery = queryOf(4L, 3L);
- @SuppressWarnings("unchecked")
- QueryResults<TestIdfiable>[] holder = new QueryResults[1];
- PagingIterator origin = new PagingIterator(
- ImmutableList.of(new TestIdfiable(IdGenerator.of(1L)),
- new TestIdfiable(IdGenerator.of(2L)),
- new TestIdfiable(IdGenerator.of(3L)),
- new TestIdfiable(IdGenerator.of(4L))),
- 2,
- query -> holder[0].setQuery(query),
- ImmutableList.of(firstQuery, secondQuery));
- holder[0] = new QueryResults<>(origin, firstQuery);
-
- Iterator<TestIdfiable> ordered =
- holder[0].keepInputOrderIfNeeded(holder[0].iterator());
-
+ CountingIterator first = new CountingIterator(1L, 2L);
+ CountingIterator second = new CountingIterator(3L, 4L);
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ ImmutableList.of(0, 1).iterator(), index -> index == 0 ?
+ new QueryResults<>(first, queryOf(2L, 1L)) :
+ new QueryResults<>(second, queryOf(4L, 3L)));
+ Iterator<TestIdfiable> ordered = results.<TestIdfiable>keepInputOrderIfNeeded().iterator();
Assert.assertEquals(IdGenerator.of(2L), ordered.next().id());
Assert.assertEquals(IdGenerator.of(1L), ordered.next().id());
- Assert.assertEquals(2, origin.consumed());
+ Assert.assertEquals(2, first.consumed);
+ Assert.assertEquals(0, second.consumed);
Assert.assertEquals(IdGenerator.of(4L), ordered.next().id());
Assert.assertEquals(IdGenerator.of(3L), ordered.next().id());
Assert.assertFalse(ordered.hasNext());
+ Assert.assertEquals(1, first.closed);
+ Assert.assertEquals(1, second.closed);
+ }
+
+ @Test
+ public void testOrderingRequirementIsLocalToEachBatch() {
+ IdQuery firstQuery = queryOf(1L);
+ firstQuery.mustSortByInput(false);
+ QueryResults<TestIdfiable> first = new QueryResults<>(
+ new CountingIterator(1L), firstQuery);
+ QueryResults<TestIdfiable> second = resultsOf(
+ ImmutableList.of(3L, 2L), ImmutableList.of(2L, 3L));
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ ImmutableList.of(first, second).iterator(), result -> result);
+ List<Id> ids = new ArrayList<>();
+ results.<TestIdfiable>keepInputOrderIfNeeded().iterator()
+ .forEachRemaining(item -> ids.add(item.id()));
+ Assert.assertEquals(ImmutableList.of(IdGenerator.of(1L),
+ IdGenerator.of(3L),
+ IdGenerator.of(2L)), ids);
+ }
+
+ @Test
+ public void testOrderingDoesNotActivateFollowingQuery() {
+ int[] fetches = {0};
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ ImmutableList.of(0, 1).iterator(), index -> {
+ fetches[0]++;
+ return resultsOf(ImmutableList.of(2L, 1L),
+ ImmutableList.of(1L, 2L));
+ });
+ Iterator<TestIdfiable> ordered =
+ results.<TestIdfiable>keepInputOrderIfNeeded().iterator();
+ Assert.assertEquals(IdGenerator.of(2L), ordered.next().id());
+ Assert.assertEquals(1, fetches[0]);
+ Assert.assertEquals(IdGenerator.of(1L), ordered.next().id());
+ Assert.assertEquals(1, fetches[0]);
+ Assert.assertTrue(ordered.hasNext());
+ Assert.assertEquals(2, fetches[0]);
+ }
+
+ @Test
+ public void testNullAndExpandedResultsKeepTheirBatchContext() {
+ int[] activated = {0};
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ ImmutableList.of(0, 1).iterator(), index -> {
+ activated[0]++;
+ return resultsOf(ImmutableList.of((long) index + 1),
+ ImmutableList.of((long) index + 1));
+ });
+ QueryResults<TestIdfiable> expanded = results.flatMap(item ->
+ Arrays.asList(null, item, item).iterator());
+ Iterator<TestIdfiable> values = expanded.filter((context, item) -> {
+ Assert.assertTrue(context.inputIds().contains(item.id()));
+ Assert.assertEquals(item.id().asLong(), (long) activated[0]);
+ return true;
+ }).iterator();
+ Assert.assertEquals(IdGenerator.of(1L), values.next().id());
+ Assert.assertEquals(IdGenerator.of(1L), values.next().id());
+ Assert.assertEquals(1, activated[0]);
+ Assert.assertEquals(IdGenerator.of(2L), values.next().id());
+ Assert.assertEquals(IdGenerator.of(2L), values.next().id());
+ Assert.assertFalse(values.hasNext());
+ }
+
+ @Test
+ public void testCloseDoesNotActivateRemainingBatch() throws Exception {
+ CountingIterator source = new CountingIterator(1L, 2L);
+ int[] activated = {0};
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ ImmutableList.of(0, 1).iterator(), index -> {
+ activated[0]++;
+ return new QueryResults<>(source, queryOf(1L, 2L));
+ });
+ Iterator<TestIdfiable> values = results.map(item -> item).iterator();
+ Assert.assertTrue(values.hasNext());
+ ((AutoCloseable) values).close();
+ ((AutoCloseable) values).close();
+ Assert.assertFalse(values.hasNext());
+ Assert.assertEquals(1, activated[0]);
+ Assert.assertEquals(1, source.closed);
+ }
+
+ @Test
+ public void testSourceFailuresCloseMappedResultsOnce() throws Exception {
+ for (boolean failHasNext : new boolean[]{true, false}) {
+ CountingIterator source = new CountingIterator(1L);
+ RuntimeException failure = new IllegalStateException("source");
+ if (failHasNext) {
+ source.hasNextFailure = failure;
+ } else {
+ source.nextFailure = failure;
+ }
+ Iterator<TestIdfiable> values = new QueryResults<>(source, queryOf(1L))
+ .map(item -> item).iterator();
+ try {
+ values.hasNext();
+ Assert.fail("Expected source failure");
+ } catch (IllegalStateException actual) {
+ Assert.assertSame(failure, actual);
+ }
+ Assert.assertEquals(1, source.closed);
+ Assert.assertFalse(values.hasNext());
+ ((AutoCloseable) values).close();
+ Assert.assertEquals(1, source.closed);
+ }
+ }
+
+ @Test
+ public void testCloseBeforeMappedSourceActivation() throws Exception {
+ CountingIterator source = new CountingIterator(1L);
+ source.hasNextFailure = new IllegalStateException("Source must not be probed");
+ Iterator<TestIdfiable> values = new QueryResults<>(source, queryOf(1L))
+ .map(item -> item).iterator();
+ ((AutoCloseable) values).close();
+ ((AutoCloseable) values).close();
+ Assert.assertEquals(0, source.consumed);
+ Assert.assertEquals(1, source.closed);
+ Assert.assertFalse(values.hasNext());
+ }
+
+ @Test
+ public void testMapperExceptionPreservesCloseFailure() {
+ CountingIterator source = new CountingIterator(1L);
+ RuntimeException failure = new IllegalArgumentException("mapper");
+ source.closeFailure = new IllegalStateException("close");
+ QueryResults<TestIdfiable> results = new QueryResults<>(source, queryOf(1L));
+ Iterator<TestIdfiable> values = results.<TestIdfiable>map(item -> {
+ throw failure;
+ }).iterator();
+ try {
+ values.hasNext();
+ Assert.fail("Expected mapper failure");
+ } catch (IllegalArgumentException actual) {
+ Assert.assertSame(failure, actual);
+ Assert.assertArrayEquals(new Throwable[]{source.closeFailure}, actual.getSuppressed());
+ }
+ Assert.assertEquals(1, source.closed);
+ }
+
+ @Test
+ public void testEmptyBatchClosesBeforeNextSupplier() {
+ CountingIterator empty = new CountingIterator();
+ CountingIterator source = new CountingIterator(1L);
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ ImmutableList.of(0, 1).iterator(), index -> {
+ if (index == 1) {
+ Assert.assertEquals(1, empty.closed);
+ }
+ return new QueryResults<>(index == 0 ? empty : source, queryOf(1L));
+ });
+ Assert.assertEquals(IdGenerator.of(1L), results.one().id());
+ Assert.assertEquals(1, empty.closed);
+ Assert.assertEquals(1, source.closed);
+ }
+
+ @Test
+ public void testExpandedChildClosesExactlyOnceOnEarlyExit() throws Exception {
+ CountingIterator source = new CountingIterator(1L, 2L);
+ CountingIterator child = new CountingIterator(3L, 4L);
+ Iterator<TestIdfiable> values = new QueryResults<>(source, queryOf(1L, 2L))
+ .flatMap(item -> child).iterator();
+ Assert.assertEquals(IdGenerator.of(3L), values.next().id());
+ ((AutoCloseable) values).close();
+ ((AutoCloseable) values).close();
+ Assert.assertEquals(1, source.closed);
+ Assert.assertEquals(1, child.closed);
+ Assert.assertEquals(1, source.consumed);
+ Assert.assertEquals(1, child.consumed);
+ }
+
+ @Test
+ public void testOrderedBatchThenUnorderedBatchPreservesBackendOrder() {
+ QueryResults<TestIdfiable> first = resultsOf(
+ ImmutableList.of(2L, 1L), ImmutableList.of(1L, 2L));
+ IdQuery secondQuery = queryOf(4L, 3L);
+ secondQuery.mustSortByInput(false);
+ QueryResults<TestIdfiable> second = new QueryResults<>(
+ new CountingIterator(3L, 4L), secondQuery);
+ QueryResults<TestIdfiable> results = QueryResults.flatMap(
+ ImmutableList.of(first, second).iterator(), result -> result);
+ List<Id> ids = new ArrayList<>();
+ results.<TestIdfiable>keepInputOrderIfNeeded().iterator()
+ .forEachRemaining(item -> ids.add(item.id()));
+ Assert.assertEquals(Arrays.asList(IdGenerator.of(2L), IdGenerator.of(1L),
+ IdGenerator.of(3L), IdGenerator.of(4L)), ids);
+ }
+
+ @Test
+ public void testContextFindsFilterAboveCopiedQueries() {
+ ConditionQuery root = new ConditionQuery(HugeType.VERTEX);
+ ConditionQuery search = root.copy();
+ ConditionQuery.ResultsFilter filter = element -> true;
+ search.registerResultsFilter(filter);
+ ConditionQuery copied = search.copy();
+ QueryResultContext context = new QueryResultContext(
+ new IdQuery(copied, IdGenerator.of(1L)));
+ Assert.assertSame(filter, context.resultsFilter());
+ Assert.assertSame(root, context.matchQuery());
+ ConditionQuery other = root.copy();
+ other.registerResultsFilter(element -> false);
+ Assert.assertSame(filter, context.resultsFilter());
+ Assert.assertNotSame(context.resultsFilter(),
+ new QueryResultContext(other).resultsFilter());
}
private static QueryResults<TestIdfiable> resultsOf(List<Long> input,
@@ -178,51 +514,43 @@ public class QueryResultsTest {
}
}
- private static final class PagingIterator
- implements Iterator<TestIdfiable> {
+ private static final class CountingIterator
+ implements Iterator<TestIdfiable>, AutoCloseable {
- private final List<TestIdfiable> results;
- private final int pageSize;
- private final Consumer<IdQuery> pageListener;
- private final List<IdQuery> queries;
+ private final Iterator<Long> values;
+ private int consumed;
+ private int closed;
+ private RuntimeException closeFailure;
+ private RuntimeException hasNextFailure;
+ private RuntimeException nextFailure;
- private int current;
- private int announcedPage;
-
- private PagingIterator(List<TestIdfiable> results, int pageSize,
- Consumer<IdQuery> pageListener,
- List<IdQuery> queries) {
- this.results = results;
- this.pageSize = pageSize;
- this.pageListener = pageListener;
- this.queries = queries;
- this.current = 0;
- this.announcedPage = 0;
+ private CountingIterator(Long... values) {
+ this.values = Arrays.asList(values).iterator();
}
@Override
public boolean hasNext() {
- if (this.current >= this.results.size()) {
- return false;
- }
- int page = this.current / this.pageSize;
- if (page != this.announcedPage) {
- this.pageListener.accept(this.queries.get(page));
- this.announcedPage = page;
+ if (this.hasNextFailure != null) {
+ throw this.hasNextFailure;
}
- return true;
+ return this.values.hasNext();
}
@Override
public TestIdfiable next() {
- if (!this.hasNext()) {
- throw new NoSuchElementException();
+ if (this.nextFailure != null) {
+ throw this.nextFailure;
}
- return this.results.get(this.current++);
+ this.consumed++;
+ return new TestIdfiable(IdGenerator.of(this.values.next()));
}
- private int consumed() {
- return this.current;
+ @Override
+ public void close() {
+ this.closed++;
+ if (this.closeFailure != null) {
+ throw this.closeFailure;
+ }
}
}
}