Fixes count issue in rel group defragmenter

where the count was stored as unsigned short, but retrieved as
signed short in some places, which may have it seen as negative.

community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/RelationshipGroupCache.java

@@ -82,7 +82,7 @@ public RelationshipGroupCache( NumberArrayFactory arrayFactory, long maxMemory,
      */
     public void incrementGroupCount( long nodeId )
     {
-        int count = groupCountCache.getShort( nodeId, 0 ) & 0xFFFF;
+        int count = groupCount( nodeId );
         count++;
         if ( (count & ~0xFFFF) != 0 )
         {
@@ -92,6 +92,11 @@ public void incrementGroupCount( long nodeId )
         groupCountCache.setShort( nodeId, 0, (short) count );
     }
 
+    int groupCount( long nodeId )
+    {
+        return groupCountCache.getShort( nodeId, 0 ) & 0xFFFF;
+    }
+
     /**
      * Getter here because we can use this already allocated data structure for other things in and
      * around places where this group cache is used.
@@ -119,7 +124,7 @@ public long prepare( long fromNodeId )
         highCacheId = 0;
         for ( long nodeId = fromNodeId; nodeId < highNodeId; nodeId++ )
         {
-            short count = groupCountCache.getShort( nodeId, 0 );
+            int count = groupCount( nodeId );
             if ( highCacheId + count > cache.length() )
             {
                 // Cannot include this one, so up until the previous is good
@@ -155,7 +160,7 @@ public boolean put( RelationshipGroupRecord groupRecord )
 
         long baseIndex = offsets.get( rebase( nodeId ) );
         // grouCount is extra validation, really
-        int groupCount = groupCountCache.getShort( nodeId, 0 );
+        int groupCount = groupCount( nodeId );
         long index = scanForFreeFrom( baseIndex, groupCount, groupRecord.getType() );
 
         // Put the group at this index
@@ -240,7 +245,7 @@ public Iterator<RelationshipGroupRecord> iterator()
         {
             private long cursor;
             private long nodeId = fromNodeId;
-            private int countLeftForThisNode = groupCountCache.getShort( nodeId, 0 );
+            private int countLeftForThisNode = groupCount( nodeId );
             {
                 findNextNodeWithGroupsIfNeeded();
             }
@@ -285,8 +290,7 @@ private void findNextNodeWithGroupsIfNeeded()
                     do
                     {
                         nodeId++;
-                        countLeftForThisNode = nodeId >= groupCountCache.length() ? 0 :
-                            groupCountCache.getShort( nodeId, 0 );
+                        countLeftForThisNode = nodeId >= groupCountCache.length() ? 0 : groupCount( nodeId );
                     }
                     while ( countLeftForThisNode == 0 && nodeId < groupCountCache.length() );
                 }

community/kernel/src/test/java/org/neo4j/unsafe/impl/batchimport/RelationshipGroupCacheTest.java

@@ -165,6 +165,37 @@ public void shouldSortOutOfOrderTypes() throws Exception
         assertEquals( cachedCount, readCount );
     }
 
+    @Test
+    public void shouldHandleGroupCountBeyondSignedShortRange() throws Exception
+    {
+        // GIVEN
+        long nodeId = 0;
+        int limit = Short.MAX_VALUE + 10;
+        RelationshipGroupCache cache = new RelationshipGroupCache( HEAP, ByteUnit.kibiBytes( 100 ), nodeId + 1 );
+
+        // WHEN first counting all groups per node
+        for ( int type = 0; type < limit; type++ )
+        {
+            cache.incrementGroupCount( nodeId );
+        }
+        // and WHEN later putting group records into the cache
+        RelationshipGroupRecord group = new RelationshipGroupRecord( -1 );
+        group.setOwningNode( nodeId );
+        for ( int type = 0; type < limit; type++ )
+        {
+            group.setId( type );
+            group.setFirstOut( type ); // just some relationship
+            group.setType( type );
+            cache.put( group );
+        }
+        long prepared = cache.prepare( nodeId );
+
+        // THEN that should work, because it used to fail inside prepare, but we can also ask
+        // the groupCount method to be sure
+        assertEquals( nodeId, prepared );
+        assertEquals( limit, cache.groupCount( nodeId ) );
+    }
+
     private int[] scrambledTypes( int count )
     {
         int[] types = new int[count];