bitflicker64 commented on code in PR #3193:
URL: https://github.com/apache/hugegraph/pull/3193#discussion_r3972183962
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResults.java:
##########
@@ -18,234 +18,362 @@
package org.apache.hugegraph.backend.query;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
-import java.util.Set;
+import java.util.function.BiPredicate;
import java.util.function.Function;
import org.apache.hugegraph.HugeException;
import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.query.QueryBatch.BatchIterator;
import org.apache.hugegraph.iterator.CIter;
-import org.apache.hugegraph.iterator.FlatMapperIterator;
import org.apache.hugegraph.iterator.ListIterator;
-import org.apache.hugegraph.iterator.MapperIterator;
-import org.apache.hugegraph.iterator.WrappedIterator;
import org.apache.hugegraph.perf.PerfUtil.Watched;
import org.apache.hugegraph.type.Idfiable;
-import org.apache.hugegraph.util.E;
-import org.apache.hugegraph.util.InsertionOrderUtil;
-import org.apache.tinkerpop.gremlin.structure.util.CloseableIterator;
+/** A lazy stream of query batches. Only the final consumer flattens the
stream. */
public class QueryResults<R> {
- private static final Iterator<?> EMPTY_ITERATOR = new EmptyIterator<>();
-
- private static final QueryResults<?> EMPTY = new QueryResults<>(
- emptyIterator(), Query.NONE);
-
- private final Iterator<R> results;
+ private final BatchIterator<QueryBatch<R>> batches;
private final List<Query> queries;
- private List<Query> currentQueries;
- private long queryVersion;
+ private final Object metadata;
+ private Iterator<R> results;
public QueryResults(Iterator<R> results, Query query) {
- this(results);
- this.addQuery(query);
+ this(results, new QueryResultContext(query));
}
- private QueryResults(Iterator<R> results) {
- this.results = results;
- this.queries = InsertionOrderUtil.newList();
- this.currentQueries = Collections.emptyList();
- this.queryVersion = 0L;
+ public QueryResults(Iterator<R> results, QueryResultContext context) {
+ QueryBatch<R> batch = new QueryBatch<>(results, context);
+ this.queries = new
ArrayList<>(Collections.singletonList(context.queries().get(0)));
+ this.batches = this.trackBatches(new BatchIterator<QueryBatch<R>>() {
+ private boolean fetched;
+
+ @Override
+ protected QueryBatch<R> fetch() {
+ if (this.fetched) {
+ return null;
+ }
+ this.fetched = true;
+ return batch;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ batch.close();
+ }
+ });
+ this.metadata = batch.results();
}
- public void setQuery(Query query) {
- if (!this.queries.isEmpty()) {
- this.queries.clear();
- }
- this.addQuery(query);
+ private QueryResults(Iterator<QueryBatch<R>> batches, Object metadata) {
+ this.queries = new ArrayList<>();
+ this.batches = this.trackBatches(batches);
+ this.metadata = metadata;
}
- private void addQuery(Query query) {
- E.checkNotNull(query, "query");
- this.addQueries(Collections.singletonList(query));
+ private BatchIterator<QueryBatch<R>> trackBatches(Iterator<QueryBatch<R>>
origin) {
+ return new BatchIterator<QueryBatch<R>>() {
+ private QueryBatch<R> active;
+
+ @Override
+ protected QueryBatch<R> fetch() throws Exception {
+ QueryBatch<R> previous = this.active;
+ this.active = null;
+ QueryBatch.closeAll(previous);
+ if (!origin.hasNext()) {
+ return null;
+ }
+ this.active = origin.next();
+ queries.clear();
+ queries.add(this.active.context().queries().get(0));
+ return this.active;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ QueryBatch<R> previous = this.active;
+ this.active = null;
+ queries.clear();
+ QueryBatch.closeAll(previous, origin);
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return QueryBatch.metadataOf(origin, meta, args);
+ }
+ };
}
- private void addQueries(List<Query> queries) {
- assert !queries.isEmpty();
- for (Query query : queries) {
- E.checkNotNull(query, "query");
- this.queries.add(query);
- }
- this.currentQueries = new ArrayList<>(queries);
- this.queryVersion++;
+ public static <R> QueryResults<R> fromBatches(Iterator<QueryBatch<R>>
batches) {
+ return new QueryResults<>(batches, batches);
+ }
+
+ public Iterator<QueryBatch<R>> batches() {
+ return this.batches;
}
public Iterator<R> iterator() {
+ if (this.results == null) {
+ this.results = new CIter<R>() {
+ private boolean closed;
+
+ @Override
+ public boolean hasNext() {
+ if (this.closed) {
+ return false;
+ }
+ try {
+ while (batches.hasNext()) {
+ // Leave the batch and its prefetched element in
the shared cursor.
+ if (batches.peek().results().hasNext()) {
+ return true;
+ }
+ batches.next().close();
+ }
+ this.close();
+ return false;
+ } catch (Throwable failure) {
+ QueryResults.close(this, failure);
+ throw QueryBatch.propagate(failure);
+ }
+ }
+
+ @Override
+ public R next() {
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ }
+ try {
+ return batches.peek().results().next();
+ } catch (Throwable failure) {
+ QueryResults.close(this, failure);
+ throw QueryBatch.propagate(failure);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (!this.closed) {
+ this.closed = true;
+ batches.close();
+ }
+ }
+
+ @Override
+ public Object metadata(String meta, Object... args) {
+ return QueryBatch.metadataOf(metadata, meta, args);
+ }
+ };
+ }
return this.results;
}
- public R one() {
- return one(this.results);
+ public <T> QueryResults<T> mapBatches(Function<QueryBatch<R>,
QueryBatch<T>> mapper) {
+ Iterator<QueryBatch<R>> origin = this.batches;
+ return new QueryResults<>(new BatchIterator<QueryBatch<T>>() {
+ private QueryBatch<?> active;
+
+ @Override
+ protected QueryBatch<T> fetch() throws Exception {
+ QueryBatch.closeAll(this.active);
+ this.active = null;
+ if (!origin.hasNext()) {
+ return null;
+ }
+ QueryBatch<R> batch = origin.next();
+ this.active = batch;
+ QueryBatch<T> mapped = mapper.apply(batch);
+ this.active = mapped;
+ return mapped;
+ }
+
+ @Override
+ protected void closeResources() throws Exception {
+ QueryBatch.closeAll(this.active, origin);
+ }
+ }, this.metadata);
}
- public QueryResults<R> toList() {
- QueryResults<R> fetched = new QueryResults<>(toList(this.results));
- fetched.addQueries(this.queries);
- return fetched;
+ public <T> QueryResults<T> map(Function<R, T> mapper) {
+ return this.mapBatches(batch -> batch.map(mapper));
}
- public List<Query> queries() {
- return Collections.unmodifiableList(this.queries);
+ public <T> QueryResults<T> flatMap(Function<R, Iterator<T>> mapper) {
+ return this.mapBatches(batch -> batch.flatMap(mapper));
}
- public <T extends Idfiable> Iterator<T> keepInputOrderIfNeeded(
- Iterator<T> origin) {
- if (!origin.hasNext()) {
- // None result found
- return origin;
- }
- if (!mustSortByInputIds(this.currentQueries)) {
- return origin;
- }
- return new InputOrderIterator<>(this, origin);
+ public QueryResults<R> filter(BiPredicate<QueryResultContext, R>
predicate) {
+ return this.mapBatches(batch -> batch.filter(predicate));
}
- private static boolean mustSortByInputIds(List<Query> queries) {
- assert !queries.isEmpty() : queries;
- for (Query query : queries) {
- if (query instanceof IdQuery &&
- ((IdQuery) query).mustSortByInput()) {
- return true;
- }
- }
- return false;
+ public <T extends Idfiable> QueryResults<T> keepInputOrderIfNeeded() {
+ return this.mapBatches(QueryBatch::keepInputOrder);
}
- @SuppressWarnings("unused")
- private boolean bigCapacity() {
- assert !this.queries.isEmpty();
- for (Query query : this.queries) {
- if (query.bigCapacity()) {
- return true;
+ public R one() {
+ return one(this.iterator());
+ }
+
+ /**
+ * Source query of the current batch. A known single source is available
before
+ * activation; composed streams start empty. Closing clears the
diagnostics.
+ */
+ public List<Query> queries() {
Review Comment:
🧹 `queries()` has no main-code caller left after this change.
At base it had exactly one: `QueryList.PageResults.query()` (base
`QueryList.java:347`), which `cf403002` deleted. Ordering now comes from
`QueryResultContext`, and base `PageEntryIterator.java:79`
(`queryResults.setQuery(pageResults.query())`) is gone. Over this head, `git
grep '\.queries()'` finds only `context.queries()` calls plus assertions in
`QueryResultsTest.java:161,164,166` and `QueryListTest.java:129,131,135`.
The bookkeeping still runs on the hot path: an `ArrayList` per
`QueryResults` (line 51), `queries.clear()` plus `queries.add(...)` on every
batch activation (lines 91-92), a clear on close (line 100), and a re-seed in
`toList()` (line 250). Same shape as the dead `PageResults.query()`/`get()`
pair you removed last round.
Requested change: drop the `queries` field and this accessor, and rework the
assertions onto `results.batches()` where the test structure allows. Note that
`java.util.Collections` (line 21) is then unused, since lines 51 and 226 are
its only uses. If it should stay, name its reader in the Javadoc at lines
221-224, because no main code reads it today.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java:
##########
@@ -316,135 +318,149 @@ private boolean needCacheVertex(HugeVertex vertex) {
@Override
@Watched(prefix = "graphcache")
- protected Iterator<HugeVertex> queryVerticesFromBackend(Query query) {
- if (this.enableCacheVertex() &&
- query.idsSize() > 0 && query.conditionsSize() == 0 &&
- !queryNeedsPostFilter(query)) {
- return this.queryVerticesByIds((IdQuery) query);
- } else {
- return super.queryVerticesFromBackend(query);
+ protected QueryResults<HugeVertex> fetchVertexBatch(Query query) {
+ if (!this.enableCacheVertex() || query.paging() ||
Review Comment:
🧹 This `paging()` test is on the leaf, while `fetchEdgeBatch` tests the
chain root.
Separate from my earlier thread that asked for this guard: now that it
exists, the two overrides disagree about what "paged" means. Line 322 tests
`query.paging()` on the `QueryList` fetcher's leaf; `fetchEdgeBatch` builds the
same context and tests `request.paging()` on `chain.get(chain.size() - 1)`, the
chain root (lines 369-372).
The difference is observable. `QueryList.IndexQuery.iterator(int, String,
long)` builds its leaf through `indexIdQuery` (`QueryList.java:326-332`) as
`new IdQuery(parent().resultType(), bindQuery)`, and `Query(HugeType, Query)`
sets `this.page = null` (`Query.java:95`), so that leaf reaches line 322 with
`paging() == false`, `idsSize() > 0` and `conditionsSize() == 0`. The vertex
cache therefore still runs inside a paged index query, and is correct only for
the reason I traced on the earlier thread: that path takes its cursor from
`pageIds.pageState()`, not from result metadata.
Requested change: state in a comment that the leaf test is deliberate and
that the index-paging leaf is safe because its cursor comes from the
`IdHolder`. Matching `fetchEdgeBatch` and testing the root instead would also
work, but it bypasses the vertex cache for every paged index query, so it is
the more expensive option.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/page/QueryList.java:
##########
@@ -25,15 +25,17 @@
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.page.IdHolder.BatchIdHolder;
import org.apache.hugegraph.backend.page.IdHolder.FixedIdHolder;
-import org.apache.hugegraph.backend.query.ConditionQuery;
import org.apache.hugegraph.backend.query.ConditionQuery.OptimizedType;
+import org.apache.hugegraph.backend.query.ConditionQuery;
import org.apache.hugegraph.backend.query.IdQuery;
import org.apache.hugegraph.backend.query.Query;
+import org.apache.hugegraph.backend.query.QueryBatch.BatchIterator;
+import org.apache.hugegraph.backend.query.QueryBatch;
Review Comment:
🧹 These two import pairs kept the ordering that `9ad4a27e` reverted in the
other two files.
This PR moved `ConditionQuery` below `ConditionQuery.OptimizedType` (lines
28-29) and added `QueryBatch.BatchIterator` above `QueryBatch` (lines 32-33).
`9ad4a27e` restored outer-class-first for exactly these two patterns elsewhere:
`Aggregate`/`Aggregate.AggregateFunc` and
`ConditionQuery`/`ConditionQuery.OptimizedType` in `GraphTransaction.java`, and
`QueryBatch`/`QueryBatch.BatchIterator` in `PageEntryIterator.java`. This file
was missed, and nothing flags it because `style/checkstyle.xml:43` has
`ImportOrder` commented out.
Requested change: swap both pairs so this file matches the ordering the same
commit restored in the other two.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
