From b56ae740d265824933c35f204c48382d077a0b23 Mon Sep 17 00:00:00 2001 From: Balmukund Trivedi Date: Tue, 11 Aug 2026 14:49:22 -0700 Subject: [PATCH 1/2] Cache a subquery result set only when it is complete (#4931) SubqueryIterator cached its collected ids on close whenever the element iterator had no further elements. That is true both when the index ran out of results and when the limit stopped the stream, so a truncated prefix of the subquery results was stored as if it were the whole answer. The limit is not part of the cache key of such a subquery. JointIndexQuery propagates a new limit to its subqueries only when it holds a single one, so for a joint query the first subquery keeps the NO_LIMIT it was built with, and two graph queries which differ only in their limit produce the same key. A later query in the same transaction asking for more results was therefore served the shorter list. Count the elements the limit let through, and cache only when fewer than the limit were emitted, which means the limit never stopped anything. A limit equal to the number of results is not cached either, because the index running out and the limit being reached cannot be told apart at that point. That costs a repeated index call, where caching a possibly short result set costs missing results. Signed-off-by: Balmukund Trivedi Co-Authored-By: Claude Opus 5 (1M context) --- .../graphdb/util/SubqueryIterator.java | 16 ++- .../util/SubqueryIteratorCacheTest.java | 115 ++++++++++++++++++ 2 files changed, 129 insertions(+), 2 deletions(-) create mode 100644 janusgraph-core/src/test/java/org/janusgraph/graphdb/util/SubqueryIteratorCacheTest.java diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/SubqueryIterator.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/SubqueryIterator.java index ea27a4f274..e3d0621ab3 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/SubqueryIterator.java +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/SubqueryIterator.java @@ -52,6 +52,10 @@ public class SubqueryIterator extends CloseableAbstractIterator function, List otherResults) { this.subQuery = subQuery; this.indexCache = indexCache; + this.limit = limit; final List cacheResponse = indexCache.getIfPresent(subQuery); final Stream stream; if (cacheResponse != null) { @@ -84,6 +89,7 @@ public SubqueryIterator(JointIndexQuery.Subquery subQuery, IndexSerializer index }) .filter(r -> r != null) // ignore invalid elements .limit(limit) + .peek(r -> emittedCount++) .iterator(); } @@ -98,13 +104,19 @@ protected JanusGraphElement computeNext() { /** * Close the iterator, stop timer and update profiler. - * Put results into cache if the underlying elementIterator is exhausted. + * Put results into cache only if the subquery results are complete, which means the index was read until it ran + * out of results rather than until the limit was reached. */ @Override public void close() { if (isTimerRunning) { assert currentIds != null; - if (!elementIterator.hasNext()) { + //Reaching the limit stops the index from being read any further, so currentIds holds a prefix of the + //subquery results rather than all of them. The cache key of a subquery of a joint query does not include + //the limit, because updateLimit only propagates the limit when there is a single subquery, so caching a + //prefix would serve too few results to a later query with a larger limit. Fewer emitted elements than the + //limit means the limit never stopped anything, so the results are complete + if (!elementIterator.hasNext() && emittedCount < limit) { indexCache.put(subQuery, currentIds); } profiler.setResultSize(currentIds.size()); diff --git a/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/SubqueryIteratorCacheTest.java b/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/SubqueryIteratorCacheTest.java new file mode 100644 index 0000000000..b04fb81ab1 --- /dev/null +++ b/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/SubqueryIteratorCacheTest.java @@ -0,0 +1,115 @@ +// Copyright 2026 JanusGraph Authors +// +// Licensed 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.janusgraph.graphdb.util; + +import org.janusgraph.core.JanusGraphElement; +import org.janusgraph.diskstorage.BackendTransaction; +import org.janusgraph.graphdb.database.IndexSerializer; +import org.janusgraph.graphdb.query.Query; +import org.janusgraph.graphdb.query.graph.JointIndexQuery; +import org.janusgraph.graphdb.query.profile.QueryProfiler; +import org.janusgraph.graphdb.transaction.StandardJanusGraphTx; +import org.janusgraph.graphdb.transaction.subquerycache.SubqueryCache; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +//The subquery cache is consulted for the results of one index of a joint query. A joint query does not propagate its +//limit to the subqueries when there is more than one of them, so the limit is not part of the cache key: a result set +//which the limit truncated must not be stored, or a later query in the same transaction asking for more results is +//served the shorter list. +public class SubqueryIteratorCacheTest { + + private static final List ALL_MATCHING_IDS = Arrays.asList(1L, 2L, 3L, 4L, 5L); + + private final JointIndexQuery.Subquery subQuery = mock(JointIndexQuery.Subquery.class); + private final SubqueryCache indexCache = mock(SubqueryCache.class); + + private void runQuery(int limit) { + when(subQuery.getProfiler()).thenReturn(QueryProfiler.NO_OP); + final IndexSerializer indexSerializer = mock(IndexSerializer.class); + when(indexSerializer.query(any(), any(), any())).thenReturn(ALL_MATCHING_IDS.stream()); + //A mock returns an empty List rather than null, which would look like a cache hit holding no results + when(indexCache.getIfPresent(any())).thenReturn(null); + + try (SubqueryIterator iterator = new SubqueryIterator(subQuery, indexSerializer, + mock(BackendTransaction.class), mock(StandardJanusGraphTx.class), indexCache, limit, + id -> mock(JanusGraphElement.class), null)) { + iterator.forEachRemaining(element -> { }); + } + } + + @Test + public void shouldNotCacheAResultSetTruncatedByTheLimit() { + runQuery(2); + verify(indexCache, never()).put(any(), any()); + } + + @Test + public void shouldCacheAResultSetWhoseIndexRanOutOfResults() { + runQuery(ALL_MATCHING_IDS.size() + 1); + + final ArgumentCaptor> cached = ArgumentCaptor.forClass(List.class); + verify(indexCache, times(1)).put(any(), cached.capture()); + assertEquals(ALL_MATCHING_IDS, cached.getValue()); + } + + @Test + public void shouldCacheAResultSetWhenThereIsNoLimit() { + runQuery(Query.NO_LIMIT); + + final ArgumentCaptor> cached = ArgumentCaptor.forClass(List.class); + verify(indexCache, times(1)).put(any(), cached.capture()); + assertEquals(ALL_MATCHING_IDS, cached.getValue()); + } + + @Test + public void shouldNotCacheWhenTheLimitIsExactlyTheNumberOfResults() { + //The index ran out at the same moment the limit was reached, so which of the two stopped the read is unknown. + //Declining to cache costs a repeated index call; caching a set which may be short costs missing results + runQuery(ALL_MATCHING_IDS.size()); + verify(indexCache, never()).put(any(), any()); + } + + @Test + public void shouldCacheAnEmptyResultSet() { + when(subQuery.getProfiler()).thenReturn(QueryProfiler.NO_OP); + final IndexSerializer indexSerializer = mock(IndexSerializer.class); + when(indexSerializer.query(any(), any(), any())).thenReturn(Collections.emptyList().stream()); + when(indexCache.getIfPresent(any())).thenReturn(null); + + try (SubqueryIterator iterator = new SubqueryIterator(subQuery, indexSerializer, + mock(BackendTransaction.class), mock(StandardJanusGraphTx.class), indexCache, 10, + id -> mock(JanusGraphElement.class), null)) { + iterator.forEachRemaining(element -> { }); + } + + //An index which matched nothing is a complete answer, and worth caching + final ArgumentCaptor> cached = ArgumentCaptor.forClass(List.class); + verify(indexCache, times(1)).put(any(), cached.capture()); + assertEquals(Collections.emptyList(), cached.getValue()); + } +} From d58379e755751ba429765a58330b16485a46796e Mon Sep 17 00:00:00 2001 From: Balmukund Trivedi Date: Wed, 12 Aug 2026 09:36:05 -0700 Subject: [PATCH 2/2] Cache a truncated subquery result set when the subquery carries the read limit The previous commit declined to cache any result set which the read limit truncated. That is wider than necessary. SubqueryCache stores a result list against the limit of the subquery which produced it, and serves that list only to a later query whose limit is no larger, so a truncated list is safe to store while the subquery carries the limit which truncated it. JointIndexQuery.updateLimit propagates the joint limit into the subquery only when there is a single subquery, so that condition holds there. It does not hold once a joint query has more than one subquery: the cache then records the wider subquery limit for a list read under the narrower joint limit, and serves too few results to a later query whose limit falls between the two. That is the case worth declining. Signed-off-by: Balmukund Trivedi Co-Authored-By: Claude Opus 5 (1M context) --- .../graphdb/util/SubqueryIterator.java | 23 ++++--- .../util/SubqueryIteratorCacheTest.java | 67 ++++++++++++------- 2 files changed, 58 insertions(+), 32 deletions(-) diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/SubqueryIterator.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/SubqueryIterator.java index e3d0621ab3..171b567298 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/SubqueryIterator.java +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/SubqueryIterator.java @@ -104,19 +104,13 @@ protected JanusGraphElement computeNext() { /** * Close the iterator, stop timer and update profiler. - * Put results into cache only if the subquery results are complete, which means the index was read until it ran - * out of results rather than until the limit was reached. + * Put results into cache only if no later query can ask for more results than the cached list holds. */ @Override public void close() { if (isTimerRunning) { assert currentIds != null; - //Reaching the limit stops the index from being read any further, so currentIds holds a prefix of the - //subquery results rather than all of them. The cache key of a subquery of a joint query does not include - //the limit, because updateLimit only propagates the limit when there is a single subquery, so caching a - //prefix would serve too few results to a later query with a larger limit. Fewer emitted elements than the - //limit means the limit never stopped anything, so the results are complete - if (!elementIterator.hasNext() && emittedCount < limit) { + if (!elementIterator.hasNext() && isSafeToCache()) { indexCache.put(subQuery, currentIds); } profiler.setResultSize(currentIds.size()); @@ -125,4 +119,17 @@ public void close() { } } + //The cache stores a result list against the limit of the subquery which produced it, and serves that list only to + //a later query whose limit is no larger. Two situations make currentIds safe to store. + //Fewer emitted elements than the limit means the limit never stopped the index being read, so currentIds holds + //every result and serves any later limit. + //Otherwise the limit truncated the read and currentIds is only a prefix. A prefix is still safe while the limit + //the cache records for it is no larger than the limit which produced it. That holds for a single subquery, because + //JointIndexQuery.updateLimit propagates the limit into it. It does not hold once a joint query has more than one + //subquery, because updateLimit then leaves the subquery limits alone: the cache would record the wider subquery + //limit for a prefix read under the narrower joint limit, and serve too few results to a later query in between. + private boolean isSafeToCache() { + return emittedCount < limit || subQuery.getLimit() <= limit; + } + } diff --git a/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/SubqueryIteratorCacheTest.java b/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/SubqueryIteratorCacheTest.java index b04fb81ab1..a62fcb814c 100644 --- a/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/SubqueryIteratorCacheTest.java +++ b/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/SubqueryIteratorCacheTest.java @@ -37,10 +37,12 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -//The subquery cache is consulted for the results of one index of a joint query. A joint query does not propagate its -//limit to the subqueries when there is more than one of them, so the limit is not part of the cache key: a result set -//which the limit truncated must not be stored, or a later query in the same transaction asking for more results is -//served the shorter list. +//The subquery cache is consulted for the results of one index of a joint query. It stores a result list against the +//limit of the subquery which produced it, and serves that list only to a later query whose limit is no larger. So a +//result list which the read limit truncated is safe to store only while the subquery carries that same limit. +//JointIndexQuery.updateLimit propagates the joint limit into the subquery only when there is a single subquery. With +//more than one the subquery keeps a wider limit, and a truncated list stored against it is served to a later query +//which asked for more results than the list holds. public class SubqueryIteratorCacheTest { private static final List ALL_MATCHING_IDS = Arrays.asList(1L, 2L, 3L, 4L, 5L); @@ -48,49 +50,68 @@ public class SubqueryIteratorCacheTest { private final JointIndexQuery.Subquery subQuery = mock(JointIndexQuery.Subquery.class); private final SubqueryCache indexCache = mock(SubqueryCache.class); - private void runQuery(int limit) { + //readLimit is the limit the iterator reads the index under, which a joint query takes from itself. subQueryLimit is + //the limit the cache records the result against, which it takes from the subquery. + private void runQuery(int readLimit, int subQueryLimit) { when(subQuery.getProfiler()).thenReturn(QueryProfiler.NO_OP); + when(subQuery.getLimit()).thenReturn(subQueryLimit); final IndexSerializer indexSerializer = mock(IndexSerializer.class); when(indexSerializer.query(any(), any(), any())).thenReturn(ALL_MATCHING_IDS.stream()); //A mock returns an empty List rather than null, which would look like a cache hit holding no results when(indexCache.getIfPresent(any())).thenReturn(null); try (SubqueryIterator iterator = new SubqueryIterator(subQuery, indexSerializer, - mock(BackendTransaction.class), mock(StandardJanusGraphTx.class), indexCache, limit, + mock(BackendTransaction.class), mock(StandardJanusGraphTx.class), indexCache, readLimit, id -> mock(JanusGraphElement.class), null)) { iterator.forEachRemaining(element -> { }); } } + private List captureCachedResult() { + final ArgumentCaptor> cached = ArgumentCaptor.forClass(List.class); + verify(indexCache, times(1)).put(any(), cached.capture()); + return cached.getValue(); + } + @Test - public void shouldNotCacheAResultSetTruncatedByTheLimit() { - runQuery(2); + public void shouldNotCacheATruncatedResultSetWhenTheSubqueryLimitIsWider() { + //This is the joint query of more than one index: the read stopped after 2 results, but the cache would record + //the list against an unlimited subquery and then serve those 2 results to a query which asked for all of them + runQuery(2, Query.NO_LIMIT); verify(indexCache, never()).put(any(), any()); } @Test - public void shouldCacheAResultSetWhoseIndexRanOutOfResults() { - runQuery(ALL_MATCHING_IDS.size() + 1); + public void shouldCacheATruncatedResultSetWhenTheSubqueryCarriesTheSameLimit() { + //This is the joint query of a single index: the cache records the same limit which truncated the read, so it + //declines to serve the list to a later query which asks for more + runQuery(2, 2); + assertEquals(ALL_MATCHING_IDS.subList(0, 2), captureCachedResult()); + } - final ArgumentCaptor> cached = ArgumentCaptor.forClass(List.class); - verify(indexCache, times(1)).put(any(), cached.capture()); - assertEquals(ALL_MATCHING_IDS, cached.getValue()); + @Test + public void shouldCacheAResultSetWhoseIndexRanOutOfResults() { + runQuery(ALL_MATCHING_IDS.size() + 1, ALL_MATCHING_IDS.size() + 1); + assertEquals(ALL_MATCHING_IDS, captureCachedResult()); } @Test public void shouldCacheAResultSetWhenThereIsNoLimit() { - runQuery(Query.NO_LIMIT); + runQuery(Query.NO_LIMIT, Query.NO_LIMIT); + assertEquals(ALL_MATCHING_IDS, captureCachedResult()); + } - final ArgumentCaptor> cached = ArgumentCaptor.forClass(List.class); - verify(indexCache, times(1)).put(any(), cached.capture()); - assertEquals(ALL_MATCHING_IDS, cached.getValue()); + @Test + public void shouldCacheAResultSetWhoseIndexRanOutAsTheLimitWasReached() { + //The index ran out at the same moment the limit was reached. Which of the two stopped the read is unknown, so + //the list counts as truncated, and the subquery carrying the same limit is what makes it safe to store + runQuery(ALL_MATCHING_IDS.size(), ALL_MATCHING_IDS.size()); + assertEquals(ALL_MATCHING_IDS, captureCachedResult()); } @Test - public void shouldNotCacheWhenTheLimitIsExactlyTheNumberOfResults() { - //The index ran out at the same moment the limit was reached, so which of the two stopped the read is unknown. - //Declining to cache costs a repeated index call; caching a set which may be short costs missing results - runQuery(ALL_MATCHING_IDS.size()); + public void shouldNotCacheAResultSetWhoseIndexRanOutAsAWiderSubqueryLimitWasReached() { + runQuery(ALL_MATCHING_IDS.size(), Query.NO_LIMIT); verify(indexCache, never()).put(any(), any()); } @@ -108,8 +129,6 @@ public void shouldCacheAnEmptyResultSet() { } //An index which matched nothing is a complete answer, and worth caching - final ArgumentCaptor> cached = ArgumentCaptor.forClass(List.class); - verify(indexCache, times(1)).put(any(), cached.capture()); - assertEquals(Collections.emptyList(), cached.getValue()); + assertEquals(Collections.emptyList(), captureCachedResult()); } }