Consistency-checker PropertyReader detects circular property chains

And reports those, instead of going OOM.

Also made a slight optimization where previously the entire property chain
would always be loaded when looking for a specific property, but now stops
loading after finding the property it's looking for.

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

@@ -71,13 +71,20 @@ public void check( NodeRecord record,
                        CheckerEngine<NodeRecord, ConsistencyReport.NodeConsistencyReport> engine,
                        RecordAccess records )
     {
-        Collection<PropertyRecord> properties = propertyReader.getPropertyRecordChain( record );
-        cacheAccess.client().putPropertiesToCache(properties);
-        if ( indexes != null )
+        try
+        {
+            Collection<PropertyRecord> properties = propertyReader.getPropertyRecordChain( record.getNextProp() );
+            cacheAccess.client().putPropertiesToCache(properties);
+            if ( indexes != null )
+            {
+                matchIndexesToNode( record, engine, records, properties );
+            }
+            checkProperty( record, engine, properties );
+        }
+        catch ( PropertyReader.CircularPropertyRecordChainException e )
         {
-            matchIndexesToNode( record, engine, records, properties );
+            engine.report().propertyChainContainsCircularReference( e.propertyRecordClosingTheCircle() );
         }
-        checkProperty( record, engine, properties );
     }
 
     /**

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

@@ -21,9 +21,11 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.LinkedList;
 import java.util.List;
 
+import org.neo4j.collection.primitive.Primitive;
+import org.neo4j.collection.primitive.PrimitiveLongSet;
+import org.neo4j.helpers.collection.Visitor;
 import org.neo4j.kernel.api.index.PropertyAccessor;
 import org.neo4j.kernel.impl.store.NodeStore;
 import org.neo4j.kernel.impl.store.PropertyStore;
@@ -37,36 +39,29 @@
 
 import static org.neo4j.kernel.impl.store.record.RecordLoad.FORCE;
 
-public class PropertyReader implements PropertyAccessor
+class PropertyReader implements PropertyAccessor
 {
     private final PropertyStore propertyStore;
     private final NodeStore nodeStore;
 
-    public PropertyReader( StoreAccess storeAccess )
+    PropertyReader( StoreAccess storeAccess )
     {
         this.propertyStore = storeAccess.getRawNeoStores().getPropertyStore();
         this.nodeStore = storeAccess.getRawNeoStores().getNodeStore();
     }
 
-    public Collection<PropertyRecord> getPropertyRecordChain( NodeRecord nodeRecord )
+    Collection<PropertyRecord> getPropertyRecordChain( long firstPropertyRecordId ) throws CircularPropertyRecordChainException
     {
-        return getPropertyRecordChain( nodeRecord.getNextProp() );
-    }
-
-    public Collection<PropertyRecord> getPropertyRecordChain( long firstId )
-    {
-        long nextProp = firstId;
-        List<PropertyRecord> toReturn = new LinkedList<>();
-        while ( nextProp != Record.NO_NEXT_PROPERTY.intValue() )
+        List<PropertyRecord> records = new ArrayList<>();
+        visitPropertyRecordChain( firstPropertyRecordId, record ->
         {
-            PropertyRecord propRecord = propertyStore.getRecord( nextProp, propertyStore.newRecord(), FORCE );
-            toReturn.add( propRecord );
-            nextProp = propRecord.getNextProp();
-        }
-        return toReturn;
+            records.add( record );
+            return false; // please continue
+        } );
+        return records;
     }
 
-    public List<PropertyBlock> propertyBlocks( Collection<PropertyRecord> records )
+    List<PropertyBlock> propertyBlocks( Collection<PropertyRecord> records )
     {
         List<PropertyBlock> propertyBlocks = new ArrayList<>();
         for ( PropertyRecord record : records )
@@ -79,18 +74,25 @@ public List<PropertyBlock> propertyBlocks( Collection<PropertyRecord> records )
         return propertyBlocks;
     }
 
-    public List<PropertyBlock> propertyBlocks( NodeRecord nodeRecord )
+    private boolean visitPropertyRecordChain( long firstPropertyRecordId, Visitor<PropertyRecord,RuntimeException> visitor )
+            throws CircularPropertyRecordChainException
     {
-        Collection<PropertyRecord> records = propertyStore.getPropertyRecordChain( nodeRecord.getNextProp() );
-        List<PropertyBlock> propertyBlocks = new ArrayList<>();
-        for ( PropertyRecord record : records )
+        PrimitiveLongSet visitedPropertyRecordIds = Primitive.longSet( 8 );
+        long nextProp = firstPropertyRecordId;
+        while ( nextProp != Record.NO_NEXT_PROPERTY.intValue() )
         {
-            for ( PropertyBlock block : record )
+            PropertyRecord propRecord = propertyStore.getRecord( nextProp, propertyStore.newRecord(), FORCE );
+            nextProp = propRecord.getNextProp();
+            if ( !Record.NO_NEXT_PROPERTY.is( nextProp ) && !visitedPropertyRecordIds.add( nextProp ) )
             {
-                propertyBlocks.add( block );
+                throw new CircularPropertyRecordChainException( propRecord );
+            }
+            if ( visitor.visit( propRecord ) )
+            {
+                return true;
             }
         }
-        return propertyBlocks;
+        return false;
     }
 
     public Value propertyValue( PropertyBlock block )
@@ -104,14 +106,61 @@ public Value getPropertyValue( long nodeId, int propertyKeyId )
         NodeRecord nodeRecord = nodeStore.newRecord();
         if ( nodeStore.getRecord( nodeId, nodeRecord, FORCE ).inUse() )
         {
-            for ( PropertyBlock block : propertyBlocks( nodeRecord ) )
+            SpecificValueVisitor visitor = new SpecificValueVisitor( propertyKeyId );
+            try
             {
-                if ( block.getKeyIndexId() == propertyKeyId )
+                if ( visitPropertyRecordChain( nodeRecord.getNextProp(), visitor ) )
                 {
-                    return propertyValue( block );
+                    return visitor.foundPropertyValue;
                 }
             }
+            catch ( CircularPropertyRecordChainException e )
+            {
+                // If we discover a circular reference and still haven't found the property then we won't find it.
+                // There are other places where this circular reference will be logged as an inconsistency,
+                // so simply catch this exception here and let this method return NO_VALUE below.
+            }
         }
         return Values.NO_VALUE;
     }
+
+    private class SpecificValueVisitor implements Visitor<PropertyRecord,RuntimeException>
+    {
+        private final int propertyKeyId;
+        private Value foundPropertyValue;
+
+        SpecificValueVisitor( int propertyKeyId )
+        {
+            this.propertyKeyId = propertyKeyId;
+        }
+
+        @Override
+        public boolean visit( PropertyRecord element ) throws RuntimeException
+        {
+            for ( PropertyBlock block : element )
+            {
+                if ( block.getKeyIndexId() == propertyKeyId )
+                {
+                    foundPropertyValue = propertyValue( block );
+                    return true;
+                }
+            }
+            return false;
+        }
+    }
+
+    static class CircularPropertyRecordChainException extends Exception
+    {
+        private final PropertyRecord propertyRecord;
+
+        CircularPropertyRecordChainException( PropertyRecord propertyRecord )
+        {
+            this.propertyRecord = propertyRecord;
+        }
+
+        PropertyRecord propertyRecordClosingTheCircle()
+        {
+            return propertyRecord;
+        }
+    }
 }

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

@@ -207,6 +207,9 @@ interface NodeConsistencyReport extends PrimitiveConsistencyReport
 
         @Documented( "The first relationship group record has another node set as owner." )
         void relationshipGroupHasOtherOwner( RelationshipGroupRecord group );
+
+        @Documented( "The property chain contains circular reference." )
+        void propertyChainContainsCircularReference( PropertyRecord propertyRecord );
     }
 
     interface RelationshipConsistencyReport

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

@@ -79,6 +79,7 @@
 import org.neo4j.kernel.configuration.Config;
 import org.neo4j.kernel.impl.annotations.Documented;
 import org.neo4j.kernel.impl.api.KernelStatement;
+import org.neo4j.kernel.impl.api.KernelTransactionImplementation;
 import org.neo4j.kernel.impl.api.index.IndexUpdateMode;
 import org.neo4j.kernel.impl.api.index.NodeUpdates;
 import org.neo4j.kernel.impl.api.index.sampling.IndexSamplingConfig;
@@ -87,6 +88,7 @@
 import org.neo4j.kernel.impl.store.AbstractDynamicStore;
 import org.neo4j.kernel.impl.store.DynamicRecordAllocator;
 import org.neo4j.kernel.impl.store.NodeLabelsField;
+import org.neo4j.kernel.impl.store.PropertyStore;
 import org.neo4j.kernel.impl.store.PropertyType;
 import org.neo4j.kernel.impl.store.RecordStore;
 import org.neo4j.kernel.impl.store.SchemaStorage;
@@ -101,6 +103,7 @@
 import org.neo4j.kernel.impl.store.record.NodeRecord;
 import org.neo4j.kernel.impl.store.record.PropertyBlock;
 import org.neo4j.kernel.impl.store.record.PropertyRecord;
+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.store.record.RelationshipTypeTokenRecord;
@@ -2100,6 +2103,46 @@ protected void transactionData( TransactionDataBuilder tx, IdGenerator next )
         assertTrue( stats.isConsistent() );
     }
 
+    @Test
+    public void shouldReportCircularPropertyRecordChain() throws Exception
+    {
+        // Given
+        fixture.apply( new GraphStoreFixture.Transaction()
+        {
+            @Override
+            protected void transactionData( TransactionDataBuilder tx, IdGenerator next )
+            {
+                // Create property chain A --> B --> C --> D
+                //                             ↑           │
+                //                             └───────────┘
+                long a = next.property();
+                long b = next.property();
+                long c = next.property();
+                long d = next.property();
+                tx.create( propertyRecordWithSingleIntProperty( a, next.propertyKey(), -1, b ) );
+                tx.create( propertyRecordWithSingleIntProperty( b, next.propertyKey(), a, c ) );
+                tx.create( propertyRecordWithSingleIntProperty( c, next.propertyKey(), b, d ) );
+                tx.create( propertyRecordWithSingleIntProperty( d, next.propertyKey(), c, b ) );
+                tx.create( new NodeRecord( next.node() ).initialize( true, a, false, -1, Record.NO_LABELS_FIELD.longValue() ) );
+            }
+
+            private PropertyRecord propertyRecordWithSingleIntProperty( long id, int propertyKeyId, long prev, long next )
+            {
+                PropertyRecord record = new PropertyRecord( id ).initialize( true, prev, next );
+                PropertyBlock block = new PropertyBlock();
+                PropertyStore.encodeValue( block, propertyKeyId, Values.intValue( 10 ), null, null );
+                record.addPropertyBlock( block );
+                return record;
+            }
+        } );
+
+        // When
+        ConsistencySummaryStatistics stats = check();
+
+        // Then report will be filed on Node inconsistent with the Property completing the circle
+        on( stats ).verify( RecordType.NODE, 1 );
+    }
+
     private ConsistencySummaryStatistics check() throws ConsistencyCheckIncompleteException
     {
         return check( fixture.directStoreAccess() );

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright (c) 2002-2018 "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.consistency.checking.full;
+
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import org.neo4j.kernel.impl.store.NeoStores;
+import org.neo4j.kernel.impl.store.PropertyStore;
+import org.neo4j.kernel.impl.store.StoreAccess;
+import org.neo4j.kernel.impl.store.StoreType;
+import org.neo4j.kernel.impl.store.record.PropertyRecord;
+import org.neo4j.test.rule.NeoStoresRule;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class PropertyReaderTest
+{
+    @Rule
+    public final NeoStoresRule storesRule = new NeoStoresRule( PropertyReaderTest.class,
+            StoreType.NODE, StoreType.COUNTS,
+            StoreType.PROPERTY, StoreType.PROPERTY_ARRAY, StoreType.PROPERTY_STRING );
+
+    @Test
+    public void shouldDetectAndAbortPropertyChainLoadingOnCircularReference() throws IOException
+    {
+        // given
+        NeoStores neoStores = storesRule.builder().build();
+
+        // Create property chain 1 --> 2 --> 3 --> 4
+        //                             ↑           │
+        //                             └───────────┘
+        PropertyStore propertyStore = neoStores.getPropertyStore();
+        PropertyRecord record = propertyStore.newRecord();
+        // 1
+        record.setId( 1 );
+        record.initialize( true, -1, 2 );
+        propertyStore.updateRecord( record );
+        // 2
+        record.setId( 2 );
+        record.initialize( true, 1, 3 );
+        propertyStore.updateRecord( record );
+        // 3
+        record.setId( 3 );
+        record.initialize( true, 2, 4 );
+        propertyStore.updateRecord( record );
+        // 4
+        record.setId( 4 );
+        record.initialize( true, 3, 2 ); // <-- completing the circle
+        propertyStore.updateRecord( record );
+
+        // when
+        PropertyReader reader = new PropertyReader( new StoreAccess( neoStores ) );
+        try
+        {
+            reader.getPropertyRecordChain( 1 );
+            fail( "Should have detected circular reference" );
+        }
+        catch ( PropertyReader.CircularPropertyRecordChainException e )
+        {
+            // then good
+            assertEquals( 4, e.propertyRecordClosingTheCircle().getId() );
+        }
+    }
+}