Avoid accumulation of index updates during recovery

Do not collect index updates during recovery since that can cause OOM.
Use iterators instead to iterate over required updates.

community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/IndexStoreView.java

@@ -59,9 +59,9 @@ <FAILURE extends Exception> StoreScan<FAILURE> visitNodes(
      * and puts those updates into {@code target}.
      *
      * @param nodeId id of node to load.
-     * @param target {@link Collection} to add updates into.
+     * @return node updates container
      */
-    void nodeAsUpdates( long nodeId, Collection<NodeUpdates> target );
+    NodeUpdates nodeAsUpdates( long nodeId );
 
     DoubleLongRegister indexUpdatesAndSize( long indexId, DoubleLongRegister output );
 
@@ -129,8 +129,9 @@ public void replaceIndexCounts( long indexId, long uniqueElements, long maxUniqu
         }
 
         @Override
-        public void nodeAsUpdates( long nodeId, Collection<NodeUpdates> target )
+        public NodeUpdates nodeAsUpdates( long nodeId )
         {
+            return null;
         }
 
         @Override

community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/IndexingService.java

@@ -26,17 +26,17 @@
 import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.function.BiConsumer;
 
 import org.neo4j.collection.primitive.Primitive;
+import org.neo4j.collection.primitive.PrimitiveLongIterator;
 import org.neo4j.collection.primitive.PrimitiveLongSet;
-import org.neo4j.collection.primitive.PrimitiveLongVisitor;
 import org.neo4j.graphdb.ResourceIterator;
-import org.neo4j.helpers.collection.Iterables;
 import org.neo4j.helpers.collection.Iterators;
 import org.neo4j.kernel.api.TokenNameLookup;
 import org.neo4j.kernel.api.exceptions.index.IndexActivationFailedKernelException;
@@ -115,7 +115,9 @@ public interface Monitor
     {
         void applyingRecoveredData( PrimitiveLongSet recoveredNodeIds );
 
-        void appliedRecoveredData( Iterable<NodeUpdates> updates );
+        void applyingRecoveredUpdate( IndexEntryUpdate<LabelSchemaDescriptor> indexUpdate );
+
+        void recoveredUpdatesApplied();
 
         void populationCompleteOn( IndexDescriptor descriptor );
 
@@ -127,7 +129,7 @@ public interface Monitor
     public static class MonitorAdapter implements Monitor
     {
         @Override
-        public void appliedRecoveredData( Iterable<NodeUpdates> updates )
+        public void recoveredUpdatesApplied()
         {   // Do nothing
         }
 
@@ -136,6 +138,12 @@ public void applyingRecoveredData( PrimitiveLongSet recoveredNodeIds )
         {   // Do nothing
         }
 
+        @Override
+        public void applyingRecoveredUpdate( IndexEntryUpdate<LabelSchemaDescriptor> indexUpdate )
+        {
+            // empty
+        }
+
         @Override
         public void populationCompleteOn( IndexDescriptor descriptor )
         {   // Do nothing
@@ -440,12 +448,6 @@ private void apply( Iterable<IndexEntryUpdate<LabelSchemaDescriptor>> updates, I
         }
     }
 
-    public void convertToIndexUpdatesAndApply( Iterable<NodeUpdates> updates, IndexUpdateMode updateMode )
-            throws IOException, IndexEntryConflictException
-    {
-        apply( Iterables.flatMap( this::convertToIndexUpdates, updates ), updateMode );
-    }
-
     @Override
     public Iterable<IndexEntryUpdate<LabelSchemaDescriptor>> convertToIndexUpdates( NodeUpdates nodeUpdates )
     {
@@ -534,28 +536,29 @@ private void applyRecoveredUpdates() throws IOException, IndexEntryConflictExcep
                     updater.remove( recoveredNodeIds );
                 }
 
-                Iterable<NodeUpdates> updates = readNodeUpdatesFromRecoveredStore();
-                convertToIndexUpdatesAndApply( updates, IndexUpdateMode.RECOVERY );
-                monitor.appliedRecoveredData( updates );
+                Iterator<NodeUpdates> updates = recoveredNodeUpdatesIterator();
+                Iterator<IndexEntryUpdate<LabelSchemaDescriptor>> indexEntryUpdates =
+                        Iterators.flatMap( nodeUpdates -> convertToIndexUpdates( nodeUpdates ).iterator(), updates );
+                while ( indexEntryUpdates.hasNext() )
+                {
+                    IndexEntryUpdate<LabelSchemaDescriptor> indexUpdate = indexEntryUpdates.next();
+                    monitor.applyingRecoveredUpdate( indexUpdate );
+                    IndexUpdater updater = updaterMap.getUpdater( indexUpdate.indexKey().schema() );
+                    if ( updater != null )
+                    {
+                        updater.process( indexUpdate );
+                    }
+                }
+                monitor.recoveredUpdatesApplied();
             }
         }
         recoveredNodeIds.clear();
     }
 
-    private Iterable<NodeUpdates> readNodeUpdatesFromRecoveredStore()
+    private Iterator<NodeUpdates> recoveredNodeUpdatesIterator()
     {
-        final List<NodeUpdates> nodeUpdates = new ArrayList<>();
-        recoveredNodeIds.visitKeys( new PrimitiveLongVisitor<RuntimeException>()
-        {
-            @Override
-            public boolean visited( long nodeId )
-            {
-                storeView.nodeAsUpdates( nodeId, nodeUpdates );
-                return false;
-            }
-        } );
-
-        return nodeUpdates;
+        PrimitiveLongIterator nodeIdIterator = recoveredNodeIds.iterator();
+        return new NodeUpdatesIterator( storeView, nodeIdIterator );
     }
 
     public void dropIndex( IndexRule rule )

community/kernel/src/main/java/org/neo4j/kernel/impl/api/index/NodeUpdatesIterator.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2002-2017 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.kernel.impl.api.index;
+
+import java.util.Iterator;
+
+import org.neo4j.collection.primitive.PrimitiveLongIterator;
+
+class NodeUpdatesIterator implements Iterator<NodeUpdates>
+{
+
+    private final IndexStoreView storeView;
+    private final PrimitiveLongIterator nodeIdIterator;
+    private NodeUpdates nextUpdate;
+
+    NodeUpdatesIterator( IndexStoreView storeView, PrimitiveLongIterator nodeIdIterator )
+    {
+        this.storeView = storeView;
+        this.nodeIdIterator = nodeIdIterator;
+    }
+
+    @Override
+    public boolean hasNext()
+    {
+        if ( nextUpdate == null )
+        {
+            while ( nodeIdIterator.hasNext() )
+            {
+                long nodeId = nodeIdIterator.next();
+                NodeUpdates updates = storeView.nodeAsUpdates( nodeId );
+                if ( updates != null )
+                {
+                    nextUpdate = updates;
+                    return true;
+                }
+            }
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public NodeUpdates next()
+    {
+        NodeUpdates update = null;
+        if ( hasNext() )
+        {
+            update = this.nextUpdate;
+            this.nextUpdate = null;
+        }
+        return update;
+    }
+}

community/kernel/src/main/java/org/neo4j/kernel/impl/transaction/state/storeview/NeoStoreIndexStoreView.java

@@ -19,17 +19,16 @@
  */
 package org.neo4j.kernel.impl.transaction.state.storeview;
 
-import java.util.Collection;
 import java.util.function.IntPredicate;
 
 import org.neo4j.collection.primitive.PrimitiveIntSet;
 import org.neo4j.helpers.collection.Visitor;
 import org.neo4j.kernel.api.exceptions.EntityNotFoundException;
-import org.neo4j.kernel.impl.api.index.NodeUpdates;
 import org.neo4j.kernel.api.labelscan.NodeLabelUpdate;
 import org.neo4j.kernel.api.properties.Property;
 import org.neo4j.kernel.impl.api.CountsAccessor;
 import org.neo4j.kernel.impl.api.index.IndexStoreView;
+import org.neo4j.kernel.impl.api.index.NodeUpdates;
 import org.neo4j.kernel.impl.api.index.StoreScan;
 import org.neo4j.kernel.impl.locking.LockService;
 import org.neo4j.kernel.impl.store.NeoStores;
@@ -107,22 +106,22 @@ public <FAILURE extends Exception> StoreScan<FAILURE> visitNodes(
     }
 
     @Override
-    public void nodeAsUpdates( long nodeId, Collection<NodeUpdates> target )
+    public NodeUpdates nodeAsUpdates( long nodeId )
     {
         NodeRecord node = nodeStore.getRecord( nodeId, nodeStore.newRecord(), FORCE );
         if ( !node.inUse() )
         {
-            return;
+            return null;
         }
         long firstPropertyId = node.getNextProp();
         if ( firstPropertyId == Record.NO_NEXT_PROPERTY.intValue() )
         {
-            return; // no properties => no updates (it's not going to be in any index)
+            return null; // no properties => no updates (it's not going to be in any index)
         }
         long[] labels = parseLabelsField( node ).get( nodeStore );
         if ( labels.length == 0 )
         {
-            return; // no labels => no updates (it's not going to be in any index)
+            return null; // no labels => no updates (it's not going to be in any index)
         }
         NodeUpdates.Builder update = NodeUpdates.forNode( nodeId, labels );
         for ( PropertyRecord propertyRecord : propertyStore.getPropertyRecordChain( firstPropertyId ) )
@@ -133,7 +132,7 @@ public void nodeAsUpdates( long nodeId, Collection<NodeUpdates> target )
                 update.added( property.getKeyIndexId(), value );
             }
         }
-        target.add( update.build() );
+        return update.build();
     }
 
     @Override

community/kernel/src/test/java/org/neo4j/kernel/impl/api/index/IndexingServiceTest.java

@@ -38,6 +38,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
@@ -52,7 +53,6 @@
 import org.neo4j.graphdb.ResourceIterator;
 import org.neo4j.graphdb.factory.GraphDatabaseSettings;
 import org.neo4j.helpers.collection.BoundedIterable;
-import org.neo4j.helpers.collection.Iterables;
 import org.neo4j.helpers.collection.Iterators;
 import org.neo4j.helpers.collection.Visitor;
 import org.neo4j.io.fs.FileSystemAbstraction;
@@ -684,17 +684,19 @@ public void recoveredUpdatesShouldBeApplied() throws Exception
 
         final NodeUpdates nodeUpdate1 = addNodeUpdate( nodeId1, "foo" );
         final NodeUpdates nodeUpdate2 = addNodeUpdate( nodeId2, "bar" );
-        final Set<NodeUpdates> nodeUpdates = asSet( nodeUpdate1, nodeUpdate2 );
 
         final AtomicBoolean applyingRecoveredDataCalled = new AtomicBoolean();
-        final AtomicBoolean appliedRecoveredDataCalled = new AtomicBoolean();
+        final AtomicBoolean recoveredUpdatesAppliedCalled = new AtomicBoolean();
 
         // Mockito not used here because it does not work well with mutable objects (set of recovered node ids in
         // this case, which is cleared at the end of recovery).
         // See https://code.google.com/p/mockito/issues/detail?id=126 and
         // https://groups.google.com/forum/#!topic/mockito/_A4BpsEAY9s
         IndexingService.Monitor monitor = new IndexingService.MonitorAdapter()
         {
+
+            private Set<IndexEntryUpdate<LabelSchemaDescriptor>> updates = new HashSet<>();
+
             @Override
             public void applyingRecoveredData( PrimitiveLongSet recoveredNodeIds )
             {
@@ -703,19 +705,26 @@ public void applyingRecoveredData( PrimitiveLongSet recoveredNodeIds )
             }
 
             @Override
-            public void appliedRecoveredData( Iterable<NodeUpdates> updates )
+            public void applyingRecoveredUpdate( IndexEntryUpdate<LabelSchemaDescriptor> indexUpdate )
             {
-                assertEquals( nodeUpdates, Iterables.asSet( updates ) );
-                appliedRecoveredDataCalled.set( true );
+                updates.add( indexUpdate );
+            }
+
+            @Override
+            public void recoveredUpdatesApplied()
+            {
+                recoveredUpdatesAppliedCalled.set( true );
+                assertEquals( 2, updates.size() );
             }
         };
 
         IndexingService indexing = newIndexingServiceWithMockedDependencies( populator, accessor, withData(), monitor );
+        indexing.createIndexes( IndexRule
+                .indexRule( 1, IndexDescriptorFactory.forLabel( labelId, index.schema().getPropertyId() ),
+                        new SchemaIndexProvider.Descriptor( "quantum-dex", "25.0" ) ) );
 
-        doAnswer( nodeUpdatesAnswer( nodeUpdate1 ) ).when( storeView )
-                .nodeAsUpdates( eq( nodeId1 ), any( Collection.class ) );
-        doAnswer( nodeUpdatesAnswer( nodeUpdate2 ) ).when( storeView )
-                .nodeAsUpdates( eq( nodeId2 ), any( Collection.class ) );
+        when( storeView.nodeAsUpdates( eq( nodeId1 ) ) ).thenReturn( nodeUpdate1 );
+        when( storeView.nodeAsUpdates( eq( nodeId2 ) ) ).thenReturn( nodeUpdate2 );
 
         // When
         life.init();
@@ -725,7 +734,7 @@ public void appliedRecoveredData( Iterable<NodeUpdates> updates )
 
         // Then
         assertTrue( "applyingRecoveredData was not called", applyingRecoveredDataCalled.get() );
-        assertTrue( "appliedRecoveredData was not called", appliedRecoveredDataCalled.get() );
+        assertTrue( "recoveredUpdatesApplied was not called", recoveredUpdatesAppliedCalled.get() );
     }
 
     private IndexUpdates nodeIdsAsIndexUpdates( PrimitiveLongSet nodeIds )
@@ -769,8 +778,7 @@ public void shouldNotLoseIndexDescriptorDueToOtherSimilarIndexDuringRecovery() t
         // GIVEN
         long nodeId = 0, indexId = 1, otherIndexId = 2;
         NodeUpdates update = addNodeUpdate( nodeId, "value" );
-        doAnswer( nodeUpdatesAnswer( update ) ).when( storeView )
-                .nodeAsUpdates( eq( nodeId ), any( Collection.class ) );
+        when( storeView.nodeAsUpdates( eq( nodeId ) ) ).thenReturn( update );
         DoubleLongRegister register = mock( DoubleLongRegister.class );
         when( register.readSecond() ).thenReturn( 42L );
         when( storeView.indexSample( anyLong(), any( DoubleLongRegister.class ) ) )
@@ -797,9 +805,7 @@ populator, accessor, withData( update ), index
         // and WHEN starting, i.e. completing recovery
         life.start();
 
-        // THEN our index should still have been recovered properly
-        // apparently we create updaters two times during recovery, get over it
-        verify( accessor, times( 2 ) ).newUpdater( RECOVERY );
+        verify( accessor ).newUpdater( RECOVERY );
     }
 
     @Test
@@ -1369,17 +1375,6 @@ public ResourceIterator<File> snapshotFiles()
         }
     }
 
-    @SuppressWarnings( "unchecked" )
-    private Answer nodeUpdatesAnswer( NodeUpdates... updates )
-    {
-        return invocation ->
-        {
-            Collection<NodeUpdates> target = (Collection<NodeUpdates>) invocation.getArguments()[1];
-            target.addAll( asList( updates ) );
-            return null;
-        };
-    }
-
     private IndexRule indexRule( long ruleId, int labelId, int propertyKeyId, SchemaIndexProvider.Descriptor
             providerDescriptor )
     {