fix: skip documents removed between an index lookup and the fetch

nitrite/src/main/java/org/dizitart/no2/collection/operation/ReadOperations.java

@@ -222,12 +222,11 @@ private RecordStream<Pair<NitriteId, Document>> findSuitableStream(FindPlan find
                 NitriteId nitriteId = idValue instanceof Long
                     ? NitriteId.createId((long) idValue)
                     : NitriteId.createId(String.valueOf(idValue));
-                if (nitriteMap.containsKey(nitriteId)) {
-                    Document document = nitriteMap.get(nitriteId);
-                    rawStream = RecordStream.single(pair(nitriteId, document));
-                } else {
-                    rawStream = RecordStream.empty();
-                }
+                // one lookup: a document removed between containsKey and get would be a null row
+                Document document = nitriteMap.get(nitriteId);
+                rawStream = document == null
+                    ? RecordStream.empty()
+                    : RecordStream.single(pair(nitriteId, document));
             } else {
                 IndexDescriptor indexDescriptor = findPlan.getIndexDescriptor();
                 if (indexDescriptor != null) {

nitrite/src/main/java/org/dizitart/no2/common/streams/FilteredStream.java

@@ -89,7 +89,8 @@ public void remove() {
         private boolean setNextId() {
             while (iterator.hasNext()) {
                 final Pair<NitriteId, Document> pair = iterator.next();
-                if (filter.apply(pair)) {
+                // a row whose document is gone is not a match for anything
+                if (pair.getSecond() != null && filter.apply(pair)) {
                     nextPair = pair;
                     nextPairSet = true;
                     return true;

nitrite/src/main/java/org/dizitart/no2/common/streams/IndexedStream.java

@@ -24,6 +24,7 @@
 import org.dizitart.no2.store.NitriteMap;
 
 import java.util.Iterator;
+import java.util.NoSuchElementException;
 import java.util.Set;
 
 /**
@@ -49,6 +50,7 @@ private static class IndexedStreamIterator implements Iterator<Pair<NitriteId, D
             SkippableIterator {
         private final Iterator<NitriteId> iterator;
         private final NitriteMap<NitriteId, Document> nitriteMap;
+        private Pair<NitriteId, Document> next;
 
         IndexedStreamIterator(Iterator<NitriteId> iterator,
                               NitriteMap<NitriteId, Document> nitriteMap) {
@@ -58,7 +60,7 @@ private static class IndexedStreamIterator implements Iterator<Pair<NitriteId, D
 
         @Override
         public boolean hasNext() {
-            return iterator.hasNext();
+            return next != null || advance();
         }
 
         /**
@@ -68,6 +70,10 @@ public boolean hasNext() {
         @Override
         public long skip(long count) {
             long skipped = 0;
+            if (next != null && count > 0) {
+                next = null;
+                skipped++;
+            }
             while (skipped < count && iterator.hasNext()) {
                 iterator.next();
                 skipped++;
@@ -77,9 +83,28 @@ public long skip(long count) {
 
         @Override
         public Pair<NitriteId, Document> next() {
-            NitriteId id = iterator.next();
-            Document document = nitriteMap.get(id);
-            return new Pair<>(id, document);
+            if (next == null && !advance()) {
+                throw new NoSuchElementException();
+            }
+            Pair<NitriteId, Document> current = next;
+            next = null;
+            return current;
+        }
+
+        /**
+         * A document removed between the index lookup and the fetch is no longer a row: it is
+         * skipped rather than handed downstream as a null for a residual filter to dereference.
+         */
+        private boolean advance() {
+            while (iterator.hasNext()) {
+                NitriteId id = iterator.next();
+                Document document = nitriteMap.get(id);
+                if (document != null) {
+                    next = new Pair<>(id, document);
+                    return true;
+                }
+            }
+            return false;
         }
     }
 }

nitrite/src/test/java/org/dizitart/no2/collection/IndexScanRemovedDocumentTest.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) 2017-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.collection;
+
+import org.dizitart.no2.Nitrite;
+import org.dizitart.no2.filters.Filter;
+import org.dizitart.no2.index.IndexOptions;
+import org.dizitart.no2.index.IndexType;
+import org.dizitart.no2.store.NitriteMap;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.dizitart.no2.filters.FluentFilter.where;
+import static org.dizitart.no2.integration.TestUtil.createDb;
+import static org.junit.Assert.*;
+
+/**
+ * An index lookup and the fetch of the documents it names are not atomic: a document removed in
+ * between must simply be absent from the result, never a null row that a residual filter, or the
+ * caller, dereferences.
+ * <p>
+ * The window is reproduced by removing the document from the collection's map behind the index's
+ * back, which is exactly what a concurrent remove looks like to a reader that already holds the ids.
+ */
+public class IndexScanRemovedDocumentTest {
+    private Nitrite db;
+    private NitriteCollection collection;
+    private NitriteId removedId;
+
+    @Before
+    public void setUp() {
+        db = createDb();
+        collection = db.getCollection("race");
+        collection.createIndex(IndexOptions.indexOptions(IndexType.NON_UNIQUE), "group");
+        for (int n = 1; n <= 3; n++) {
+            collection.insert(Document.createDocument("group", "a").put("n", n));
+        }
+        removedId = collection.find(where("n").eq(2)).firstOrNull().getId();
+
+        NitriteMap<NitriteId, Document> map = db.getStore().openMap("race", NitriteId.class, Document.class);
+        assertNotNull(map.remove(removedId));
+    }
+
+    @After
+    public void tearDown() {
+        db.close();
+    }
+
+    @Test
+    public void testIndexScanWithResidualFilterSkipsRemovedDocument() {
+        List<Document> found = collection.find(where("group").eq("a").and(where("n").gte(1))).toList();
+        assertEquals(2, found.size());
+        assertTrue(found.stream().noneMatch(d -> d == null));
+        assertTrue(found.stream().noneMatch(d -> d.getId().equals(removedId)));
+    }
+
+    @Test
+    public void testIndexScanWithoutResidualFilterSkipsRemovedDocument() {
+        List<Document> found = collection.find(where("group").eq("a")).toList();
+        assertEquals(2, found.size());
+        assertTrue(found.stream().noneMatch(d -> d == null));
+    }
+
+    @Test
+    public void testIndexScanSizeExcludesRemovedDocument() {
+        assertEquals(2, collection.find(where("group").eq("a").and(where("n").gte(1))).size());
+    }
+
+    @Test
+    public void testByIdOfRemovedDocumentIsEmpty() {
+        assertNull(collection.find(Filter.byId(removedId)).firstOrNull());
+        assertNull(collection.getById(removedId));
+    }
+}