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()); + } + } +}