Ability to fixate dynamic NumberArray

when it has done all it's dynamic growing. Fixating a dynamic NumberArray
returns a new instance with better performance, but one which cannot
dynamically grow anymore.

ParallelBatchImporter now fixates the node->relationship cache after
relationship import for faster relationship->relationship stage.

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

@@ -175,6 +175,7 @@ public void doImport( Input input ) throws IOException
                     relationships.supportsMultiplePasses() ? relationships : inputCache.relationships(),
                     idMapper, neoStore, nodeRelationshipLink, input.specificRelationshipIds() );
             executeStages( relationshipStage );
+            nodeRelationshipLink.fixate();
 
             // Prepare for updating
             neoStore.flush();

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

@@ -0,0 +1,117 @@
+/**
+ * Copyright (c) 2002-2015 "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.unsafe.impl.batchimport.cache;
+
+/**
+ * A {@link NumberArray} base built up out of smaller chunks.
+ */
+abstract class ChunkedNumberArray<N extends NumberArray> implements NumberArray
+{
+    protected final long chunkSize;
+    protected NumberArray[] chunks;
+
+    ChunkedNumberArray( long chunkSize )
+    {
+        this.chunkSize = chunkSize;
+    }
+
+    @Override
+    public long length()
+    {
+        return chunks.length * chunkSize;
+    }
+
+    @Override
+    public long size()
+    {
+        long size = 0;
+        for ( int i = 0; i < chunks.length; i++ )
+        {
+            size += chunks[i].size();
+        }
+        return size;
+    }
+
+    @Override
+    public void clear()
+    {
+        for ( NumberArray chunk : chunks )
+        {
+            chunk.clear();
+        }
+    }
+
+    @Override
+    public long highestSetIndex()
+    {
+        for ( int i = chunks.length-1; i >= 0; i-- )
+        {
+            NumberArray chunk = chunks[i];
+            long highestSetInChunk = chunk.highestSetIndex();
+            if ( highestSetInChunk > -1 )
+            {
+                return i*chunkSize + highestSetInChunk;
+            }
+        }
+        return -1;
+    }
+
+    @Override
+    public void acceptMemoryStatsVisitor( MemoryStatsVisitor visitor )
+    {
+        for ( NumberArray chunk : chunks )
+        {
+            chunk.acceptMemoryStatsVisitor( visitor );
+        }
+    }
+
+    @SuppressWarnings( "unchecked" )
+    protected N chunkAt( long index )
+    {
+        int chunkIndex = chunkIndex( index );
+        return (N) chunks[chunkIndex];
+    }
+
+    @SuppressWarnings( "unchecked" )
+    protected N chunkOrNullAt( long index )
+    {
+        int chunkIndex = chunkIndex( index );
+        return chunkIndex < chunks.length ? (N) chunks[chunkIndex] : null;
+    }
+
+    protected int chunkIndex( long index )
+    {
+        return (int) (index/chunkSize);
+    }
+
+    protected long index( long index )
+    {
+        return index % chunkSize;
+    }
+
+    @Override
+    public void close()
+    {
+        for ( NumberArray chunk : chunks )
+        {
+            chunk.close();
+        }
+    }
+}

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

@@ -38,7 +38,7 @@ public DynamicIntArray( NumberArrayFactory factory, long chunkSize, int defaultV
     @Override
     public int get( long index )
     {
-        IntArray chunk = chunkAt( index );
+        IntArray chunk = chunkOrNullAt( index );
         return chunk != null ? chunk.get( index( index ) ) : defaultValue;
     }
 
@@ -65,4 +65,10 @@ protected IntArray addChunk( long chunkSize )
     {
         return factory.newIntArray( chunkSize, defaultValue );
     }
+
+    @Override
+    public IntArray fixate()
+    {
+        return new FixedIntArray( chunks, chunkSize, defaultValue );
+    }
 }

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

@@ -38,7 +38,7 @@ public DynamicLongArray( NumberArrayFactory factory, long chunkSize, long defaul
     @Override
     public long get( long index )
     {
-        LongArray chunk = chunkAt( index );
+        LongArray chunk = chunkOrNullAt( index );
         return chunk != null ? chunk.get( index( index ) ) : defaultValue;
     }
 
@@ -65,4 +65,10 @@ protected LongArray addChunk( long chunkSize )
     {
         return factory.newLongArray( chunkSize, defaultValue );
     }
+
+    @Override
+    public LongArray fixate()
+    {
+        return new FixedLongArray( chunks, chunkSize, defaultValue );
+    }
 }

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

@@ -28,69 +28,17 @@
  * @see NumberArrayFactory#newDynamicLongArray(long, long)
  * @see NumberArrayFactory#newDynamicIntArray(long, int)
  */
-abstract class DynamicNumberArray<N extends NumberArray> implements NumberArray
+abstract class DynamicNumberArray<N extends NumberArray> extends ChunkedNumberArray<N>
 {
     protected final NumberArrayFactory factory;
-    private final long chunkSize;
-    private NumberArray[] chunks = new NumberArray[0];
 
     DynamicNumberArray( NumberArrayFactory factory, long chunkSize )
     {
+        super( chunkSize );
         this.factory = factory;
-        this.chunkSize = chunkSize;
+        this.chunks = new NumberArray[0];
     }
 
-    @Override
-    public long length()
-    {
-        return chunks.length * chunkSize;
-    }
-
-    @Override
-    public long size()
-    {
-        long size = 0;
-        for ( int i = 0; i < chunks.length; i++ )
-        {
-            size += chunks[i].size();
-        }
-        return size;
-    }
-
-    @Override
-    public void clear()
-    {
-        for ( NumberArray chunk : chunks )
-        {
-            chunk.clear();
-        }
-    }
-
-    @Override
-    public long highestSetIndex()
-    {
-        for ( int i = chunks.length-1; i >= 0; i-- )
-        {
-            NumberArray chunk = chunks[i];
-            long highestSetInChunk = chunk.highestSetIndex();
-            if ( highestSetInChunk > -1 )
-            {
-                return i*chunkSize + highestSetInChunk;
-            }
-        }
-        return -1;
-    }
-
-    @Override
-    public void acceptMemoryStatsVisitor( MemoryStatsVisitor visitor )
-    {
-        for ( NumberArray chunk : chunks )
-        {
-            chunk.acceptMemoryStatsVisitor( visitor );
-        }
-    }
-
-    @SuppressWarnings( "unchecked" )
     protected N ensureChunkAt( long index )
     {
         while ( index >= length() )
@@ -105,34 +53,8 @@ protected N ensureChunkAt( long index )
                 }
             }
         }
-        return (N) chunks[chunkIndex( index )];
+        return chunkAt( index );
     }
 
     protected abstract N addChunk( long chunkSize );
-
-    @SuppressWarnings( "unchecked" )
-    protected N chunkAt( long index )
-    {
-        int chunkIndex = chunkIndex( index );
-        return chunkIndex < chunks.length ? (N) chunks[chunkIndex] : null;
-    }
-
-    private int chunkIndex( long index )
-    {
-        return (int) (index/chunkSize);
-    }
-
-    protected long index( long index )
-    {
-        return index % chunkSize;
-    }
-
-    @Override
-    public void close()
-    {
-        for ( NumberArray chunk : chunks )
-        {
-            chunk.close();
-        }
-    }
 }

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

@@ -0,0 +1,62 @@
+/**
+ * Copyright (c) 2002-2015 "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.unsafe.impl.batchimport.cache;
+
+class FixedIntArray extends FixedNumberArray<IntArray> implements IntArray
+{
+    private final int defaultValue;
+
+    FixedIntArray( NumberArray[] chunks, long chunkSize, int defaultValue )
+    {
+        super( chunks, chunkSize );
+        this.defaultValue = defaultValue;
+    }
+
+    @Override
+    public void swap( long fromIndex, long toIndex, int numberOfEntries )
+    {
+        // Let's just do this the stupid way. There's room for optimization here
+        for ( int i = 0; i < numberOfEntries; i++ )
+        {
+            int intermediary = get( fromIndex+i );
+            set( fromIndex+i, get( toIndex+i ) );
+            set( toIndex+i, intermediary );
+        }
+    }
+
+    @Override
+    public int get( long index )
+    {
+        IntArray chunk = chunkOrNullAt( index );
+        return chunk != null ? chunk.get( index( index ) ) : defaultValue;
+    }
+
+    @Override
+    public void set( long index, int value )
+    {
+        chunkAt( index ).set( index( index ), value );
+    }
+
+    @Override
+    public IntArray fixate()
+    {
+        return this;
+    }
+}