Add support for secondary unit in commands

Add support for secondary unit during RelationshipCommand, RelationshipGroupCommand, PropertyCommands serialization,
support secondary unit in PhysicalLogCommandReaderV3_0.

community/kernel/src/main/java/org/neo4j/kernel/impl/api/TransactionApplierFacade.java

@@ -28,7 +28,8 @@
 
 /**
  * Wraps several {@link TransactionApplier}s. In this case, each individual visit-call will delegate to {@link
- * #visit(Command)} instead, which will call each wrapped {@link TransactionApplier} in turn. In {@link #close()},
+ * #visit(StorageCommand)} instead, which will call each wrapped {@link TransactionApplier} in turn. In
+ * {@link #close()},
  * the appliers are closed in reversed order.
  */
 public class TransactionApplierFacade extends TransactionApplier.Adapter

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

@@ -355,7 +355,7 @@ private BatchTransactionApplierFacade applier( TransactionApplicationMode mode )
 
         // Schema index application
         appliers.add( new IndexBatchTransactionApplier( indexingService, labelScanStoreSync, indexUpdatesSync,
-                neoStores.getNodeStore(), neoStores.getPropertyStore(), new PropertyLoader( neoStores ),
+                neoStores.getNodeStore(), new PropertyLoader( neoStores ),
                 indexUpdatesConverter, mode ) );
 
         // Legacy index application

community/kernel/src/main/java/org/neo4j/kernel/impl/store/record/AbstractBaseRecord.java

@@ -87,7 +87,7 @@ public void setId( long id )
     }
 
     /**
-     * Sets a secondary record unit ID for this record. If this is set to something other than {@code -1}
+     * Sets a secondary record unit ID for this record. If this is set to something other than {@link #NO_ID}
      * then {@link #requiresSecondaryUnit()} will return {@code true}.
      * Setting this id is separate from setting {@link #requiresSecondaryUnit()} since this secondary unit id
      * may be used to just free that id at the time of updating in the store if a record goes from two to one unit.

community/kernel/src/main/java/org/neo4j/kernel/impl/store/record/Record.java

@@ -45,16 +45,19 @@ public enum Record
     NO_LABELS_FIELD( (byte)0, 0 );
 
     public static final byte CREATED_IN_TX = 2;
+    public static final byte REQUIRE_SECONDARY_UNIT = 4;
+    public static final byte HAS_SECONDARY_UNIT = 8;
+
 
     private byte byteValue;
     private int intValue;
 
-    private Record( Record from )
+    Record( Record from )
     {
         this( from.byteValue, from.intValue );
     }
 
-    private Record( byte byteValue, int intValue )
+    Record( byte byteValue, int intValue )
     {
         this.byteValue = byteValue;
         this.intValue = intValue;

community/kernel/src/main/java/org/neo4j/kernel/impl/transaction/command/Command.java

@@ -156,7 +156,10 @@ void writeDynamicRecord( WritableChannel channel, DynamicRecord record ) throws
             {
                 inUse |= Record.FIRST_IN_CHAIN.byteValue();
             }
-            channel.putLong( record.getId() ).putInt( record.getType() ).put( inUse ).putInt( record.getLength() )
+            channel.putLong( record.getId() )
+                   .putInt( record.getType() )
+                   .put( inUse )
+                   .putInt( record.getLength() )
                    .putLong( record.getNextBlock() );
             byte[] data = record.getData();
             assert data != null;
@@ -165,7 +168,9 @@ void writeDynamicRecord( WritableChannel channel, DynamicRecord record ) throws
         else
         {
             byte inUse = Record.NOT_IN_USE.byteValue();
-            channel.putLong( record.getId() ).putInt( record.getType() ).put( inUse );
+            channel.putLong( record.getId() )
+                   .putInt( record.getType() )
+                   .put( inUse );
         }
     }
 
@@ -262,7 +267,9 @@ public void serialize( WritableChannel channel ) throws IOException
         private void writeRelationshipRecord( WritableChannel channel, RelationshipRecord record ) throws IOException
         {
             byte flags = bitFlags( bitFlag( record.inUse(), Record.IN_USE.byteValue() ),
-                    bitFlag( record.isCreated(), Record.CREATED_IN_TX ) );
+                                   bitFlag( record.isCreated(), Record.CREATED_IN_TX ),
+                                   bitFlag( record.requiresSecondaryUnit(), Record.REQUIRE_SECONDARY_UNIT ),
+                                   bitFlag( record.hasSecondaryUnitId(), Record.HAS_SECONDARY_UNIT ));
             channel.put( flags );
             if ( record.inUse() )
             {
@@ -271,6 +278,10 @@ private void writeRelationshipRecord( WritableChannel channel, RelationshipRecor
                        .putLong( record.getSecondPrevRel() ).putLong( record.getSecondNextRel() )
                        .putLong( record.getNextProp() )
                        .put( (byte) ((record.isFirstInFirstChain() ? 1 : 0) | (record.isFirstInSecondChain() ? 2 : 0)) );
+                if ( record.hasSecondaryUnitId() )
+                {
+                    channel.putLong( record.getSecondaryUnitId() );
+                }
             }
             else
             {
@@ -304,13 +315,20 @@ public void serialize( WritableChannel channel ) throws IOException
         private void writeRelationshipGroupRecord( WritableChannel channel, RelationshipGroupRecord record )
                 throws IOException
         {
-            channel.put( (byte) (record.inUse() ? Record.IN_USE.intValue() : Record.NOT_IN_USE.intValue()) );
+            byte flags = bitFlags( bitFlag( record.inUse(), Record.IN_USE.byteValue() ),
+                                   bitFlag( record.requiresSecondaryUnit(), Record.REQUIRE_SECONDARY_UNIT ),
+                                   bitFlag( record.hasSecondaryUnitId(), Record.HAS_SECONDARY_UNIT ) );
+            channel.put( flags );
             channel.putShort( (short) record.getType() );
             channel.putLong( record.getNext() );
             channel.putLong( record.getFirstOut() );
             channel.putLong( record.getFirstIn() );
             channel.putLong( record.getFirstLoop() );
             channel.putLong( record.getOwningNode() );
+            if ( record.hasSecondaryUnitId() )
+            {
+                channel.putLong( record.getSecondaryUnitId() );
+            }
         }
     }
 
@@ -375,13 +393,12 @@ public void serialize( WritableChannel channel ) throws IOException
 
         private void writePropertyRecord( WritableChannel channel, PropertyRecord record ) throws IOException
         {
-            byte inUse = record.inUse() ? Record.IN_USE.byteValue() : Record.NOT_IN_USE.byteValue();
-            if ( record.getRelId() != -1 )
-            {
-                // Here we add 2, i.e. set the second lsb.
-                inUse += Record.REL_PROPERTY.byteValue();
-            }
-            channel.put( inUse ); // 1
+            byte flags = bitFlags( bitFlag( record.inUse(), Record.IN_USE.byteValue() ),
+                                   bitFlag( record.getRelId() != -1, Record.REL_PROPERTY.byteValue() ),
+                                   bitFlag( record.requiresSecondaryUnit(), Record.REQUIRE_SECONDARY_UNIT ),
+                                   bitFlag( record.hasSecondaryUnitId(), Record.HAS_SECONDARY_UNIT ) );
+
+            channel.put( flags ); // 1
             channel.putLong( record.getNextProp() ).putLong( record.getPrevProp() ); // 8 + 8
             long nodeId = record.getNodeId();
             long relId = record.getRelId();
@@ -399,6 +416,10 @@ else if ( relId != -1 )
                 // prop chain
                 channel.putLong( -1 ); // 8
             }
+            if ( record.hasSecondaryUnitId() )
+            {
+                channel.putLong( record.getSecondaryUnitId() );
+            }
             channel.put( (byte) record.numberOfProperties() ); // 1
             for ( PropertyBlock block : record )
             {

community/kernel/src/main/java/org/neo4j/kernel/impl/transaction/command/IndexBatchTransactionApplier.java

@@ -68,15 +68,15 @@ public class IndexBatchTransactionApplier extends BatchTransactionApplier.Adapte
     public IndexBatchTransactionApplier( IndexingService indexingService,
             WorkSync<Supplier<LabelScanWriter>,LabelUpdateWork> labelScanStoreSync,
             WorkSync<IndexingService,IndexUpdatesWork> indexUpdatesSync,
-            NodeStore nodeStore, PropertyStore propertyStore, PropertyLoader propertyLoader,
+            NodeStore nodeStore, PropertyLoader propertyLoader,
             PropertyPhysicalToLogicalConverter indexUpdateConverter,
             TransactionApplicationMode mode )
     {
         this.indexingService = indexingService;
         this.labelScanStoreSync = labelScanStoreSync;
         this.indexUpdatesSync = indexUpdatesSync;
         this.indexUpdateConverter = indexUpdateConverter;
-        this.transactionApplier = new SingleTransactionApplier( nodeStore, propertyStore, propertyLoader, mode );
+        this.transactionApplier = new SingleTransactionApplier( nodeStore, propertyLoader, mode );
     }
 
     @Override
@@ -117,17 +117,15 @@ public void close() throws Exception
     private class SingleTransactionApplier extends TransactionApplier.Adapter
     {
         private final NodeStore nodeStore;
-        private final PropertyStore propertyStore;
         private final PropertyLoader propertyLoader;
         private final TransactionApplicationMode mode;
         private final NodePropertyCommandsExtractor indexUpdatesExtractor = new NodePropertyCommandsExtractor();
         private List<IndexRule> createdIndexes;
 
-        public SingleTransactionApplier( NodeStore nodeStore, PropertyStore propertyStore,
-                PropertyLoader propertyLoader, TransactionApplicationMode mode )
+        public SingleTransactionApplier( NodeStore nodeStore, PropertyLoader propertyLoader,
+                TransactionApplicationMode mode )
         {
             this.nodeStore = nodeStore;
-            this.propertyStore = propertyStore;
             this.propertyLoader = propertyLoader;
             this.mode = mode;
         }
@@ -164,7 +162,7 @@ private IndexUpdates indexUpdates()
         private IndexUpdates createIndexUpdates()
         {
             return mode == TransactionApplicationMode.RECOVERY ? new RecoveryIndexUpdates() :
-                new OnlineIndexUpdates( nodeStore, propertyStore, propertyLoader, indexUpdateConverter );
+                new OnlineIndexUpdates( nodeStore, propertyLoader, indexUpdateConverter );
         }
 
         @Override

community/kernel/src/main/java/org/neo4j/kernel/impl/transaction/command/PhysicalLogCommandReaderV3_0.java

@@ -58,7 +58,6 @@
 import static org.neo4j.kernel.impl.transaction.command.CommandReading.PROPERTY_DELETED_DYNAMIC_RECORD_ADDER;
 import static org.neo4j.kernel.impl.transaction.command.CommandReading.PROPERTY_INDEX_DYNAMIC_RECORD_ADDER;
 import static org.neo4j.kernel.impl.util.Bits.bitFlag;
-import static org.neo4j.kernel.impl.util.Bits.notFlag;
 import static org.neo4j.kernel.impl.util.IoPrimitiveUtils.read2bLengthAndString;
 import static org.neo4j.kernel.impl.util.IoPrimitiveUtils.read2bMap;
 import static org.neo4j.kernel.impl.util.IoPrimitiveUtils.read3bLengthAndString;
@@ -206,12 +205,11 @@ private Command visitRelationshipGroupCommand( ReadableChannel channel ) throws
     private RelationshipGroupRecord readRelationshipGroupRecord( long id, ReadableChannel channel )
             throws IOException
     {
-        byte inUseByte = channel.get();
-        boolean inUse = inUseByte == Record.IN_USE.byteValue();
-        if ( inUseByte != Record.IN_USE.byteValue() && inUseByte != Record.NOT_IN_USE.byteValue() )
-        {
-            throw new IOException( "Illegal in use flag: " + inUseByte );
-        }
+        byte flags = channel.get();
+        boolean inUse = bitFlag( flags, Record.IN_USE.byteValue() );
+        boolean requireSecondaryUnit = bitFlag( flags, Record.REQUIRE_SECONDARY_UNIT );
+        boolean hasSecondaryUnit = bitFlag( flags, Record.HAS_SECONDARY_UNIT );
+
         int type = channel.getShort();
         RelationshipGroupRecord record = new RelationshipGroupRecord( id, type );
         record.setInUse( inUse );
@@ -220,6 +218,11 @@ private RelationshipGroupRecord readRelationshipGroupRecord( long id, ReadableCh
         record.setFirstIn( channel.getLong() );
         record.setFirstLoop( channel.getLong() );
         record.setOwningNode( channel.getLong() );
+        record.setRequiresSecondaryUnit( requireSecondaryUnit );
+        if ( hasSecondaryUnit )
+        {
+            record.setSecondaryUnitId( channel.getLong() );
+        }
         return record;
     }
 
@@ -430,20 +433,15 @@ record = new NodeRecord( id );
     private RelationshipRecord readRelationshipRecord( long id, ReadableChannel channel ) throws IOException
     {
         byte flags = channel.get();
-        boolean inUse = false;
-        if ( notFlag( notFlag( flags, Record.IN_USE.byteValue() ), Record.CREATED_IN_TX ) != 0 )
-        {
-            throw new IOException( "Illegal in use flag: " + flags );
-        }
-        if ( bitFlag( flags, Record.IN_USE.byteValue() ) )
-        {
-            inUse = true;
-        }
+        boolean inUse = bitFlag( flags, Record.IN_USE.byteValue() );
+        boolean requiresSecondaryUnit = bitFlag( flags, Record.REQUIRE_SECONDARY_UNIT );
+        boolean hasSecondaryUnit = bitFlag( flags, Record.HAS_SECONDARY_UNIT );
         RelationshipRecord record;
         if ( inUse )
         {
             record = new RelationshipRecord( id, channel.getLong(), channel.getLong(), channel.getInt() );
             record.setInUse( true );
+            record.setRequiresSecondaryUnit( requiresSecondaryUnit );
             record.setFirstPrevRel( channel.getLong() );
             record.setFirstNextRel( channel.getLong() );
             record.setSecondPrevRel( channel.getLong() );
@@ -452,6 +450,10 @@ record = new RelationshipRecord( id, channel.getLong(), channel.getLong(), chann
             byte extraByte = channel.get();
             record.setFirstInFirstChain( (extraByte & 0x1) > 0 );
             record.setFirstInSecondChain( (extraByte & 0x2) > 0 );
+            if (hasSecondaryUnit)
+            {
+                record.setSecondaryUnitId( channel.getLong() );
+            }
         }
         else
         {
@@ -517,21 +519,20 @@ private PropertyRecord readPropertyRecord( long id, ReadableChannel channel ) th
         // in_use(byte)+type(int)+key_indexId(int)+prop_blockId(long)+
         // prev_prop_id(long)+next_prop_id(long)
         PropertyRecord record = new PropertyRecord( id );
-        byte inUseFlag = channel.get(); // 1
+        byte flags = channel.get(); // 1
+
+        boolean inUse = bitFlag( flags, Record.IN_USE.byteValue() );
+        boolean nodeProperty = !bitFlag( flags, Record.REL_PROPERTY.byteValue() );
+        boolean requireSecondaryUnit = bitFlag( flags, Record.REQUIRE_SECONDARY_UNIT );
+        boolean hasSecondaryUnit = bitFlag( flags, Record.HAS_SECONDARY_UNIT );
+
+        record.setRequiresSecondaryUnit( requireSecondaryUnit );
+
         long nextProp = channel.getLong(); // 8
         long prevProp = channel.getLong(); // 8
         record.setNextProp( nextProp );
         record.setPrevProp( prevProp );
-        boolean inUse = false;
-        if ( (inUseFlag & Record.IN_USE.byteValue()) == Record.IN_USE.byteValue() )
-        {
-            inUse = true;
-        }
-        boolean nodeProperty = true;
-        if ( (inUseFlag & Record.REL_PROPERTY.byteValue()) == Record.REL_PROPERTY.byteValue() )
-        {
-            nodeProperty = false;
-        }
+
         long primitiveId = channel.getLong(); // 8
         if ( primitiveId != -1 && nodeProperty )
         {
@@ -541,6 +542,10 @@ else if ( primitiveId != -1 )
         {
             record.setRelId( primitiveId );
         }
+        if (hasSecondaryUnit)
+        {
+            record.setSecondaryUnitId( channel.getLong() );
+        }
         int nrPropBlocks = channel.get();
         assert nrPropBlocks >= 0;
         if ( nrPropBlocks > 0 )

community/kernel/src/main/java/org/neo4j/kernel/impl/transaction/log/entry/LogEntryWriter.java

@@ -42,16 +42,7 @@ public class LogEntryWriter
     public LogEntryWriter( FlushableChannel channel )
     {
         this.channel = channel;
-        this.serializer = new Visitor<StorageCommand,IOException>()
-        {
-            @Override
-            public boolean visit( StorageCommand command ) throws IOException
-            {
-                writeLogEntryHeader( COMMAND );
-                command.serialize( channel );
-                return false;
-            }
-        };
+        this.serializer = new StorageCommandSerializer( channel );
     }
 
     private void writeLogEntryHeader( byte type ) throws IOException
@@ -92,4 +83,22 @@ public void writeCheckPointEntry( LogPosition logPosition ) throws IOException
         channel.putLong( logPosition.getLogVersion() ).
                 putLong( logPosition.getByteOffset() );
     }
+
+    private class StorageCommandSerializer implements Visitor<StorageCommand,IOException>
+    {
+        private final FlushableChannel channel;
+
+        public StorageCommandSerializer( FlushableChannel channel )
+        {
+            this.channel = channel;
+        }
+
+        @Override
+        public boolean visit( StorageCommand command ) throws IOException
+        {
+            writeLogEntryHeader( COMMAND );
+            command.serialize( channel );
+            return false;
+        }
+    }
 }

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

@@ -64,19 +64,16 @@
 public class OnlineIndexUpdates implements IndexUpdates
 {
     private final NodeStore nodeStore;
-    private final PropertyStore propertyStore;
     private final PropertyLoader propertyLoader;
     private final PropertyPhysicalToLogicalConverter converter;
     private final Collection<NodePropertyUpdate> updates = new ArrayList<>();
     private NodeRecord nodeRecord;
 
     public OnlineIndexUpdates( NodeStore nodeStore,
-                             PropertyStore propertyStore,
                              PropertyLoader propertyLoader,
                              PropertyPhysicalToLogicalConverter converter )
     {
         this.nodeStore = nodeStore;
-        this.propertyStore = propertyStore;
         this.propertyLoader = propertyLoader;
         this.converter = converter;
     }

community/kernel/src/test/java/org/neo4j/kernel/impl/transaction/command/IndexBatchTransactionApplierTest.java

@@ -61,8 +61,8 @@ public void shouldProvideLabelScanStoreUpdatesSortedByNodeId() throws Exception
         TransactionToApply tx = mock( TransactionToApply.class );
         PropertyStore propertyStore = mock( PropertyStore.class );
         try ( IndexBatchTransactionApplier applier = new IndexBatchTransactionApplier( indexing, labelScanSync,
-                indexUpdatesSync, mock( NodeStore.class ), propertyStore,
-                mock( PropertyLoader.class ), new PropertyPhysicalToLogicalConverter( propertyStore ),
+                indexUpdatesSync, mock( NodeStore.class ), mock( PropertyLoader.class ),
+                new PropertyPhysicalToLogicalConverter( propertyStore ),
                 TransactionApplicationMode.INTERNAL ) )
         {
             try ( TransactionApplier txApplier = applier.startTx( tx ) )

community/kernel/src/test/java/org/neo4j/kernel/impl/transaction/command/NeoStoreTransactionApplierTest.java

@@ -931,7 +931,7 @@ private BatchTransactionApplier newApplierFacade( BatchTransactionApplier... app
     private BatchTransactionApplier newIndexApplier()
     {
         return new IndexBatchTransactionApplier( indexingService, labelScanStoreSynchronizer,
-                indexUpdatesSync, nodeStore, propertyStore, new PropertyLoader( neoStores ),
+                indexUpdatesSync, nodeStore, new PropertyLoader( neoStores ),
                 new PropertyPhysicalToLogicalConverter( propertyStore ), INTERNAL );
     }
 }