Use directly a PageCursor in NodeCursor

community/kernel/src/main/java/org/neo4j/kernel/impl/api/store/NodeCursor.java

@@ -19,18 +19,21 @@
  */
 package org.neo4j.kernel.impl.api.store;
 
+import java.io.IOException;
 import java.util.Iterator;
 import java.util.function.Consumer;
 
 import org.neo4j.collection.primitive.PrimitiveIntSet;
 import org.neo4j.cursor.Cursor;
+import org.neo4j.io.pagecache.PageCursor;
 import org.neo4j.kernel.impl.locking.Lock;
 import org.neo4j.kernel.impl.locking.LockService;
 import org.neo4j.kernel.impl.store.NodeLabelsField;
+import org.neo4j.kernel.impl.store.NodeStore;
 import org.neo4j.kernel.impl.store.RecordCursors;
+import org.neo4j.kernel.impl.store.UnderlyingStorageException;
 import org.neo4j.kernel.impl.store.record.NodeRecord;
 import org.neo4j.kernel.impl.store.record.Record;
-import org.neo4j.kernel.impl.store.record.RecordLoad;
 import org.neo4j.kernel.impl.util.IoPrimitiveUtils;
 import org.neo4j.storageengine.api.NodeItem;
 import org.neo4j.storageengine.api.txstate.NodeState;
@@ -48,6 +51,7 @@ public class NodeCursor implements NodeItem, Cursor<NodeItem>
 {
     private final NodeRecord nodeRecord;
     private final Consumer<NodeCursor> instanceCache;
+    private final NodeStore nodeStore;
     private final RecordCursors recordCursors;
     private final LockService lockService;
 
@@ -56,18 +60,21 @@ public class NodeCursor implements NodeItem, Cursor<NodeItem>
     private boolean fetched;
     private long[] labels;
     private Iterator<Long> added;
+    private PageCursor nodePageCursor;
 
-    NodeCursor( NodeRecord nodeRecord, Consumer<NodeCursor> instanceCache, RecordCursors recordCursors,
-            LockService lockService )
+    NodeCursor( NodeRecord nodeRecord, Consumer<NodeCursor> instanceCache, NodeStore nodeStore,
+            RecordCursors recordCursors, LockService lockService )
     {
         this.nodeRecord = nodeRecord;
         this.instanceCache = instanceCache;
+        this.nodeStore = nodeStore;
         this.recordCursors = recordCursors;
         this.lockService = lockService;
     }
 
     public Cursor<NodeItem> init( Progression progression, ReadableTransactionState state )
     {
+        this.nodePageCursor = nodeStore.newPageCursor();
         this.progression = progression;
         this.state = state;
         this.added = state != null && progression.mode() == APPEND
@@ -88,8 +95,7 @@ private boolean fetchNext()
         long id;
         while ( (id = progression.nextId()) >= 0 )
         {
-            if ( (state == null || !state.nodeIsDeletedInThisTx( id )) &&
-                    recordCursors.node().next( id, nodeRecord, RecordLoad.CHECK ) )
+            if ( (state == null || !state.nodeIsDeletedInThisTx( id )) && readNodeRecord( id ) )
             {
                 return true;
             }
@@ -110,6 +116,20 @@ private boolean fetchNext()
         return false;
     }
 
+    private boolean readNodeRecord( long id )
+    {
+        try
+        {
+            nodeRecord.clear();
+            nodeStore.readIntoRecord( id, nodeRecord, CHECK, nodePageCursor );
+            return nodeRecord.inUse();
+        }
+        catch ( IOException e )
+        {
+            throw new UnderlyingStorageException( e );
+        }
+    }
+
     private void recordFromTxState( long id )
     {
         nodeRecord.clear();
@@ -122,6 +142,8 @@ public void close()
         labels = null;
         added = null;
         state = null;
+        nodePageCursor.close();
+        nodePageCursor = null;
         instanceCache.accept( this );
     }
 
@@ -222,7 +244,7 @@ private Lock acquireLock()
             try
             {
                 // It's safer to re-read the node record here, specifically nextProp, after acquiring the lock
-                if ( !recordCursors.node().next( nodeRecord.getId(), nodeRecord, CHECK ) )
+                if ( !readNodeRecord( nodeRecord.getId() ) )
                 {
                     // So it looks like the node has been deleted. The current behavior of NodeStore#loadRecord
                     // is to only set the inUse field on loading an unused record. This should (and will)

community/kernel/src/main/java/org/neo4j/kernel/impl/api/store/StoreStatement.java

@@ -89,7 +89,7 @@ public StoreStatement( NeoStores neoStores, Supplier<IndexReaderFactory> indexRe
             @Override
             protected NodeCursor create()
             {
-                return new NodeCursor( nodeStore.newRecord(), this, recordCursors, lockService );
+                return new NodeCursor( nodeStore.newRecord(), this, nodeStore, recordCursors, lockService );
             }
         };
         singleRelationshipCursor = new InstanceCache<StoreSingleRelationshipCursor>()

community/kernel/src/main/java/org/neo4j/kernel/impl/store/CommonAbstractStore.java

@@ -1028,7 +1028,7 @@ public RECORD getRecord( long id, RECORD record, RecordLoad mode )
         }
     }
 
-    void readIntoRecord( long id, RECORD record, RecordLoad mode, PageCursor cursor ) throws IOException
+    public void readIntoRecord( long id, RECORD record, RecordLoad mode, PageCursor cursor ) throws IOException
     {
         // Mark the record with this id regardless of whether or not we load the contents of it.
         // This is done in this method since there are multiple call sites and they all want the id
@@ -1123,6 +1123,18 @@ public Collection<RECORD> getRecords( long firstId, RecordLoad mode )
         }
     }
 
+    public PageCursor newPageCursor()
+    {
+        try
+        {
+            return storeFile.io( getNumberOfReservedLowIds(), PF_SHARED_READ_LOCK );
+        }
+        catch ( IOException e )
+        {
+            throw new UnderlyingStorageException( e );
+        }
+    }
+
     @Override
     public RecordCursor<RECORD> newRecordCursor( final RECORD record )
     {

community/kernel/src/main/java/org/neo4j/kernel/impl/store/RecordCursors.java

@@ -24,7 +24,6 @@
 import org.neo4j.kernel.impl.store.record.DynamicRecord;
 import org.neo4j.kernel.impl.store.record.NodeRecord;
 import org.neo4j.kernel.impl.store.record.PropertyRecord;
-import org.neo4j.kernel.impl.store.record.RecordLoad;
 import org.neo4j.kernel.impl.store.record.RelationshipGroupRecord;
 import org.neo4j.kernel.impl.store.record.RelationshipRecord;
 
@@ -35,7 +34,6 @@
  */
 public class RecordCursors implements AutoCloseable
 {
-    private final RecordCursor<NodeRecord> node;
     private final RecordCursor<RelationshipRecord> relationship;
     private final RecordCursor<RelationshipGroupRecord> relationshipGroup;
     private final RecordCursor<PropertyRecord> property;
@@ -45,7 +43,6 @@ public class RecordCursors implements AutoCloseable
 
     public RecordCursors( NeoStores neoStores )
     {
-        node = newCursor( neoStores.getNodeStore() );
         relationship = newCursor( neoStores.getRelationshipStore() );
         relationshipGroup = newCursor( neoStores.getRelationshipGroupStore() );
         property = newCursor( neoStores.getPropertyStore() );
@@ -62,13 +59,8 @@ private static <R extends AbstractBaseRecord> RecordCursor<R> newCursor( RecordS
     @Override
     public void close()
     {
-        IOUtils.closeAll( RuntimeException.class,
-                node, relationship, relationshipGroup, property, propertyArray, propertyString, label );
-    }
-
-    public RecordCursor<NodeRecord> node()
-    {
-        return node;
+        IOUtils.closeAll( RuntimeException.class, relationship, relationshipGroup, property, propertyArray,
+                propertyString, label );
     }
 
     public RecordCursor<RelationshipRecord> relationship()