Relationship Chain check for non used record in a chain.

community/consistency-check/src/main/java/org/neo4j/consistency/checking/RelationshipRecordCheck.java

@@ -35,6 +35,7 @@
 
 import static org.neo4j.consistency.checking.cache.CacheSlots.RelationshipLink.NEXT;
 import static org.neo4j.consistency.checking.cache.CacheSlots.RelationshipLink.PREV;
+import static org.neo4j.consistency.checking.cache.CacheSlots.RelationshipLink.SLOT_IN_USE;
 import static org.neo4j.consistency.checking.cache.CacheSlots.RelationshipLink.SLOT_REFERENCE;
 import static org.neo4j.consistency.checking.cache.CacheSlots.RelationshipLink.SLOT_RELATIONSHIP_ID;
 import static org.neo4j.consistency.checking.cache.CacheSlots.RelationshipLink.SLOT_SOURCE_OR_TARGET;
@@ -168,6 +169,7 @@ RelationshipRecord populateRelationshipFromCache( long nodeId, RelationshipRecor
                 {
                     rel.setSecondNextRel( cacheAccess.getFromCache( nodeId, SLOT_REFERENCE ) );
                 }
+                rel.setInUse( cacheAccess.getBooleanFromCache( nodeId, SLOT_IN_USE ) );
                 return rel;
             }
 
@@ -222,6 +224,7 @@ RelationshipRecord populateRelationshipFromCache( long nodeId, RelationshipRecor
                 {
                     rel.setSecondPrevRel( cacheAccess.getFromCache( nodeId, SLOT_REFERENCE ) );
                 }
+                rel.setInUse( cacheAccess.getBooleanFromCache( nodeId, SLOT_IN_USE ) );
                 return rel;
             }
 
@@ -276,6 +279,7 @@ RelationshipRecord populateRelationshipFromCache( long nodeId, RelationshipRecor
                 {
                     rel.setSecondNextRel( cacheAccess.getFromCache( nodeId, SLOT_REFERENCE ) );
                 }
+                rel.setInUse( cacheAccess.getBooleanFromCache( nodeId, SLOT_IN_USE ) );
                 return rel;
             }
 
@@ -330,6 +334,7 @@ RelationshipRecord populateRelationshipFromCache( long nodeId, RelationshipRecor
                 {
                     rel.setSecondPrevRel( cacheAccess.getFromCache( nodeId, SLOT_REFERENCE ) );
                 }
+                rel.setInUse( cacheAccess.getBooleanFromCache( nodeId, SLOT_IN_USE ) );
                 return rel;
             }
 
@@ -406,13 +411,13 @@ public void checkConsistency( RelationshipRecord relationship,
                     if ( cacheAccess.withinBounds( relationship.getFirstNode() ) )
                     {
                         cacheAccess.putToCache( relationship.getFirstNode(), SOURCE, PREV,
-                                relationship.getId(), relationship.getFirstPrevRel() );
+                                relationship.getId(), relationship.getFirstPrevRel(), 1 );
                         updateCacheCounts( cache1Free, cacheAccess );
                     }
                     if ( cacheAccess.withinBounds( relationship.getSecondNode() ) )
                     {
                         cacheAccess.putToCache( relationship.getSecondNode(), TARGET, PREV,
-                                relationship.getId(), relationship.getSecondPrevRel() );
+                                relationship.getId(), relationship.getSecondPrevRel(), 1 );
                         updateCacheCounts( cache2Free, cacheAccess );
                     }
                 }
@@ -421,13 +426,13 @@ public void checkConsistency( RelationshipRecord relationship,
                     if ( cacheAccess.withinBounds( relationship.getFirstNode() ) )
                     {
                         cacheAccess.putToCache( relationship.getFirstNode(), SOURCE, NEXT,
-                                relationship.getId(), relationship.getFirstNextRel() );
+                                relationship.getId(), relationship.getFirstNextRel() , 1 );
                         updateCacheCounts( cache1Free, cacheAccess );
                     }
                     if ( cacheAccess.withinBounds( relationship.getSecondNode() ) )
                     {
                         cacheAccess.putToCache( relationship.getSecondNode(), TARGET, NEXT,
-                                relationship.getId(), relationship.getSecondNextRel() );
+                                relationship.getId(), relationship.getSecondNextRel() , 1 );
                         updateCacheCounts( cache2Free, cacheAccess );
                     }
                 }
@@ -598,7 +603,12 @@ public void checkReference( RelationshipRecord record, RelationshipRecord referr
                     }
                 }
                 else
-                {   // successfully checked
+                {
+                    if ( !referred.inUse() )
+                    {
+                        engine.report().notUsedRelationshipReferencedInChain( referred );
+                    }
+                    // successfully checked
                     // clear cache only if cache is used - meaning referred was built using cache.
                     if ( referred.isCreated() )
                     {

community/consistency-check/src/main/java/org/neo4j/consistency/checking/cache/CacheAccess.java

@@ -56,6 +56,17 @@ interface Client
          */
         long getFromCache( long id, int slot );
 
+        /**
+         * Gets a cached value, put there with {@link #putToCache(long, long...)} or
+         * {@link #putToCacheSingle(long, int, long)} and interpret field value as a boolean.
+         * 0 will be treated as false all the rest as true.
+         *
+         * @param id the entity id this cached value is tied to.
+         * @param slot which cache slot for this id.
+         * @return false if slot value is 0, true otherwise.
+         */
+        boolean getBooleanFromCache(long id, int slot);
+
         /**
          * Caches all values for an id, i.e. fills all slots.
          *
@@ -200,6 +211,12 @@ public long getFromCache( long id, int slot )
             return 0;
         }
 
+        @Override
+        public boolean getBooleanFromCache( long id, int slot )
+        {
+            return false;
+        }
+
         @Override
         public boolean withinBounds( long id )
         {

community/consistency-check/src/main/java/org/neo4j/consistency/checking/cache/CacheSlots.java

@@ -43,6 +43,7 @@ interface RelationshipLink
         int SLOT_PREV_OR_NEXT = 1;
         int SLOT_RELATIONSHIP_ID = 2;
         int SLOT_REFERENCE = 3;
+        int SLOT_IN_USE = 4;
         long SOURCE = 0;
         long TARGET = -1;
         long PREV = 0;

community/consistency-check/src/main/java/org/neo4j/consistency/checking/cache/DefaultCacheAccess.java

@@ -107,6 +107,12 @@ public long getFromCache( long id, int slot )
             return cache.get( id, slot );
         }
 
+        @Override
+        public boolean getBooleanFromCache( long id, int slot )
+        {
+            return cache.get( id, slot ) != 0;
+        }
+
         @Override
         public void putToCache( long id, long... values )
         {

community/consistency-check/src/main/java/org/neo4j/consistency/checking/full/CheckStage.java

@@ -32,8 +32,8 @@ public enum CheckStage implements Stage
     Stage3_NS_NextRel( false, true, "NodeStore pass - just cache nextRel and inUse", 1, 1, ID_SLOT_SIZE ),
     Stage4_RS_NextRel( true, true, "RelationshipStore pass - check nodes inUse, FirstInFirst, FirstInSecond using cached info", 1, 1, ID_SLOT_SIZE ),
     Stage5_Check_NextRel( false, true, "NodeRelationship cache pass - check nextRel", 1, 1, ID_SLOT_SIZE ),
-    Stage6_RS_Forward( true, true, "RelationshipStore pass - forward scan of source chain using the cache", 1, 1, ID_SLOT_SIZE, ID_SLOT_SIZE ),
-    Stage7_RS_Backward( true, false, "RelationshipStore pass - reverse scan of source chain using the cache", 1, 1, ID_SLOT_SIZE, ID_SLOT_SIZE ),
+    Stage6_RS_Forward( true, true, "RelationshipStore pass - forward scan of source chain using the cache", 1, 1, ID_SLOT_SIZE, ID_SLOT_SIZE, 1 ),
+    Stage7_RS_Backward( true, false, "RelationshipStore pass - reverse scan of source chain using the cache", 1, 1, ID_SLOT_SIZE, ID_SLOT_SIZE, 1 ),
     Stage8_PS_Props( true, true, "PropertyStore and Node to Index check pass" ),
     Stage9_NS_LabelCounts( true, true, "NodeStore pass - Label counts" ),
     Stage10_NS_PropertyRelocator( true, true, "Property store relocation" );
@@ -43,7 +43,7 @@ public enum CheckStage implements Stage
     private final String purpose;
     private final int[] cacheSlotSizes;
 
-    private CheckStage( boolean parallel, boolean forward, String purpose, int... cacheFields )
+    CheckStage( boolean parallel, boolean forward, String purpose, int... cacheFields )
     {
         this.parallel = parallel;
         this.forward = forward;

community/consistency-check/src/main/java/org/neo4j/consistency/checking/labelscan/LabelScanCheck.java

@@ -36,8 +36,8 @@ public void check( LabelScanDocument record, CheckerEngine<LabelScanDocument,
         NodeLabelRange range = record.getNodeLabelRange();
         for ( long nodeId : range.nodes() )
         {
-            engine.comparativeCheck( records.node( nodeId ), new NodeInUseWithCorrectLabelsCheck<LabelScanDocument,ConsistencyReport.LabelScanConsistencyReport>(
-                    record.getNodeLabelRange().labels( nodeId ) ) );
+            engine.comparativeCheck( records.node( nodeId ),
+                    new NodeInUseWithCorrectLabelsCheck<>( record.getNodeLabelRange().labels( nodeId ) ) );
         }
     }
 }

community/consistency-check/src/main/java/org/neo4j/consistency/report/ConsistencyReport.java

@@ -204,6 +204,9 @@ interface NodeConsistencyReport extends PrimitiveConsistencyReport
     interface RelationshipConsistencyReport
             extends PrimitiveConsistencyReport
     {
+        @Documented( "The relationship record is not in use, but referenced from relationships chain." )
+        void notUsedRelationshipReferencedInChain( RelationshipRecord relationshipRecord );
+
         @Documented( "The relationship type field has an illegal value." )
         void illegalRelationshipType();
 

community/consistency-check/src/test/java/org/neo4j/consistency/ConsistencyCheckServiceIntegrationTest.java

@@ -19,18 +19,19 @@
  */
 package org.neo4j.consistency;
 
+import org.apache.commons.lang3.StringUtils;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.RuleChain;
+
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.Files;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
 
-import org.apache.commons.lang3.StringUtils;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.RuleChain;
-
 import org.neo4j.consistency.ConsistencyCheckService.Result;
 import org.neo4j.consistency.checking.GraphStoreFixture;
 import org.neo4j.consistency.checking.full.ConsistencyCheckIncompleteException;
@@ -46,17 +47,23 @@
 import org.neo4j.kernel.api.exceptions.TransactionFailureException;
 import org.neo4j.kernel.configuration.Config;
 import org.neo4j.kernel.configuration.Settings;
+import org.neo4j.kernel.impl.storageengine.impl.recordstorage.RecordStorageEngine;
+import org.neo4j.kernel.impl.store.NeoStores;
+import org.neo4j.kernel.impl.store.RelationshipStore;
 import org.neo4j.kernel.impl.store.record.NodeRecord;
+import org.neo4j.kernel.impl.store.record.RecordLoad;
+import org.neo4j.kernel.impl.store.record.RelationshipRecord;
+import org.neo4j.kernel.internal.GraphDatabaseAPI;
 import org.neo4j.logging.NullLogProvider;
 import org.neo4j.test.TestGraphDatabaseFactory;
 import org.neo4j.test.rule.TestDirectory;
 
 import static java.lang.String.format;
-
+import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
-
 import static org.neo4j.helpers.collection.MapUtil.stringMap;
 import static org.neo4j.test.Property.property;
 import static org.neo4j.test.Property.set;
@@ -83,6 +90,26 @@ protected void generateInitialData( GraphDatabaseService graphDb )
     @Rule
     public final RuleChain chain = RuleChain.outerRule( testDirectory ).around( fixture );
 
+    @Test
+    public void reportNotUsedRelationshipReferencedInChain() throws Exception
+    {
+        prepareDbWithDeletedRelationshipPartOfTheChain();
+
+        Date timestamp = new Date();
+        ConsistencyCheckService service = new ConsistencyCheckService( timestamp );
+        Config configuration = new Config( settings(), GraphDatabaseSettings.class, ConsistencyCheckSettings.class );
+
+        ConsistencyCheckService.Result result = runFullConsistencyCheck( service, configuration );
+
+        assertFalse( result.isSuccessful() );
+
+        File reportFile = result.reportFile();
+        assertTrue( "Consistency check report file should be generated.", reportFile.exists() );
+        assertThat( "Expected to see report about not deleted relationship record present as part of a chain",
+                Files.readAllLines( reportFile.toPath() ).toString(),
+                containsString( "The relationship record is not in use, but referenced from relationships chain.") );
+    }
+
     @Test
     public void shouldSucceedIfStoreIsConsistent() throws Exception
     {
@@ -189,6 +216,43 @@ private GraphDatabaseService getGraphDatabaseService()
         return builder.newGraphDatabase();
     }
 
+    private void prepareDbWithDeletedRelationshipPartOfTheChain()
+    {
+        GraphDatabaseAPI db = (GraphDatabaseAPI) new TestGraphDatabaseFactory().newEmbeddedDatabaseBuilder( testDirectory.graphDbDir() )
+                .setConfig( GraphDatabaseSettings.record_format, getRecordFormatName() )
+                .newGraphDatabase();
+        try
+        {
+
+            RelationshipType relationshipType = RelationshipType.withName( "testRelationshipType" );
+            try ( Transaction tx = db.beginTx() )
+            {
+                Node node1 = set( db.createNode() );
+                Node node2 = set( db.createNode(), property( "key", "value" ) );
+                node1.createRelationshipTo( node2, relationshipType );
+                node1.createRelationshipTo( node2, relationshipType );
+                node1.createRelationshipTo( node2, relationshipType );
+                node1.createRelationshipTo( node2, relationshipType );
+                node1.createRelationshipTo( node2, relationshipType );
+                node1.createRelationshipTo( node2, relationshipType );
+                tx.success();
+            }
+
+            RecordStorageEngine recordStorageEngine = db.getDependencyResolver().resolveDependency( RecordStorageEngine.class );
+
+            NeoStores neoStores = recordStorageEngine.testAccessNeoStores();
+            RelationshipStore relationshipStore = neoStores.getRelationshipStore();
+            RelationshipRecord relationshipRecord = new RelationshipRecord( -1 );
+            RelationshipRecord record = relationshipStore.getRecord( 4, relationshipRecord, RecordLoad.FORCE );
+            record.setInUse( false );
+            relationshipStore.updateRecord( relationshipRecord );
+        }
+        finally
+        {
+            db.shutdown();
+        }
+    }
+
     protected Map<String,String> settings( String... strings )
     {
         Map<String, String> defaults = new HashMap<>();

community/consistency-check/src/test/java/org/neo4j/consistency/checking/RecordCheckTestBase.java

@@ -99,21 +99,12 @@ public void check( RelationshipRecord record,
 
     public static RecordCheck<PropertyRecord, ConsistencyReport.PropertyConsistencyReport> dummyPropertyChecker()
     {
-        return new RecordCheck<PropertyRecord, ConsistencyReport.PropertyConsistencyReport>()
-        {
-            @Override
-            public void check( PropertyRecord record,
-                               CheckerEngine<PropertyRecord, ConsistencyReport.PropertyConsistencyReport> engine,
-                               RecordAccess records )
-            {
-            }
-        };
+        return ( record, engine, records ) -> {};
     }
 
     public static PrimitiveRecordCheck<NeoStoreRecord, ConsistencyReport.NeoStoreConsistencyReport> dummyNeoStoreCheck()
     {
-        return new NeoStoreCheck( new PropertyChain<NeoStoreRecord,ConsistencyReport.NeoStoreConsistencyReport>(
-                from -> null ) )
+        return new NeoStoreCheck( new PropertyChain<>( from -> null ) )
         {
             @Override
             public void check( NeoStoreRecord record,