TreeNodeDynamicSize impl part 11 - Pass SeekCursorTest

TreeNodeDynamicSize fail more gracefully when reading
bad key or value size.

Also
- SeekCursors use correct Type
- SeekCursor read key and value together
- Fix some problems with structure propagation in some SeekCursorTests

community/index/src/main/java/org/neo4j/index/internal/gbptree/SeekCursor.java

@@ -513,7 +513,7 @@ private boolean internalNext() throws IOException
                 if ( verifyExpectedFirstAfterGoToNext )
                 {
                     pos = seekForward ? 0 : keyCount - 1;
-                    bTreeNode.keyAt( cursor,firstKeyInNode, pos, INTERNAL );
+                    bTreeNode.keyAt( cursor,firstKeyInNode, pos, LEAF );
                 }
 
                 if ( concurrentWriteHappened )
@@ -545,8 +545,7 @@ private boolean internalNext() throws IOException
                 if ( 0 <= pos && pos < keyCount )
                 {
                     // Read the next value in this leaf
-                    bTreeNode.keyAt( cursor, mutableKey, pos, INTERNAL );
-                    bTreeNode.valueAt( cursor, mutableValue, pos );
+                    bTreeNode.keyValueAt( cursor, mutableKey, mutableValue, pos );
                 }
             }
             while ( concurrentWriteHappened = cursor.shouldRetry() );

community/index/src/main/java/org/neo4j/index/internal/gbptree/TreeNode.java

@@ -226,6 +226,8 @@ static void removeSlotAt( PageCursor cursor, int pos, int itemCount, int baseOff
 
     abstract KEY keyAt( PageCursor cursor, KEY into, int pos, Type type );
 
+    abstract void keyValueAt( PageCursor cursor, KEY intoKey, VALUE intoValue, int pos );
+
     abstract void insertKeyAndRightChildAt( PageCursor cursor, KEY key, long child, int pos, int keyCount,
             long stableGeneration, long unstableGeneration );
 

community/index/src/main/java/org/neo4j/index/internal/gbptree/TreeNodeDynamicSize.java

@@ -105,6 +105,7 @@ KEY keyAt( PageCursor cursor, KEY into, int pos, Type type )
         {
             cursor.setCursorException( format( "Read unreliable key, keySize=%d, keyValueSizeCap=%d, keyHasTombstone=%b",
                     keySize, keyValueSizeCap, hasTombstone( keySize ) ) );
+            return into;
         }
         if ( type == LEAF )
         {
@@ -114,6 +115,23 @@ KEY keyAt( PageCursor cursor, KEY into, int pos, Type type )
         return into;
     }
 
+    @Override
+    void keyValueAt( PageCursor cursor, KEY intoKey, VALUE intoValue, int pos )
+    {
+        placeCursorAtActualKey( cursor, pos, LEAF );
+
+        int keySize = readKeySize( cursor );
+        int valueSize = readValueSize( cursor );
+        if ( keySize + valueSize > keyValueSizeCap || keySize < 0 || valueSize < 0 )
+        {
+            cursor.setCursorException( format( "Read unreliable key, keySize=%d, valueSize=%d, keyValueSizeCap=%d, keyHasTombstone=%b",
+                    keySize, valueSize, keyValueSizeCap, hasTombstone( keySize ) ) );
+            return;
+        }
+        layout.readKey( cursor, intoKey, keySize );
+        layout.readValue( cursor, intoValue, valueSize );
+    }
+
     @Override
     void insertKeyAndRightChildAt( PageCursor cursor, KEY key, long child, int pos, int keyCount, long stableGeneration,
             long unstableGeneration )
@@ -256,11 +274,12 @@ VALUE valueAt( PageCursor cursor, VALUE into, int pos )
         // Read value
         int keySize = readKeySize( cursor );
         int valueSize = readValueSize( cursor );
-        if ( keySize + valueSize > keyValueSizeCap )
+        if ( keySize + valueSize > keyValueSizeCap || keySize < 0 || valueSize < 0 )
         {
             cursor.setCursorException(
                     format( "Read unreliable key, value size greater than cap: keySize=%d, valueSize=%d, keyValueSizeCap=%d",
                             keySize, valueSize, keyValueSizeCap ) );
+            return into;
         }
         progressCursor( cursor, keySize );
         layout.readValue( cursor, into, valueSize );

community/index/src/main/java/org/neo4j/index/internal/gbptree/TreeNodeFixedSize.java

@@ -109,6 +109,13 @@ KEY keyAt( PageCursor cursor, KEY into, int pos, Type type )
         return into;
     }
 
+    @Override
+    void keyValueAt( PageCursor cursor, KEY intoKey, VALUE intoValue, int pos )
+    {
+        keyAt( cursor, intoKey, pos, LEAF );
+        valueAt( cursor, intoValue, pos );
+    }
+
     @Override
     void insertKeyAndRightChildAt( PageCursor cursor, KEY key, long child, int pos, int keyCount, long stableGeneration,
             long unstableGeneration )

community/index/src/test/java/org/neo4j/index/internal/gbptree/SeekCursorDynamicSizeTest.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2002-2017 "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.index.internal.gbptree;
+
+public class SeekCursorDynamicSizeTest extends SeekCursorTestBase<RawBytes,RawBytes>
+{
+    @Override
+    TestLayout<RawBytes,RawBytes> getLayout()
+    {
+        return new SimpleByteArrayLayout();
+    }
+
+    @Override
+    TreeNode<RawBytes,RawBytes> getTreeNode( int pageSize, TestLayout<RawBytes,RawBytes> layout )
+    {
+        return new TreeNodeDynamicSize<>( pageSize, layout );
+    }
+}

community/index/src/test/java/org/neo4j/index/internal/gbptree/SeekCursorFixedSizeTest.java

@@ -23,12 +23,10 @@
 
 public class SeekCursorFixedSizeTest extends SeekCursorTestBase<MutableLong,MutableLong>
 {
-    private SimpleLongLayout layout = new SimpleLongLayout();
-
     @Override
     TestLayout<MutableLong,MutableLong> getLayout()
     {
-        return layout;
+        return new SimpleLongLayout();
     }
 
     @Override

community/index/src/test/java/org/neo4j/index/internal/gbptree/SeekCursorTestBase.java

@@ -306,6 +306,7 @@ public boolean next( long pageId ) throws IOException
 
         // GIVEN
         long i = fullLeaf();
+        long left = createRightSibling( cursor );
         long j = fullLeaf( i );
 
         long fromInclusive = j - 1;
@@ -1198,7 +1199,7 @@ public void mustRereadHeadersOnRetry() throws Exception
         {
             // reading a couple of keys
             assertTrue( cursor.next() );
-            assertEquals( key( 0 ), cursor.get().key() );
+            assertEqualsKey( key( 0 ), cursor.get().key() );
 
             // and WHEN a change happens
             append( keyCount );
@@ -1208,11 +1209,11 @@ public void mustRereadHeadersOnRetry() throws Exception
             assertTrue( cursor.next() );
 
             // and the new key should be found in the end as well
-            assertEquals( key( 1 ), cursor.get().key() );
+            assertEqualsKey( key( 1 ), cursor.get().key() );
             long lastFoundKey = 1;
             while ( cursor.next() )
             {
-                assertEquals( key( lastFoundKey + 1 ), cursor.get().key() );
+                assertEqualsKey( key( lastFoundKey + 1 ), cursor.get().key() );
                 lastFoundKey = getSeed( cursor.get().key() );
             }
             assertEquals( keyCount, lastFoundKey );
@@ -1733,25 +1734,26 @@ public void shouldFailSuccessorIfReadFailureNotCausedByCheckpointInInternal() th
         }
 
         // a checkpoint
+        long oldRootId = rootId;
         long oldStableGeneration = stableGeneration;
         long oldUnstableGeneration = unstableGeneration;
         checkpoint();
         int keyCount = TreeNode.keyCount( cursor );
 
         // and update root with an insert in new generation
-        while ( TreeNode.keyCount( cursor ) == keyCount )
+        while ( keyCount( rootId ) == keyCount )
         {
             insert( i );
             i++;
         }
 
         // and corrupt successor pointer
-        cursor.next( rootId );
+        cursor.next( oldRootId );
         corruptGSPP( cursor, TreeNode.BYTE_POS_SUCCESSOR );
 
         // when
         // starting a seek on the old root with generation that is not up to date, simulating a concurrent checkpoint
-        PageAwareByteArrayCursor pageCursorForSeeker = cursor.duplicate( rootId );
+        PageAwareByteArrayCursor pageCursorForSeeker = cursor.duplicate( oldRootId );
         pageCursorForSeeker.next();
         try ( SeekCursor<KEY,VALUE> ignored = seekCursor(
                 i, i + 1, pageCursorForSeeker, oldStableGeneration, oldUnstableGeneration ) )
@@ -1861,7 +1863,7 @@ public void shouldCatchupRootWhenRootNodeHasTooNewGeneration() throws Exception
 
         // when
         //noinspection EmptyTryBlock
-        try ( SeekCursor<KEY,VALUE> ignored = new SeekCursor<>( cursor, node, layout.newKey(), layout.newKey(), layout,
+        try ( SeekCursor<KEY,VALUE> ignored = new SeekCursor<>( cursor, node, key( 0 ), key( 1 ), layout,
                 stableGeneration, unstableGeneration, generationSupplier, rootCatchup, generation - 1,
                 exceptionDecorator ) )
         {
@@ -2083,20 +2085,41 @@ private void triggerUnderflowAndSeekRange( SeekCursor<KEY,VALUE> seeker,
 
     private void triggerUnderflow( long nodeId ) throws IOException
     {
+        // On underflow keys will move from left to right
+        // and key count of the right will increase.
+        // We don't know if keys will move from nodeId to
+        // right sibling or to nodeId from left sibling.
+        // So we monitor both nodeId and rightSibling.
         PageCursor readCursor = cursor.duplicate( nodeId );
         readCursor.next();
+        int midKeyCount = TreeNode.keyCount( readCursor );
+        int prevKeyCount = midKeyCount + 1;
 
-        int keyCount = TreeNode.keyCount( readCursor );
-        int prevKeyCount;
+        PageCursor rightSiblingCursor = null;
+        long rightSibling = TreeNode.rightSibling( readCursor, stableGeneration, unstableGeneration );
+        int rightKeyCount = 0;
+        int prevRightKeyCount = 1;
+        boolean monitorRight = TreeNode.isNode( rightSibling );
+        if ( monitorRight )
+        {
+            rightSiblingCursor = cursor.duplicate( GenerationSafePointerPair.pointer( rightSibling ) );
+            rightSiblingCursor.next();
+            rightKeyCount = TreeNode.keyCount( rightSiblingCursor );
+            prevRightKeyCount = rightKeyCount + 1;
+        }
 
-        do
+        while ( midKeyCount < prevKeyCount && rightKeyCount <= prevRightKeyCount )
         {
             long toRemove = keyAt( readCursor, 0, LEAF );
             remove( toRemove );
-            prevKeyCount = keyCount;
-            keyCount = TreeNode.keyCount( readCursor );
+            prevKeyCount = midKeyCount;
+            midKeyCount = TreeNode.keyCount( readCursor );
+            if ( monitorRight )
+            {
+                prevRightKeyCount = rightKeyCount;
+                rightKeyCount = TreeNode.keyCount( rightSiblingCursor );
+            }
         }
-        while ( keyCount < prevKeyCount );
     }
 
     private void checkpoint()