TreeNodeDynamicSize impl part 7 - rebalance leaf

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

@@ -1044,11 +1044,12 @@ private void underflowInLeaf( PageCursor cursor, StructurePropagation<KEY> struc
                 leftSiblingCursor.next();
                 int leftSiblingKeyCount = TreeNode.keyCount( leftSiblingCursor );
 
-                if ( bTreeNode.canRebalanceLeaves( leftSiblingKeyCount, keyCount ) )
+                int keysToRebalance = bTreeNode.canRebalanceLeaves( leftSiblingCursor, leftSiblingKeyCount, cursor, keyCount );
+                if ( keysToRebalance != -1 )
                 {
                     createSuccessorIfNeeded( leftSiblingCursor, structurePropagation,
                             StructurePropagation.UPDATE_LEFT_CHILD, stableGeneration, unstableGeneration );
-                    rebalanceLeaf( leftSiblingCursor, leftSiblingKeyCount, cursor, keyCount, structurePropagation );
+                    rebalanceLeaf( leftSiblingCursor, leftSiblingKeyCount, cursor, keyCount, keysToRebalance , structurePropagation );
                 }
                 else
                 {
@@ -1153,15 +1154,11 @@ private void merge( PageCursor leftSiblingCursor, int leftSiblingKeyCount, PageC
     }
 
     private void rebalanceLeaf( PageCursor leftCursor, int leftKeyCount, PageCursor rightCursor, int rightKeyCount,
-            StructurePropagation<KEY> structurePropagation )
+            int numberOfKeysToMove, StructurePropagation<KEY> structurePropagation )
     {
-        int totalKeyCount = rightKeyCount + leftKeyCount;
-        int moveFromPosition = totalKeyCount / 2;
-        int numberOfKeysToMove = leftKeyCount - moveFromPosition;
+        bTreeNode.moveKeyValuesFromLeftToRight( leftCursor, leftKeyCount, rightCursor, rightKeyCount, leftKeyCount - numberOfKeysToMove );
 
-        bTreeNode.moveKeyValuesFromLeftToRight( leftCursor, leftKeyCount, rightCursor, rightKeyCount, moveFromPosition );
-
-        TreeNode.setKeyCount( leftCursor, moveFromPosition );
+        TreeNode.setKeyCount( leftCursor, leftKeyCount - numberOfKeysToMove );
         TreeNode.setKeyCount( rightCursor, rightKeyCount + numberOfKeysToMove );
 
         // Propagate change

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

@@ -336,7 +336,11 @@ static void goTo( PageCursor cursor, String messageOnError, long nodeId )
 
     abstract boolean leafUnderflow( PageCursor cursor, int keyCount );
 
-    abstract boolean canRebalanceLeaves( int leftKeyCount, int rightKeyCount );
+    /**
+     * Can we move keys from underflowing left to right so that none of them underflow?
+     * @return number of keys to move or -1 if not possible.
+     */
+    abstract int canRebalanceLeaves( PageCursor leftCursor, int leftKeyCount, PageCursor rightCursor, int rightKeyCount );
 
     abstract boolean canMergeLeaves( int leftKeyCount, int rightKeyCount );
 
@@ -362,7 +366,9 @@ abstract void doSplitInternal( PageCursor leftCursor, int leftKeyCount, PageCurs
             long newRightChild, int middlePos, long stableGeneration, long unstableGeneration );
 
     /**
-     * Move all rightmost keys and values in left node from given position to right node.
+     * Move all rightmost keys and values in left leaf from given position to right node.
+     *
+     * Right leaf will be defragmented.
      *
      * Key count is NOT updated.
      */

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

@@ -415,9 +415,40 @@ boolean leafUnderflow( PageCursor cursor, int keyCount )
     }
 
     @Override
-    boolean canRebalanceLeaves( int leftKeyCount, int rightKeyCount )
+    int canRebalanceLeaves( PageCursor leftCursor, int leftKeyCount, PageCursor rightCursor, int rightKeyCount )
     {
-        throw new UnsupportedOperationException( "Implement me" );
+        // Know that right has underflow
+        int leftActiveSpace = totalActiveSpace( leftCursor, leftKeyCount );
+        int rightActiveSpace = totalActiveSpace( rightCursor, rightKeyCount );
+
+        if ( leftActiveSpace + rightActiveSpace < totalSpace( pageSize ) )
+        {
+            // There is no way to split this up without one of them having an underflow
+            return -1;
+        }
+
+        int prevDelta;
+        int currentDelta = Math.abs( leftActiveSpace - rightActiveSpace );
+        int keysToMove = 0;
+        int lastChunkSize;
+        do
+        {
+            keysToMove++;
+            lastChunkSize = totalSpaceOfKeyValue( leftCursor, leftKeyCount - keysToMove );
+            leftActiveSpace -= lastChunkSize;
+            rightActiveSpace += lastChunkSize;
+
+            prevDelta = currentDelta;
+            currentDelta = Math.abs( leftActiveSpace - rightActiveSpace );
+        }
+        while ( currentDelta < prevDelta );
+        keysToMove--; // Move back to optimal split
+        leftActiveSpace += lastChunkSize;
+        rightActiveSpace -= lastChunkSize;
+
+        int halfSpace = halfSpace();
+        boolean canRebalance = leftActiveSpace > halfSpace && rightActiveSpace > halfSpace;
+        return canRebalance ? keysToMove : -1;
     }
 
     @Override
@@ -574,6 +605,13 @@ private int middle( PageCursor leftCursor, int insertPos, KEY newKey, VALUE newV
         return middle;
     }
 
+    private int totalActiveSpace( PageCursor cursor, int keyCount )
+    {
+        int deadSpace = getDeadSpace( cursor );
+        int allocSpace = getAllocSpace( cursor, keyCount, LEAF );
+        return totalSpace( pageSize ) - deadSpace - allocSpace;
+    }
+
     private int totalSpace( int pageSize )
     {
         return pageSize - HEADER_LENGTH_DYNAMIC;
@@ -613,7 +651,14 @@ void doSplitInternal( PageCursor leftCursor, int leftKeyCount, PageCursor rightC
     void moveKeyValuesFromLeftToRight( PageCursor leftCursor, int leftKeyCount, PageCursor rightCursor, int rightKeyCount,
             int fromPosInLeftNode )
     {
-        throw new UnsupportedOperationException( "Implement me" );
+        defragmentLeaf( rightCursor );
+        int numberOfKeysToMove = leftKeyCount - fromPosInLeftNode;
+
+        // Push keys and values in right sibling to the right
+        insertSlotsAt( rightCursor, 0, numberOfKeysToMove, rightKeyCount, keyPosOffsetLeaf( 0 ), bytesKeySize() );
+
+        // Move
+        moveKeysAndValues( leftCursor, fromPosInLeftNode, rightCursor, 0, numberOfKeysToMove );
     }
 
     private void setAllocOffset( PageCursor cursor, int allocOffset )

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

@@ -279,9 +279,15 @@ boolean leafUnderflow( PageCursor cursor, int keyCount )
     }
 
     @Override
-    boolean canRebalanceLeaves( int leftKeyCount, int rightKeyCount )
+    int canRebalanceLeaves( PageCursor leftCursor, int leftKeyCount, PageCursor rightCursor, int rightKeyCount )
     {
-        return leftKeyCount + rightKeyCount >= leafMaxKeyCount();
+        if ( leftKeyCount + rightKeyCount >= leafMaxKeyCount() )
+        {
+            int totalKeyCount = rightKeyCount + leftKeyCount;
+            int moveFromPosition = totalKeyCount / 2;
+            return leftKeyCount - moveFromPosition;
+        }
+        return -1;
     }
 
     @Override

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

@@ -499,10 +499,12 @@ public void modifierMustRemoveFromLeftChild() throws Exception
     public void modifierMustRemoveFromRightChildButNotFromInternalWithHitOnInternalSearch() throws Exception
     {
         initialize();
-        for ( int i = 0; numberOfRootSplits == 0; i++ )
+        int i;
+        for ( i = 0; numberOfRootSplits == 0; i++ )
         {
             insert( key( i ), value( i ) );
         }
+        insert( key( i ), value( i ) ); // And one more to avoid rebalance
 
         // when key to remove exists in internal
         KEY keyToRemove = structurePropagation.rightKey;
@@ -566,10 +568,12 @@ public void modifierMustNotRemoveWhenKeyOnlyExistInInternal() throws Exception
     {
         // given
         initialize();
-        for ( int i = 0; numberOfRootSplits == 0; i++ )
+        int i;
+        for ( i = 0; numberOfRootSplits == 0; i++ )
         {
             insert( key( i ), value( i ) );
         }
+        insert( key( i ), value( i ) ); // And an extra to not cause rebalance
 
         // when key to remove exists in internal
         KEY keyToRemove = structurePropagation.rightKey;
@@ -773,7 +777,8 @@ public void mustPropagateStructureOnMergeFromLeft() throws Exception
         // new left child contain keys from old left and old middle
         goTo( readCursor, oldRightChild );
         KEY firstKeyOfOldRightChild = keyAt( 0, LEAF );
-        List<KEY> expectedKeysInNewLeftChild = allKeys.subList( 0, allKeys.indexOf( firstKeyOfOldRightChild ) );
+        int index = indexOf( firstKeyOfOldRightChild, allKeys, layout );
+        List<KEY> expectedKeysInNewLeftChild = allKeys.subList( 0, index );
         goTo( readCursor, newLeftChild );
         assertNodeContainsExpectedKeys( expectedKeysInNewLeftChild, LEAF );
 
@@ -1405,6 +1410,20 @@ public void mustThrowIfReachingNodeWithValidSuccessor() throws Exception
         }
     }
 
+    private int indexOf( KEY theKey, List<KEY> keys, Layout<KEY,VALUE> layout )
+    {
+        int i = 0;
+        for ( KEY key : keys )
+        {
+            if ( layout.compare( theKey, key ) == 0 )
+            {
+                return i;
+            }
+            i++;
+        }
+        return -1;
+    }
+
     private void giveSuccessor( PageCursor cursor, long nodeId ) throws IOException
     {
         goTo( cursor, nodeId );