Strips a bunch of logic from store classes

Which is a beginning towards having "naked" stores, ideally only a single store
class accepting a record format and all logic living outside. We're not there
yet, but:
- Moved a bunch of things from NodeStore --> DynamicNodeLabels
- Removes some stuff from RelationshipGroupStore, it's now naked
- Attempt to refer to RecordStore directly instead of RelationshipGroupStore
- Removes physical --> logical converter from PropertyStore

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

@@ -33,6 +33,7 @@
 import org.neo4j.kernel.impl.store.InvalidRecordException;
 import org.neo4j.kernel.impl.store.NeoStores;
 import org.neo4j.kernel.impl.store.NodeStore;
+import org.neo4j.kernel.impl.store.RecordStore;
 import org.neo4j.kernel.impl.store.RelationshipGroupStore;
 import org.neo4j.kernel.impl.store.RelationshipStore;
 import org.neo4j.kernel.impl.store.record.NodeRecord;
@@ -59,7 +60,7 @@ public abstract class StoreAbstractNodeCursor extends NodeItemHelper implements
 {
     protected final NodeRecord nodeRecord;
     protected NodeStore nodeStore;
-    protected RelationshipGroupStore relationshipGroupStore;
+    protected RecordStore<RelationshipGroupRecord> relationshipGroupStore;
     protected RelationshipStore relationshipStore;
     protected final LockService lockService;
     protected StoreStatement storeStatement;

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

@@ -24,9 +24,10 @@
 import org.neo4j.kernel.impl.locking.Lock;
 import org.neo4j.kernel.impl.locking.LockService;
 import org.neo4j.kernel.impl.store.NeoStores;
-import org.neo4j.kernel.impl.store.RelationshipGroupStore;
+import org.neo4j.kernel.impl.store.RecordStore;
 import org.neo4j.kernel.impl.store.RelationshipStore;
 import org.neo4j.kernel.impl.store.record.Record;
+import org.neo4j.kernel.impl.store.record.RelationshipGroupRecord;
 import org.neo4j.kernel.impl.store.record.RelationshipRecord;
 import org.neo4j.kernel.impl.util.InstanceCache;
 import org.neo4j.storageengine.api.PropertyItem;
@@ -43,7 +44,7 @@ public abstract class StoreAbstractRelationshipCursor extends EntityItemHelper
 {
     protected final RelationshipRecord relationshipRecord;
     protected final RelationshipStore relationshipStore;
-    protected final RelationshipGroupStore relationshipGroupStore;
+    protected final RecordStore<RelationshipGroupRecord> relationshipGroupStore;
     private final LockService lockService;
     protected StoreStatement storeStatement;
 

community/kernel/src/main/java/org/neo4j/kernel/impl/storageengine/impl/recordstorage/RecordStorageEngine.java

@@ -52,6 +52,7 @@
 import org.neo4j.kernel.impl.api.TransactionApplierFacade;
 import org.neo4j.kernel.impl.api.index.IndexingService;
 import org.neo4j.kernel.impl.api.index.IndexingServiceFactory;
+import org.neo4j.kernel.impl.api.index.PropertyPhysicalToLogicalConverter;
 import org.neo4j.kernel.impl.api.index.sampling.IndexSamplingConfig;
 import org.neo4j.kernel.impl.api.scan.LabelScanStoreProvider;
 import org.neo4j.kernel.impl.api.store.CacheLayer;
@@ -150,6 +151,7 @@ public class RecordStorageEngine implements StorageEngine, Lifecycle
     private final WorkSync<IndexingService,IndexUpdatesWork> indexUpdatesSync;
     private final NeoStoreIndexStoreView indexStoreView;
     private final LegacyIndexProviderLookup legacyIndexProviderLookup;
+    private final PropertyPhysicalToLogicalConverter indexUpdatesConverter;
 
     // Immutable state for creating/applying commands
     private final Loaders loaders;
@@ -199,6 +201,7 @@ public RecordStorageEngine(
 
         try
         {
+            indexUpdatesConverter = new PropertyPhysicalToLogicalConverter( neoStores.getPropertyStore() );
             schemaCache = new SchemaCache( constraintSemantics, Collections.<SchemaRule>emptyList() );
             schemaStorage = new SchemaStorage( neoStores.getSchemaStore() );
 
@@ -358,7 +361,8 @@ private BatchTransactionApplierFacade applier( TransactionApplicationMode mode )
 
         // Schema index application
         appliers.add( new IndexBatchTransactionApplier( indexingService, labelScanStoreSync, indexUpdatesSync,
-                neoStores.getNodeStore(), neoStores.getPropertyStore(), new PropertyLoader( neoStores ), mode ) );
+                neoStores.getNodeStore(), neoStores.getPropertyStore(), new PropertyLoader( neoStores ),
+                indexUpdatesConverter, mode ) );
 
         // Legacy index application
         appliers.add(

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

@@ -1178,6 +1178,12 @@ public String toString()
         return getClass().getSimpleName();
     }
 
+    @Override
+    public int getStoreHeaderInt()
+    {
+        throw new UnsupportedOperationException( "No header" );
+    }
+
     public static abstract class Configuration
     {
         public static final Setting<Boolean> rebuild_idgenerators_fast =

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

@@ -111,6 +111,12 @@ protected boolean isInUse( PageCursor cursor )
         return recordFormat.isInUse( cursor );
     }
 
+    @Override
+    public int getStoreHeaderInt()
+    {
+        return ((IntStoreHeader) storeHeader).value();
+    }
+
     @Override
     public <FAILURE extends Exception> void
             accept( org.neo4j.kernel.impl.store.RecordStore.Processor<FAILURE> processor, RECORD record ) throws FAILURE

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

@@ -19,10 +19,12 @@
  */
 package org.neo4j.kernel.impl.store;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Iterator;
 
+import org.neo4j.helpers.collection.Pair;
 import org.neo4j.kernel.impl.store.record.DynamicRecord;
 import org.neo4j.kernel.impl.store.record.NodeRecord;
 
@@ -36,7 +38,6 @@
 import static org.neo4j.kernel.impl.store.NodeLabelsField.fieldPointsToDynamicRecordOfLabels;
 import static org.neo4j.kernel.impl.store.NodeLabelsField.firstDynamicLabelRecordId;
 import static org.neo4j.kernel.impl.store.NodeLabelsField.parseLabelsBody;
-import static org.neo4j.kernel.impl.store.NodeStore.getDynamicLabelsArrayFromHeavyRecords;
 import static org.neo4j.kernel.impl.store.PropertyType.ARRAY;
 
 public class DynamicNodeLabels implements NodeLabels
@@ -62,7 +63,7 @@ public static long[] get( NodeRecord node, NodeStore nodeStore )
         {
             nodeStore.ensureHeavy( node, firstDynamicLabelRecordId( node.getLabelField() ) );
         }
-        return nodeStore.getDynamicLabelsArray( node.getUsedDynamicLabelRecords() );
+        return getDynamicLabelsArray( node.getUsedDynamicLabelRecords(), nodeStore.getDynamicLabelStore() );
     }
 
     @Override
@@ -104,7 +105,7 @@ public static Collection<DynamicRecord> putSorted( NodeRecord node, long[] label
         {
             Iterator<DynamicRecord> recycledRecords = changedDynamicRecords.iterator();
             Collection<DynamicRecord> allocatedRecords =
-                    NodeStore.allocateRecordsForDynamicLabels( node.getId(), labelIds,
+                    allocateRecordsForDynamicLabels( node.getId(), labelIds,
                             recycledRecords, allocator );
             // Set the rest of the previously set dynamic records as !inUse
             while ( recycledRecords.hasNext() )
@@ -125,10 +126,10 @@ public Collection<DynamicRecord> add( long labelId, NodeStore nodeStore, Dynamic
     {
         nodeStore.ensureHeavy( node, firstDynamicLabelRecordId( labelField ) );
         Collection<DynamicRecord> existingRecords = node.getDynamicLabelRecords();
-        long[] existingLabelIds = nodeStore.getDynamicLabelsArray( existingRecords );
+        long[] existingLabelIds = getDynamicLabelsArray( existingRecords, nodeStore.getDynamicLabelStore() );
         long[] newLabelIds = LabelIdArray.concatAndSort( existingLabelIds, labelId );
         Collection<DynamicRecord> changedDynamicRecords =
-                NodeStore.allocateRecordsForDynamicLabels( node.getId(), newLabelIds, existingRecords.iterator(), allocator );
+                allocateRecordsForDynamicLabels( node.getId(), newLabelIds, existingRecords.iterator(), allocator );
         node.setLabelField( dynamicPointer( changedDynamicRecords ), changedDynamicRecords );
         return changedDynamicRecords;
     }
@@ -137,7 +138,8 @@ public Collection<DynamicRecord> add( long labelId, NodeStore nodeStore, Dynamic
     public Collection<DynamicRecord> remove( long labelId, NodeStore nodeStore )
     {
         nodeStore.ensureHeavy( node, firstDynamicLabelRecordId( labelField ) );
-        long[] existingLabelIds = nodeStore.getDynamicLabelsArray( node.getUsedDynamicLabelRecords() );
+        long[] existingLabelIds = getDynamicLabelsArray( node.getUsedDynamicLabelRecords(),
+                nodeStore.getDynamicLabelStore() );
         long[] newLabelIds = filter( existingLabelIds, labelId );
         Collection<DynamicRecord> existingRecords = node.getDynamicLabelRecords();
         if ( InlineNodeLabels.tryInlineInNodeRecord( node, newLabelIds, existingRecords ) )
@@ -146,8 +148,8 @@ public Collection<DynamicRecord> remove( long labelId, NodeStore nodeStore )
         }
         else
         {
-            Collection<DynamicRecord> newRecords =
-                    nodeStore.allocateRecordsForDynamicLabels( node.getId(), newLabelIds, existingRecords.iterator() );
+            Collection<DynamicRecord> newRecords = allocateRecordsForDynamicLabels( node.getId(),
+                    newLabelIds, existingRecords.iterator(), nodeStore.getDynamicLabelStore() );
             node.setLabelField( dynamicPointer( newRecords ), existingRecords );
             if ( !newRecords.equals( existingRecords ) )
             {   // One less dynamic record, mark that one as not in use
@@ -197,4 +199,42 @@ public String toString()
         return format( "Dynamic(id:%d,[%s])", firstDynamicLabelRecordId( node.getLabelField() ),
                 Arrays.toString( getDynamicLabelsArrayFromHeavyRecords( node.getUsedDynamicLabelRecords() ) ) );
     }
+
+    public static Collection<DynamicRecord> allocateRecordsForDynamicLabels( long nodeId, long[] labels,
+            Iterator<DynamicRecord> useFirst, AbstractDynamicStore dynamicLabelStore )
+    {
+        return allocateRecordsForDynamicLabels( nodeId, labels, useFirst, (DynamicRecordAllocator)dynamicLabelStore );
+    }
+
+    public static Collection<DynamicRecord> allocateRecordsForDynamicLabels( long nodeId, long[] labels,
+            Iterator<DynamicRecord> useFirst, DynamicRecordAllocator allocator )
+    {
+        long[] storedLongs = LabelIdArray.prependNodeId( nodeId, labels );
+        Collection<DynamicRecord> records = new ArrayList<>();
+        DynamicArrayStore.allocateRecords( records, storedLongs, useFirst, allocator );
+        return records;
+    }
+
+    public static long[] getDynamicLabelsArray( Iterable<DynamicRecord> records,
+            AbstractDynamicStore dynamicLabelStore )
+    {
+        long[] storedLongs = (long[])
+            DynamicArrayStore.getRightArray( dynamicLabelStore.readFullByteArray( records, PropertyType.ARRAY ) );
+        return LabelIdArray.stripNodeId( storedLongs );
+    }
+
+    public static long[] getDynamicLabelsArrayFromHeavyRecords( Iterable<DynamicRecord> records )
+    {
+        long[] storedLongs = (long[])
+            DynamicArrayStore.getRightArray( readFullByteArrayFromHeavyRecords( records, PropertyType.ARRAY ) );
+        return LabelIdArray.stripNodeId( storedLongs );
+    }
+
+    public static Pair<Long, long[]> getDynamicLabelsArrayAndOwner( Iterable<DynamicRecord> records,
+            AbstractDynamicStore dynamicLabelStore )
+    {
+        long[] storedLongs = (long[])
+                DynamicArrayStore.getRightArray( dynamicLabelStore.readFullByteArray( records, PropertyType.ARRAY ) );
+        return Pair.of(storedLongs[0], LabelIdArray.stripNodeId( storedLongs ));
+    }
 }

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

@@ -42,6 +42,7 @@
 import org.neo4j.kernel.impl.store.id.IdGeneratorFactory;
 import org.neo4j.kernel.impl.store.id.IdType;
 import org.neo4j.kernel.impl.store.kvstore.DataInitializer;
+import org.neo4j.kernel.impl.store.record.RelationshipGroupRecord;
 import org.neo4j.kernel.info.DiagnosticsManager;
 import org.neo4j.logging.Log;
 import org.neo4j.logging.LogProvider;
@@ -326,10 +327,7 @@ private DynamicStringStore getPropertyKeyTokenNamesStore()
         return (DynamicStringStore) getStore( StoreType.PROPERTY_KEY_TOKEN_NAME );
     }
 
-    /**
-     * @return the {@link RelationshipGroupStore}
-     */
-    public RelationshipGroupStore getRelationshipGroupStore()
+    public RecordStore<RelationshipGroupRecord> getRelationshipGroupStore()
     {
         return (RelationshipGroupStore) getStore( StoreType.RELATIONSHIP_GROUP );
     }

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

@@ -20,12 +20,8 @@
 package org.neo4j.kernel.impl.store;
 
 import java.io.File;
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collection;
-import java.util.Iterator;
 
-import org.neo4j.helpers.collection.Pair;
 import org.neo4j.io.pagecache.PageCache;
 import org.neo4j.kernel.configuration.Config;
 import org.neo4j.kernel.impl.store.format.RecordFormat;
@@ -37,7 +33,6 @@
 import org.neo4j.kernel.impl.util.Bits;
 import org.neo4j.logging.LogProvider;
 
-import static org.neo4j.kernel.impl.store.AbstractDynamicStore.readFullByteArrayFromHeavyRecords;
 import static org.neo4j.kernel.impl.store.NoStoreHeaderFormat.NO_STORE_HEADER_FORMAT;
 
 /**
@@ -121,42 +116,6 @@ public DynamicArrayStore getDynamicLabelStore()
         return dynamicLabelStore;
     }
 
-    public Collection<DynamicRecord> allocateRecordsForDynamicLabels( long nodeId, long[] labels,
-            Iterator<DynamicRecord> useFirst )
-    {
-        return allocateRecordsForDynamicLabels( nodeId, labels, useFirst, dynamicLabelStore );
-    }
-
-    public static Collection<DynamicRecord> allocateRecordsForDynamicLabels( long nodeId, long[] labels,
-            Iterator<DynamicRecord> useFirst, DynamicRecordAllocator allocator )
-    {
-        long[] storedLongs = LabelIdArray.prependNodeId( nodeId, labels );
-        Collection<DynamicRecord> records = new ArrayList<>();
-        DynamicArrayStore.allocateRecords( records, storedLongs, useFirst, allocator );
-        return records;
-    }
-
-    public long[] getDynamicLabelsArray( Iterable<DynamicRecord> records )
-    {
-        long[] storedLongs = (long[])
-            DynamicArrayStore.getRightArray( dynamicLabelStore.readFullByteArray( records, PropertyType.ARRAY ) );
-        return LabelIdArray.stripNodeId( storedLongs );
-    }
-
-    public static long[] getDynamicLabelsArrayFromHeavyRecords( Iterable<DynamicRecord> records )
-    {
-        long[] storedLongs = (long[])
-            DynamicArrayStore.getRightArray( readFullByteArrayFromHeavyRecords( records, PropertyType.ARRAY ) );
-        return LabelIdArray.stripNodeId( storedLongs );
-    }
-
-    public Pair<Long, long[]> getDynamicLabelsArrayAndOwner( Iterable<DynamicRecord> records )
-    {
-        long[] storedLongs = (long[])
-                DynamicArrayStore.getRightArray( dynamicLabelStore.readFullByteArray( records, PropertyType.ARRAY ) );
-        return Pair.of(storedLongs[0], LabelIdArray.stripNodeId( storedLongs ));
-    }
-
     public void updateDynamicLabelRecords( Iterable<DynamicRecord> dynamicLabelRecords )
     {
         for ( DynamicRecord record : dynamicLabelRecords )

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

@@ -34,9 +34,7 @@
 import org.neo4j.io.pagecache.PageCache;
 import org.neo4j.io.pagecache.PageCursor;
 import org.neo4j.io.pagecache.PagedFile;
-import org.neo4j.kernel.api.index.NodePropertyUpdate;
 import org.neo4j.kernel.configuration.Config;
-import org.neo4j.kernel.impl.api.index.PropertyPhysicalToLogicalConverter;
 import org.neo4j.kernel.impl.api.store.PropertyRecordCursor;
 import org.neo4j.kernel.impl.store.format.RecordFormat;
 import org.neo4j.kernel.impl.store.id.IdGeneratorFactory;
@@ -47,7 +45,6 @@
 import org.neo4j.kernel.impl.store.record.PropertyRecord;
 import org.neo4j.kernel.impl.store.record.Record;
 import org.neo4j.kernel.impl.store.record.RecordLoad;
-import org.neo4j.kernel.impl.transaction.state.PropertyRecordChange;
 import org.neo4j.logging.LogProvider;
 
 import static org.neo4j.helpers.collection.IteratorUtil.first;
@@ -70,7 +67,6 @@ public static abstract class Configuration extends CommonAbstractStore.Configura
     private final DynamicStringStore stringStore;
     private final PropertyKeyTokenStore propertyKeyTokenStore;
     private final DynamicArrayStore arrayStore;
-    private final PropertyPhysicalToLogicalConverter physicalToLogicalConverter;
 
     public PropertyStore(
             File fileName,
@@ -88,7 +84,6 @@ public PropertyStore(
         this.stringStore = stringPropertyStore;
         this.propertyKeyTokenStore = propertyKeyTokenStore;
         this.arrayStore = arrayPropertyStore;
-        this.physicalToLogicalConverter = new PropertyPhysicalToLogicalConverter( this );
     }
 
     @Override
@@ -428,35 +423,6 @@ public Collection<PropertyRecord> getPropertyRecordChain( long firstRecordId,
         return toReturn;
     }
 
-    public void toLogicalUpdates( Collection<NodePropertyUpdate> target,
-            Iterable<PropertyRecordChange> changes,
-            long[] nodeLabelsBefore,
-            long[] nodeLabelsAfter )
-    {
-        physicalToLogicalConverter.apply( target, changes, nodeLabelsBefore, nodeLabelsAfter );
-    }
-
-    /**
-     * For property records there's no "inUse" byte and we need to read the whole record to
-     * see if there are any PropertyBlocks in use in it.
-     */
-    @Override
-    protected boolean isInUse( PageCursor cursor )
-    {
-        int offsetAtBeginning = cursor.getOffset();
-        cursor.setOffset( offsetAtBeginning + 1/*mod*/ + 4/*prev*/ + 4/*next*/ );
-        int recordSize = getRecordSize();
-        while ( cursor.getOffset() - offsetAtBeginning < recordSize )
-        {
-            long block = cursor.getLong();
-            if ( PropertyType.getPropertyType( block, true ) != null )
-            {
-                return true;
-            }
-        }
-        return false;
-    }
-
     @Override
     public PropertyRecord newRecord()
     {