From 583a01b8f85419526424112c32ed7ceb6661968f Mon Sep 17 00:00:00 2001 From: fleisch Date: Fri, 21 Aug 2026 11:41:47 +0200 Subject: [PATCH] feat: seek to the paging offset instead of scanning to it BoundedStream reaches the offset by pulling and discarding that many records, each one a document read and deserialized, so a page costs O(skip) and paging a collection costs O(n^2/pageSize). Add a default NitriteMap.entries(long skipCount) that keeps the discard behaviour, leaving InMemoryMap, RocksDBMap and TransactionalMap unaffected, and override it in NitriteMVMap, where MVStore's per-page entry counts locate the offset in O(log n). ReadOperations uses it only where the offset means the same at the source as at the end of the pipeline: no collection scan filter, no blocking sort, no OR sub-plans, and not while the index sorted stream supplies the order. An index scan skips over the id set instead, so skipped ids never turn into document reads. The seek pins one RootReference and walks it instead when another commit lands mid-lookup, and avoids Cursor.skip(long): past the end that leaves the cursor at the first entry rather than exhausting it. Paging 20k documents of 1KB went from about ten full scans to 0.6-0.9x of one. --- .../dizitart/no2/mvstore/NitriteMVMap.java | 59 +++++ .../no2/integration/PagedFindTest.java | 219 ++++++++++++++++ .../NitriteMVMapSkippedEntriesTest.java | 241 ++++++++++++++++++ .../collection/operation/ReadOperations.java | 56 +++- .../no2/common/streams/IndexedStream.java | 7 +- .../org/dizitart/no2/store/NitriteMap.java | 24 ++ .../operation/ReadOperationsPagingTest.java | 198 ++++++++++++++ .../memory/InMemoryMapSkippedEntriesTest.java | 103 ++++++++ 8 files changed, 901 insertions(+), 6 deletions(-) create mode 100644 nitrite-mvstore-adapter/src/test/java/org/dizitart/no2/integration/PagedFindTest.java create mode 100644 nitrite-mvstore-adapter/src/test/java/org/dizitart/no2/mvstore/NitriteMVMapSkippedEntriesTest.java create mode 100644 nitrite/src/test/java/org/dizitart/no2/collection/operation/ReadOperationsPagingTest.java create mode 100644 nitrite/src/test/java/org/dizitart/no2/store/memory/InMemoryMapSkippedEntriesTest.java diff --git a/nitrite-mvstore-adapter/src/main/java/org/dizitart/no2/mvstore/NitriteMVMap.java b/nitrite-mvstore-adapter/src/main/java/org/dizitart/no2/mvstore/NitriteMVMap.java index 7df7f6c60..5883104f8 100644 --- a/nitrite-mvstore-adapter/src/main/java/org/dizitart/no2/mvstore/NitriteMVMap.java +++ b/nitrite-mvstore-adapter/src/main/java/org/dizitart/no2/mvstore/NitriteMVMap.java @@ -18,11 +18,15 @@ import org.dizitart.no2.common.RecordStream; import org.dizitart.no2.common.tuples.Pair; +import org.dizitart.no2.exceptions.ValidationException; import org.dizitart.no2.store.NitriteMap; import org.dizitart.no2.store.NitriteStore; +import org.h2.mvstore.Cursor; import org.h2.mvstore.MVMap; import org.h2.mvstore.MVStore; +import org.h2.mvstore.RootReference; +import java.util.Collections; import java.util.Iterator; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; @@ -149,6 +153,61 @@ public Pair next() { }; } + @Override + public RecordStream> entries(long skipCount) { + if (skipCount < 0) { + throw new ValidationException("skip count cannot be negative"); + } + if (skipCount == 0) { + return entries(); + } + return () -> { + // pin one root, so the offset and the scan that follows it see the same map + RootReference rootReference = mvMap.flushAndGetRoot(); + if (skipCount >= rootReference.root.getTotalCount()) { + return Collections.emptyIterator(); + } + + // MVMap keeps an entry count per page, so the key at a position is found in + // O(log n) without reading what is skipped. getKey() resolves against whatever + // root the map holds while it runs, so its answer counts as an offset into the + // pinned snapshot only if no other thread committed meanwhile - a commit always + // installs a new RootReference, so an unchanged one proves it did not. If one + // did, walk the snapshot instead: slower, but never a window from another tree. + Key fromKey = mvMap.getKey(skipCount); + Cursor cursor = fromKey != null && mvMap.getRoot() == rootReference + ? mvMap.cursor(rootReference, fromKey, null, false) + : scanTo(rootReference, skipCount); + + return new Iterator>() { + @Override + public boolean hasNext() { + return cursor.hasNext(); + } + + @Override + public Pair next() { + Key key = cursor.next(); + return new Pair<>(key, cursor.getValue()); + } + }; + }; + } + + /** + * Positions a cursor on the given snapshot by walking to {@code skipCount}. Deliberately + * not {@link Cursor#skip(long)}: skipping past the end leaves that cursor at the first + * entry instead of exhausting it, which would answer a page beyond the end with the start + * of the map. + */ + private Cursor scanTo(RootReference rootReference, long skipCount) { + Cursor cursor = mvMap.cursor(rootReference, null, null, false); + for (long i = 0; i < skipCount && cursor.hasNext(); i++) { + cursor.next(); + } + return cursor; + } + @Override public RecordStream> reversedEntries() { return () -> new ReverseIterator<>(mvMap); diff --git a/nitrite-mvstore-adapter/src/test/java/org/dizitart/no2/integration/PagedFindTest.java b/nitrite-mvstore-adapter/src/test/java/org/dizitart/no2/integration/PagedFindTest.java new file mode 100644 index 000000000..dcabe62d3 --- /dev/null +++ b/nitrite-mvstore-adapter/src/test/java/org/dizitart/no2/integration/PagedFindTest.java @@ -0,0 +1,219 @@ +/* + * Copyright (c) 2017-2021 Nitrite author or 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.dizitart.no2.integration; + +import org.dizitart.no2.Nitrite; +import org.dizitart.no2.collection.Document; +import org.dizitart.no2.collection.FindOptions; +import org.dizitart.no2.collection.NitriteCollection; +import org.dizitart.no2.common.SortOrder; +import org.dizitart.no2.filters.Filter; +import org.dizitart.no2.mvstore.MVStoreModule; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.dizitart.no2.filters.FluentFilter.where; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * End-to-end tests for paged reads on an MVStore-backed database: every plan + * shape must return exactly the same pages as slicing the full result list. + */ +public class PagedFindTest { + private static final int SIZE = 5000; + + private final String fileName = TestUtil.getRandomTempDbFile(); + private Nitrite db; + private NitriteCollection collection; + + @Before + public void setUp() { + MVStoreModule storeModule = MVStoreModule.withConfig() + .filePath(fileName) + .build(); + db = Nitrite.builder() + .fieldSeparator(".") + .loadModule(storeModule) + .openOrCreate(); + + collection = db.getCollection("paged"); + for (int i = 0; i < SIZE; i++) { + collection.insert(Document.createDocument("value", i).put("group", i % 5)); + } + } + + @After + public void tearDown() { + if (db != null && !db.isClosed()) { + db.close(); + } + TestUtil.deleteDb(fileName); + } + + private static List values(Iterable documents) { + List values = new ArrayList<>(); + for (Document document : documents) { + values.add(document.get("value", Integer.class)); + } + return values; + } + + private static FindOptions options(String sortField, SortOrder sortOrder) { + return sortField == null ? new FindOptions() : FindOptions.orderBy(sortField, sortOrder); + } + + private void assertPagingMatchesFullResult(Filter filter, String sortField, + SortOrder sortOrder, int pageSize) { + List expected = values(collection.find(filter, options(sortField, sortOrder))); + + List paged = new ArrayList<>(); + long offset = 0; + while (true) { + FindOptions pageOptions = options(sortField, sortOrder) + .skip(offset) + .limit((long) pageSize); + List page = values(collection.find(filter, pageOptions)); + if (page.isEmpty()) { + break; + } + assertTrue("page must not exceed page size", page.size() <= pageSize); + paged.addAll(page); + offset += pageSize; + } + + assertEquals(expected, paged); + } + + @Test + public void testNaturalOrderPagedIterationMatchesFullIteration() { + for (int pageSize : new int[]{1, 7, 100, SIZE, SIZE + 100}) { + assertPagingMatchesFullResult(Filter.ALL, null, null, pageSize); + } + } + + @Test + public void testSkipBeyondEndIsEmpty() { + assertTrue(collection.find(Filter.ALL, + new FindOptions().skip(SIZE + 1).limit(10)).toList().isEmpty()); + assertTrue(collection.find(Filter.ALL, + new FindOptions().skip(SIZE).limit(10)).toList().isEmpty()); + } + + @Test + public void testLastPageIsPartial() { + List lastPage = values(collection.find(Filter.ALL, + new FindOptions().skip(SIZE - 3).limit(10))); + assertEquals(3, lastPage.size()); + } + + @Test + public void testPagingAfterRemovalsMatchesFullIteration() { + collection.remove(where("group").eq(2)); + assertPagingMatchesFullResult(Filter.ALL, null, null, 97); + } + + @Test + public void testPagingWithOrderByMatchesFullIteration() { + assertPagingMatchesFullResult(Filter.ALL, "value", SortOrder.Descending, 131); + } + + @Test + public void testPagingWithIndexedFilterMatchesFullIteration() { + collection.createIndex("value"); + assertPagingMatchesFullResult(where("value").gte(SIZE / 2), null, null, 89); + } + + @Test + public void testPagingWithIndexedFilterAndIndexOrderMatchesFullIteration() { + collection.createIndex("value"); + assertPagingMatchesFullResult(where("value").gte(SIZE / 2), + "value", SortOrder.Ascending, 89); + } + + @Test + public void testPagingWithNonIndexedFilterMatchesFullIteration() { + assertPagingMatchesFullResult(where("group").eq(3), null, null, 53); + } + + @Test + public void testPagingWithOrFilterMatchesFullIteration() { + assertPagingMatchesFullResult( + Filter.or(where("group").eq(1), where("group").eq(4)), + null, null, 71); + } + + /** + * Paging through the whole collection must not be quadratic. The dataset is + * larger than the page cache, so without the store-level skip every page + * request re-reads and deserializes all skipped entries from disk (~25 full + * scans in total, measured at ~10x the duration of one full scan); with it, + * the paged lap costs about one full scan plus per-page seek overhead + * (measured at ~1-2x). The 5x bound sits between the two, to stay robust + * on slow or busy machines. + */ + @Test + public void testPagedIterationIsNotQuadratic() { + int size = 20_000; + int pageSize = 400; + String payload = "x".repeat(1024); + String perfFileName = TestUtil.getRandomTempDbFile(); + + MVStoreModule storeModule = MVStoreModule.withConfig() + .filePath(perfFileName) + .cacheSize(1) // MB, much smaller than the data, to defeat the page cache + .build(); + try (Nitrite perfDb = Nitrite.builder() + .fieldSeparator(".") + .loadModule(storeModule) + .openOrCreate()) { + + NitriteCollection perfCollection = perfDb.getCollection("paged-perf"); + for (int i = 0; i < size; i++) { + perfCollection.insert(Document.createDocument("value", i).put("payload", payload)); + } + + long fullScanStart = System.nanoTime(); + int count = values(perfCollection.find()).size(); + long fullScanNanos = System.nanoTime() - fullScanStart; + assertEquals(size, count); + + long pagedStart = System.nanoTime(); + int pagedCount = 0; + for (long offset = 0; offset < size; offset += pageSize) { + pagedCount += values(perfCollection.find(Filter.ALL, + new FindOptions().skip(offset).limit((long) pageSize))).size(); + } + long pagedNanos = System.nanoTime() - pagedStart; + assertEquals(size, pagedCount); + + double ratio = (double) pagedNanos / fullScanNanos; + System.out.printf("paged iteration took %.1fx of a full scan%n", ratio); + assertTrue(String.format( + "paged iteration took %.1fx of a full scan; skip does not seem to be pushed down", + ratio), + ratio < 5); + } finally { + TestUtil.deleteDb(perfFileName); + } + } +} diff --git a/nitrite-mvstore-adapter/src/test/java/org/dizitart/no2/mvstore/NitriteMVMapSkippedEntriesTest.java b/nitrite-mvstore-adapter/src/test/java/org/dizitart/no2/mvstore/NitriteMVMapSkippedEntriesTest.java new file mode 100644 index 000000000..026ebdf99 --- /dev/null +++ b/nitrite-mvstore-adapter/src/test/java/org/dizitart/no2/mvstore/NitriteMVMapSkippedEntriesTest.java @@ -0,0 +1,241 @@ +/* + * Copyright (c) 2019-2020. Nitrite author or 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.dizitart.no2.mvstore; + +import org.dizitart.no2.common.RecordStream; +import org.dizitart.no2.common.tuples.Pair; +import org.dizitart.no2.exceptions.ValidationException; +import org.h2.mvstore.MVMap; +import org.h2.mvstore.MVStore; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; +import java.util.stream.LongStream; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertTrue; + +/** + * Tests the MVStore-backed override of + * {@link org.dizitart.no2.store.NitriteMap#entries(long)} which must seek via + * {@link MVMap#getKey(long)} instead of iterating over the skipped entries. + */ +public class NitriteMVMapSkippedEntriesTest { + private static final int SIZE = 1000; + + private MVStore mvStore; + private NitriteMVMap nitriteMVMap; + + @Before + public void setUp() { + mvStore = MVStore.open(null); // in-memory store + MVMap mvMap = mvStore.openMap("test"); + // populate the backing map directly, out of order, to prove natural key order + for (long i = SIZE - 1; i >= 0; i--) { + mvMap.put(i, "value-" + i); + } + nitriteMVMap = new NitriteMVMap<>(mvMap, null); + } + + @After + public void tearDown() { + mvStore.close(); + } + + private static List keysOf(RecordStream> stream) { + List keys = new ArrayList<>(); + for (Pair pair : stream) { + keys.add(pair.getFirst()); + } + return keys; + } + + private static List range(long fromInclusive, long toExclusive) { + return LongStream.range(fromInclusive, toExclusive).boxed().collect(Collectors.toList()); + } + + @Test + public void testSkipZeroReturnsAllEntriesInOrder() { + assertEquals(range(0, SIZE), keysOf(nitriteMVMap.entries(0))); + assertEquals(keysOf(nitriteMVMap.entries()), keysOf(nitriteMVMap.entries(0))); + } + + @Test + public void testSkipReturnsSuffixInNaturalOrder() { + assertEquals(range(750, SIZE), keysOf(nitriteMVMap.entries(750))); + } + + @Test + public void testSkipOneEntryShort() { + assertEquals(List.of((long) (SIZE - 1)), keysOf(nitriteMVMap.entries(SIZE - 1))); + } + + @Test + public void testSkipEqualToSizeReturnsEmpty() { + assertTrue(keysOf(nitriteMVMap.entries(SIZE)).isEmpty()); + } + + @Test + public void testSkipBeyondSizeReturnsEmpty() { + assertTrue(keysOf(nitriteMVMap.entries(SIZE + 500)).isEmpty()); + } + + @Test(expected = ValidationException.class) + public void testNegativeSkipThrows() { + nitriteMVMap.entries(-1); + } + + @Test + public void testStreamIsReIterable() { + RecordStream> stream = nitriteMVMap.entries(998); + assertEquals(List.of(998L, 999L), keysOf(stream)); + assertEquals(List.of(998L, 999L), keysOf(stream)); + } + + @Test + public void testStreamReflectsRemovalsOnReIteration() { + RecordStream> stream = nitriteMVMap.entries(SIZE - 2); + assertEquals(List.of((long) (SIZE - 2), (long) (SIZE - 1)), keysOf(stream)); + + // removing an entry shifts the suffix; a fresh iteration must see the new state + mvStore.openMap("test").remove(0L); + assertEquals(List.of((long) (SIZE - 1)), keysOf(stream)); + } + + @Test + public void testSkipOnEmptyMapReturnsEmpty() { + MVMap emptyMap = mvStore.openMap("empty"); + NitriteMVMap emptyNitriteMap = new NitriteMVMap<>(emptyMap, null); + assertTrue(keysOf(emptyNitriteMap.entries(5)).isEmpty()); + assertTrue(keysOf(emptyNitriteMap.entries(0)).isEmpty()); + } + + @Test + public void testValuesArePairedWithCorrectKeys() { + for (Pair pair : nitriteMVMap.entries(123)) { + assertEquals("value-" + pair.getFirst(), pair.getSecond()); + } + } + + @Test + public void testEveryOffsetMatchesFullScanSuffix() { + List full = keysOf(nitriteMVMap.entries()); + // covers both branches of MVStore's positional seek (small and large offsets) + for (long skip : new long[]{1, 5, 9, 10, 63, 64, 65, 500, 999}) { + assertEquals("skip=" + skip, full.subList((int) skip, SIZE), + keysOf(nitriteMVMap.entries(skip))); + } + } + + /** + * A page past the end must be empty. MVStore's own {@code Cursor.skip(long)} + * resets to the first entry instead of exhausting the cursor when it skips + * past the end, which would turn a page past the end into the whole + * collection and make a "read pages until one comes back empty" loop + * never terminate. + */ + @Test + public void testSkipPastEndNeverRestartsFromTheBeginning() { + for (long skip : new long[]{SIZE, SIZE + 1, SIZE + 10, 2L * SIZE, Integer.MAX_VALUE}) { + assertTrue("skip=" + skip, keysOf(nitriteMVMap.entries(skip)).isEmpty()); + } + } + + /** + * The scan must read one consistent snapshot: entries committed after the + * iterator was created must not appear part-way through it. + */ + @Test + public void testIterationReadsASingleSnapshot() { + MVMap mvMap = mvStore.openMap("test"); + + Iterator> iterator = nitriteMVMap.entries(SIZE - 4).iterator(); + assertEquals(Long.valueOf(SIZE - 4), iterator.next().getFirst()); + + // committed after the iterator pinned its snapshot, inside the range still to be read + mvMap.put((long) SIZE - 3, "changed"); + mvMap.put((long) SIZE + 1, "appended"); + + List remaining = new ArrayList<>(); + while (iterator.hasNext()) { + remaining.add(iterator.next().getFirst()); + } + assertEquals(List.of((long) SIZE - 3, (long) SIZE - 2, (long) SIZE - 1), remaining); + } + + /** + * Under a writer that repeatedly shrinks the map below the paged offset, a + * page must never come back starting at the first key: that is the + * signature of a seek that fell back to the beginning of the tree, and it + * would feed a pager the same rows forever. + */ + @Test + public void testConcurrentShrinkNeverYieldsTheFirstKey() throws Exception { + MVMap mvMap = mvStore.openMap("test"); + long skip = SIZE / 2; + + AtomicBoolean stopped = new AtomicBoolean(false); + AtomicReference writerFailure = new AtomicReference<>(); + Thread writer = new Thread(() -> { + try { + while (!stopped.get()) { + // never touches key 0, so the minimum key is stable + for (long i = 1; i < SIZE - 1; i++) { + mvMap.remove(i); + } + for (long i = 1; i < SIZE - 1; i++) { + mvMap.put(i, "value-" + i); + } + } + } catch (Throwable t) { + writerFailure.set(t); + } + }); + + writer.start(); + try { + for (int i = 0; i < 500; i++) { + List page = keysOf(nitriteMVMap.entries(skip)); + if (!page.isEmpty()) { + assertNotEquals("page restarted at the first key", Long.valueOf(0), page.get(0)); + assertTrue("page starts before the requested offset", page.get(0) >= skip); + // a window stitched together from two snapshots would repeat or + // go backwards here + for (int k = 1; k < page.size(); k++) { + assertTrue("page keys are not strictly increasing at " + k, + page.get(k) > page.get(k - 1)); + } + } + } + } finally { + stopped.set(true); + writer.join(30_000); + } + + if (writerFailure.get() != null) { + throw new AssertionError("writer failed", writerFailure.get()); + } + } +} diff --git a/nitrite/src/main/java/org/dizitart/no2/collection/operation/ReadOperations.java b/nitrite/src/main/java/org/dizitart/no2/collection/operation/ReadOperations.java index 7d9c838e7..6170be2c7 100644 --- a/nitrite/src/main/java/org/dizitart/no2/collection/operation/ReadOperations.java +++ b/nitrite/src/main/java/org/dizitart/no2/collection/operation/ReadOperations.java @@ -33,6 +33,7 @@ import java.text.Collator; import java.util.ArrayList; import java.util.Collection; +import java.util.Iterator; import java.util.LinkedHashSet; import java.util.List; @@ -197,6 +198,11 @@ private RecordStream> findSuitableStream(FindPlan find RecordStream> rawStream; RecordStream> indexSortedStream = null; + // the offset can be taken at the source, before any document is fetched, as long as + // nothing between the source and the bound stage drops or reorders records + long skip = findPlan.getSkip() == null ? 0 : findPlan.getSkip(); + boolean skipApplied = false; + if (!findPlan.getSubPlans().isEmpty()) { // or filters get all sub stream by finding suitable stream of all sub plans List>> subStreams = new ArrayList<>(); @@ -240,10 +246,26 @@ private RecordStream> findSuitableStream(FindPlan find indexedIdCount[0] = nitriteIds.size(); // create indexed stream from optimized filter - rawStream = new IndexedStream(nitriteIds, nitriteMap); + if (skip > 0 && canPushDownSkip(findPlan)) { + // walking the id set is cheap, fetching a document is not, so drop the + // skipped ids before they reach the map + rawStream = new IndexedStream(skippedIds(nitriteIds, skip), nitriteMap); + skipApplied = true; + } else { + rawStream = new IndexedStream(nitriteIds, nitriteMap); + } } else { indexSortedStream = indexSortedStream(findPlan); - rawStream = indexSortedStream != null ? indexSortedStream : nitriteMap.entries(); + if (indexSortedStream != null) { + rawStream = indexSortedStream; + } else if (skip > 0 && canPushDownSkip(findPlan)) { + // let the store seek to the offset instead of reading and + // deserializing everything in front of it + rawStream = nitriteMap.entries(skip); + skipApplied = true; + } else { + rawStream = nitriteMap.entries(); + } } } @@ -262,13 +284,37 @@ private RecordStream> findSuitableStream(FindPlan find rawStream = new SortedDocumentStream(findPlan, rawStream); } - if (findPlan.getLimit() != null || findPlan.getSkip() != null) { + if (findPlan.getLimit() != null || (findPlan.getSkip() != null && !skipApplied)) { long limit = findPlan.getLimit() == null ? Long.MAX_VALUE : findPlan.getLimit(); - long skip = findPlan.getSkip() == null ? 0 : findPlan.getSkip(); - rawStream = new BoundedStream<>(skip, limit, rawStream); + rawStream = new BoundedStream<>(skipApplied ? 0 : skip, limit, rawStream); } } return rawStream; } + + /** + * Indicates whether the offset can be taken at the record source. A post-filter or a + * blocking sort between the source and the bound stage changes which records the skip + * removes, and an OR plan de-duplicates its concatenated sub-streams, so in those cases + * the skip has to stay in the bound stage and count result rows. + */ + private boolean canPushDownSkip(FindPlan findPlan) { + return findPlan.getSubPlans().isEmpty() + && findPlan.getCollectionScanFilter() == null + && (findPlan.getBlockingSortOrder() == null || findPlan.getBlockingSortOrder().isEmpty()); + } + + /** + * A lazy view of {@code nitriteIds} without its first {@code skipCount} elements. + */ + private static Iterable skippedIds(Collection nitriteIds, long skipCount) { + return () -> { + Iterator iterator = nitriteIds.iterator(); + for (long i = 0; i < skipCount && iterator.hasNext(); i++) { + iterator.next(); + } + return iterator; + }; + } } diff --git a/nitrite/src/main/java/org/dizitart/no2/common/streams/IndexedStream.java b/nitrite/src/main/java/org/dizitart/no2/common/streams/IndexedStream.java index ab86c9fdd..04e93132f 100644 --- a/nitrite/src/main/java/org/dizitart/no2/common/streams/IndexedStream.java +++ b/nitrite/src/main/java/org/dizitart/no2/common/streams/IndexedStream.java @@ -32,10 +32,15 @@ */ public class IndexedStream implements RecordStream> { private final NitriteMap nitriteMap; - private final Set nitriteIds; + private final Iterable nitriteIds; public IndexedStream(Set nitriteIds, NitriteMap nitriteMap) { + this((Iterable) nitriteIds, nitriteMap); + } + + public IndexedStream(Iterable nitriteIds, + NitriteMap nitriteMap) { this.nitriteIds = nitriteIds; this.nitriteMap = nitriteMap; } diff --git a/nitrite/src/main/java/org/dizitart/no2/store/NitriteMap.java b/nitrite/src/main/java/org/dizitart/no2/store/NitriteMap.java index 07b26ce27..f0def5ffa 100644 --- a/nitrite/src/main/java/org/dizitart/no2/store/NitriteMap.java +++ b/nitrite/src/main/java/org/dizitart/no2/store/NitriteMap.java @@ -19,7 +19,9 @@ import org.dizitart.no2.common.meta.Attributes; import org.dizitart.no2.common.meta.AttributesAware; import org.dizitart.no2.common.RecordStream; +import org.dizitart.no2.common.streams.BoundedStream; import org.dizitart.no2.common.tuples.Pair; +import org.dizitart.no2.exceptions.ValidationException; import static org.dizitart.no2.common.Constants.META_MAP_NAME; import static org.dizitart.no2.common.util.StringUtils.isNullOrEmpty; @@ -194,6 +196,28 @@ public interface NitriteMap extends AttributesAware, AutoCloseable { */ RecordStream> entries(); + /** + * Gets a {@link RecordStream} view of the mappings contained in this map, + * skipping the first {@code skipCount} entries of {@link #entries()}. + * + *

The default implementation iterates over {@link #entries()} and discards the + * skipped entries. A store whose backing structure can locate an entry by position + * should override this and seek instead, so that a paged read does not pay for the + * entries it skips. + * + * @param skipCount the number of leading entries to skip; must not be negative. + * @return a view of the mappings after the first {@code skipCount} entries. + */ + default RecordStream> entries(long skipCount) { + if (skipCount < 0) { + throw new ValidationException("skip count cannot be negative"); + } + if (skipCount == 0) { + return entries(); + } + return new BoundedStream<>(skipCount, Long.MAX_VALUE, entries()); + } + /** * Gets a reversed {@link RecordStream} view of the mappings contained in this map. * diff --git a/nitrite/src/test/java/org/dizitart/no2/collection/operation/ReadOperationsPagingTest.java b/nitrite/src/test/java/org/dizitart/no2/collection/operation/ReadOperationsPagingTest.java new file mode 100644 index 000000000..3c089e6e3 --- /dev/null +++ b/nitrite/src/test/java/org/dizitart/no2/collection/operation/ReadOperationsPagingTest.java @@ -0,0 +1,198 @@ +/* + * Copyright (c) 2017-2021 Nitrite author or 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.dizitart.no2.collection.operation; + +import org.dizitart.no2.NitriteConfig; +import org.dizitart.no2.collection.Document; +import org.dizitart.no2.collection.FindOptions; +import org.dizitart.no2.collection.NitriteId; +import org.dizitart.no2.common.Fields; +import org.dizitart.no2.common.SortOrder; +import org.dizitart.no2.common.processors.ProcessorChain; +import org.dizitart.no2.filters.Filter; +import org.dizitart.no2.index.IndexDescriptor; +import org.dizitart.no2.index.IndexType; +import org.dizitart.no2.index.NitriteIndexer; +import org.dizitart.no2.store.NitriteStore; +import org.dizitart.no2.store.memory.InMemoryMap; +import org.junit.Before; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.LinkedHashSet; +import java.util.List; + +import static org.dizitart.no2.filters.FluentFilter.where; +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.*; + +/** + * Verifies that skip is pushed down to the store layer for paged reads, + * so that skipped entries are neither read nor deserialized. + */ +public class ReadOperationsPagingTest { + private static final String COLLECTION = "test"; + private static final int SIZE = 10; + + private InMemoryMap nitriteMap; + private IndexOperations indexOperations; + + @Before + public void setUp() { + nitriteMap = spy(new InMemoryMap<>(COLLECTION, mock(NitriteStore.class))); + for (int i = 0; i < SIZE; i++) { + nitriteMap.put(id(i), Document.createDocument("value", i)); + } + clearInvocations(nitriteMap); + + indexOperations = mock(IndexOperations.class); + when(indexOperations.listIndexes()).thenReturn(new ArrayList<>()); + } + + private static NitriteId id(long value) { + return NitriteId.createId(value); + } + + private ReadOperations readOperations(NitriteConfig config) { + return new ReadOperations(COLLECTION, indexOperations, config, nitriteMap, new ProcessorChain()); + } + + private static List values(Iterable documents) { + List values = new ArrayList<>(); + for (Document document : documents) { + values.add(document.get("value", Integer.class)); + } + return values; + } + + @Test + public void testPureScanWithSkipDelegatesSkipToMap() { + ReadOperations operations = readOperations(new NitriteConfig()); + + List result = values(operations.find(Filter.ALL, + new FindOptions().skip(6).limit(2))); + + assertEquals(List.of(6, 7), result); + verify(nitriteMap).entries(6L); + } + + @Test + public void testPureScanWithSkipOnlyDelegatesSkipToMap() { + ReadOperations operations = readOperations(new NitriteConfig()); + + List result = values(operations.find(Filter.ALL, + new FindOptions().skip(4))); + + assertEquals(List.of(4, 5, 6, 7, 8, 9), result); + verify(nitriteMap).entries(4L); + } + + @Test + public void testPureScanSkipBeyondSizeIsEmpty() { + ReadOperations operations = readOperations(new NitriteConfig()); + + List result = values(operations.find(Filter.ALL, + new FindOptions().skip(SIZE + 5).limit(2))); + + assertEquals(List.of(), result); + verify(nitriteMap).entries((long) (SIZE + 5)); + } + + @Test + public void testPureScanWithZeroSkipUsesPlainEntries() { + ReadOperations operations = readOperations(new NitriteConfig()); + + List result = values(operations.find(Filter.ALL, + new FindOptions().skip(0).limit(3))); + + assertEquals(List.of(0, 1, 2), result); + verify(nitriteMap, never()).entries(anyLong()); + } + + @Test + public void testNoPushDownWithBlockingSort() { + ReadOperations operations = readOperations(new NitriteConfig()); + + // sort must be applied before skip, so skip cannot be delegated to the map + List result = values(operations.find(Filter.ALL, + FindOptions.orderBy("value", SortOrder.Descending).skip(6).limit(2))); + + assertEquals(List.of(3, 2), result); + verify(nitriteMap, never()).entries(anyLong()); + } + + @Test + public void testNoPushDownWithCollectionScanFilter() { + ReadOperations operations = readOperations(new NitriteConfig()); + + // the filter must see every entry, so skip cannot be delegated to the map + List result = values(operations.find(where("value").gte(2), + new FindOptions().skip(3).limit(2))); + + assertEquals(List.of(5, 6), result); + verify(nitriteMap, never()).entries(anyLong()); + } + + @Test + public void testIndexedScanSkipsWithoutFetchingDocuments() { + ReadOperations operations = readOperations(configWithIndexOnValueField()); + + List result = values(operations.find(where("value").eq(3), + new FindOptions().skip(7).limit(2))); + + assertEquals(List.of(7, 8), result); + // only the two returned documents may be fetched, none of the seven skipped ones + verify(nitriteMap, times(2)).get(any()); + verify(nitriteMap).get(id(7)); + verify(nitriteMap).get(id(8)); + } + + @Test + public void testIndexedScanWithoutSkipFetchesOnlyLimitedDocuments() { + ReadOperations operations = readOperations(configWithIndexOnValueField()); + + List result = values(operations.find(where("value").eq(3), + new FindOptions().limit(2))); + + assertEquals(List.of(0, 1), result); + verify(nitriteMap, times(2)).get(any()); + } + + /** + * Sets up an index descriptor on the "value" field backed by a stub indexer + * that reports every document as matching, in natural id order. + */ + private NitriteConfig configWithIndexOnValueField() { + IndexDescriptor indexDescriptor = + new IndexDescriptor(IndexType.UNIQUE, Fields.withNames("value"), COLLECTION); + when(indexOperations.listIndexes()).thenReturn(List.of(indexDescriptor)); + + LinkedHashSet matchingIds = new LinkedHashSet<>(); + for (int i = 0; i < SIZE; i++) { + matchingIds.add(id(i)); + } + NitriteIndexer indexer = mock(NitriteIndexer.class); + when(indexer.findByFilter(any(), any())).thenReturn(matchingIds); + + NitriteConfig nitriteConfig = mock(NitriteConfig.class); + when(nitriteConfig.findIndexer(IndexType.UNIQUE)).thenReturn(indexer); + return nitriteConfig; + } +} diff --git a/nitrite/src/test/java/org/dizitart/no2/store/memory/InMemoryMapSkippedEntriesTest.java b/nitrite/src/test/java/org/dizitart/no2/store/memory/InMemoryMapSkippedEntriesTest.java new file mode 100644 index 000000000..40d523806 --- /dev/null +++ b/nitrite/src/test/java/org/dizitart/no2/store/memory/InMemoryMapSkippedEntriesTest.java @@ -0,0 +1,103 @@ +/* + * Copyright (c) 2017-2021 Nitrite author or 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.dizitart.no2.store.memory; + +import org.dizitart.no2.common.RecordStream; +import org.dizitart.no2.common.tuples.Pair; +import org.dizitart.no2.exceptions.ValidationException; +import org.dizitart.no2.store.NitriteStore; +import org.junit.Before; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; + +/** + * Tests the default {@link org.dizitart.no2.store.NitriteMap#entries(long)} + * contract via {@link InMemoryMap}. + */ +public class InMemoryMapSkippedEntriesTest { + private InMemoryMap map; + + @Before + public void setUp() { + map = new InMemoryMap<>("test", mock(NitriteStore.class)); + for (int i = 0; i < 10; i++) { + map.put(i, "value-" + i); + } + } + + private static List keysOf(RecordStream> stream) { + List keys = new ArrayList<>(); + for (Pair pair : stream) { + keys.add(pair.getFirst()); + } + return keys; + } + + @Test + public void testSkipZeroReturnsAllEntriesInOrder() { + assertEquals(keysOf(map.entries()), keysOf(map.entries(0))); + assertEquals(10, keysOf(map.entries(0)).size()); + } + + @Test + public void testSkipReturnsSuffixInNaturalOrder() { + List keys = keysOf(map.entries(6)); + assertEquals(List.of(6, 7, 8, 9), keys); + } + + @Test + public void testSkipEqualToSizeReturnsEmpty() { + assertTrue(keysOf(map.entries(10)).isEmpty()); + } + + @Test + public void testSkipBeyondSizeReturnsEmpty() { + assertTrue(keysOf(map.entries(100)).isEmpty()); + } + + @Test(expected = ValidationException.class) + public void testNegativeSkipThrows() { + map.entries(-1); + } + + @Test + public void testStreamIsReIterable() { + RecordStream> stream = map.entries(8); + assertEquals(List.of(8, 9), keysOf(stream)); + assertEquals(List.of(8, 9), keysOf(stream)); + } + + @Test + public void testSkipOnEmptyMapReturnsEmpty() { + InMemoryMap emptyMap = new InMemoryMap<>("empty", mock(NitriteStore.class)); + assertTrue(keysOf(emptyMap.entries(5)).isEmpty()); + } + + @Test + public void testValuesArePairedWithCorrectKeys() { + for (Pair pair : map.entries(4)) { + assertEquals("value-" + pair.getFirst(), pair.getSecond()); + } + } +}